Documentation
¶
Overview ¶
Package wal turns the buffer's queue primitives into a replication log.
The buffer is a destructive single-consumer queue; replication needs a non-destructively *tailed* log with a base snapshot for bootstrap. This package reuses the buffer's object-store abstraction and append-friendly manifest, and extends the manifest with a snapshot pointer (footer v2) and per-record sequencing (footer v3) so a single CAS-protected object describes both the live WAL tail and the base a new or lagging replica restores from, and lets a consumer address playback to an individual record sequence.
Sequence model (footer v3): each entry owns a contiguous half-open range of record sequences [Sequence, Sequence+Count). A segment with N framed records advances the log's nextSequence by N, and record i in that segment has sequence Sequence+i. Entries tile the sequence space with no gaps or overlaps, so a consumer can read only the manifest, find the entry whose range contains a target sequence (Locate), and start reading segments from there.
Legacy footers (v1 buffer, v2) used one sequence per entry; their entries are read back with Count==0, meaning "this segment occupies a single sequence slot and all its records share that sequence" - the original whole-segment semantics, preserved so old manifests still replay correctly.
Index ¶
- Constants
- Variables
- type Applier
- type ApplyFunc
- type Compression
- type CursorStore
- type DecodeFunc
- type Durability
- type Entry
- type FileCursorStore
- type HandleFunc
- type Manifest
- func (m *Manifest) Append(location string, md []RecordMeta, count int) (uint64, error)
- func (m *Manifest) Bytes() ([]byte, error)
- func (m *Manifest) Count() int
- func (m *Manifest) Entries() ([]Entry, error)
- func (m *Manifest) EntriesAfter(afterSeq uint64) ([]Entry, error)
- func (m *Manifest) EntriesContaining(seq uint64) ([]Entry, error)
- func (m *Manifest) EntriesFrom(minSeq uint64) ([]Entry, error)
- func (m *Manifest) Epoch() uint64
- func (m *Manifest) Locate(seq uint64) (Entry, int, bool)
- func (m *Manifest) NextSequence() uint64
- func (m *Manifest) SetEpoch(e uint64)
- func (m *Manifest) SetSnapshot(p SnapshotPointer)
- func (m *Manifest) Snapshot() SnapshotPointer
- func (m *Manifest) TailLocations(limit int) (map[string]Entry, error)
- func (m *Manifest) TruncateThrough(throughSeq uint64) (int, error)
- type MemCursorStore
- type Producer
- type ProducerConfig
- type Record
- type RecordMeta
- type Replica
- type ReplicaConfig
- type SeqRange
- type SnapshotPointer
- type Store
Constants ¶
const ( // the whole object. The manifest is the ordering authority, so a silent // flip here misdirects the whole log, not one batch of records. FooterVersion uint16 = 4 )
Footer/format constants.
Variables ¶
var ErrCorrupt = errors.New("wal: checksum mismatch (corrupt object)")
ErrCorrupt reports that stored bytes failed their integrity check. Distinct from a transport error: retrying the GET re-reads the same bad bytes, so replication should stop rather than spin.
var ErrFenced = errors.New("wal: producer fenced by a newer epoch")
ErrFenced is returned once a newer primary has claimed the log (bumped the epoch). A fenced producer is permanently halted; construct a fresh one only after re-establishing that this node is the primary.
var ErrHalted = errors.New("wal: producer halted")
ErrHalted is returned by Append after the producer has stopped (a fatal commit error, fencing, or Close).
var ErrNoRecords = errors.New("wal: Append requires at least one record")
ErrNoRecords is returned by Append when given no records. An empty group would produce an invalid manifest entry and halt the producer.
Functions ¶
This section is empty.
Types ¶
type Applier ¶
Applier is the engine-specific path in the replication path. The replica hands it each framed record in sequence order, and it applies the change to the target system: a Bitcask applier appends the record to its active file and updates its keydir; a SQLite applier replays a frame; a cache applier updates an in-memory map. The wal package depends only on this interface and never on any particular engine.
Apply MUST be idempotent. Produce is at-least-once, and the tailer re-applies a whole segment if an earlier record in it failed or if the replica restarts before persisting its cursor, so the same record may arrive more than once. For a key-value store, idempotent put/delete satisfies this with no extra state.
func TypedApplier ¶
func TypedApplier[T any](decode DecodeFunc[T], handle HandleFunc[T]) Applier
TypedApplier is the recommended way to plug an engine in: supply a decoder that turns record bytes into your operation type and a handler that applies it. The wal layer owns the boilerplate (pull the record, decode, dispatch) and stays ignorant of both the frame format and the engine - neither the decode nor the apply logic lives in this package.
applier := wal.TypedApplier(
decodeKVOp, // []byte -> KVOp (yours)
func(ctx context.Context, seq uint64, op KVOp) error {
switch op.Kind {
case OpPut: return db.Put(op.Key, op.Value) // db is yours
case OpDelete: return db.Delete(op.Key)
}
return nil
},
)
Decode/handle errors surface from Poll; because Apply must be idempotent, a failure simply causes the segment to be re-delivered on the next pass.
type Compression ¶
type Compression uint8
Compression selects the codec applied to a segment's record block.
const ( CompressionNone Compression = 0 CompressionZstd Compression = 1 )
type CursorStore ¶
type CursorStore interface {
// Load returns the persisted next-to-apply cursor. ok is false if nothing
// has been persisted yet.
Load(ctx context.Context) (next uint64, ok bool, err error)
// Save persists the next-to-apply cursor durably.
Save(ctx context.Context, next uint64) error
}
CursorStore persists a replica's next-to-apply cursor so a restart resumes instead of replaying from the beginning. The cursor is saved after each applied segment; because Apply is idempotent, a crash between apply and save at worst re-applies one segment.
(A future v2 retention policy could publish cursors to the object store so GC can keep min(cursor) across replicas; that is a different use of the same idea and is not built here.)
type DecodeFunc ¶
DecodeFunc parses an opaque record's bytes into a typed operation. You own this: it mirrors however the writer framed the record.
type Durability ¶
type Durability struct {
// contains filtered or unexported fields
}
Durability resolves when the records of an Append have been committed to the manifest (or failed permanently).
func (*Durability) Count ¶
func (d *Durability) Count() int
Count is the number of records this Append wrote. They occupy the contiguous record-sequence range [first, first+Count), where first is what Wait returns.
type Entry ¶
type Entry struct {
Sequence uint64
Count uint32
Location string
Metadata []RecordMeta
}
Entry is one committed manifest entry: a segment object plus the metadata ranges describing the framed records it holds. The entry owns the record sequence range [Sequence, Sequence+RangeSize()); record i has sequence Sequence+i. Count is the number of framed records in the segment; Count==0 is the legacy per-entry sentinel (one sequence slot, all records share Sequence).
type FileCursorStore ¶
type FileCursorStore struct {
// contains filtered or unexported fields
}
FileCursorStore persists the cursor to a local file as 8 little-endian bytes, written atomically via a temp file and rename. Suitable for a full local replica that keeps its cursor next to its data.
func NewFileCursorStore ¶
func NewFileCursorStore(path string) *FileCursorStore
type HandleFunc ¶
HandleFunc applies a decoded operation to the target system, e.g. by calling into the Bitcask (or other) engine's API. You own this too.
type Manifest ¶
type Manifest struct {
// contains filtered or unexported fields
}
Manifest is the in-memory, mutable view of a replication-log manifest. It preserves the buffer's O(1) append shape: existing entries are held as raw bytes (base) and new entries accumulate in a side buffer (appended); only a snapshot change or truncation rewrites base. base/appended are always held in the current (v3) entry encoding; a parsed legacy manifest is normalized on load.
Manifest is not safe for concurrent mutation; the writer owns one instance.
func NewManifest ¶
func NewManifest() *Manifest
NewManifest returns an empty manifest at epoch 0 with no snapshot.
func ParseManifest ¶
ParseManifest decodes a serialized manifest. It accepts v4 (this package), v3, v2 and v1. Legacy (v1/v2) entries are normalized in memory to the v3 encoding with Count==0, so downstream decoding is uniform; a re-commit upgrades the object to v4.
func (*Manifest) Append ¶
Append assigns a record-sequence range to a new entry pointing at a segment object holding count framed records, and returns the sequence of the first record (the entry's base sequence). nextSequence advances by count. count must be >= 1.
func (*Manifest) EntriesAfter ¶
EntriesAfter returns live entries with Sequence > afterSeq, in order.
func (*Manifest) EntriesContaining ¶
EntriesContaining returns the entries needed to replay from record sequence seq onward: every entry whose range end is > seq, in order. The first entry returned is the one whose range contains seq (or, if seq is below all live entries, the earliest entry); a consumer skips records before seq in that first segment. This is the record-addressable read path.
func (*Manifest) EntriesFrom ¶
EntriesFrom returns live entries with Sequence >= minSeq, in order.
func (*Manifest) Locate ¶
Locate returns the entry whose record-sequence range contains seq, along with the index of that record within the segment (seq - entry.Sequence) and a found flag. A consumer reads only the manifest, calls Locate(seq) to learn which segment to fetch and how many leading records to skip, then streams forward. For a legacy (Count==0) entry the offset is always 0. Not found if seq is outside [firstLiveSequence, NextSequence).
func (*Manifest) NextSequence ¶
NextSequence returns the sequence the next appended record will receive.
func (*Manifest) SetSnapshot ¶
func (m *Manifest) SetSnapshot(p SnapshotPointer)
SetSnapshot records a new base snapshot. The next Bytes/commit carries it.
func (*Manifest) Snapshot ¶
func (m *Manifest) Snapshot() SnapshotPointer
Snapshot returns the current snapshot pointer (zero value if none).
func (*Manifest) TailLocations ¶
TailLocations maps the last `limit` live entries by segment location. Locations are unique per (runID, ordinal), so a writer that lost a CAS response can check whether its entries landed. Only the tail is scanned: those entries are necessarily last.
func (*Manifest) TruncateThrough ¶
TruncateThrough removes every entry whose records are all <= throughSeq (its End()-1 <= throughSeq), i.e. entries fully superseded by a snapshot covering up to throughSeq. An entry with any record > throughSeq is kept whole (a segment is never split). Returns the number of entries removed.
type MemCursorStore ¶
type MemCursorStore struct {
// contains filtered or unexported fields
}
MemCursorStore is an in-memory CursorStore for tests.
type Producer ¶
type Producer struct {
// contains filtered or unexported fields
}
Producer is the epoch-fenced single-writer primary. It accepts opaque framed records, group-commits them into segment objects, and CAS-appends manifest entries. It knows nothing about record contents.
func NewProducer ¶
func NewProducer(ctx context.Context, os objectstore.ObjectStore, cfg ProducerConfig) (*Producer, error)
NewProducer constructs a producer and claims the log by bumping the manifest epoch. A successful return means this node owns the log at its claimed epoch.
func (*Producer) Append ¶
Append enqueues a group of framed records with an optional metadata payload. It BLOCKS when the in-flight byte (or batch-count) budget is exhausted, until a flush frees space or ctx is cancelled - this is the producer's backpressure onto the caller. The returned Durability resolves when the group is committed. Records are not copied; do not mutate them until it resolves.
type ProducerConfig ¶
type ProducerConfig struct {
// ManifestPath is the manifest object key.
ManifestPath string
// SegmentPrefix is the key prefix for segment objects; segments are named
// "<SegmentPrefix>/<runID>/<ordinal:016x>".
SegmentPrefix string
// FlushInterval bounds how long records wait before a segment is sealed.
FlushInterval time.Duration
// FlushBytes seals a segment early once buffered record bytes reach it.
// Zero disables size-based flushing.
FlushBytes int
// Compression applied to segment record blocks.
Compression Compression
// LegacySegmentFormat writes unchecksummed batch-v1 segments instead of
// v2, for downstream consumers that read them as upstream buffer batches.
// Gives up corruption detection. Reading v1 is always supported.
LegacySegmentFormat bool
// MaxInFlightBytes caps the total bytes Appended but not yet durably
// committed. Append BLOCKS once a further record would exceed it - the
// primary backpressure signal (default 256 MiB). A single Append larger
// than the cap is admitted only when nothing else is in flight, to avoid
// deadlock.
MaxInFlightBytes int
// MaxInFlightBatches caps the number of un-committed Append calls in
// flight - a secondary safety stop against many tiny Appends (default
// 4096).
MaxInFlightBatches int
// SegmentMaxBytes caps the size of a single segment object. When a flush
// drains more than this, it rotates into multiple segments (whole Append
// groups are kept intact; an oversized group gets its own segment). Zero
// means one segment per flush (no rotation).
SegmentMaxBytes int
// ManifestAppendBatchSize caps how many segment entries are coalesced into
// one manifest CAS. Zero coalesces all of a flush's segments into a single
// CAS. Coalescing cuts manifest write rate/cost and raises the throughput
// ceiling (bytes-per-commit), at the cost of one CAS covering several
// segments.
ManifestAppendBatchSize int
// MaxClaimAttempts bounds epoch-claim CAS retries (default 8).
MaxClaimAttempts int
// UploadMaxAttempts bounds segment-upload retries (default 6).
UploadMaxAttempts int
// UploadInitialBackoff is the first upload retry backoff (default 100ms).
UploadInitialBackoff time.Duration
// UploadConcurrency bounds how many segment uploads run in parallel within
// a flush (default 4). Uploads overlap, but commits remain strictly serial
// and in ordinal order, so total log order is preserved. Set to 1 for fully
// sequential uploads.
UploadConcurrency int
// ManifestMaxAttempts bounds retries of *transient* manifest-commit errors
// (default 6). A precondition conflict is a free re-plan (not counted), and
// the re-load's epoch check converts a competing writer into ErrFenced.
ManifestMaxAttempts int
// ManifestInitialBackoff is the first backoff for transient commit retries
// (default 100ms).
ManifestInitialBackoff time.Duration
// MaxCommitConflicts bounds how many times a commit may lose the manifest
// CAS race (412) before giving up (default 64), so a misclassified error
// cannot spin forever.
MaxCommitConflicts int
// Logger receives structured lifecycle events: claims, fencing, commit
// retries/conflicts, upload failures, halts. Defaults to a text logger on
// stderr filtered to Warn, so a healthy producer is silent; pass your own
// *slog.Logger (any handler, any level, including Debug/Info for routine
// flow) to change verbosity or destination.
Logger *slog.Logger
// Meter records batch/commit/retry/fencing metrics (see producerMetrics).
// Defaults to otel.GetMeterProvider()'s meter, which is a no-op until the
// process calls otel.SetMeterProvider - so metrics cost nothing unless
// wired up, here or globally.
Meter metric.Meter
}
ProducerConfig configures the primary's write path.
type Record ¶
type Record struct {
// Sequence is the manifest sequence of the segment this record came from.
// All records sharing a segment share its sequence; the cursor advances at
// segment granularity.
Sequence uint64
// GroupMeta is the metadata payload of the Append group this record
// belonged to (may be nil).
GroupMeta []byte
// Data is the opaque framed record bytes. The wal package never decodes
// these; the frame format is the application's own.
Data []byte
}
Record is one framed record delivered to an Applier.
type RecordMeta ¶
RecordMeta is a per-range annotation attached to a manifest entry, delimiting the run of framed records beginning at StartIndex.
type Replica ¶
type Replica struct {
// contains filtered or unexported fields
}
Replica tails the WAL and applies records to a local state machine via an Applier. Many replicas may tail the same manifest concurrently; readers are not epoch-fenced.
func NewReplica ¶
func NewReplica(os objectstore.ObjectStore, apply Applier, cfg ReplicaConfig) *Replica
NewReplica constructs a replica positioned at cfg.StartAfter.
func (*Replica) Next ¶
Next returns the next sequence the replica expects to apply (one past the highest applied entry).
func (*Replica) Poll ¶
Poll runs one tail pass: it loads the manifest, fetches and applies every segment with Sequence > cursor in order, and advances the cursor (per segment). It returns the number of records applied. The cursor advances past a segment only after all of its records have been applied, so a mid-segment failure re-applies the whole segment next time (hence the idempotency requirement).
type ReplicaConfig ¶
type ReplicaConfig struct {
// ManifestPath is the manifest object key (must match the producer).
ManifestPath string
// PollInterval is how often Run polls the manifest for new entries.
PollInterval time.Duration
// StartAt is the initial cursor: the next sequence to apply. The replica
// applies entries with Sequence >= StartAt. Without snapshots it defaults
// to 0 (replay from the beginning); bootstrapping from a snapshot would
// set it to the snapshot's ThroughSeq + 1.
StartAt uint64
// Cursor, if set, persists the next-to-apply cursor so a restart resumes
// instead of replaying from StartAt. A persisted value overrides StartAt.
Cursor CursorStore
// CursorSaveInterval batches cursor saves: the cursor is persisted at most
// once per this many applied segments, plus once at the end of each poll
// pass that applied anything. 0 or 1 saves after every segment (safest;
// default). Larger values trade a few extra re-applied segments after a
// crash (idempotent) for fewer fsyncs.
CursorSaveInterval int
// MaxRecordsPerPoll bounds how many records a single Poll applies: once a
// poll has applied at least this many, it stops at the next segment boundary
// (cursor saved there) and returns, leaving the rest for the following Poll.
// 0 means unbounded (drain everything available). Granularity is a whole
// segment, so the effective cap is "the first segment boundary at or past
// this count". Useful for pacing ingestion so progress is observable.
MaxRecordsPerPoll int
// Logger receives structured lifecycle events: poll/apply errors. Defaults
// to a text logger on stderr filtered to Warn, so a healthy replica is
// silent; pass your own *slog.Logger to change verbosity or destination.
Logger *slog.Logger
// Meter records applied-record and poll-duration metrics. Defaults to
// otel.GetMeterProvider()'s meter, a no-op until the process calls
// otel.SetMeterProvider.
Meter metric.Meter
}
ReplicaConfig configures the read replica's tail loop.
type SeqRange ¶
SeqRange is the contiguous record-sequence range a single Append occupies: its records have sequences [First, First+Count). Use it instead of recomputing First+i offsets at call sites, which is an easy place to introduce off-by-one errors.
func (SeqRange) All ¶
All materializes every sequence in the range. Prefer First/Count/At for large batches; this allocates a slice.
func (SeqRange) At ¶
At returns the sequence of the i-th record in the Append (0-indexed). It panics if i is outside [0, Count).
type SnapshotPointer ¶
SnapshotPointer identifies the base snapshot a replica bootstraps from. The zero value (empty Location) means "no snapshot has been taken yet".
func (SnapshotPointer) IsZero ¶
func (s SnapshotPointer) IsZero() bool
IsZero reports whether no snapshot has been recorded.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is the CAS read/commit handle for a manifest object. The writer uses Commit (PutUpdate against the version observed by Load); the reader uses Load to tail. It is deliberately thin: on a precondition failure the caller re-Loads to get the fresh version and replans, exactly as the buffer does.
func NewStore ¶
func NewStore(os objectstore.ObjectStore, path string) *Store
NewStore binds a Store to a manifest path in the given object store.
func (*Store) Commit ¶
func (s *Store) Commit(ctx context.Context, m *Manifest, version *objectstore.UpdateVersion) error
Commit writes the manifest only if its stored version still matches (PutUpdate). Returns objectstore.ErrPreconditionFailed on a lost CAS race; the caller should re-Load and replan.
func (*Store) Create ¶
Create writes the manifest only if it does not already exist (PutCreate). Returns objectstore.ErrAlreadyExists if another writer created it first.
func (*Store) Load ¶
func (s *Store) Load(ctx context.Context) (m *Manifest, version *objectstore.UpdateVersion, ok bool, err error)
Load fetches and parses the manifest, returning its current version for use as a CAS precondition. If the manifest does not exist yet it returns a fresh empty manifest, a nil version, and ok=false.
Source Files