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 Handlers
- type SegmentWriter
- func (sw *SegmentWriter) Checkpoint() error
- func (sw *SegmentWriter) Close() error
- func (sw *SegmentWriter) Epoch() uint64
- 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) 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 complete record from data and dispatches it to h. It stops cleanly at end-of-log or a torn final record (returning nil), and returns an ErrCorrupt-wrapping error on a complete record whose CRC fails. Records already applied before the stopping point are kept.
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.
Types ¶
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
}
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 opens a fresh segment numbered beyond the existing ones (never truncating them), so ReplayDir can still recover the prior segments before the next SegmentWriter.Checkpoint discards them.
func (*SegmentWriter) Checkpoint ¶ added in v0.2.0
func (sw *SegmentWriter) Checkpoint() error
Checkpoint discards every segment written so far. Call it after a full head flush, whose part durably supersedes those records. It closes the current segment and best-effort deletes all segments (the next write lazily opens a fresh one carrying the current epoch). 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.
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) 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).
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.
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