Share the fsyncs: group commit, AddBatch, AddWait, TryPeek

Per-op durability used to serialize every producer on the disk: two
fsyncs per record, under the queue mutex, with readers stalled behind
them. The whole change hangs on one new invariant — a STAGED record's
bytes are written past the published extent while the header, the
cursors and the counts do not move — so a flush leader can release the
mutex during its fsyncs and nothing (readers, commits, Stats, a crash)
can see a record whose bytes are not yet durable. Publication is a
span: data fsync for everyone staged, one header write, one header
fsync. Every staged record lives in the active file (cycling, batch,
Requeue, Sync and Close quiesce first), so a failed publication
discards exactly one contiguous tail.

Group commit: an Add that arrives while a flush is in flight stages
its record and rides the next span; a solo Add leads a span of one and
allocates nothing, so the single-producer path stays 0 allocs/op and
within noise of its old cost. Concurrent durable Adds go ~7x faster on
this machine, each still individually durable when it returns. The
span's failure arms mirror the solo append's exactly — data-fsync
latches with nothing published, header-write discards with nothing
latched, header-fsync latches with the span real in the page cache —
and faults_test.go pins ordering and arms for all three shapes (solo,
group, batch) under the same injection points.

AddBatch(items) amortizes the same span across a batch from one
goroutine (~50x per record here) and reports how many leading items
were placed and durable when it stops early. AddWait(ctx, v) is the
blocking half of MaxBytes backpressure: it parks on a space signal
that every commit-capable consume op raises, and still fails fast on
ErrRecordTooLarge. TryPeek inspects the head without consuming — no
cursor moves, and a damaged head previews as an UNCOUNTED ErrCorrupt
so the consume that follows books the event exactly once.

Add and AddWait now marshal BEFORE taking the lock, into pooled
buffers, so codecs run concurrently across producers and a slow codec
no longer stalls consumers; AddBatch marshals under the lock. The
failure arms unreachable without a seam are covered in the faults
build (unpublishSpan, discardStaged, rollbackAppend at 100% there).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: JohanLindvall

CLAUDE.md

@@ -24,9 +24,11 @@ store using plain `pread`/`pwrite`/`fsync` (no mmap).
   deliberately *not* split further: pulling `writeHeader`/`flushFile` away from `append` and
   `commitTo` would separate the calls whose order is the whole point (see the
   data-before-header invariant below).
