spool

README

spool

A disk-backed FIFO byte queue for Go. Records are appended durably to segment files and consumed in order, surviving process restarts and power loss.

It exists for the store-and-forward problem: your process produces records faster than a remote endpoint can accept them, or the endpoint is simply down. Buffering in memory bounds you by RAM and loses everything on restart. This spools to disk instead, bounded by a byte cap, with an explicit commit so a crash re-delivers rather than drops.

q, _ := spool.Open("/var/lib/myagent/spool", spool.Options{MaxBytes: 512 << 20})

q.Append(record) // returns only once the record is on disk

data, commit, ok, _ := q.Pop()
if ok && send(data) == nil {
    commit() // not before: a crash here re-delivers
}

No dependencies outside the standard library except cespare/xxhash for the frame checksum.

Install

go get github.com/JohanLindvall/spool

Requires Go 1.21 or newer. CI runs the suite on amd64 and arm64 against current stable, plus the go.mod floor on amd64.

The durability contract

This is the part worth reading carefully, because it is the whole reason to pick this over a chan []byte and a bufio.Writer.

Append fsyncs before it returns. Once it returns nil, the record is on the physical medium. A producer that tails files or reads from a socket may therefore advance its own checkpoint on the strength of a successful Append — that is the entire point, and it is why the fsync is not optional.

if err := q.Append(rec); err != nil {
    return err // do NOT advance the checkpoint
}
checkpoint.Advance(offset) // safe: rec is durable

Pop does not remove. It hands back the record and a commit function. Delivery is at-least-once: a crash between Pop and commit re-delivers the record on the next open. commit is idempotent, becomes a no-op after Close, and a commit whose segment has since been skipped is a no-op rather than a silent retirement of somebody else's records.

Group commit for small records. One fsync per record is roughly a millisecond on commodity hardware, which caps you near a thousand records per second. If your records are small and you can batch, AppendNoSync enqueues without forcing durability and Sync makes the whole group safe with one fsync:

for _, rec := range batch {
    if err := q.AppendNoSync(rec); err != nil {
        return err
    }
}
if err := q.Sync(); err != nil {
    return err // errors.Is(err, spool.ErrGroupLost): re-append the group
}
checkpoint.Advance(offset) // the rule moves to Sync, it never weakens

The checkpoint rule moves from Append to Sync — it does not soften. On a Sync failure the entire unsynced group is rolled back and must be re-appended; duplicates are possible if some frames were popped in the meantime, which is within at-least-once. Close flushes any open group, and reports the error if it could not.

The consumer has its own fsync, and it is off by default. commit rewrites a 24-byte cursor file. Syncing that costs an fsync per record, which puts the consumer under the same ~1 ms ceiling the producer just escaped. But the cursor is a progress hint, not a durability record: a stale, torn or missing cursor causes only redelivery, which at-least-once already permits, bounded by one segment. So Options.CommitSync defaults to false. Set it when duplicates after a power cut cost more than consumer throughput.

Measured on an AMD Ryzen 7 8840HS against ext4 on NVMe (make bench — these numbers describe your disk far more than they describe this code):

Path	ns/record	records/s
Append (256 B, one fsync per record)	1 145 000	~875
AppendNoSync ×10 + Sync	119 000	~8 400
AppendNoSync ×100 + Sync	14 048	~71 000
AppendNoSync ×1000 + Sync	2 795	~358 000
Pop alone (no commit)	1 341	~746 000
Pop + commit, CommitSync: false	3 024	~331 000
Pop + commit, CommitSync: true	1 137 000	~880

Read the last two rows together with the group-commit rows: a producer doing 358 000 records/s into a consumer pinned at 880/s grows the backlog until the cap refuses. Both sides need the fsync amortised, or neither does.

Framing itself is free by comparison: the per-frame xxhash64 runs at ~18 GB/s with zero allocations, some three orders of magnitude cheaper than the fsync it rides along with. Integrity here is not a throughput trade.

Append and AppendNoSync are allocation-free. Pop costs three allocations per record: the payload it hands you, plus the commit closure and its captured state. Replacing the closure with a Record type would save two allocations sitting next to two ReadAt syscalls, at the cost of breaking every caller — not a trade worth making.

Bounding and backpressure

MaxBytes caps the undelivered backlog — not the disk footprint. A delivered but unretired prefix keeps its segment alive, so physical usage can exceed the cap by up to one segment; Stats().DiskBytes reports the real number.

The zero value is bounded. Options{} means DefaultMaxBytes (64 MiB), not unlimited. Uncapped is still available, but you have to ask:

spool.Options{}                            // 64 MiB cap, 8 MiB segments
spool.Options{MaxBytes: spool.Unbounded}   // no cap — fills the volume if the consumer stops

