wal

Documentation

Overview

Package wal provides a small segmented write-ahead log with grouped fsync.

Index

• func Replay(dir string, from uint64, maxRecord int, fn func(Record) error) error
• func ReplayFromCursor(dir string, cursor Cursor, maxRecord int, fn func(Record, Cursor) error) error
• type Cursor
  • func CursorAfter(record Record) Cursor
• type Log
  • func Open(dir string, opts Options) (*Log, error)
  • func (l *Log) Append(ctx context.Context, payload []byte) (RecordID, error)
  • func (l *Log) Close() error
  • func (l *Log) CompactBefore(seq uint64) error
  • func (l *Log) NextSeq() uint64
• type Options
• type Record
• type RecordID

Functions

func Replay

func Replay(dir string, from uint64, maxRecord int, fn func(Record) error) error

Replay invokes fn for every record with seq >= from, in order.

func ReplayFromCursor

func ReplayFromCursor(dir string, cursor Cursor, maxRecord int, fn func(Record, Cursor) error) error

ReplayFromCursor invokes fn for every record at or after cursor, in order, passing alongside each record the cursor to resume just past it.

Types

type Cursor

type Cursor struct {
	SegmentBase uint64
	Offset      int64
	Seq         uint64
}

Cursor is a resume position for ReplayFromCursor. Replay starts at Offset within the segment whose base is SegmentBase and delivers records with seq >= Seq.

func CursorAfter

func CursorAfter(record Record) Cursor

CursorAfter returns the cursor positioned immediately after record, suitable for resuming replay without re-reading it.

type Log

type Log struct {
	// contains filtered or unexported fields
}

Log is a segmented write-ahead log. Concurrent appends are staged into a shared buffer and flushed by a background sync loop with a single write+fsync per batch; each Append blocks until its batch is durable.

Locking: mu guards the append state (buffer, seq counters, pending batch). fileOps guards the active file handle across write/fsync and segment rolls. Methods with a Locked suffix require mu to be held.

func Open

func Open(dir string, opts Options) (*Log, error)

Open opens (or creates) the log in dir, recovers the next sequence number from the existing segments, truncates any torn tail from the active segment, and starts the background sync loop.

func (*Log) Append

func (l *Log) Append(ctx context.Context, payload []byte) (RecordID, error)

Append writes payload as one record and blocks until it is durably synced, returning the record's ID. The returned error is the sync outcome of the whole batch the record was flushed in.

func (*Log) Close

func (l *Log) Close() error

Close stops the sync loop, flushes any buffered records, and closes the active segment. It returns the latched sync error, if any.

func (*Log) CompactBefore

func (l *Log) CompactBefore(seq uint64) error

CompactBefore deletes segment files that contain only records with sequence numbers below seq. A segment is deletable when it is not the active segment and the next segment's base proves every record in it is below seq; the last listed segment is therefore always kept. When EVERY record in the log is below seq, the active segment is first rolled (past a size floor) so it too becomes deletable — otherwise a fully-compacted active segment stays on disk until new appends overflow it, which never happens once its writers go quiet.

func (*Log) NextSeq

func (l *Log) NextSeq() uint64

NextSeq returns the next sequence number the log will assign: one past the newest record (durable or staged), as recovered at Open — where the last segment's base is a floor, so an empty active segment left by compaction rotation cannot regress the sequence space — and advanced by appends since.

type Options

type Options struct {
	// SegmentBytes caps a segment file's size; an append that would
	// exceed it rolls the log to a new segment.
	SegmentBytes int64

	// SyncInterval is the backstop timer for the sync loop. Every Append
	// wakes the loop immediately (group commit), so this only bounds how
	// long buffered records can wait if a wakeup is ever missed.
	SyncInterval time.Duration

	// MaxRecord is the maximum payload size accepted by Append and the
	// upper bound trusted when validating frame lengths during recovery.
	MaxRecord int

	// CompactRotateBytes is the minimum active-segment size at which
	// CompactBefore rolls a fully-compactable active segment so its file
	// can be reclaimed. Without rotation, an active segment whose records
	// are all below the compaction point is pinned on disk forever — it
	// only seals when NEW appends overflow it, which never happens once
	// its topics stop producing. <=0 uses the default (1 MiB); the floor
	// bounds how much fully-compacted data may linger rather than churning
	// a segment roll on every compaction under light traffic.
	CompactRotateBytes int64
}

Options configures a Log. Zero values pick sensible defaults.

type Record

type Record struct {
	ID      RecordID
	Payload []byte
}

Record is a single log entry as delivered by replay.

type RecordID

type RecordID struct {
	SegmentBase uint64
	Offset      int64
	Seq         uint64
}

RecordID locates a record in the log. Seqs are dense and strictly increasing across segments; SegmentBase is the seq of the first record the containing segment may hold, and Offset is the byte position of the record's frame within that segment file.