-- [diskqueue.go](diskqueue.go) — the generic `Queue[T]` writer/owner (Add, Empty/Count/Size,
-  Stats/Rewind/Sync/Err/Close, NewReader) on top of `store`.
-- [reader.go](reader.go) — `Reader[T]`: all consume ops (Reserve/Take/Commit/
+- [diskqueue.go](diskqueue.go) — the generic `Queue[T]` writer/owner (Add/AddWait/AddBatch,
+  Empty/Count/Size, Stats/Rewind/Sync/Err/Close, NewReader) on top of `store`,
+  including the group-commit leader/follower machinery (`addDurableLocked`,
+  `leadFlushLocked`, the quiesce/space signals).
+- [reader.go](reader.go) — `Reader[T]`: all consume ops (TryPeek/Reserve/Take/Commit/
   Drain/Follow/Err); copies each record into its own scratch buffer.
 - [prealloc_linux.go](prealloc_linux.go) / [prealloc_other.go](prealloc_other.go) — `preallocate`:
   `syscall.Fallocate` (stdlib, Linux) with an `ftruncate` fallback everywhere else and on
@@ -350,8 +352,57 @@ mirrors its own `written`/`committed` counts into its header.
   reclaims. The arm is unreachable through the public API (both callers pass a
   cursor `commitTo` keeps inside a live segment) and cannot be deleted either —
   dropping the guard nil-derefs on the next line.
+- **Staged records are invisible until their span publishes.** The per-op
+  append path (Queue-level: `addDurableLocked`/`leadFlushLocked`/
+  `publishBatchLocked` over the store's `stagePending`/`takeSpan`/
+  `publishSpan`) writes record bytes past the published extent and tracks them
+  only in `pendingBytes`/`inFlight`: `df.hdr`, `af.size`, `writeOff` and
+  `nWritten` move at publication, never at staging. That single rule is what
+  lets the flush leader release the queue mutex during its fsyncs — readers,
+  `commitTo`, `Stats` and a crash all see the published extent only, so
+  nothing can deliver, commit or recover a record whose bytes are not yet
+  durable. A header write racing the data fsync is the torn state the solo
+  append's ordering exists to prevent; staging keeps the header out of reach
+  by construction.
+- **Every staged record lives in the active file** — cycling, `AddBatch`,
+  `Requeue`, `Sync` and `Close` quiesce first (`waitFlushQuiescedLocked`) — so
+  a failed publication discards exactly one contiguous tail
+  (`discardStaged`/`unpublishSpan`) and the in-memory view snaps back to the
+  header that never changed. The span failure arms mirror the solo append's:
+  data-fsync failure latches with nothing published, header-write failure
+  discards with nothing latched, header-fsync failure latches with the span
+  real in the page cache. `faults_test.go` pins the ordering for all three
+  shapes (solo, group, batch) under the same injection-point names.
+- **Group commit shares the fsyncs, not the contract.** A solo Add leads a span
+  of one and allocates nothing (the `flushGroup` is created only by a follower
+  that arrives mid-flush), which is what keeps `BenchmarkAddTake` at 0
+  allocs/op. Followers wait on their group's `done` channel and take the
+  span's verdict; the leader drains spans until nothing is pending, then
+  clears `flushing` and wakes the quiesce waiters. After ANY wait that
+  released the lock (quiesce, space, follower), re-check `closed` — staging
+  onto a closed store would reopen handles Close just released. `Requeue`
+  quiesces BEFORE its `takeHead`, because its rotation must be atomic under
+  one continuous lock hold (its `rewindHead` on failure may not undo another
+  reader's progress). The leader deliberately leaves `af.dirty` set after its
+  header fsync — a commit may have rewritten the header while the lock was
+  away, and clearing it would skip that write's flush.
+- **Marshal runs outside the lock for Add/AddWait** (pooled `marshalBuf`, so
+  codecs run concurrently across producers and a slow codec cannot stall
+  consumers); `AddBatch` marshals under the lock through `w.scratch`. Both
+  keep the no-callback rule: `MarshalFunc` must never call back into the
+  queue.
+- **`peekHead` previews, it never books.** `Reader.TryPeek` moves no cursor,
+  pays down no `pendingCorrupt`, and its `ErrCorrupt` is uncounted — the
+  consume op that steps past the damage books the event exactly once. Don't
+  add accounting to the peek path; double-counting a loss is how operators
+  stop trusting `Corruptions`.
 - **Blocking waiters.** `waitLocked`/`signal` use a lazily-created `notify`
-  channel, nil when nobody waits, so `Add` stays allocation-free.
+  channel, nil when nobody waits, so `Add` stays allocation-free. The same
+  pattern runs the other two waits: `spaceNotify` (consume ops wake `AddWait`
+  producers; every commit-capable Reader op signals it, spurious wakes are
+  re-checked) and `flushDrained` (the flush leader wakes quiesce waiters).
+  `Close` closes `notify` AND `spaceNotify`, then quiesces the flush before
+  `st.close()` so a leader mid-fsync finishes on live handles.
 - **`Reader.Drain`/`Follow` consume** via the shared `headOff` (like iterator-
   shaped `Take`): read **and `commitTo` under the lock**, then release and yield —
   so they commit-on-read (at-most-once) and are safe for concurrent cooperating

README.md

@@ -179,7 +179,9 @@ processing (at-least-once), use `Reserve`/`Commit`.
 | Method | Description |
 | --- | --- |
 | `New[T](path, marshal, unmarshal, ...Options)` | Open/create a Queue at `path` (segment cap via `Options.MaxSegments`, default 32). |
-| `Add(v T) error` | Append an item. Returns `ErrFull` at `MaxSegments` or `MaxBytes` (transient), `ErrRecordTooLarge` if it can never fit under `MaxBytes` (permanent). A record too big for one segment gets a segment of its own. |
+| `Add(v T) error` | Append an item. Returns `ErrFull` at `MaxSegments` or `MaxBytes` (transient), `ErrRecordTooLarge` if it can never fit under `MaxBytes` (permanent). A record too big for one segment gets a segment of its own. Serialization runs before the lock, and under the default policy concurrent `Add`s share their fsyncs (group commit) while each record stays individually durable. |
+| `AddWait(ctx, v T) error` | `Add` with backpressure: blocks while the queue is full, until a commit frees capacity or `ctx` is done. `ErrRecordTooLarge` still returns immediately. |
+| `AddBatch(items []T) (int, error)` | Append items in order with the fsyncs amortized over the whole batch (per segment: one data fsync, one header write, one header fsync). Returns how many leading items were placed — on error the first `n` are in and durable, the rest are not. |
 | `NewReader() *Reader[T]` | Create a Reader to consume items (one per consuming goroutine). |
 | `Empty() bool` | Whether anything is available to read. |
 | `Count() int` | Number of items added but not yet committed. |
@@ -194,6 +196,7 @@ processing (at-least-once), use `Reserve`/`Commit`.
 
 | Method | Description |
 | --- | --- |
+| `TryPeek() (T, bool, error)` | Look at the front item without consuming it: no cursor moves, and the next read (by any Reader) returns the same item. A damaged head previews as `ErrCorrupt` with nothing dropped or counted. |
 | `TryReserve() (T, bool, int64, error)` | Non-blocking read; returns the item and its offset. |
 | `TryTake() (T, bool, error)` | Non-blocking read + commit. |
 | `Reserve(ctx) (T, bool, int64, error)` | Block until an item is available, then read it. |

bench_test.go

@@ -164,3 +164,58 @@ func BenchmarkBatchCommit(b *testing.B) {
 		b.StartTimer()
 	}
 }
+
+// BenchmarkAddBatch measures the amortized durable append against its per-Add
+// equivalent: one lock round and (per segment) one data fsync + one header
+// write + one header fsync for the whole batch. Must stay 0 allocs/op like
+// every other hot path.
+func BenchmarkAddBatch(b *testing.B) {
+	const batch = 100
+	m, u := bytesCodec()
+	w, err := New[[]byte](b.TempDir(), m, u, Options{MaxSegments: -1}) // per-op durability
+	if err != nil {
+		b.Fatal(err)
+	}
+	defer func() { _ = w.Close() }()
+	r := w.NewReader()
+	items := make([][]byte, batch)
+	for i := range items {
+		items[i] = benchPayload
+	}
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		if n, err := w.AddBatch(items); n != batch || err != nil {
+			b.Fatalf("AddBatch: n=%d err=%v", n, err)
+		}
+		b.StopTimer()
+		for j := 0; j < batch; j++ { // drain so the backlog stays flat
+			if _, ok, err := r.TryTake(); !ok || err != nil {
+				b.Fatalf("drain: ok=%v err=%v", ok, err)
+			}
+		}
+		b.StartTimer()
+	}
+}
+
+// BenchmarkAddParallel is the group-commit case: many producers on the per-op
+// policy, each Add individually durable, the fsyncs shared across whoever
+// arrived while the previous span was in flight. Compare against
+// BenchmarkAddDurable/SyncEvery1 to see what the sharing buys.
+func BenchmarkAddParallel(b *testing.B) {
+	m, u := bytesCodec()
+	w, err := New[[]byte](b.TempDir(), m, u, Options{MaxSegments: -1}) // per-op durability
+	if err != nil {
+		b.Fatal(err)
+	}
+	defer func() { _ = w.Close() }()
+	b.ReportAllocs()
+	b.ResetTimer()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			if err := w.Add(benchPayload); err != nil {
+				b.Fatal(err)
+			}
+		}
+	})
+}