That default is deliberate. A disk buffer whose unconfigured state is "grow until the volume dies" takes down whatever else shares the mount, and the failure is silent until something unrelated breaks. Size it for your outage budget.

Past the cap, Append returns ErrFull and leaves the queue untouched, so the caller decides: block, drop, or shed load upstream. Nothing is discarded behind your back.

switch err := q.Append(rec); {
case errors.Is(err, spool.ErrExceedsCap):
    // Permanent: this record cannot fit under this cap on an empty queue
    // either. Drop it or split it — retrying will fail identically forever.
    metrics.Oversized.Inc()
case errors.Is(err, spool.ErrFull):
    metrics.Dropped.Inc() // transient: clears as the consumer drains
case err != nil:
    return err
}

A record fits iff len(rec) <= MaxBytes - spool.FrameOverhead. Records larger than spool.MaxRecordBytes (4 GiB − 1, the frame length field's ceiling) are refused with ErrRecordTooLarge regardless of the cap.

Observability

Stats() returns a snapshot: gauges for now, monotonic counters since Open.

st := q.Stats()
// Gauges:   BacklogBytes, MaxBytes, DiskBytes, Segments, UnsyncedBytes
// Counters: Appended, Popped, Committed, Full, LostBytes, LostSegments,
//           ForeignSegments, ForeignBytes, CursorErrors, DiscardedBytes

It is a plain struct on purpose — no prometheus.Collector (which would pull six modules into a package with one dependency and impose a registry model on callers who may use OTel), no expvar (process-global names break the moment you open two spools), no callback interface (which would run your code under the mutex, next to an fsync).

The four numbers that answer real pages:

• BacklogBytes / MaxBytes — utilisation. "70% and climbing" is the signal; a bare byte count is not.
• DiskBytes — what actually fills the volume, including the delivered-but- unretired prefix that BacklogBytes deliberately excludes.
• UnsyncedBytes — what a crash would lose right now, i.e. how far AppendNoSync has run ahead of the last Sync.
• rate(Appended) vs rate(Committed) — which side is stuck. Popped > Committed with Committed flat and a non-empty backlog is a poison head record; Requeue moves it to the back so the rest can drain.

increase(LostBytes) > 0 means corruption destroyed data, with a magnitude the ErrCorrupt return cannot carry. CursorErr() reports a cursor that cannot be persisted — commits still advance in memory, but a restart will redeliver.

Signal() returns a channel that fires after each append and whenever Pop advances past unreadable data, so one wait path covers both reasons Pop returns ok == false:

for {
    data, commit, ok, err := q.Pop()
    switch {
    case errors.Is(err, spool.ErrClosed):
        return
    case err != nil:
        // Corruption was dropped and the queue advanced. Count it and go round
        // again — Signal fires for these too, so the wait below stays correct.
        log.Warn("spool read", "err", err)
    case !ok:
        select {
        case <-q.Signal():
        case <-ctx.Done():
            return
        }
    default:
        if send(data) == nil {
            commit()
        }
    }
}

What a crash costs

The recovery behaviour is the design, not an afterthought — every case below has a test, and the fault-injection seams make the fsync and truncate failure paths reachable rather than theoretical.

Damage	Result
Crash mid-Append (torn frame at the tail)	truncated at load; costs nothing (the Append never returned)
Crash with an open AppendNoSync group	the group is lost; the next Sync returns ErrGroupLost instead of falsely acking
Failed fsync mid-append	tail rolled back to its last durable size, read position clamped with it, group invalidated
Crash between commit and cursor write	the record is re-delivered (at-least-once)
Torn cursor rewrite	fails its checksum, redelivers from the oldest segment
Cursor permanently unwritable	commits still advance; CursorErr() reports it, restart redelivers ≤ one segment
One flipped byte in a payload	one record lost: Pop returns ErrCorrupt and advances
One flipped byte in a frame length	up to one segment lost — see below
Damaged segment header	segment dropped at load, surfaced as one ErrCorrupt, bytes counted in LostBytes
Zero-length segment (aborted create)	removed silently — it never held a record, so it is not a loss event
Segment file deleted underneath	head skipped, fs.ErrNotExist surfaced, queue keeps making progress
Unknown format version	segment dropped silently, counted in ForeignSegments

"One bad byte costs one record" is true of payloads, not of the four length bytes. The checksum covers the length alongside the payload, so a damaged length is always detected — that is the guarantee. But it cannot be repaired: a length damaged upward overruns the segment, a length damaged downward desynchronizes the walk into the middle of a payload, and either way the frame boundaries from that point on are gone. Pop skips the whole segment, so one bad byte there can cost up to SegmentBytes.

Two rules run through all of it. Corruption degrades to reported loss, never to corrupt output and never to a wedged queue — a damaged record is dropped and counted, not handed onward as plausible-looking telemetry, and there is no failure mode where the consumer waits forever on bytes that will never parse. And every loss path is observable: Stats().LostBytes and LostSegments for corruption, Discarded() for a tail truncated at open, ErrCorrupt per event.

On-disk format

A directory of segment files (<seq>.seg, zero-padded, ascending) plus a cursor file.

segment: "KSPOOL" | version u16be | frame*
frame:   len u32be | xxhash64(len‖payload) u64be | payload
cursor:  seq u64be | offset u64be | fnv64a u64be     (rewritten in place)

Twelve bytes of framing per record (exported as FrameOverhead), eight per segment. The checksum covers the length bytes together with the payload, which is the load-bearing detail: a flipped length byte is caught rather than mis-framing everything behind it silently.

The format version is per segment rather than per spool, so a future bump does not invalidate what is already on disk — segments of both versions coexist in one directory, each read with the framing it was written in. A segment naming a version this build does not know is discarded at open; the spool is a transient buffer, and carrying a reader for every past and future format would not pay for itself.

(The magic still reads KSPOOL from this code's origin as an agent's log-export buffer. It is kept as-is so existing spool directories stay readable.)

Concurrency

Any number of goroutines may call the Append family, Sync, Pop, Close and the accessors concurrently; all state is mutex-guarded.

The real constraint is that Pop carries no per-consumer position: two Pop calls without an intervening commit return the same record. Serialize your consumer. A group-commit producer should likewise own the AppendNoSync/Sync pairing, or the group boundaries stop meaning anything.

Nothing coordinates two processes on one directory, and the failure is not graceful — each would repair the tail, retire the other's segments and overwrite the other's cursor. One process per spool directory.

Comparison with other spoolers

Everything below is an agent — a whole binary with inputs, parsing, routing, and a disk buffer somewhere inside. This is a library: the buffer alone, with no opinion about what you put in it. So the interesting comparison is not features, it is the durability trade each one made, because that is what you would otherwise be reimplementing.

Defaults as of July 2026, read from the source rather than the marketing:

	On-disk structure	Integrity check	fsync (default)	Loss window on power cut	Cap	When full
spool	segment files + cursor	xxhash64 over len‖payload, per frame	per Append, or per Sync in group mode	none (or the open group)	bytes, bounded by default	ErrFull to the caller
Prometheus WAL	128 MiB segments, 32 KiB pages	CRC32 per record	on segment rotation only	up to a segment	WAL truncation at head compaction	n/a — not a queue
Grafana Alloy loki.write WAL	Prometheus wlog + Snappy	CRC32 per record	on segment rotation only	up to a segment	max_segment_age (1 h)	old segments deleted, sent or not
Promtail WAL	same code lineage as Alloy	same	same	same	same	same
Fluentd buf_file	chunk files + .meta	none	never — file_chunk.rb contains no fsync	whatever the kernel has not written back	total_limit_size, 64 GiB	throw_exception (default)
Fluent Bit storage.type filesystem	mmap'd chunks (chunkio)	CRC32, off by default	msync(MS_ASYNC) unless storage.sync full	whatever the kernel has not written back	storage.total_limit_size	discards the oldest chunk
Filebeat queue.disk	segment files	CRC32 per frame + duplicated length	per write batch, before the producer ACK	none acked	max_size (required)	producer blocks
Vector buffer.type disk	mmap ledger + 128 MiB data files	CRC32C per record	every 500 ms	up to 500 ms	max_size, min ~256 MiB	block or drop_newest
Telegraf disk_write_through	tidwall/wal, one per output	CRC per entry	per Add batch, often one metric (buffer_disk_sync defaults true)	none	none — metric_buffer_limit is not applied to the disk buffer	grows until the disk fills
OTel Collector persistent queue	bbolt B+tree via file_storage	bbolt page checksums	off — NoSync: !cfg.FSync, fsync defaults false	whatever bbolt has not flushed	queue_size, 1000 requests by default	reject, or block

Three things fall out of that table.

"Durable" almost always means process-durable, not power-durable. A write() that reached the page cache survives kill -9, an OOM kill, and a container restart, because the kernel owns those pages. It does not survive a power cut or a kernel panic. Most of the buffers above are betting, reasonably, that the crash you actually get is a process crash. If a node losing power may not lose the records it already accepted, that bet is the wrong one, and Append here fsyncs for exactly that reason.

The cap is often not in bytes — sometimes there isn't one. The OTel Collector's persistent queue defaults to 1000 requests: batch count, not disk footprint, so the bytes on disk are whatever those batches happen to weigh (sizer: bytes exists, but you have to ask for it). Alloy's loki.write WAL caps in time — max_segment_age defaults to 1 h, and segments past it are deleted whether or not they were ever sent. Telegraf's disk buffer enforces no limit at all: metric_buffer_limit governs the memory buffer, and buffer_disk.go's Add writes every metric to the WAL unconditionally, so it grows until the volume does. If your operational question is "will this fill the disk," expect to answer it yourself in most of these. MaxBytes here is bytes, and Bytes()/Cap() tells you how close you are.

In fairness: this library shipped its first tag with MaxBytes: 0 meaning unbounded, and its own benchmark then wrote 12 GB of segments in 90 seconds onto a volume that also held an etcd. The zero value is DefaultMaxBytes now, and Unbounded is a thing you have to type. The criticism above is not a claim of superior taste — it is a fixed bug.

Full means different things. Fluent Bit discards the oldest chunk — the data that has waited longest, which under a sustained outage is precisely the data you were buffering for. Vector's drop_newest discards the other end. Filebeat's producer blocks, as does Vector under when_full: block, pushing the problem upstream. This package returns ErrFull and changes nothing, which is neither: it is the caller's decision, made where the caller has the context to make it.

Also worth knowing when picking among the agents: Promtail reached end of life on 2 March 2026 and its source has been removed from the Loki repository — Alloy is the successor, and the two share WAL code. Alloy's loki.write WAL is off unless you add a wal block. Telegraf's disk buffer shipped as experimental in 1.32. The OTel Collector's bbolt store needs compaction configured or the file keeps the high-water mark of your worst outage forever.

None of this makes those buffers wrong. Fluent Bit's mmap chunks are fast precisely because they skip the fsync; Prometheus batches into 32 KiB pages because it ingests millions of samples a second and a per-record fsync would be absurd there. They are tuned for volume in an agent that owns the whole pipeline. This is tuned for a producer that must be able to say "that record is safe, I can forget my copy of it" — and then hands you the group-commit knob for when per-record is too strict.

When not to use this

• You need multi-process or multi-consumer access. One process, one consumer. Use a real broker.

• You need random access, keys, or a secondary index. This is a FIFO byte queue. bbolt or SQLite fit that shape; this does not.

• You need exactly-once. You get at-least-once. Make the consumer idempotent.

• You are already running one of the agents above and its buffer's trade suits you. Don't add a second buffer to a pipeline that has one.

• Your records are tiny and you cannot batch. A per-record fsync will cap you near a thousand records a second. Group commit fixes this, but only if the producer can hold a group.

• Your consumer commits per tiny record and duplicates are unacceptable. CommitSync: true puts an fsync on every commit, which caps the consumer near a thousand records a second. If you need both throughput and a narrow duplicate window, commit per batch rather than per record.

Development

make check   # lint + race tests
make bench   # bounded; fsync-dominated, so it describes your disk
make fuzz    # corruption recovery, beyond the committed seed corpus

CI runs vet, -race tests on amd64 and arm64 plus the go.mod floor on amd64, golangci-lint, and a 90 s fuzz run against the corruption-recovery paths on every push. A fuzz failure uploads the crasher so it can be committed to testdata/ as a regression seed.

Every benchmark passes an explicit SegmentBytes and MaxBytes and drains inside b.StopTimer. This is not tidiness: an uncapped Pop benchmark once wrote 12 GB of segments in 90 seconds. Options{} is bounded now, but a benchmark that leans on the default measures the cap rather than the code.

License

MIT

Documentation

Overview

Package spool is a disk-backed FIFO byte queue. Records are appended durably to segment files and consumed in order, surviving process restarts and power loss.

It exists for store-and-forward: a producer that outruns a remote endpoint, or outlives its outage, spools records to disk (bounded by a byte cap) instead of holding them in memory or pinning its own input. Delivery is at-least-once, with an explicit commit, so a crash re-delivers rather than drops.

Durability contract

Append fsyncs the frame before returning. Once it returns nil the record is on the physical medium, so a producer that has observed a successful Append may safely advance its own checkpoint and forget its copy.

Producers with many SMALL records can group-commit instead: AppendNoSync enqueues without forcing durability, and Sync makes everything since the last durable point safe with ONE fsync. The checkpoint rule moves to Sync ("advance only after Sync returns"); it never silently weakens. On a Sync failure the whole unsynced group is rolled back and must be re-appended, which Sync reports as ErrGroupLost. Duplicates are possible if some frames were popped meanwhile — within at-least-once.

Pop does not remove. It returns the record with a commit function, and nothing is retired until commit runs, so a crash in between re-delivers.

On-disk format

A directory of segment files (<seq>.seg, zero-padded, ascending) plus a cursor file:

segment: "KSPOOL" | version u16be | frame*
frame:   len u32be | xxhash64(len‖payload) u64be | payload
cursor:  seq u64be | offset u64be | fnv64a u64be   (rewritten in place)

Checksumming the length alongside the payload means a flipped length byte is caught instead of mis-framing the rest of the segment, and a damaged record is dropped and reported rather than handed onward as plausible-looking data.

A segment whose header is absent or names a version this build does not know is discarded at open: the spool is a transient buffer, so carrying a reader for every past or future format would not pay for itself. The version is per segment rather than per spool, so when a new format is added (bump formatVersion, extend knownVersions and the frame read/write paths) segments of both versions coexist in one directory, each read with the framing it was written in. The magic reads KSPOOL for historical reasons — this package was extracted from an agent named kubescrape — and is kept so existing spool directories stay readable.

Recovery and what corruption costs

On restart the newest segment's torn tail — a frame a crash left incomplete — is truncated away and counted in Discarded. A frame that is whole but fails its checksum is left in place and dropped by Pop when it is reached, so one damaged byte inside a PAYLOAD costs one record.

That bound does not extend to the four bytes of the frame length. A length damaged upward overruns the segment and a length damaged downward desynchronizes the walk; either way the frame boundaries from that point on are unrecoverable, so Pop skips the whole segment. One bad byte there costs up to SegmentBytes. The checksum makes this detected rather than silent, which is the guarantee — not that the blast radius is always one record.

A torn cursor fails its checksum on the next load and redelivers from the oldest surviving segment (duplicates, within at-least-once).

Every loss path increments Stats().LostBytes or Discarded; none is silent.

Platform notes

Durability rests on os.File.Sync. On Linux that is fsync(2); on macOS the Go runtime issues fcntl(F_FULLFSYNC), which does flush the drive write cache. Neither is trustworthy on a network filesystem — do not put a spool on NFS, SMB or a similar remote mount, where an fsync may be acknowledged by a client cache.

A failed fsync must not be retried on the same descriptor: on Linux the kernel may have already marked the pages clean, so a second fsync can report success for data that was never written. This package therefore rolls the tail back to its last durable size rather than retrying.

One process per directory; see Open.

Example

package main

import (
	"fmt"
	"os"

	"github.com/JohanLindvall/spool"
)

func main() {
	dir, _ := os.MkdirTemp("", "spool")
	defer func() { _ = os.RemoveAll(dir) }()

	q, err := spool.Open(dir, spool.Options{MaxBytes: 64 << 20})
	if err != nil {
		panic(err)
	}
	defer func() { _ = q.Close() }()

	// Append is durable: once it returns, the record survives a crash.
	if err := q.Append([]byte("batch-1")); err != nil {
		panic(err)
	}

	// Pop hands back the record with a commit function; commit only after the
	// record is truly handled (e.g. the collector acked it) — a crash before
	// commit re-delivers it.
	data, commit, ok, err := q.Pop()
	fmt.Printf("%q ok=%v err=%v\n", data, ok, err)
	commit()
}

Output:
"batch-1" ok=true err=<nil>

Example (ConsumerLoop)

A consumer loop that blocks instead of polling, stays cancellable, and does not spin on the error returns. Pop's errors are advancing — the queue has already moved past the damage — so they must not be retried in a tight loop without also waiting.

package main

import (
	"errors"
	"fmt"
	"os"

	"github.com/JohanLindvall/spool"
)

func main() {
	dir, _ := os.MkdirTemp("", "spool")
	defer func() { _ = os.RemoveAll(dir) }()

	q, err := spool.Open(dir, spool.Options{MaxBytes: 64 << 20})
	if err != nil {
		panic(err)
	}
	defer func() { _ = q.Close() }()
	_ = q.Append([]byte("only"))

	done := make(chan struct{})
	close(done) // stand-in for ctx.Done(); this example stops after one record

	for {
		data, commit, ok, err := q.Pop()
		switch {
		case errors.Is(err, spool.ErrClosed):
			return
		case err != nil:
			// Corruption was dropped and the queue advanced. Count it and go
			// round again; Signal fires for these too, so the wait below is
			// still correct if this was the last thing in the queue.
			fmt.Println("read error:", err)
		case !ok:
			select {
			case <-q.Signal():
			case <-done:
				fmt.Println("stopped with", q.Bytes(), "bytes queued")
				return
			}
		default:
			fmt.Printf("send %q\n", data)
			commit() // only after the send succeeded
		}
	}
}

Output:
send "only"
stopped with 0 bytes queued

Example (ErrFull)

ErrFull is backpressure and clears as the consumer drains. ErrExceedsCap wraps it but never clears: the record cannot fit under this cap at all.

package main

import (
	"errors"
	"fmt"
	"os"

	"github.com/JohanLindvall/spool"
)

func main() {
	dir, _ := os.MkdirTemp("", "spool")
	defer func() { _ = os.RemoveAll(dir) }()

	q, err := spool.Open(dir, spool.Options{MaxBytes: 64})
	if err != nil {
		panic(err)
	}
	defer func() { _ = q.Close() }()

	oversized := q.Append(make([]byte, 64))
	fmt.Println("oversized permanent:", errors.Is(oversized, spool.ErrExceedsCap))

	// A record that fits: at most MaxBytes-FrameOverhead bytes.
	fmt.Println("largest that fits:", q.Cap()-spool.FrameOverhead)
}

Output:
oversized permanent: true
largest that fits: 52

Example (GroupCommit)

Group commit trades one fsync per record for one per group. The checkpoint rule moves with it: advance only after Sync returns, never after AppendNoSync. This is the API whose misuse loses data.

package main

import (
	"errors"
	"fmt"
	"os"

	"github.com/JohanLindvall/spool"
)

func main() {
	dir, _ := os.MkdirTemp("", "spool")
	defer func() { _ = os.RemoveAll(dir) }()

	q, err := spool.Open(dir, spool.Options{MaxBytes: 64 << 20})
	if err != nil {
		panic(err)
	}
	defer func() { _ = q.Close() }()

	batch := [][]byte{[]byte("a"), []byte("b"), []byte("c")}
	for _, rec := range batch {
		// NOT durable yet. A crash here loses the whole group.
		if err := q.AppendNoSync(rec); err != nil {
			panic(err)
		}
	}
	if err := q.Sync(); err != nil {
		// The group was rolled back and must be re-appended. Do NOT advance
		// any upstream checkpoint.
		if errors.Is(err, spool.ErrGroupLost) {
			fmt.Println("group lost; re-append")
		}
		return
	}
	// Only now may the producer forget its copy of a, b and c.
	fmt.Println("durable:", q.Stats().Appended, "records")
}

Output:
durable: 3 records

Index

• Constants
• Variables
• type Options
• type Spool
  • func Open(dir string, opts Options) (*Spool, error)
  • func (s *Spool) Append(data []byte) error
  • func (s *Spool) AppendNoSync(data []byte) error
  • func (s *Spool) Bytes() int64
  • func (s *Spool) Cap() int64
  • func (s *Spool) Close() error
  • func (s *Spool) CursorErr() error
  • func (s *Spool) Discarded() int64
  • func (s *Spool) Pop() (data []byte, commit func(), ok bool, err error)
  • func (s *Spool) Requeue(data []byte, commit func()) (rotated bool, err error)
  • func (s *Spool) Segments() int
  • func (s *Spool) Signal() <-chan struct{}
  • func (s *Spool) Stats() Stats
  • func (s *Spool) Sync() error
• type Stats

Examples

• Package
• Package (ConsumerLoop)
• Package (ErrFull)
• Package (GroupCommit)

Constants

const (
	// DefaultSegmentBytes is the segment size used when Options.SegmentBytes is
	// zero. It also bounds the work repairTail does at Open, and the redelivery
	// window when the cursor cannot be persisted.
	DefaultSegmentBytes = 8 << 20

	// DefaultMaxBytes is the backlog cap used when Options.MaxBytes is zero.
	// Deliberately a modest placeholder and not a sizing recommendation: a disk
	// buffer whose unconfigured state is "grow until the volume dies" fills the
	// disk its own process runs on, and the failure lands on whatever else
	// shares the mount. Size it for your outage budget.
	DefaultMaxBytes = 64 << 20

	// Unbounded disables the backlog cap. It must be requested explicitly.
	Unbounded = -1
)

const FrameOverhead = 12

FrameOverhead is the per-record framing cost in bytes for the current format, so callers can size backlog comparisons against record lengths. A record fits under MaxBytes iff len(record) <= MaxBytes-FrameOverhead.

const MaxRecordBytes = frameLenMax

MaxRecordBytes is the largest record this frame format can represent: the length field is a big-endian uint32. Append rejects anything larger with ErrRecordTooLarge rather than truncating the length silently.

Variables

var ErrClosed = errors.New("spool: closed")

ErrClosed is returned by Append, AppendNoSync, Requeue, Sync and Pop after Close. Commit functions from earlier Pops become no-ops: call the last commit before Close.

var ErrCorrupt = errors.New("spool: corrupt")

ErrCorrupt is returned by Pop when the head frame fails its integrity check: a checksum mismatch, a length overrunning the segment, or a segment tail too short to hold a frame header. The damaged data is dropped — the frame alone when the framing is still trustworthy, otherwise the whole segment — and the queue advances, so corruption degrades to reported loss rather than to corrupt telemetry or a wedged queue. Stats().LostBytes carries the magnitude the error itself cannot.

var ErrExceedsCap = fmt.Errorf("%w: record exceeds MaxBytes", ErrFull)

ErrExceedsCap is returned when a single record can never fit under MaxBytes: the check fails on an empty queue and will fail identically forever. It wraps ErrFull so existing errors.Is(err, ErrFull) callers keep working, but unlike plain ErrFull it will not clear with time — drop the record or split it.

var ErrFull = errors.New("spool: full")

ErrFull is returned by Append when the queue is at its size cap. It is transient: it clears as the consumer drains.

var ErrGroupLost = errors.New("spool: unsynced records lost; re-append the group")

ErrGroupLost reports that records accepted by AppendNoSync were invalidated before any Sync could cover them. The records were never persisted: re-append the group. Returned by Sync and joined into Close's error.

var ErrRecordTooLarge = errors.New("spool: record exceeds MaxRecordBytes")

ErrRecordTooLarge is returned for a record the frame format cannot represent (see MaxRecordBytes). Unlike ErrExceedsCap it is not a capacity condition and does not wrap ErrFull: no cap setting makes it succeed.

Types

type Options

type Options struct {
	// SegmentBytes caps one segment file (0 = DefaultSegmentBytes). A single
	// record may exceed it; segments rotate lazily once non-empty.
	SegmentBytes int64

	// MaxBytes caps the undelivered BACKLOG — the sum of every segment's bytes
	// minus the consumed prefix of the read head, which is what Bytes reports.
	// It is not the on-disk footprint: a delivered but unretired prefix keeps
	// its segment alive, so physical usage can exceed this by up to one
	// segment, and every segment's 8-byte header is charged except the
	// consumed part of the head's. Stats().DiskBytes reports the real
	// footprint.
	//
	// Append returns ErrFull past the cap, leaving the queue unchanged.
	// 0 = DefaultMaxBytes; Unbounded (-1) disables the cap entirely.
	MaxBytes int64

	// CommitSync makes every commit fsync the read cursor.
	//
	// The cursor is a progress hint, not a durability record: a stale, torn or
	// missing cursor causes only REDELIVERY, which the at-least-once contract
	// already permits, bounded by one segment. Syncing it narrows the duplicate
	// window after a power cut, and costs one fsync per commit — on this
	// package's own benchmarks, the difference between roughly 900 and 300,000
	// commits per second.
	//
	// The default (false) does not sync. Set it when duplicates cost more than
	// consumer throughput.
	CommitSync bool
}

Options configure a Spool.

type Spool

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

Spool is a durable FIFO byte queue.

Any number of goroutines may call the Append family, Sync, Pop, Close and the accessors concurrently; all state is mutex-guarded. The real constraint is that Pop carries no per-consumer position: two Pops without an intervening commit return the SAME record, so the consumer must be serialized. One process per directory — see Open.

func Open

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

Open opens or creates the spool rooted at dir.

One process per directory. Nothing enforces this: two processes on one spool each repair the tail, retire the other's segments and overwrite the other's cursor, which destroys records with no error surfacing on either side.

func (*Spool) Append

func (s *Spool) Append(data []byte) error

Append durably enqueues one record. It returns ErrFull when the size cap would be exceeded, leaving the queue unchanged so the caller can apply backpressure, and ErrExceedsCap when the record can never fit.

Once it returns nil the record is on the physical medium, so a producer may advance its own checkpoint on the strength of it.

func (*Spool) AppendNoSync

func (s *Spool) AppendNoSync(data []byte) error

AppendNoSync enqueues one record WITHOUT forcing it to disk. The record is readable immediately (same-process visibility) but not durable — a crash may lose it — until a Sync (or a later synced Append on the same tail) returns. The group-commit half of the durability contract: producers with many small records call AppendNoSync per record and Sync once per group, paying one fsync per group instead of per record, and advance their checkpoint only after Sync.

func (*Spool) Bytes

func (s *Spool) Bytes() int64

Bytes is the current undelivered backlog in bytes (what MaxBytes caps).

func (*Spool) Cap

func (s *Spool) Cap() int64

Cap is the configured MaxBytes (0 = uncapped). Exposed so a monitor can report utilisation rather than a bare byte count: "the buffer is at 70% and climbing" is the signal that a collector is degrading, and it is only expressible against the cap.

func (*Spool) Close

func (s *Spool) Close() error

Close flushes any unsynced group (a clean shutdown must not lose records AppendNoSync accepted) and releases the file handles. It does not delete queued data; a flush failure is returned so the producer knows the group never became durable. Close is idempotent.

func (*Spool) CursorErr

func (s *Spool) CursorErr() error

CursorErr reports the last failure to persist the read cursor, sticky until a later persist succeeds. Non-nil means commits are advancing in memory only: a restart redelivers every record back to the last cursor that reached disk, bounded by one segment. It is a disk fault rather than a queue state, so it is reported here instead of being returned from commit.

func (*Spool) Discarded

func (s *Spool) Discarded() int64

Discarded is the number of bytes a damaged or torn tail cost, so the caller can log and count it. It is almost always fixed at Open, but can also grow at runtime when a failed append reopens and re-repairs the tail. A crash mid-append leaves one incomplete frame (small and expected); anything larger means fsynced records were destroyed by corruption, which must never be invisible.

func (*Spool) Pop

func (s *Spool) Pop() (data []byte, commit func(), ok bool, err error)

Pop returns the next record and a commit function that removes it. ok false with a nil error means the queue is empty; a non-nil error means the head frame could not be read — the caller should surface it and retry. ErrCorrupt means damaged data was dropped, fs.ErrNotExist that the head segment's file was gone, and ErrClosed that the spool is closed; in the first two cases the queue has advanced past the problem, so the next Pop makes progress.

Signal also fires whenever Pop advances past unreadable data, so a consumer may use one wait path for both ok == false cases.

The record is not removed until commit is called, so a crash before commit re-delivers it (at-least-once). commit is idempotent, and becomes a no-op after Close — call the last commit before closing.

func (*Spool) Requeue

func (s *Spool) Requeue(data []byte, commit func()) (rotated bool, err error)

Requeue moves the record just popped to the back of the queue and commits it at the head, so one undeliverable record cannot block everything behind it. Pass the data and the commit function that Pop returned.

It ignores MaxBytes because the rotation is size-neutral — an equal-or-larger head frame is committed immediately after — and the cap must not be able to wedge a full spool whose head is undeliverable. It returns rotated=false without writing when the record is the only one queued, since rewriting it would be pointless churn.

It is not atomic: the append is fsynced before the commit runs, so a crash between them re-delivers the record — within at-least-once.

func (*Spool) Segments

func (s *Spool) Segments() int

Segments is the number of segment files currently on disk. Physical footprint can exceed the backlog by up to one segment (a delivered but unreclaimed prefix), so this is the number to watch for disk pressure.

func (*Spool) Signal

func (s *Spool) Signal() <-chan struct{}

Signal fires (non-blocking) after each Append, and whenever Pop advances past unreadable data, so a consumer can wait for work without polling.

func (*Spool) Stats

func (s *Spool) Stats() Stats

Stats returns a snapshot of the queue's gauges and this process's counters.

func (*Spool) Sync

func (s *Spool) Sync() error

Sync makes every record accepted by AppendNoSync durable — the group-commit barrier. A producer using AppendNoSync may advance its checkpoint only after Sync returns nil; on error the whole unsynced group was rolled back (or was already lost with a damaged tail) and must be re-appended. Duplicates are possible when frames of the group were popped before the failure — within at-least-once, exactly like the synced-append failure paths.

type Stats

type Stats struct {
	// Gauges.
	BacklogBytes  int64 // undelivered bytes — what MaxBytes caps (same as Bytes)
	MaxBytes      int64 // configured cap; 0 when Unbounded was requested
	DiskBytes     int64 // sum of segment file sizes: the number that fills the volume
	Segments      int
	UnsyncedBytes int64 // accepted by AppendNoSync, not yet fsynced: what a crash loses now

	// Counters since Open.
	Appended        uint64 // records accepted by any Append form
	Popped          uint64 // records handed out, including redeliveries
	Committed       uint64 // records retired by commit
	Full            uint64 // appends refused with ErrFull or ErrExceedsCap
	LostBytes       uint64 // destroyed by corruption; a LOWER BOUND
	LostSegments    uint64
	ForeignSegments uint64 // dropped for a format version this build cannot read
	ForeignBytes    uint64
	CursorErrors    uint64 // failed cursor persists (see CursorErr)
	DiscardedBytes  int64  // same value as Discarded
}

Stats is a point-in-time snapshot. Gauges describe the queue now; counters are monotonic for the life of this process and are not persisted. Fields may be added — construct it only with keyed literals.