Documentation
¶
Overview ¶
Package wal is the write-ahead log: CRC-framed records appended to numbered segment files, replayed in order to reconstruct the in-memory index after a crash. A record frames a length, a body (type + payload) and a CRC32C of the body; on replay a truncated or torn final record ends the log cleanly (crash recovery), while a complete record with a bad CRC is surfaced as corruption. Append-style and group-commit-friendly.
Index ¶
- Constants
- Variables
- func Replay(data []byte, h Handlers) error
- func ReplayDir(dir string, h Handlers) error
- func ReplayDirFrom(dir string, minEpoch uint64, h Handlers) error
- type Damage
- type Handlers
- type SegmentWriter
- func (sw *SegmentWriter) Bytes() int64
- func (sw *SegmentWriter) Checkpoint() error
- func (sw *SegmentWriter) CheckpointThrough(obsolete int) error
- func (sw *SegmentWriter) Close() error
- func (sw *SegmentWriter) Epoch() uint64
- func (sw *SegmentWriter) Seal(epoch uint64) (int, error)
- func (sw *SegmentWriter) Segments() int
- func (sw *SegmentWriter) Seq() int
- func (sw *SegmentWriter) SetEpoch(epoch uint64)
- func (sw *SegmentWriter) SetLogger(l *zap.Logger)
- func (sw *SegmentWriter) SetObs(m *obs.WAL)
- func (sw *SegmentWriter) SetSync(on bool)
- func (sw *SegmentWriter) Size() int
- func (sw *SegmentWriter) Sync() error
- func (sw *SegmentWriter) Write(p []byte) (int, error)
- func (sw *SegmentWriter) WriteFrames(p []byte) error
- func (sw *SegmentWriter) WriteRecords(id signal.SeriesID, payload []byte) error
- func (sw *SegmentWriter) WriteSamples(id signal.SeriesID, ts []int64, values []float64) error
- func (sw *SegmentWriter) WriteSamplesSF(id signal.SeriesID, ts []int64, values, sf []float64) error
- func (sw *SegmentWriter) WriteSeries(id signal.SeriesID, s signal.Series) error
- func (sw *SegmentWriter) WriteSide(payload []byte) error
- type Writer
- func (wr *Writer) WriteRecords(id signal.SeriesID, payload []byte) error
- func (wr *Writer) WriteSamples(id signal.SeriesID, ts []int64, values []float64) error
- func (wr *Writer) WriteSamplesSF(id signal.SeriesID, ts []int64, values, sf []float64) error
- func (wr *Writer) WriteSeries(id signal.SeriesID, s signal.Series) error
- func (wr *Writer) WriteSide(payload []byte) error
Constants ¶
const ( // DefaultMaxSegmentBytes is the size at which the writer rotates to a new segment. DefaultMaxSegmentBytes = 32 << 20 // 32 MiB )
Variables ¶
var ErrCorrupt = errors.New("wal: corrupt record")
ErrCorrupt is returned when a complete record fails its CRC check.
Functions ¶
func Replay ¶
Replay reads every record from data and dispatches it to h. data must be a *complete* log — a replication payload, or a segment read whole — so a record that does not fit inside it is truncation, and Replay returns an ErrCorrupt-wrapping error rather than a short read. A complete record whose CRC fails is the same error. Records already applied before the stopping point are kept.
Only the last segment of a WAL directory may legitimately end mid-record (a crash was appending to it); ReplayDirFrom is the one caller that tolerates it, and only there.
func ReplayDir ¶
ReplayDir replays every segment in dir (all epochs). See ReplayDirFrom.
func ReplayDirFrom ¶ added in v0.3.0
ReplayDirFrom replays the segments in dir whose epoch is greater than minEpoch, in ascending segment order, dispatching each record to h. Segments at or below minEpoch are skipped — their records are already durable in a flushed part (the watermark), so skipping them makes recovery exactly-once. A torn final record in the **last** replayed segment ends replay cleanly, but only once nothing whole is found after it; a torn record anywhere earlier, or one followed by a CRC-valid frame, is a hole in the middle of history and returns an ErrCorrupt-wrapping error.
Types ¶
type Damage ¶ added in v0.47.0
type Damage struct {
Segment string // the segment's file name
Offset int // where the unreadable region starts
Length int // how many bytes replay skipped
Err error // why the region is unreadable
// Kept names the copy of the segment kept for inspection, empty when none could be written.
Kept string
}
Damage is a region of a segment that replay could not read and skipped.
type Handlers ¶
type Handlers struct {
OnSeries func(id signal.SeriesID, s signal.Series) error
OnSamples func(id signal.SeriesID, ts []int64, values []float64) error
// OnSamplesSF receives samples that carry per-sample lossy-sampling scale factors (len(sf) ==
// len(ts)). If unset, such records fall back to OnSamples (the weights are dropped, i.e. read as
// 1) so a reader that does not care about sampling still recovers the samples.
OnSamplesSF func(id signal.SeriesID, ts []int64, values, sf []float64) error
OnRecords func(id signal.SeriesID, payload []byte) error
OnSide func(payload []byte) error
// OnDamage, when set, makes directory replay ([ReplayDirFrom]) salvage a damaged log: each region
// it cannot read is reported here and skipped, and replay goes on with the next valid frame and
// every later segment. Returning an error stops replay with it. When nil, damage fails replay with
// an [ErrCorrupt]-wrapping error. [Replay] of a single payload is always strict.
OnDamage func(Damage) error
}
Handlers receives decoded records during Replay. Unset handlers skip their record type.
type SegmentWriter ¶
type SegmentWriter struct {
// contains filtered or unexported fields
}
SegmentWriter appends WAL records to numbered segment files in a directory, rotating to a fresh segment once the current one reaches the size limit. Replaying the directory in order (ReplayDir) reconstructs the logged state. Not safe for concurrent use.
func Create ¶
func Create(dir string, maxBytes int) (*SegmentWriter, error)
Create opens (creating the directory if needed) a segmented WAL writer. A non-positive maxBytes uses DefaultMaxSegmentBytes. If the directory already holds segments from a prior run, Create **resumes**: it repairs the last segment's torn tail (see [repair]) and opens a fresh segment numbered beyond the existing ones, so ReplayDir can still recover the prior segments before the next SegmentWriter.Checkpoint discards them.
func CreateFS ¶ added in v0.47.0
func CreateFS(fsys vfs.FS, maxBytes int) (*SegmentWriter, error)
CreateFS is Create over an already-rooted filesystem, for tests outside this package that inject faults through a fake one.
func (*SegmentWriter) Bytes ¶ added in v0.47.0
func (sw *SegmentWriter) Bytes() int64
Bytes returns the bytes the on-disk segments hold (see SegmentWriter.Segments). A cheap in-memory read for introspection; not safe for concurrent use.
func (*SegmentWriter) Checkpoint ¶ added in v0.2.0
func (sw *SegmentWriter) Checkpoint() error
Checkpoint discards every segment written so far. Call it only when a durable part supersedes every record logged up to now — a flush that ran concurrently with ingest must use SegmentWriter.CheckpointThrough with the sequence its SegmentWriter.Seal returned instead.
func (*SegmentWriter) CheckpointThrough ¶ added in v0.45.0
func (sw *SegmentWriter) CheckpointThrough(obsolete int) error
CheckpointThrough discards the segments up to and including through, whose records a flushed part durably supersedes; later segments (holding records appended while that part was being written) are kept. The flush advances the epoch and persists it as the bucket-index watermark *before* this call, so even a crash between the part committing and this deletion replays nothing already flushed (exactly-once — the watermark and the part list advance atomically; see ReplayDirFrom).
func (*SegmentWriter) Close ¶
func (sw *SegmentWriter) Close() error
Close syncs and closes the current segment and releases the writer's handle on the segment directory. The writer is spent: a later append cannot open a segment. Idempotent. Like SegmentWriter.Sync, it fails when a torn segment cannot be restored.
func (*SegmentWriter) Epoch ¶ added in v0.12.0
func (sw *SegmentWriter) Epoch() uint64
Epoch returns the flush generation stamped into new segments (see SegmentWriter.SetEpoch). A cheap in-memory read for introspection; not safe for concurrent use.
func (*SegmentWriter) Seal ¶ added in v0.45.0
func (sw *SegmentWriter) Seal(epoch uint64) (int, error)
Seal closes the current segment and stamps subsequent ones with epoch, returning the sequence number it sealed through — the argument SegmentWriter.CheckpointThrough takes once the flush of those records commits. Like SegmentWriter.Sync, it fails while a torn segment cannot be restored.
A flush seals at the instant it detaches the head, not when it publishes: the part it goes on to write off-lock holds exactly the records logged before that instant. A record appended during the part write then lands in a segment beyond the sealed number, carrying the generation past the watermark the flush is about to commit — so the checkpoint does not delete it and replay does not skip it.
func (*SegmentWriter) Segments ¶ added in v0.47.0
func (sw *SegmentWriter) Segments() int
Segments returns the number of segment files on disk: the closed ones no checkpoint has discarded yet, plus the open one. A cheap in-memory read for introspection; not safe for concurrent use.
func (*SegmentWriter) Seq ¶ added in v0.12.0
func (sw *SegmentWriter) Seq() int
Seq returns the current segment sequence number — the count of segments opened so far (0 before the first write opens one). A cheap in-memory read for introspection; not safe for concurrent use.
func (*SegmentWriter) SetEpoch ¶ added in v0.3.0
func (sw *SegmentWriter) SetEpoch(epoch uint64)
SetEpoch stamps subsequent segments with epoch (a flush generation). Because SegmentWriter.Checkpoint closes the current segment without opening a new one, the next write starts a segment carrying the epoch set here — so each segment self-describes the generation of its records, and ReplayDirFrom can skip whole segments already superseded by a flushed part.
func (*SegmentWriter) SetLogger ¶ added in v0.5.0
func (sw *SegmentWriter) SetLogger(l *zap.Logger)
SetLogger attaches a logger that records segment lifecycle events (open/rotate/checkpoint) at Debug. The WAL append path takes no context, so these lines are not trace-correlated. nil ⇒ no-op.
func (*SegmentWriter) SetObs ¶ added in v0.5.0
func (sw *SegmentWriter) SetObs(m *obs.WAL)
SetObs attaches the WAL metrics handle (append/fsync/rotation counters). nil disables metering.
func (*SegmentWriter) SetSync ¶ added in v0.3.0
func (sw *SegmentWriter) SetSync(on bool)
SetSync enables (or disables) an fsync after every framed write — power-loss durability at a throughput cost. The default is off (records reach the OS page cache, surviving a process crash but not necessarily a power loss).
func (*SegmentWriter) Size ¶ added in v0.12.0
func (sw *SegmentWriter) Size() int
Size returns the byte size of the current open segment (0 when none is open). A cheap in-memory read for introspection; not safe for concurrent use.
func (*SegmentWriter) Sync ¶
func (sw *SegmentWriter) Sync() error
Sync flushes the current segment to stable storage (no-op when no segment is open). It first finishes restoring a segment a failed write tore, and fails while it cannot: the records before the failed write are durable only once the restore is. It also fails, once, when a segment a rotation closed failed to sync since the last call, so a nil return covers every record acknowledged since then.
func (*SegmentWriter) Write ¶
func (sw *SegmentWriter) Write(p []byte) (int, error)
Write implements io.Writer, appending to the current segment and tracking its size so the writer knows when to rotate.
A failed write can still land a prefix of p — a full or failing disk returns a short write — and a later write appending behind it would turn a torn tail, which recovery discards, into corruption in the middle of the log, which fails every replay. So a failure is healed before anything else is written: see [SegmentWriter.heal].
func (*SegmentWriter) WriteFrames ¶ added in v0.38.0
func (sw *SegmentWriter) WriteFrames(p []byte) error
WriteFrames logs a run of records that is **already framed** by a Writer — the form the cluster write path builds to replicate the accepted set. It is byte-identical to writing those records one call at a time, without re-encoding them. p must end on a frame boundary; an empty p is a no-op.
func (*SegmentWriter) WriteRecords ¶ added in v0.2.0
func (sw *SegmentWriter) WriteRecords(id signal.SeriesID, payload []byte) error
WriteRecords logs a stream's opaque engine-encoded record payload.
func (*SegmentWriter) WriteSamples ¶
WriteSamples logs a run of samples for one series.
func (*SegmentWriter) WriteSamplesSF ¶ added in v0.10.0
WriteSamplesSF logs a run of samples for one series that also carry per-sample scale factors.
func (*SegmentWriter) WriteSeries ¶
WriteSeries logs a series registration (opening/rotating the segment first as needed).
func (*SegmentWriter) WriteSide ¶ added in v0.2.0
func (sw *SegmentWriter) WriteSide(payload []byte) error
WriteSide logs an opaque engine-encoded side-store delta.
type Writer ¶
type Writer struct {
// contains filtered or unexported fields
}
Writer appends framed records to an io.Writer (typically a segment file). It reuses internal buffers, so it is not safe for concurrent use.
func (*Writer) WriteRecords ¶ added in v0.2.0
WriteRecords logs a run of records for one stream: its signal.SeriesID then an opaque, engine-encoded payload (the record engine owns the column encoding, keyed by its schema; the WAL stays signal-agnostic). Replaying it hands the payload back to the engine via Handlers.OnRecords.
func (*Writer) WriteSamples ¶
WriteSamples logs a run of samples for one series: its signal.SeriesID then the (timestamp, value) pairs. ts and values must have the same length.
func (*Writer) WriteSamplesSF ¶ added in v0.10.0
WriteSamplesSF logs a run of samples for one series that also carry per-sample lossy-sampling scale factors: its signal.SeriesID then the (timestamp, value, sf) triples. ts, values, and sf must have the same length. Used only when sampling actually weighted the batch; the unsampled path stays on Writer.WriteSamples (no per-sample sf on the wire).
func (*Writer) WriteSeries ¶
WriteSeries logs a series registration: its signal.SeriesID and full identity (Resource + Scope + data-point attributes). Replaying it reconstructs the series.
Source Files