wal

package
v0.0.0-...-8df5048 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 17, 2026 License: AGPL-3.0 Imports: 30 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// DefaultDiskReserveMin is the least default reserve: below this much
	// free space on the WAL volume, live writes are refused.
	DefaultDiskReserveMin = int64(512) << 20

	// DiskPressureMinOn is the least time the gate stays on once it trips, so
	// clients and MQTT subscriptions are not flapped by every purge.
	DiskPressureMinOn = 5 * time.Second
)

WAL volume disk pressure (#676). The writer PVC holds the WAL, Raft, the registration journal, arc.db, compaction temp and the standby store, so the constraint is the volume's free space, not the WAL's bytes. A WAL write that fails for space loses an entry the client was already told was accepted; refusing the write up front (503, retryable) loses nothing.

View Source 
const (
	// ReserveFileName is the reserve's name in the WAL directory. It must
	// never end in ".wal": every WAL listing, the purge and recovery match
	// that suffix, and the reserve is not a WAL file.
	ReserveFileName = ".arc-wal-reserve"
	// ReserveFileSize is how much the reserve holds.
	ReserveFileSize = int64(64) << 20
	// ReserveBootMinFree: a start that finds less free space than this on the
	// WAL volume removes the reserve before anything writes there.
	ReserveBootMinFree = uint64(32) << 20
)

The WAL reserve file (#676): space held on the WAL volume so a node whose volume filled up can still start. A full volume made the first WAL header write fail with ENOSPC, and the pod crash-looped; releasing the reserve at boot gives the start, and recovery, room to free space.

View Source 
const (
	WALChecksumCRC32 = 0x01 // CRC32 checksum type

	// Entry format: [Length: 4 bytes] [Timestamp: 8 bytes] [Checksum: 4 bytes] [Payload: N bytes]
	WALEntryHeaderSize = 16
	WALFileHeaderSize  = 7 // Magic(4) + Version(2) + ChecksumType(1)

	// MaxWALPayloadSize is the maximum allowed payload size for a single WAL entry.
	// This limit prevents integer overflow during buffer allocation (CWE-190) and
	// aligns with the replication protocol limit (100MB).
	MaxWALPayloadSize = 100 * 1024 * 1024 // 100MB

	// WALEnvelopeMarker is the first byte of an enveloped WAL payload.
	// Enveloped format: [0x01][2-byte db name length][db name][original msgpack]
	// Since msgpack maps/arrays always start with bytes >= 0x80, 0x01 is unambiguous.
	WALEnvelopeMarker   = 0x01
	WALCheckpointMarker = 0x02
	WALTrackedMarker    = 0x03
	// WALCoveredMarker is a tracked entry that also carries the ingest batch's
	// coverage id in a trailer (#1207); see coverage.go for the layout.
	WALCoveredMarker = 0x04
)
View Source 
const (

	// MaxCoverageKeyLen bounds the buffer key a covered entry carries; a
	// longer key is written as a plain tracked entry, without coverage.
	MaxCoverageKeyLen = 1 << 12
)

Covered entry layout (marker 0x04): 

[0x04][instance u64][walseq u64][logical payload][key][keyLen u16][chunk u16][chunks u16][cseq u64]

The logical payload is what a 0x03 entry carries (an enveloped msgpack value), so everything that reads a tracked entry reads a covered one the same way once the header and trailer are stripped, and the identity token is the same "instance+walseq" form — checkpoints, the purge floor and recovery identities are unchanged.

CSeq is last so the CRC of everything before it can be computed outside the shard lock: Commit only writes 8 bytes and folds them into the CRC.

Variables

View Source 
var (
	WALMagic   = []byte{'A', 'R', 'C', 'W'} // Magic bytes
	WALVersion = uint16(0x0001)             // Version 1
)

WAL file format constants

View Source 
var ErrCoverageTrailer = errors.New("WAL covered entry has a malformed coverage trailer")

ErrCoverageTrailer reports a 0x04 entry whose trailer does not fit.

View Source 
var ErrPayloadTooLarge = errors.New("WAL payload exceeds maximum allowed size")

ErrPayloadTooLarge indicates the payload exceeds MaxWALPayloadSize.

View Source 
var ErrWALDropped = errors.New("WAL entry dropped: async buffer full")

ErrWALDropped is returned by Append/AppendRaw/AppendRawWithMeta when the async entry channel is full and the entry is dropped. Previous behavior returned nil and silently incremented DroppedEntries — callers logging "data preserved in WAL for recovery" downstream were reporting durability they did not actually have. Returning a sentinel lets callers (the ingestion buffer in particular) increment their own error counters and surface accurate operator-facing messages. Use errors.Is to detect.

Functions

func DefaultDiskReserve 

func DefaultDiskReserve(maxSizeBytes int64) int64

DefaultDiskReserve is the reserve wal.disk_min_free_mb = 0 selects: max(512 MiB, two WAL files).

func DirectoryBytes 

func DirectoryBytes(walDir string) (int64, error)

DirectoryBytes reports the bytes occupied by every file in the WAL directory except the reserve file (ReserveFileName, #676).

Every file but the reserve, not a suffix list. The names that are not `*.wal` — `.wal.recovery` attempt sidecars, `.wal-recovery-*` temporaries leaked by a crash, legacy `.wal.recovered` files — are exactly what accumulates when recovery is failing, which is the state this gauge's capacity alert exists for. A suffix list also has to be revisited every time the WAL grows a new artifact name, and was not.

It therefore measures DIRECTORY occupancy. On a volume dedicated to the WAL, which is how `wal.directory` is meant to be pointed, that is the figure an operator wants; aimed at a shared path it will include whatever else is there.

The reserve file is left out: it is a fixed 64 MiB held for a start on a full volume, not WAL growth, and wal.max_directory_mb is enforced on this figure — counting the reserve would spend the cap on the space set aside for the case the cap exists to prevent.

func DiskPressureLevels 

func DiskPressureLevels(reserve, maxSizeBytes int64, total uint64) (on, offPoint int64, clamped bool)

DiskPressureLevels returns the effective free-space levels of the gate on a volume of total bytes: it turns on below reserve and off at or above offPoint. reserve <= 0 selects DefaultDiskReserve. The reserve and the band above it (two WAL files) are each clamped to a fifth of the volume; clamped reports that the reserve was. total 0 (unknown) clamps nothing.

The standby store's own free-space reserve on the same volume (#1266) must stay above offPoint, or a standby fills the volume to where its promotion starts under disk pressure.

func InstancesOnDisk 

func InstancesOnDisk(dir, skip string, logger zerolog.Logger) ([]uint64, error)

InstancesOnDisk returns the instances of the intact tracked and covered entries of every WAL file in dir (*.wal and quarantined *.failed) except skip (the active file, whose instance the caller knows), sorted. A file's answer is cached by its size and modification time. A file that is gone, or not a WAL file at all (a damaged quarantined one), holds nothing; only an I/O error fails the call.

func MeasurementName 

func MeasurementName(m interface{}) (string, bool)

MeasurementName reads the "m" of a columnar payload the way ingest does (ingest.extractMeasurement): a string as is, an integer as "measurement_<n>". The WAL keeps the payload the client sent, so a reader that accepted only a string failed every entry written with an integer measurement (#1225).

func ParseEnvelope 

func ParseEnvelope(payload []byte, defaultDB string) (database string, msgpackData []byte)

ParseEnvelope extracts the database name and msgpack payload from a WAL entry. If the payload uses the envelope format [0x01][2-byte dbLen][dbName]msgpack, it returns the database name and the inner msgpack bytes. Otherwise, it returns defaultDB and the original payload unchanged.

func ParseRecoveryRowIdentity 

func ParseRecoveryRowIdentity(identity string) (parent string, start, end int, ok bool)

ParseRecoveryRowIdentity recognizes a checkpoint for the half-open row range [start,end) of one tracked WAL entry. The parent is a writer-instance/sequence identity, never a legacy content hash: identical legacy payloads can represent distinct writes. The encoding is independent of the configured batch size.

func PrepareReserveAtBoot 

func PrepareReserveAtBoot(walDir string, logger zerolog.Logger) bool

PrepareReserveAtBoot runs before anything writes to the WAL volume at start (#676). It keeps an existing reserve file, and removes it when the volume has less than ReserveBootMinFree free, logging an Error: the start then has the reserve's room for the WAL header, the registration journal and SQLite. It never creates the reserve (the disk pressure monitor does, once there is room) and never refuses a start: a statfs error or a failed remove is logged and the start goes on. It reports whether it released.

func ReserveFilePath 

func ReserveFilePath(walDir string) string

ReserveFilePath is the reserve file's path in walDir.

func WithReplayCoverage 

func WithReplayCoverage(ctx context.Context, rc ReplayCoverage) context.Context

WithReplayCoverage attaches rc to the context recovery passes to its replay callbacks.

Types

type ColumnarEntry 

type ColumnarEntry struct {
	Database    string // From envelope metadata (empty = "default")
	Measurement string
	Columns     map[string][]interface{}
	TagColumns  []string
}

ColumnarEntry represents a columnar WAL entry written via the zero-copy path

type ColumnarRecoveryCallback 

type ColumnarRecoveryCallback func(ctx context.Context, database, measurement string, columns map[string][]interface{}, walIdentity string) error

ColumnarRecoveryCallback is called for columnar WAL entries during recovery.

walIdentity is the identity of the entry being replayed, so the re-buffered batch can inherit it and have its eventual flush checkpoint the ORIGINAL entry — which is what stops a later pass replaying it again. It is empty for an entry that carries no tracked identity. database may be empty if the WAL entry predates the envelope format (defaults to "default").

type ColumnarRecoveryCallbackWithTagColumns 

type ColumnarRecoveryCallbackWithTagColumns func(ctx context.Context, database, measurement string, columns map[string][]interface{}, walIdentity string, tagColumns []string) error

ColumnarRecoveryCallbackWithTagColumns also receives tag metadata preserved in raw MessagePack entries. When set, it takes precedence over ColumnarCallback.

type Coverage 

type Coverage struct {
	Instance uint64
	Key      string
	CSeq     uint64
	Chunk    uint16
	Chunks   uint16
}

Coverage identifies the ingest batch a WAL entry belongs to (#1207).

The ingest buffer mints CSeq per (writer instance, buffer key) under the shard lock at the moment the batch is committed, so the batches of one key commit in CSeq order and a flush of that key covers one contiguous range. A batch too large for one entry is split into Chunks entries that share the CSeq; Chunk is this entry's index.

type CoverageHold 

type CoverageHold interface {
	HoldsAny(instance uint64, key string, lo, hi uint64) bool
}

CoverageHold reports whether any batch of instance and buffer key with a coverage sequence in [lo, hi] is still waiting for its file to register. Implementations are immutable snapshots, safe for concurrent use.

type CoverageSpan 

type CoverageSpan struct {
	Instance uint64
	Key      string
	Lo, Hi   uint64
}

CoverageSpan is a run of coverage sequences of one instance and key.

type DiskMonitor 

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

DiskMonitor decides WAL volume disk pressure (#676). A goroutine of its own samples the volume every second and at once on a rotation or a disk-full write: the WAL maintenance goroutine can block for minutes in recovery. Between samples the gate is exact: every byte admitted to the WAL queue since the sample is charged against the free space it saw, so a burst cannot run past the reserve before the next sample.

It trips when free - bytes since the sample - reservations < reserve, when the directory cap is set and directory bytes + bytes since + reservations > cap, or on any write that failed for space (DiskFullObserved only trips). Requests reserve their size at admission (Reserve), so a large request, or several at once, is refused before it can cross the reserve. It clears only at or above the off-point (reserve + two WAL files, both clamped) and after DiskPressureMinOn: a trip is never delayed by a minimum time in the off state, because the trip is what keeps the volume from filling.

func NewDiskMonitor 

func NewDiskMonitor(w *Writer, cfg DiskMonitorConfig) *DiskMonitor

NewDiskMonitor builds the disk pressure monitor of w. Start it before serving writes.

func (*DiskMonitor) Close 

func (m *DiskMonitor) Close() error

Close stops the sampler goroutine. The gate keeps its last state.

func (*DiskMonitor) Levels 

func (m *DiskMonitor) Levels() (reserve, offPoint int64)

Levels returns the effective reserve (the gate is on below it) and off-point (it clears at or above it), in free bytes on the WAL volume, as of the last sample.

func (*DiskMonitor) OnChange 

func (m *DiskMonitor) OnChange(fn func(on bool))

OnChange registers fn for every change of the gate, called from the sampler goroutine; fn(true) is called at once when the gate is already on. fn must not block for long and must not call OnChange.

func (*DiskMonitor) Pressure 

func (m *DiskMonitor) Pressure() bool

Pressure reports whether live writes are refused now. The hot path: an atomic flag, one snapshot load and two counter loads. Writes that would cross a budget trip the gate here, at once, and wake the sampler.

func (*DiskMonitor) Release 

func (m *DiskMonitor) Release(n int64)

Release returns n reserved bytes: queued since (now charged as queued), or never written.

func (*DiskMonitor) Reserve 

func (m *DiskMonitor) Reserve(n int64) bool

Reserve admits a request of about n WAL bytes (#676): it fits when the bytes queued since the sample, every reservation not yet queued and n stay within both budgets; it is then charged until Release. A request that does not fit is refused without turning the gate on, unless what is already charged is over a budget. Concurrent reservations see each other.

func (*DiskMonitor) Start 

func (m *DiskMonitor) Start()

Start takes the first sample, so the gate is armed before the first write, and starts the sampler goroutine. Idempotent.

type DiskMonitorConfig 

type DiskMonitorConfig struct {
	// Dir is the WAL directory; its volume is the one sampled.
	Dir string
	// ReserveBytes is wal.disk_min_free_mb in bytes; 0 = DefaultDiskReserve.
	ReserveBytes int64
	// MaxSizeBytes is wal.max_size_mb in bytes (the hysteresis band is two).
	MaxSizeBytes int64
	// MaxDirectoryBytes is wal.max_directory_mb in bytes; 0 = no cap. For a
	// volume whose statfs does not describe the limit that applies (an
	// emptyDir with a sizeLimit reports the node's disk).
	MaxDirectoryBytes int64
	// ReserveFile keeps the reserve file (ReserveFileName) in Dir.
	ReserveFile bool
	Logger      zerolog.Logger

	// ActiveBytes and Rotate let the gate clear when the active WAL file is
	// what holds it on (NewDiskMonitor sets both from the writer): while
	// writes are refused nothing rotates that file, so its bytes, flushed or
	// not, would never be purged. Nil: never rotate.
	ActiveBytes func() int64
	Rotate      func() error
}

DiskMonitorConfig configures a DiskMonitor.

type Entry 

type Entry struct {
	TimestampUS      uint64                   // Microseconds since epoch
	PayloadHash      string                   // Entry identity: the writer-assigned "instanceSeq" (32 hex) for a tracked entry, else a SHA-256 of the logical payload (64 hex)
	CheckpointHashes []string                 // Flush checkpoint identities, when present
	Records          []map[string]interface{} // Row format (from Append path)
	ColumnarData     *ColumnarEntry           // Columnar format (from AppendRaw path)
	Coverage         *Coverage                // Ingest batch coverage, for a covered (0x04) entry (#1207)
	// DecodeErr is set for an entry whose bytes are intact (checksum valid)
	// but whose payload this build cannot decode. Its identity is still known,
	// so recovery can tell whether it was already flushed (#1225).
	DecodeErr error
}

Entry represents a single WAL entry

func DecodeLogical 

func DecodeLogical(logicalPayload []byte) *Entry

DecodeLogical decodes a logical WAL payload ([0x01][dbLen][db]msgpack or bare msgpack): row form into Records, columnar form into ColumnarData, and anything else into an entry carrying DecodeErr. The standby store keeps these payloads as the origin wrote them (#1207).

type PreparedAppend 

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

PreparedAppend is a covered WAL append built outside the ingest shard lock and committed under it (#1207). Prepare allocates the WAL sequences — they are pending in the purge floor from that moment — and assembles every entry except the coverage sequence. Commit stamps the sequence, fires the replication hook and enqueues. Abandon releases the sequences: nothing was written, so a write refused to the client never reaches the WAL or a replica.

Exactly one of Commit or Abandon must be called; Abandon after Commit is a no-op, so a deferred Abandon is the intended pattern.

func (*PreparedAppend) Abandon 

func (p *PreparedAppend) Abandon()

Abandon releases the sequences of a prepared append that was never committed. Safe to defer: a no-op after Commit.

func (*PreparedAppend) Chunks 

func (p *PreparedAppend) Chunks() int

Chunks is the number of WAL entries the batch was split into.

func (*PreparedAppend) Commit 

func (p *PreparedAppend) Commit(cseq uint64) ([]string, error)

Commit stamps cseq, fires the replication hook and enqueues every entry in order. It is cheap — no allocation proportional to the payload — because it runs under the ingest shard lock.

On a full queue it stops at the first dropped entry and returns ErrWALDropped with the tokens of the entries already enqueued: those are in the WAL and must be checkpointed by the flush like any other, while the dropped and unsent ones are released here. The batch itself is committed to the buffer either way; the WAL is for durability, not correctness.

func (*PreparedAppend) Instance 

func (p *PreparedAppend) Instance() uint64

Instance is the writer instance the prepared entries belong to.

type Reader 

type Reader struct {

	// Metrics
	TotalEntries     int64
	TotalBytes       int64
	CorruptedEntries int64
	// TruncatedEntries counts incomplete entries at EOF after an interrupted
	// append. These are not corrupt complete entries and cannot be replayed.
	TruncatedEntries int64
	// UndecodableEntries counts intact entries whose payload this build
	// cannot decode. They are returned with DecodeErr set, and are included
	// in TotalEntries.
	UndecodableEntries int64
	// contains filtered or unexported fields
}

Reader reads WAL files for recovery operations

func NewReader 

func NewReader(filePath string, logger zerolog.Logger) *Reader

NewReader creates a new WAL reader

func (*Reader) ReadAll 

func (r *Reader) ReadAll() ([]Entry, error)

ReadAll reads all entries from the WAL file

func (*Reader) ReadCheckpointHashes 

func (r *Reader) ReadCheckpointHashes() ([]string, error)

ReadCheckpointHashes scans a WAL file for flush checkpoints without materializing or decoding ordinary data entries. Non-checkpoint payloads are skipped by seeking over their validated on-disk lengths; the recovery pass still reads and checksum-validates entries before replaying them.

func (*Reader) ReadTrackedIdentities 

func (r *Reader) ReadTrackedIdentities() ([]string, error)

ReadTrackedIdentities returns the identity tokens of the intact tracked (0x03) and covered (0x04) entries of a WAL file, without decoding their payloads (#1225). A damaged entry is counted in CorruptedEntries and skipped; a torn tail ends the scan.

func (*Reader) ScanCovered 

func (r *Reader) ScanCovered(match func(Coverage) bool, fn func(*Entry) error) error

ScanCovered streams the intact covered (0x04) entries of the file whose coverage match accepts, in file order; only those payloads are decoded (#1207 S4b1c2b: an origin republishing parked batches reads them back). A damaged entry is counted and skipped; a torn tail ends the scan. An error from fn stops it and is returned.

type Recovery 

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

Recovery manages WAL recovery operations

func NewRecovery 

func NewRecovery(walDir string, logger zerolog.Logger) *Recovery

NewRecovery creates a new WAL recovery manager

func (*Recovery) CleanupOldWALs 

func (r *Recovery) CleanupOldWALs(maxAge time.Duration) (int, int64, error)

CleanupOldWALs removes legacy .recovered WAL files older than the specified age. Note: As of the current implementation, WAL files are deleted immediately after successful recovery, so this function is primarily for cleaning up legacy files from previous versions that renamed files to .recovered instead of deleting them.

func (*Recovery) ListWALFiles 

func (r *Recovery) ListWALFiles() (active []string, recovered []string, err error)

ListWALFiles lists all WAL files in the directory. Returns active (pending) WAL files and legacy .recovered files. Note: As of the current implementation, WAL files are deleted immediately after successful recovery, so the recovered list will typically be empty or contain only legacy files from previous versions.

func (*Recovery) Recover 

func (r *Recovery) Recover(ctx context.Context, callback RecoveryCallback) (*RecoveryStats, error)

Recover scans the WAL directory and replays all WAL files

func (*Recovery) RecoverWithOptions 

func (r *Recovery) RecoverWithOptions(ctx context.Context, callback RecoveryCallback, opts *RecoveryOptions) (*RecoveryStats, error)

RecoverWithOptions scans the WAL directory and replays WAL files with configurable options

type RecoveryCallback 

type RecoveryCallback func(ctx context.Context, records []map[string]interface{}) error

RecoveryCallback is called for each batch of records during recovery (row format)

type RecoveryOptions 

type RecoveryOptions struct {
	// SkipActiveFile is the path to the currently active WAL file that should be skipped
	// during periodic recovery (to avoid reading a file being actively written)
	SkipActiveFile string

	// AdditionalCheckpointHashes contains checkpoints read safely from the
	// active file, which is intentionally excluded from recovery scans.
	AdditionalCheckpointHashes []string

	// BatchSize limits how many records are replayed per callback invocation
	// This provides backpressure during mass recovery after prolonged outages
	// 0 means no limit (all records in an entry replayed at once)
	BatchSize int

	// ColumnarCallback handles columnar WAL entries from the zero-copy write path
	ColumnarCallback ColumnarRecoveryCallback

	// ColumnarCallbackWithTagColumns preserves raw MessagePack tag metadata.
	ColumnarCallbackWithTagColumns ColumnarRecoveryCallbackWithTagColumns

	// TrackedRowCallback handles row-format entries in batches bounded by
	// BatchSize. Legacy content-hash entries inherit no checkpoint identity.
	TrackedRowCallback TrackedRecoveryCallback

	// ValidateTrackedRows validates the complete row entry before any of its
	// batches are submitted. This preserves whole-entry validation when batching.
	ValidateTrackedRows func([]map[string]interface{}) error

	// CheckpointRecovered checkpoints original parent identities after all of
	// their row ranges are durable and before deleting their files. Production
	// callers use Writer.MarkFlushed to release the original pending sequence.
	// A failure retains the files for retry. Nil is suitable for synchronous
	// callbacks with no live writer pending identities to release.
	CheckpointRecovered func(context.Context, []string) error

	// MinFileAge, when > 0, skips WAL files modified more recently than this.
	// Defense against the #594 class beyond the SkipActiveFile name match:
	// the periodic recovery reads CurrentFile() and then scans — a rotation
	// landing between those instants would put the NEW active (header-only)
	// file in the scan, and deleting it re-creates the unlinked-inode data
	// loss. Production call sites pass a few seconds; a genuinely
	// recoverable young file is picked up by the next pass. Zero disables
	// the guard (tests recover freshly-written files).
	MinFileAge time.Duration

	// MinFileAgeExemptFiles are known-closed files that may be recovered even
	// when their modification time is recent. The periodic recovery path uses
	// this for the file that Writer.Rotate just closed; MinFileAge still
	// protects files that rotate while recovery is scanning.
	MinFileAgeExemptFiles []string

	// BeforeDelete is an epoch barrier that must make every replayed entry in a
	// batch durable before the corresponding WAL files are removed. The
	// callback runs after BarrierBatchFiles files or BarrierBatchRows records
	// have replayed, whichever threshold is reached first. A nil callback means
	// the recovery callback itself provides synchronous durability.
	// A batch containing only entries covered by durable checkpoints needs no
	// flush; an earlier replay still pending in that batch must be fenced.
	BeforeDelete func(context.Context) error

	// BarrierBatchFiles bounds how many successfully replayed files are covered
	// by one BeforeDelete call. Defaults to 16 when BeforeDelete is configured.
	BarrierBatchFiles int

	// BarrierBatchRows optionally triggers BeforeDelete after this many rows
	// have replayed, allowing callers to align barriers with max_buffer_size.
	BarrierBatchRows int

	// MaxReplayFailures is the number of failed replay passes before a poison
	// WAL file is quarantined as .wal.failed. Defaults to 3.
	MaxReplayFailures int

	// Held returns the registration hold (#1207 S4c). It is consulted after
	// each barrier, so the batches this pass just replayed are in it. Nil
	// holds nothing.
	Held func() CoverageHold

	// OnQuarantine is called after a WAL file is moved to quarantinedAs, so
	// the writer can stop waiting for the entries in it (#1225).
	OnQuarantine func(path, quarantinedAs string)
}

RecoveryOptions configures WAL recovery behavior

type RecoveryStats 

type RecoveryStats struct {
	RecoveredFiles   int
	RecoveredBatches int
	RecoveredEntries int
	// CorruptedEntries counts entries this pass could not replay: damaged
	// ones, ones whose replay failed, and intact ones that cannot be decoded
	// and were not flushed. An undecodable entry already flushed is skipped
	// like any other flushed entry, and not counted.
	CorruptedEntries int
	SkippedFiles     int
	KeptFiles        int
	BarrierFailures  int
	QuarantinedFiles int
	// PartialRowEntries counts the WAL files this pass left on disk for which at
	// least one row RANGE of a parent entry was handed off. Their checkpoints
	// land on the next flush, and a binary without row-range support treats
	// those identities as inert and replays the parent entry in full — so this
	// is the state the restart compatibility guidance is about, made checkable.
	//
	// It describes what THIS pass observed, so it is a floor on the directory's
	// residue rather than a census: a file whose replay failed before any range
	// was handed off contributes nothing, and so does the out-of-range
	// checkpoint case, which breaks before the ranges are computed even though
	// a bad coverage map is itself evidence that ranges exist.
	PartialRowEntries int
	RecoveryDuration  time.Duration
	// Retained lists the files this pass replayed (or found checkpointed) but
	// kept because a batch in them is still waiting for its file to register
	// (#1207 S4c), and every file after the first such one: a later file can
	// hold the only checkpoint for an earlier one's entries. Not failures: they
	// are not counted in KeptFiles; the writer deletes them through
	// ReleaseRetained once nothing in them is held.
	Retained []RetainedFile
}

RecoveryStats holds statistics about WAL recovery

type ReplayCoverage 

type ReplayCoverage struct {
	Coverage Coverage
	RowTotal int
}

ReplayCoverage travels with a replayed covered entry from recovery to the ingest buffer, so the file it is flushed into registers with the batch's original coverage (#1207). RowTotal is the entry's row count for a row-format entry, which recovery may replay in row ranges; zero for a columnar entry, which is always replayed whole.

func ReplayCoverageFrom 

func ReplayCoverageFrom(ctx context.Context) (ReplayCoverage, bool)

ReplayCoverageFrom returns the coverage of the entry being replayed, if any.

type ReplicationEntry 

type ReplicationEntry struct {
	// TimestampUS is the entry timestamp in microseconds since epoch
	TimestampUS uint64

	// Payload is the raw msgpack data
	Payload []byte

	// Coverage identifies the ingest batch the entry belongs to, for entries
	// appended through PrepareCovered (#1207). Nil otherwise.
	Coverage *Coverage
}

ReplicationEntry is a WAL entry as the replication hook receives it: the origin stream (#1207) sends it to standby writers.

type ReplicationHook 

type ReplicationHook func(entry *ReplicationEntry)

ReplicationHook is called for each WAL entry before it's written locally: the origin stream's, on a primary writer with WAL coverage (#1207).

type RetainedFile 

type RetainedFile struct {
	Path     string
	Coverage []CoverageSpan
}

RetainedFile is a WAL file recovery kept for registrations still pending, or because an earlier file of the same pass was kept for them. Coverage is the covered batches its entries belong to (possibly none held, for a file kept for its position).

type SyncMode 

type SyncMode string

SyncMode defines how WAL syncs to disk 

const (
	SyncModeFsync     SyncMode = "fsync"     // Full sync: data + metadata (safest)
	SyncModeFdatasync SyncMode = "fdatasync" // Data sync only (balanced, default)
	SyncModeAsync     SyncMode = "async"     // No explicit sync (fastest, least safe)
)

type TrackedRecoveryCallback 

type TrackedRecoveryCallback func(ctx context.Context, records []map[string]interface{}, walIdentity string) error

TrackedRecoveryCallback receives an entry identity, or a recovery row-range identity when BatchSize splits a tracked entry. A flush must checkpoint exactly the supplied identity, never the parent of a partially flushed entry.

type Writer 

type Writer struct {

	// Metrics (atomic for lock-free reads)
	TotalEntries   int64
	TotalBytes     int64
	TotalSyncs     int64
	FailedSyncs    int64 // Syncs that returned an error
	TotalRotations int64
	DroppedEntries int64 // Entries dropped due to full buffer
	FailedWrites   int64 // Entries the writer loop could not write (lost)
	DiskFullWrites int64 // FailedWrites whose error was ENOSPC or EDQUOT
	// RecoveredWrites counts failed entry writes the retry on a new WAL file
	// recovered: nothing was lost, but the WAL file or its device failed.
	RecoveredWrites int64
	// contains filtered or unexported fields
}

Writer is a Write-Ahead Log writer with configurable durability

func NewWriter 

func NewWriter(cfg *WriterConfig) (*Writer, error)

NewWriter creates a new WAL writer

func (*Writer) ActiveFileBytes 

func (w *Writer) ActiveFileBytes() int64

ActiveFileBytes is the size of the WAL file being written, header included (#676).

func (*Writer) AdoptRetained 

func (w *Writer) AdoptRetained(files []RetainedFile)

AdoptRetained takes over the files recovery retained. Files this process created are left to PurgeFlushed; a file already adopted keeps its first summary (a later pass reads the same entries).

func (*Writer) Append 

func (w *Writer) Append(records []map[string]interface{}) error

Append writes records to the WAL asynchronously (non-blocking)

func (*Writer) AppendRaw 

func (w *Writer) AppendRaw(payload []byte) error

AppendRaw writes raw (already serialized) msgpack bytes to the WAL asynchronously This is a zero-copy optimization - use this when you already have msgpack bytes

func (*Writer) AppendRawTracked 

func (w *Writer) AppendRawTracked(payload []byte) ([]string, error)

AppendRawTracked writes a raw msgpack payload with a process-local sequence identity and returns the identities the eventual flush must checkpoint. It is used by replication followers so their local WAL participates in the same sequence-floor purge as primary ingest.

func (*Writer) AppendRawWithMeta 

func (w *Writer) AppendRawWithMeta(database string, payload []byte) error

AppendRawWithMeta writes raw msgpack bytes with database metadata envelope. Format: [0x01 marker][2-byte db name length][db name][original msgpack] This preserves the database name for correct recovery routing.

Unlike calling AppendRaw with a pre-built envelope, this method builds the WAL entry in a single allocation to avoid copying the payload twice.

A payload whose logical entry exceeds MaxWALPayloadSize is split into multiple entries, each carrying the same database envelope (#677) — a wide ingest request previously landed here as a wholesale ErrPayloadTooLarge rejection, so the WAL silently recorded nothing while ingest reported healthy.

func (*Writer) AppendRawWithMetaTracked 

func (w *Writer) AppendRawWithMetaTracked(database string, payload []byte) ([]string, error)

AppendRawWithMetaTracked is the tracked counterpart to AppendRawWithMeta.

func (*Writer) AppendTracked 

func (w *Writer) AppendTracked(records []map[string]interface{}) ([]string, error)

AppendTracked writes a row-format entry and returns its payload identity. The identity is used by flush checkpoints to avoid replaying data that was already durably written to storage.

func (*Writer) Close 

func (w *Writer) Close() error

Close closes the WAL writer

func (*Writer) CoverageLost 

func (w *Writer) CoverageLost(key string, cseq uint64) bool

CoverageLost reports whether a covered entry of this writer's batch cseq under buffer key was lost to a failed write (#676): the WAL holds no complete copy of the batch, so recovery will never replay it.

func (*Writer) CurrentCheckpointHashes 

func (w *Writer) CurrentCheckpointHashes() ([]string, error)

CurrentCheckpointHashes returns flush checkpoints in the active WAL file. The writer lock keeps the file stable while the reader scans it.

func (*Writer) CurrentFile 

func (w *Writer) CurrentFile() string

CurrentFile returns the current WAL file path

func (*Writer) DiskFullObserved 

func (w *Writer) DiskFullObserved() bool

DiskFullObserved reports that the most recent entry write or rotation failed because the volume or the quota was full (ENOSPC, EDQUOT), and no entry has been written since (#676).

func (*Writer) ForgetTracked 

func (w *Writer) ForgetTracked(hashes []string)

ForgetTracked drops identities whose data will never be flushed, because the write that produced them was abandoned after its WAL append.

This is not the same as MarkFlushed: no checkpoint is written, because nothing reached storage. It only stops the floor waiting for data no buffer holds. Without it a single rejected write — a type-mismatched column, a client that disconnects during a schema-change flush, the schema-churn guard, a closing shard — would pin the floor for the life of the process, and because PurgeFlushed stops at the first retained file, the WAL would never reclaim anything again (#676).

Callers must only pass identities they are certain no buffer and no flush in flight still owns.

func (*Writer) HeldFiles 

func (w *Writer) HeldFiles() int

HeldFiles reports the WAL files kept because batches in them are waiting for their files to register: files of this process holding a held batch, and files recovery retained. While registrations cannot land (a lost quorum, a start without the cluster) every later file is kept too, by the purge order; arc_wal_dir_bytes shows that growth.

func (*Writer) MarkFlushed 

func (w *Writer) MarkFlushed(hashes []string) error

MarkFlushed appends a checkpoint after the corresponding data entries have reached durable storage. A failed checkpoint is safe: it can only cause a replay duplicate, never data loss.

Admission is non-blocking — a full queue returns ErrWALDropped — because this is the production ingest flush path (ArrowBuffer.markWALFlushed). Completion already blocks on the writer loop's reply. Use MarkFlushedContext from recovery, which needs the opposite trade.

func (*Writer) MarkFlushedContext 

func (w *Writer) MarkFlushedContext(ctx context.Context, hashes []string) error

MarkFlushedContext is MarkFlushed with a cancellable, blocking admission wait instead of the drop-on-full enqueue.

Recovery's parent finalization runs under the same sustained queue pressure that made the forced maintenance rotation unreliable (#1009) — same pass, same queue — but was left on the dropping path when rotation moved off it. A dropped checkpoint retains the replayed file for another pass, which is safe but is one of the ways a file stays retained long enough to matter.

ctx bounds ADMISSION only. Once an entry is admitted the checkpoint will be written, so the reply is waited for unconditionally — see markFlushed. On a cancelled admission or a closed writer this returns an error, so the caller retains the WAL and replays later: at-least-once, never at-most-once.

func (*Writer) MarkFlushedRetained 

func (w *Writer) MarkFlushedRetained(hashes []string) error

MarkFlushedRetained writes the same durable checkpoint as MarkFlushed — so recovery does not replay the entries — but keeps their sequences pending in the purge floor. A covered flush (#1207) holds its WAL entries until the files they became are registered; the caller releases them with ReleaseRegistered once that happens.

func (*Writer) MinUnflushedSequence 

func (w *Writer) MinUnflushedSequence() uint64

func (*Writer) PendingUnflushedCount 

func (w *Writer) PendingUnflushedCount() int

MinUnflushedSequence returns the lowest tracked sequence this process has appended that no durable flush checkpoint covers. With nothing pending it returns one past the highest sequence issued.

Deliberately NOT one past infinity. The caller reads the floor and then calls PurgeFlushed with it, under a different lock acquisition, so an append and a rotation can land in between; a MaxUint64 floor is above every sequence the writer can ever issue, and would purge that file. highest+1 is above everything pending and below anything issued later, which closes that window at no cost.

Deliberately a plain scan of the pending set, too. An earlier revision walked a forward-only cursor over the sequence space to avoid it, which was both unsound and slower: unsound because a sequence allocated but not yet published let the cursor park above it and never return to it, and slower because every pending sequence lies in [min, highest], so the range the cursor walks is always at least as large as the set it is scanning. PendingUnflushedCount reports how many tracked sequences are awaiting a durable flush checkpoint. It is the floor's working set, and it grows for as long as flushes keep failing — during a prolonged object-store outage that is one entry per write request, which is worth being able to see rather than infer. Detect and report rather than cap, as with the ingest buffers.

func (*Writer) PrepareCovered 

func (w *Writer) PrepareCovered(database, key string, payload []byte) (*PreparedAppend, error)

PrepareCovered prepares a covered append of payload (msgpack) for database under buffer key key. An oversized payload is split into chunks exactly as AppendRawWithMetaTracked splits it.

func (*Writer) PurgeAll 

func (w *Writer) PurgeAll() (int, error)

PurgeAll deletes all WAL files in the directory. Call this after a clean shutdown where all data has been flushed to storage, so that recovery on next startup doesn't replay already-persisted data.

func (*Writer) PurgeFlushed 

func (w *Writer) PurgeFlushed(minUnflushedSeq uint64) (int, error)

PurgeFlushed deletes rotated WAL files whose data is known to have reached storage, where "known" means every sequence they hold is below minUnflushedSeq — the lowest sequence that is still only in memory.

This replaces purging by modification time, which assumed flush latency was at most a fixed age and deleted acknowledged-but-unflushed data whenever that assumption broke (#966, #1009; PR #997 measured 48,500 records lost this way). It is clock-free: no mtime granularity, no NTP steps.

Three rules, each of which is a data-loss or duplicate bug if dropped:

  • Only files THIS process created are candidates. A file from a previous process is absent from w.fileSeqs, and nothing here knows whether its data was flushed; only recovery may delete those.
  • A file holding an untracked data entry is never a candidate. Nothing will ever report that entry flushed, so its sequence bound means nothing.
  • Walk in rotation order and STOP at the first file that must be retained. A flush checkpoint is appended like any other entry, so a checkpoint covering file R's entries can live in R or in any later file. Deleting a later file while R is retained can therefore destroy the only record that R's entries were flushed, and recovery would replay them — permanent duplicate rows for a measurement without tags.

Returns the number of files deleted.

func (*Writer) PurgeHint 

func (w *Writer) PurgeHint() <-chan struct{}

PurgeHint is signalled when a purge could now delete something: a rotation finished a file, or the purge floor or a pin moved (#676). Hints coalesce (capacity 1); the receiver runs PurgeFlushed and ReleaseRetained, rate limited — the WAL directory otherwise grows until the periodic pass.

func (*Writer) PurgeInactive 

func (w *Writer) PurgeInactive() (int, error)

PurgeInactive deletes all WAL files except the currently active one. Use this during normal operation (unlike PurgeAll which is for shutdown) to clean up rotated WAL files after their data has been flushed to storage.

func (*Writer) Quarantined 

func (w *Writer) Quarantined(path, quarantinedAs string) int

Quarantined tells the writer that recovery moved path to quarantinedAs after repeated replay failures (#1225). If this process wrote it, entries in it may still be waiting for a flush: recovery can no longer replay them from it, so keeping them pending would hold the purge floor at this file for the life of the process, and no WAL file would be reclaimed again (#676). Their sequences are released. Returns how many were.

Every pending entry of the file is released, including any this pass replayed into a buffer: if that flush fails too, nothing can replay it again. Nothing else owns them: only the periodic recovery pass quarantines a file this process wrote, after a rotation and a flush of every buffer.

The sequences come from the quarantined file itself. An entry whose bytes are damaged has no readable identity; its sequence stays pending until a restart.

func (*Writer) ReleaseRegistered 

func (w *Writer) ReleaseRegistered(hashes []string)

ReleaseRegistered drops identities checkpointed with MarkFlushedRetained from the purge floor once their files are registered (#1207). Also the release for a checkpoint that failed, exactly as ForgetTracked is today.

func (*Writer) ReleaseRetained 

func (w *Writer) ReleaseRetained() (int, error)

ReleaseRetained deletes adopted files whose batches are no longer held. It walks every WAL file this process did not create in name order — the order recovery replays and retains them — and stops at the first one it must keep: a held file, or an earlier-process file it was never given (recovery kept it for a failure and will process it again). Stopping, not skipping, keeps the proof ordering: a later file can hold the only checkpoint for an earlier one's entries. Returns the number of files deleted.

func (*Writer) RetainedFiles 

func (w *Writer) RetainedFiles() int

RetainedFiles reports how many files recovery retained that are still kept.

func (*Writer) Rotate 

func (w *Writer) Rotate() error

Rotate closes the active WAL file and starts a new one, and does not return until every entry enqueued before the call has been written to the OLD file.

That ordering is the whole point. A forced rotation exists so recovery can reach data that is only in the active file (#1009): the file is rotated away and then replayed. A rotation that simply took w.mu and swapped the handle would leave every entry already sitting in the queue — up to BufferSize of them — to be written to the NEW file, where the pass that rotated would not look. Going through the writer loop puts the rotation behind those entries.

func (*Writer) RotateContext 

func (w *Writer) RotateContext(ctx context.Context) error

RotateContext waits for queue capacity instead of dropping a maintenance command when writers fill the queue. The command uses the same FIFO as data, so all preceding accepted entries are processed before the rotation.

Cancellation bounds both admission and completion waits. Once admitted, the command remains owned by the writer loop and can complete after cancellation; callers must not infer a rotation boundary from an error. The buffered reply lets the writer finish even after its caller has stopped waiting.

func (*Writer) SetCoverageHold 

func (w *Writer) SetCoverageHold(hold func() CoverageHold)

SetCoverageHold installs the source of the hold. hold is called once per purge pass, outside the writer lock; nil (or a nil result) holds nothing.

func (*Writer) SetReplicationHook 

func (w *Writer) SetReplicationHook(hook ReplicationHook)

SetReplicationHook sets the hook function called for each WAL entry (the origin stream's, #1207). The hook is called synchronously before the entry is written locally.

func (*Writer) Stats 

func (w *Writer) Stats() map[string]interface{}

Stats returns WAL statistics

func (*Writer) TrackedInstance 

func (w *Writer) TrackedInstance() uint64

TrackedInstance is this writer's instance id: the prefix of every identity token it mints and the instance of every coverage id it writes.

type WriterConfig 

type WriterConfig struct {
	WALDir       string        // Directory for WAL files
	SyncMode     SyncMode      // Sync mode: fsync, fdatasync, async
	MaxSizeBytes int64         // Rotate WAL when it reaches this size (default: 100MB)
	MaxAge       time.Duration // Rotate WAL after this duration (default: 1 hour)
	SyncInterval time.Duration // Request a sync at least this often while bytes are unsynced (default: 100ms; 0 = default)
	SyncBytes    int64         // Request a sync once this many bytes are unsynced (default: 1MB; 0 = default)
	BufferSize   int           // Size of async write buffer (default: 10000)
	Logger       zerolog.Logger
}

WriterConfig holds configuration for WAL writer

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.