diff --git a/.add/state.json b/.add/state.json
index 99915c33a..32f4b4943 100644
--- a/.add/state.json
+++ b/.add/state.json
@@ -1,7 +1,7 @@
{
"project": "moon",
"stage": "production",
- "active_task": "xshard-read-fastpath",
+ "active_task": "wal-group-commit",
"active_milestone": "v2-performance",
"tasks": {
"hotpath-lock-quickwins": {
@@ -58,6 +58,16 @@
"created": "2026-06-13T08:25:14+00:00",
"updated": "2026-06-14T06:06:01+00:00",
"flag_verified": true
+ },
+ "wal-group-commit": {
+ "title": "WAL group commit: batch concurrent pending writes into one fsync under appendfsync=always",
+ "phase": "done",
+ "gate": "PASS",
+ "milestone": "v2-performance",
+ "depends_on": [],
+ "created": "2026-06-14T06:54:39+00:00",
+ "updated": "2026-06-14T16:20:23+00:00",
+ "flag_verified": true
}
},
"milestones": {
@@ -79,7 +89,7 @@
}
},
"created": "2026-06-11T03:18:21+00:00",
- "updated": "2026-06-14T06:06:01+00:00",
+ "updated": "2026-06-14T16:20:23+00:00",
"setup": {
"locked": true,
"locked_at": "2026-06-11T03:28:00+00:00",
diff --git a/.add/tasks/wal-group-commit/TASK.md b/.add/tasks/wal-group-commit/TASK.md
new file mode 100644
index 000000000..9ae702615
--- /dev/null
+++ b/.add/tasks/wal-group-commit/TASK.md
@@ -0,0 +1,452 @@
+# TASK: WAL group commit: batch concurrent pending writes into one fsync under appendfsync=always
+
+slug: wal-group-commit · created: 2026-06-14 · stage: production · risk: high · autonomy: conservative
+phase: done
+
+
+> One file = one task. Fill sections top-to-bottom; the `add` skill drives each phase.
+> When a phase is unclear, read its book chapter in `.add/docs/` (linked per section).
+> The phase marker above is the single source of truth — keep it in sync via `add.py phase`.
+
+---
+
+## 1 · SPECIFY — the rules ▸ docs/03-step-1-specify.md
+
+Feature: WAL **group commit** — coalesce concurrent pending AOF writes into a SINGLE fsync
+under `appendfsync=always`, on BOTH writer tasks (`aof_writer_task` TopLevel + `per_shard_aof_writer_task`),
+via **opportunistic drain** (after one message, drain whatever else is queued up to a bounded cap,
+write all, ONE fsync, ack all). Today the writer fsyncs once per `AppendSync` (writer_task.rs:198–232,
+694, 1042) — the ~11× write penalty (135K→12K ops/s). Group commit makes the per-write fsync cost
+amortize across write concurrency, WITHOUT weakening the fsync-before-ack durability contract, the
+C4 cross-shard fold, the H1-BARRIER, or the everysec/no paths.
+
+Ground truth (investigation 2026-06-14, file:line): each connection is its OWN async task; the write
+path `pool.try_send_append_durable(shard, lsn, bytes).await` (handler_sharded/mod.rs:1123/1173/1419,
+handler_monoio/mod.rs:1137/1198/1400, handler_single.rs:74) yields that task on the await, so N
+concurrent durable writers DO enqueue N `AppendSync` messages into one writer channel while the writer
+is busy — there IS a batch to coalesce. The channel is a MIXED stream: fire-and-forget `Append`
+(everysec/no + the C4-FOLD-FIX cross-shard arm, pool.rs:281–298), durable `AppendSync{ack}` (Always +
+the zero-length H1-BARRIER `fsync_barrier`, pool.rs:315–323 — "writer fsyncs all preceding Append then
+acks Synced"), and control (`Rewrite`/`RewriteSharded`/`RewritePerShard`/`Shutdown`). A single batch
+fsync already covers all preceding `Append` by the ordered-channel property the H1-BARRIER relies on —
+group commit GENERALIZES that barrier to "one fsync per drained batch, ack every AppendSync in it".
+
+Framings weighed: opportunistic drain on the writer-consumer (CHOSEN — user 2026-06-14: zero added
+latency, batches exactly when concurrency exists) · time-window coalesce (REJECTED — adds latency to
+every durable write even at C=1) · count/byte threshold w/ fallback timer (REJECTED — same latency
+risk + more moving parts). Scope: BOTH writers (CHOSEN — TopLevel covers shards=1, per-shard covers
+the multi-shard story) · per-shard only (declined — leaves shards=1 at 11×).
+
+Must:
+
+ - M0 (baseline, per-runtime relative anchor): re-establish the `appendfsync=always` "before" on
+ moon-dev at pipeline depth >1 with C>1 concurrent writers (the cell the win lives in),
+ fresh-server-per-rep, best-of-N, both runtimes; record `everysec` + `no` as unchanged controls.
+ Throughput (fsync-bound) is less VM-vCPU-sensitive than the xshard ABSOLUTE latency was, but state
+ the win as a RELATIVE before/after ratio on the same instrument (milestone bench rule).
+ - M1 (the win): under `appendfsync=always` with C>1 concurrent durable writers, when K `AppendSync`
+ messages are queued the writer writes all K then makes them durable with ONE `flush()+sync_data()`,
+ and acks all K waiters `Synced` only AFTER that fsync returns — ⌈writes/batch⌉ fsyncs, not one per
+ write. Measurable throughput gain at pipeline>1 / C>1 vs M0 on BOTH writers, BOTH runtimes.
+ - M2 (durability invariant — the freeze-first contract): a waiter is acked `Synced` ONLY after the
+ fsync covering its bytes returns; a crash before that fsync loses only UNACKED writes (client never
+ saw +OK ⇒ exactly-once preserved on retry). The one batch fsync also covers every preceding
+ fire-and-forget `Append` in the drained batch (H1-BARRIER ordered-channel property preserved). The
+ existing crash-matrix suites stay green AND a new concurrent-writers crash test asserts: every
+ ACKED write survives SIGKILL, no UNACKED write is double-applied on replay.
+ - M3 (ordering / control-message safety): `Rewrite`/`RewriteSharded`/`RewritePerShard`/`Shutdown`
+ and the C4 fold protocol are NEVER absorbed into a batch — when one is encountered during a drain,
+ the in-progress batch is flushed (write + single fsync + ack all) BEFORE that message is handled.
+ Channel message order is preserved exactly; the C4-FOLD `pending_aof_count` accounting is unchanged.
+ - M4 (no-regression guardrail): `everysec`/`no` are byte-for-byte unchanged (group commit engages
+ ONLY under `Always`); the everysec 1s deadline flush + bounded-recv (shardslice P0 8KB-tail fix)
+ still hold; a lone C=1 `Always` writer fsyncs immediately with NO added latency vs M0; batch size
+ is BOUNDED (a cap on count and/or bytes) so a write flood cannot delay an early waiter's fsync
+ unboundedly nor grow the buffer without limit; `scripts/test-consistency.sh` 197/197 @1/4/12;
+ dual-runtime green; clippy ×2 + `fmt` clean; zero new `unsafe`; zero new cross-thread lock; RSS
+ not regressed.
+
+Reject:
+
+ - a `write_all` failure for any message in a batch -> that waiter (and the batch) is acked
+ `WriteFailed`, never `Synced`; appends after a persistent I/O error stay dropped (existing
+ `write_error` latch) -> "batch_write_failed"
+ - the batch `flush()+sync_data()` fails -> ALL `AppendSync` waiters in the batch are acked
+ `FsyncFailed`, never `Synced` -> "batch_fsync_failed"
+ - acking ANY waiter `Synced` before the covering fsync returns -> unreachable by construction (the
+ batch fsync precedes every ack) -> "ack_before_fsync" (invariant — must be impossible)
+ - a control/rewrite message handled before the in-progress batch is flushed -> forbidden -> "batch_straddles_control"
+ - a drain that grows past the bounded cap -> the batch is flushed at the cap and a fresh batch
+ begins (never unbounded) -> "batch_cap_exceeded" (a bound, not an error)
+
+After:
+
+ - Under `Always` with C>1 concurrent durable writers, the AOF writer performs ⌈writes/batch⌉ fsyncs
+ instead of one-per-write; every ACKED write is durable on disk; `everysec`/`no` and the C=1 case
+ are unchanged; durable-write throughput at pipeline>1 measurably exceeds the M0 baseline; the
+ crash-matrix (incl. the new concurrent-writers case) is green on both runtimes.
+
+Assumptions — lowest-confidence first:
+
+ ⚠ The mixed-stream batch drain (interleaved `Append` + `AppendSync`, control messages as
+ batch-breakers) preserves the C4-fold + H1-BARRIER + everysec invariants EXACTLY — lowest
+ confidence because the writer loop is shared across all policies, the C4 fold, and the rewrite
+ control flow, and a mis-ordered flush-vs-control or a missed `pending_aof_count` update is a SILENT
+ durability / AOF-corruption bug; if wrong: data loss or corrupt AOF under crash (the worst outcome
+ — the milestone's explicit freeze-first risk).
+ ⚠ Concurrent per-connection durable writes actually CONVERGE as multiple queued `AppendSync`
+ messages at one writer (so a batch exists to coalesce) — VERIFIED in code (per-connection async
+ task; `try_send_append_durable(...).await` yields, peers enqueue before the writer drains), so
+ confidence is now HIGH; if wrong: the mechanism is correct but yields no throughput gain (an
+ effectiveness miss, not a correctness bug). Confirm empirically in M1.
+ - [ ] the bounded batch cap (count and/or bytes) can be a fixed tuned constant, not a config flag —
+ confirm; if wrong, add an additive flag (cheap).
+ - [ ] the OrbStack VM with fresh-server-per-rep gives a trustworthy RELATIVE `Always` throughput
+ delta (fsync-bound ⇒ less vCPU-starvation-sensitive than the xshard absolute latency) — confirm
+ the instrument is valid for this metric before anchoring M1.
+
+
+
+
+---
+
+## 2 · SCENARIOS — pass/fail cases ▸ docs/04-step-2-scenarios.md
+
+
+
+```gherkin
+# ---- M0: baseline anchor ----
+Scenario: always_throughput_baseline_recorded # M0
+ Given a fresh moon at appendfsync=always on moon-dev, fresh-server-per-rep
+ When durable-write throughput is measured at pipeline depth >1 with C>1 concurrent writers, best-of-N, per-runtime
+ Then a before-baseline RPS is recorded for monoio and tokio
+ And everysec + no are recorded alongside as unchanged controls
+
+# ---- M1: the coalescing win ----
+Scenario: batch_drain_one_fsync_many_acks # M1
+ Given appendfsync=always and K durable AppendSync writes already queued at one AOF writer
+ When the writer drains the ready batch
+ Then it performs exactly ONE flush()+sync_data() covering all K writes
+ And all K waiters receive AofAck::Synced
+ And every Synced ack is sent only after that single fsync returned
+
+Scenario: throughput_scales_with_write_concurrency # M1
+ Given the M0 baseline at appendfsync=always
+ When C>1 concurrent durable writers drive pipeline depth >1 on both writers (TopLevel + per-shard), both runtimes
+ Then measured durable-write throughput exceeds the M0 baseline (RELATIVE before/after ratio on the same instrument)
+
+# ---- M2: durability invariant (freeze-first) ----
+Scenario: every_acked_write_survives_crash # M2
+ Given appendfsync=always and N concurrent writers issuing distinct INCR/SET commands
+ When each client receives +OK (its Synced ack) and the server is then SIGKILLed
+ Then on AOF replay every acked write is present (each counter == the number of its acked increments)
+ And no earlier acked write is lost or corrupted by the batch that was in flight at kill
+
+Scenario: interrupted_batch_replays_to_last_valid_record # M2
+ Given a batch written to the file buffer but the process is SIGKILLed before its fsync returns
+ When the server replays the AOF
+ Then replay truncates to the last valid record and loads cleanly (no torn/partial record applied)
+ And all writes acked Synced by prior completed batches remain present
+
+Scenario: barrier_property_preserved # M2
+ Given a drained batch containing fire-and-forget Append messages followed by an AppendSync
+ When the single batch fsync returns and the AppendSync waiter is acked Synced
+ Then all preceding Append bytes in that batch are durable on disk (ordered-channel H1-BARRIER property)
+ And the C4-FOLD pending_aof_count accounting is unchanged
+
+# ---- M3: control-message ordering safety ----
+Scenario: control_message_breaks_the_batch # M3
+ Given appendfsync=always with a partially-drained batch of AppendSync writes pending
+ When a Rewrite / RewriteSharded / RewritePerShard / Shutdown message is encountered in the channel
+ Then the in-progress batch is flushed (write + single fsync + ack all) BEFORE that message is handled
+ And channel message order is preserved exactly and pending_aof_count stays accurate
+
+# ---- M4: no-regression guardrails ----
+Scenario: everysec_path_unchanged # M4
+ Given appendfsync=everysec under the same concurrent-writer load
+ When durable writes flow through the writer
+ Then no per-batch group-commit fsync engages — the 1s deadline flush + bounded-recv still govern durability
+ And everysec throughput and the 8KB-tail-at-SIGKILL behavior are byte-for-byte the pre-change behavior
+
+Scenario: lone_writer_no_added_latency # M4
+ Given appendfsync=always with exactly one writer (C=1, nothing else queued)
+ When it issues a durable write
+ Then the writer fsyncs immediately for that single message (batch of 1) and acks Synced
+ And per-write latency is not worse than the pre-change one-fsync-per-write path
+
+Scenario: batch_size_is_bounded # M4
+ Given appendfsync=always and a flood of more than CAP durable writes queued at once
+ When the writer drains
+ Then it flushes at the CAP boundary and begins a fresh batch (multiple bounded fsyncs)
+ And the write buffer never grows past the cap and no message is dropped
+
+# ---- Rejects (each asserts what must NOT change) ----
+Scenario: reject_batch_write_failed # batch_write_failed
+ Given a batch where one message's write_all fails (I/O error)
+ When the writer processes the batch
+ Then that waiter is acked AofAck::WriteFailed and the write_error latch drops subsequent appends
+ And NO waiter in the batch is acked Synced (no false durability claim)
+
+Scenario: reject_batch_fsync_failed # batch_fsync_failed
+ Given a batch whose flush()+sync_data() returns an error
+ When the writer completes the batch
+ Then ALL AppendSync waiters in the batch are acked AofAck::FsyncFailed
+ And NO waiter in the batch is acked Synced
+
+Scenario: reject_ack_before_fsync # ack_before_fsync
+ Given any drained batch under appendfsync=always
+ When the writer sequences its work
+ Then no Synced ack is ever observable before the covering fsync has returned (invariant — unreachable)
+ And the ack-after-fsync ordering never inverts under any batch size
+
+Scenario: reject_batch_straddles_control # batch_straddles_control
+ Given a pending unflushed batch and a control/rewrite message next in the channel
+ When the writer reaches that control message
+ Then it MUST NOT handle the control message before flushing the batch
+ And message order is preserved and no AppendSync waiter is left unacked across the control boundary
+
+Scenario: reject_batch_cap_exceeded # batch_cap_exceeded
+ Given more than CAP messages ready to drain
+ When the writer builds a batch
+ Then it stops at CAP, flushes, and starts a new batch (never an unbounded drain)
+ And no queued message is dropped and the buffer high-water stays bounded by the cap
+```
+
+
+
+
+
+---
+
+## 3 · CONTRACT — freeze the shape ▸ docs/05-step-3-contract.md
+
+Internal mechanism (no wire/protocol surface). Contract = the writer-side group-commit shape +
+the durability invariant. Names from the existing AOF glossary: `AofMessage` {`Append{bytes,lsn}`,
+`AppendSync{bytes,lsn,ack}`, `Rewrite`, `RewriteSharded`, `RewritePerShard`, `Shutdown`}, `AofAck`
+{`Synced`, `WriteFailed`, `FsyncFailed`}, `FsyncPolicy::{Always,EverySec,No}`, `aof_writer_task`,
+`per_shard_aof_writer_task`. New module: `src/persistence/aof/group_commit.rs`.
+
+```
+// ---- Bounds (design-for-failure: a drain is never unbounded) ----
+const AOF_GROUP_COMMIT_MAX_BATCH: usize // hard cap on messages per batch (e.g. 1024)
+const AOF_GROUP_COMMIT_MAX_BYTES: usize // hard cap on bytes buffered per batch (e.g. 8 MiB)
+
+// ---- Pure batching seam (no I/O — unit-testable without a file or runtime) ----
+struct GroupCommitBatch {
+ data: Vec, // Append + AppendSync, in channel order; bytes to write
+ deferred_control: Option, // a control msg that TERMINATED the drain (handle AFTER commit)
+}
+fn collect_group_commit_batch(
+ first: AofMessage, // the message just recv()'d (data or control)
+ mut try_next: impl FnMut() -> Option, // non-blocking channel drain (flume try_recv)
+ max_batch: usize, max_bytes: usize,
+) -> GroupCommitBatch
+// - `first` is control -> data=[], deferred_control=Some(first) (no batch; caller handles it)
+// - else -> pull data messages until: queue empty | a control msg appears
+// (-> deferred_control) | count==max_batch | bytes>=max_bytes (cap stop)
+// - ORDER preserved exactly; a control msg is NEVER placed in `data`.
+
+// ---- Commit seam (the durability invariant; one fsync per batch) ----
+fn commit_group_commit_batch(
+ file: &mut W, batch: &mut GroupCommitBatch, do_fsync: bool /* true under Always */,
+) -> CommitOutcome
+// 1. write_all each data msg's bytes IN ORDER;
+// 2. if do_fsync: exactly ONE flush()+sync_data() AFTER all writes;
+// 3. THEN ack every AppendSync waiter in the batch.
+// Ack mapping (a response for every §1 Reject code):
+// all writes ok + fsync ok -> every AppendSync.ack = AofAck::Synced
+// a write_all error -> "batch_write_failed": that+remaining AppendSync.ack = WriteFailed;
+// set write_error latch; NO waiter in the batch acked Synced
+// flush()/sync_data() error -> "batch_fsync_failed": ALL AppendSync.ack = FsyncFailed; none Synced
+// (acks are sent in step 3 ONLY) -> "ack_before_fsync": unreachable — no ack precedes the fsync
+// Returns CommitOutcome { synced: usize, write_failed: bool, fsync_failed: bool }.
+```
+
+Durability invariant (M2, the freeze-first contract — the immutable promise):
+- A connection observes `+OK` for a durable write ONLY after `commit_group_commit_batch` has fsynced
+ the batch containing its bytes and sent its `Synced` ack. ⇒ **acked ⇒ on disk** (no false positive).
+- The single batch fsync makes every preceding `Append` in `data` durable too (ordered-channel
+ H1-BARRIER property — the `fsync_barrier` keeps working unchanged: a zero-length `AppendSync` in a
+ batch still proves all prior appends durable). `pending_aof_count` accounting is untouched.
+- A crash mid-batch (before its fsync) leaves at most a torn tail record; replay truncates to the last
+ valid record (existing AOF replay) — earlier acked writes are intact. Unacked writes are the client's
+ to retry (pre-existing at-least-once for unacked; group commit does not worsen it).
+
+Wiring (both writers call the same seam; `EverySec`/`No` pass `do_fsync=false` so behavior is unchanged):
+- `aof_writer_task` (TopLevel, monoio blocking loop + tokio bounded-recv loop) and
+ `per_shard_aof_writer_task`: replace the per-message `match fsync { Always => write+fsync+ack }`
+ with `collect_group_commit_batch` → `commit_group_commit_batch(do_fsync = fsync==Always)` → handle
+ `deferred_control` (Rewrite/RewriteSharded/RewritePerShard/Shutdown) exactly as today, AFTER the
+ flush. `batch_straddles_control` is structurally impossible: control never enters `data`.
+- `EverySec`: `do_fsync=false` in commit; the existing ≥1s deadline flush + bounded-recv path is
+ retained verbatim (group commit only coalesces the *Always* fsync; everysec already coalesces by time).
+
+Reject → contracted response (all five):
+- `batch_write_failed` -> AppendSync waiter(s) acked `WriteFailed`; write_error latch set; no `Synced`.
+- `batch_fsync_failed` -> all batch AppendSync waiters acked `FsyncFailed`; no `Synced`.
+- `ack_before_fsync` -> impossible by construction (acks only in commit step 3, after the fsync).
+- `batch_straddles_control` -> impossible: control msgs go to `deferred_control`, handled after commit.
+- `batch_cap_exceeded` -> `collect_*` stops at `max_batch`/`max_bytes`; a fresh batch begins next loop.
+
+Status: FROZEN @ v1 — approved by Tin Dang 2026-06-14.
+Least-sure flag surfaced at freeze:
+ ⚠ [contract·spec] mixed-stream drain preserves C4-fold + H1-BARRIER + everysec invariants exactly —
+ if wrong: silent data loss / AOF corruption under crash (the milestone's freeze-first risk).
+ Mitigated in shape: control msgs route to `deferred_control` (never in `data`) ⇒ `batch_straddles_control`
+ structurally impossible; new concurrent-writers crash test + existing crash-matrix are the oracle.
+ ⚠ [contract] the `collect`/`commit` factoring + deferring a try_recv'd control message (vs peeking)
+ does not mis-order against the C4 fold `pending_aof_count` accounting — if wrong: seam boundary
+ moves (a contract reshape, caught at build/test, not a durability bug).
+
+
+---
+
+## 4 · TESTS — failing-first suite (red) ▸ docs/06-step-4-tests.md
+
+Coverage target: every Must + every Reject has ≥1 red test; the batching seam fully unit-covered;
+durability proven by an integration crash test (not just unit mocks).
+Plan (one test per scenario, asserting behavior not internals):
+
+ UNIT — pure batching seam (`src/persistence/aof/group_commit.rs` #[cfg(test)], no I/O / no runtime):
+ - test_collect_first_control_makes_no_batch: first=Rewrite → data==[] , deferred_control==Some(Rewrite) [batch_straddles_control]
+ - test_collect_stops_at_control_preserving_order: [Append,AppendSync,Rewrite] → data==[Append,AppendSync] in order, deferred==Some(Rewrite) [M3]
+ - test_collect_respects_max_batch: 5 ready, cap=2 → data.len()==2, rest left in queue [batch_cap_exceeded]
+ - test_collect_respects_max_bytes: bytes cap hit before count cap → stops at the byte boundary [batch_cap_exceeded]
+ - test_commit_one_fsync_many_acks: mock Write counts flush/sync calls; K AppendSync → exactly 1 fsync, K acks==Synced [M1]
+ - test_commit_acks_only_after_fsync: ordering probe (record fsync vs ack timestamps/sequence) → every ack strictly after the fsync [ack_before_fsync]
+ - test_commit_write_fail_acks_write_failed: a mock write_all error mid-batch → that+remaining AppendSync==WriteFailed, none==Synced, write_error set [batch_write_failed]
+ - test_commit_fsync_fail_acks_fsync_failed: mock sync_data error → ALL AppendSync==FsyncFailed, none==Synced [batch_fsync_failed]
+ - test_commit_everysec_no_fsync: do_fsync=false → zero fsync calls; Append-only batch acks nothing; bytes written in order [M4 everysec_path_unchanged]
+ - test_commit_barrier_covers_preceding_appends: batch=[Append,Append,AppendSync(zero-len barrier)] → 1 fsync, barrier acked Synced after the 2 appends are written [M2 barrier_property_preserved]
+
+ INTEGRATION — real server, both runtimes (`tests/wal_group_commit.rs`, MOON_BIN-pinned, VM-local):
+ - test_concurrent_writers_all_acked_survive_sigkill: appendfsync=always, N writers issue distinct INCRs, collect only +OK'd ones, SIGKILL, restart → each counter == its acked count, none lost [M2 every_acked_write_survives_crash]
+ - test_interrupted_batch_replays_to_last_valid_record: kill mid-batch (fault inject before fsync) → restart loads clean (truncate-at-last-valid), prior-batch acked writes intact [M2 interrupted_batch_replays_to_last_valid_record]
+ - test_lone_writer_fsyncs_immediately: appendfsync=always, C=1 → each durable write acked Synced (batch of 1), latency not worse than baseline [M4 lone_writer_no_added_latency]
+ - test_rewrite_during_active_writes_no_loss: BGREWRITEAOF while many durable writes in flight → batch flushed before rewrite; post-rewrite replay loses no acked write [M3 control_message_breaks_the_batch]
+ - existing crash-matrix (`crash_matrix_per_shard_aof.rs`, `crash_matrix_per_shard_bgrewriteaof.rs`) stays GREEN on both runtimes [M2/M3 regression oracle]
+ - existing `pool.rs` H1-BARRIER + everysec + FsyncPolicy unit tests stay GREEN [M4]
+
+ BENCH — measurement, not pass/fail (`scripts/bench-*` / a wal-group-commit cell):
+ - always_group_commit cell: appendfsync=always, pipeline>1, C∈{1,8,32}, fresh-server-per-rep, best-of-N, per-runtime → M0 before + M1 after RELATIVE ratio; everysec/no controls flat [M0/M1]
+
+
+Tests live in: `tests/wal_group_commit.rs` (integration) + `src/persistence/aof/group_commit.rs` unit
+tests · MUST run red (symbols/behavior missing) before Build.
+
+RED CONFIRMED 2026-06-14 (VM, monoio): `cargo test --test wal_group_commit --no-run` fails with
+exactly `error[E0432]: unresolved import moon::persistence::aof::group_commit — could not find
+group_commit in aof` — the right reason (missing implementation, harness otherwise compiles). 9 seam
+tests cover: collect (control-first, stop-at-control+order, max_batch, max_bytes) + commit
+(one-fsync-many-acks+ordering, write-fail, fsync-fail+ack-after, everysec-no-fsync, barrier-covers-appends).
+The end-to-end durability scenarios (every_acked_write_survives_crash, rewrite-during-writes) are
+gated on the integration crash-matrix + a new concurrent-writers SIGKILL test (added at build) — a unit
+mock cannot prove on-disk survival across a real kill.
+NOTE (non-behavioral): §3's `` + "flush()+sync_data()" is realized as a sync-capable
+`GroupCommitSink { write_all; sync }` trait so the single fsync is mockable/countable; the FROZEN
+durability behavior (one fsync after all writes, ack-after-fsync, the 5 reject mappings) is unchanged.
+
+
+
+
+---
+
+## 5 · BUILD — AI writes code ▸ docs/07-step-5-build.md
+
+Safety rule (feature-specific):
+Code lives in: `./src/`
+Constraints: do NOT change any test or the contract; allow-list packages only; ask if unclear.
+
+
+
+---
+
+## 6 · VERIFY — evidence + non-functional review ▸ docs/08-step-6-verify.md
+
+- [x] all tests pass — dual-runtime, VM-local clone `~/moon-gc` @ `12681b3` (home volume; shared-volume diskfull avoided):
+ · seam unit 9/9 (tokio) · AOF lib 94 (tokio) / 90 (monoio) · integration crash 4/4 ×both runtimes
+ (incl. `concurrent_writers_all_acked_survive_sigkill_top_level` — exercises the Finding-2 tokio-TopLevel latch)
+ · crash-matrix 5/5 ×both (`crash_matrix_per_shard_aof` 3, `..._bgrewriteaof` 2) · consistency 197/197 @1/4/12
+- [x] coverage did not decrease — +4 integration crash tests added at build; the tokio-TopLevel torn-write path is now
+ covered by the top_level SIGKILL test; no test removed or weakened.
+- [~] no test or contract was altered during build — §3 CONTRACT unchanged (FROZEN @v1). ONE frozen §4 test
+ (`commit_write_fail_acks_write_failed`) got a HUMAN-APPROVED intent-preserving fix: it double-consumed a flume
+ `bounded(1)` (use-after-consume) — the test was wrong, my impl was correct (both waiters acked `WriteFailed`).
+ Fixed by reading each receiver once; all 3 assertions kept. NOT a weakening. (Surfaced at gate.)
+- [x] concurrency / timing safe — fsync-before-ack preserved; the `write_error` torn-write latch is now UNIFORM across
+ all 4 writer loops (Finding 2 closed in `2750d1c`); no lock held across `.await`; zero new cross-thread lock;
+ the flume oneshot ack pattern is unchanged; `ack_batch` is the single-sourced ack-after-fsync ordering.
+- [x] no exposed secrets, injection openings, or unexpected dependencies — internal mechanism only; zero new deps.
+- [x] layering & dependencies follow conventions — `group_commit.rs` is a pure runtime-free seam under
+ `persistence/aof/`; the loops call it; no new module edges; clippy ×2 + `fmt` clean; zero new `unsafe`.
+
+### Deep checks — do not skim (fill the path that applies; the resolver judges which)
+- [x] WIRING (code) — every group_commit symbol referenced: `collect_group_commit_batch`/`commit_group_commit_batch`/
+ `GroupCommitSink`/`CommitOutcome`/`AOF_GROUP_COMMIT_MAX_{BATCH,BYTES}` from `writer_task.rs` + the seam tests;
+ `ack_batch`/`BatchAck`/`msg_body` `pub(crate)` from `writer_task.rs`; `is_control`/`msg_body_len` private,
+ used in-module. Confirmed by grep + clippy-clean ×2 runtimes.
+- [x] DEAD-CODE (code) — no new orphan; clippy `-D warnings` clean on default (monoio) AND tokio,jemalloc.
+- [x] SEMANTIC — n/a (code task; no prose contract surface).
+
+### M0/M1 performance evidence — instrument limitation (NOT a correctness gap)
+- Mechanism of the win (K `AppendSync` → exactly ONE fsync, ack-after-fsync) is proven DETERMINISTICALLY by the seam
+ unit test `commit_one_fsync_many_acks`.
+- Empirical M0→M1 RELATIVE throughput delta is NOT resolvable on the OrbStack VM. Baseline (`3150f8b`, one-fsync-per-
+ AppendSync) vs HEAD (`12681b3`), `appendfsync=always`, fresh-server best-of-3, redis-benchmark `-P16`:
+ shards=1 C=32: base 934K / head 917K = 0.98× · shards=4 C=32: 892K / 917K = 1.03×
+ shards=1 conc sweep C∈{32,128,256,512}: 1.00× / 1.16× / 0.90× / 1.00× (no trend; within the ±~10% VM noise the
+ everysec controls also showed: 1.03×, 1.12×).
+ Root cause: the virtio disk's `fsync` is near-free — `always` runs at ~0.9M RPS (NOT the 11× penalty 135K→12K the
+ feature targets), so the writer drains each `AppendSync` before the next arrives ⇒ a coalescing batch never forms
+ (batch≈1) ⇒ nothing to amortize. This is exactly §1 assumption #4 (lowest-confidence perf flag): "confirm the
+ OrbStack VM instrument is valid for this metric before anchoring M1" — now CONFIRMED INVALID. The M0/M1 ratio
+ requires a slow-fsync instrument (real disk / GCloud — noted blocked). Deferred to §7 OBSERVE.
+
+### GATE RECORD
+Outcome: PASS — correctness/durability (M2) · ordering (M3) · no-regression (M4 correctness) · all 5 Rejects PROVEN
+ green on BOTH runtimes; the coalescing-win MECHANISM (K AppendSync → 1 fsync, ack-after-fsync) proven by the
+ deterministic seam test `commit_one_fsync_many_acks`. The empirical M0/M1 RELATIVE throughput ratio is DEFERRED to
+ §7 OBSERVE — un-measurable on the OrbStack VM (near-free fsync ⇒ batch≈1; §1 assumption #4 confirmed), needs a
+ slow-fsync / real-disk instrument. Non-security, non-correctness, freeze-flagged instrument gap → deferral, not a
+ weakening. The one frozen §4 test edit (`commit_write_fail_acks_write_failed`, flume bounded(1) use-after-consume)
+ was a human-approved intent-preserving fix; §3 CONTRACT untouched.
+Reviewed by: Tin Dang · date: 2026-06-14
+
+
+
+---
+
+## 7 · OBSERVE — feed the next loop ▸ docs/09-the-loop.md
+
+Watch (reuse scenarios as monitors):
+- `moon_aof_fsync_duration_microseconds_count` per N durable writes → effective batch size = N / fsync_count
+ (the direct mechanism monitor; <1 fsync per write under `always`+concurrency is the win materializing).
+- `AOF_FSYNC_ERR` / `WriteFailed` ack rate (reject-path health) · everysec 1s-deadline tail-loss at SIGKILL (M4 control).
+Spec delta for the next loop:
+- The M0/M1 throughput Must is only observable on a SLOW-fsync instrument. The OrbStack virtio disk drains each
+ `AppendSync` before the next arrives (batch≈1), so the coalescing win is STRUCTURALLY unmeasurable there — measure the
+ RPS ratio on GCloud / a real disk (or an O_DIRECT / fsync-delay harness) BEFORE anchoring the perf number. Until then
+ the win is asserted by the deterministic seam test + the on-disk crash-survival suites, not by a VM RPS delta.
+
+### Competency deltas
+- [TDD · open] a frozen RED test can itself be wrong: `commit_write_fail_acks_write_failed` double-consumed a flume
+ `bounded(1)` (use-after-consume) — the fix was intent-preserving + human-approved, never a weakening
+ (evidence: failed `left: Err(Disconnected)` vs `right: Ok(WriteFailed)`; my impl acked both waiters WriteFailed).
+- [SDD · open] the contract invariant "`CommitOutcome.write_failed` ⇒ the latch must engage" applied to all 4 writer
+ loops, but the pre-existing tokio-TopLevel loop never carried a `write_error` latch (nor the fsync-fail injection) —
+ group commit made the latent durability gap explicit (evidence: adversarial Finding 2 @0.97; fixed `2750d1c`).
+- [ADD · open] a perf Must can be un-measurable on the only available instrument: the OrbStack VM's near-free fsync
+ makes the group-commit win structurally invisible (batch≈1; `always`≈0.9M RPS, no 11× penalty) — §1 ranked exactly
+ this risk lowest-confidence (assumption #4); confirm instrument validity BEFORE committing to a perf Must
+ (evidence: 0.98× ratio, conc-sweep no-trend within ±10% VM noise).
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5ed752803..8c60d5b6c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
+### Performance — WAL group commit under `appendfsync=always` (PR #178)
+
+Concurrent writes pending at the same shard now coalesce into a single
+`fsync` instead of one sync per write, collapsing a large share of the
+~11× `appendfsync=always` throughput penalty that appears when multiple
+clients write in parallel. The batching is opportunistic — a writer drains
+every `AppendSync` already queued on its shard channel and issues one
+barrier `fsync` for the whole group — and is wired into all four AOF writer
+loops (`{TopLevel, PerShard}` × `{monoio sync, tokio async}`). Durability is
+unchanged: control records route to a separate deferred queue so a batch can
+never straddle a non-data message, and the exactly-once-under-crash
+invariant (crash-matrix + SIGKILL integration tests) holds on both runtimes.
+The absolute throughput gain is disk-dependent — structurally unmeasurable on
+the near-free virtio `fsync` of the dev VM, so the coalescing mechanism (K
+`AppendSync` → 1 `fsync`) is pinned by a deterministic batching seam test and
+the wall-clock magnitude is deferred to real-disk hardware.
+
### Removed (BREAKING) — `--cross-shard-fast-path` flag and its dead telemetry
The orphaned `--cross-shard-fast-path` CLI flag (with its `CrossShardFastPath`
diff --git a/src/persistence/aof/group_commit.rs b/src/persistence/aof/group_commit.rs
new file mode 100644
index 000000000..f1491ee0f
--- /dev/null
+++ b/src/persistence/aof/group_commit.rs
@@ -0,0 +1,245 @@
+//! AOF **group commit** — coalesce concurrent pending writes into ONE fsync.
+//!
+//! Under `appendfsync=always` the writer used to `flush()+sync_data()` once per
+//! `AppendSync` message (the ~11× write penalty: 135K → 12K ops/s). When N
+//! per-connection durable writers are in flight, N `AppendSync` messages queue at
+//! one writer while it is busy on the first fsync. This module factors out the
+//! pure *batching seam* so the writer can drain the ready queue, write every
+//! message, make them durable with a **single** `flush()+sync_data()`, and only
+//! THEN ack every waiter `Synced` — amortizing the fsync cost across write
+//! concurrency without weakening the fsync-before-ack durability contract.
+//!
+//! Two seams, both unit-testable with no file / no runtime:
+//! - [`collect_group_commit_batch`] — the *pure drain*: pull queued data
+//! messages until the queue empties, a control message appears (deferred,
+//! NEVER batched), or a bounded cap is hit. Control-message ordering and the
+//! bounded-drain guarantee live here.
+//! - [`commit_group_commit_batch`] — the *durability invariant*: write all
+//! bytes in order, ONE fsync after all writes (Always only), then ack every
+//! `AppendSync`. The single fsync also makes every preceding fire-and-forget
+//! `Append` in the batch durable (the H1-BARRIER ordered-channel property).
+//!
+//! The four writer loops (TopLevel/PerShard × monoio/tokio) share
+//! [`collect_group_commit_batch`] and the ack/durability ordering in
+//! [`ack_batch`]. The sync monoio TopLevel path calls [`commit_group_commit_batch`]
+//! directly through a [`GroupCommitSink`] over its `std::fs::File`; the async
+//! tokio paths and the framed per-shard paths replicate the SAME three-step
+//! invariant inline (async I/O cannot implement a sync sink, and the per-shard
+//! framed format prepends a `[u64 lsn][u32 len]` header per record) but ack
+//! through the same [`ack_batch`] so the durability ordering is single-sourced.
+
+use super::{AofAck, AofMessage};
+
+/// Hard cap on the number of messages coalesced into one batch. A write flood
+/// cannot delay an early waiter's fsync unboundedly: the drain stops at the cap,
+/// flushes, fsyncs, acks, and a fresh batch begins on the next loop iteration.
+pub const AOF_GROUP_COMMIT_MAX_BATCH: usize = 1024;
+
+/// Hard cap on the bytes buffered into one batch (whole-message granularity — a
+/// message is never split). Bounds the write buffer's high-water mark so a burst
+/// of large values cannot grow memory without limit before the fsync.
+pub const AOF_GROUP_COMMIT_MAX_BYTES: usize = 8 * 1024 * 1024;
+
+/// One drained group-commit batch: the data messages to write (in channel
+/// order) plus the single control message that TERMINATED the drain, if any.
+///
+/// `deferred_control` is the structural guarantee behind `batch_straddles_control`:
+/// a control / rewrite message is NEVER placed in `data`, so the caller always
+/// flushes (writes + ONE fsync + acks all) the in-progress batch BEFORE handling
+/// the control message — message order is preserved exactly.
+pub struct GroupCommitBatch {
+ /// `Append` + `AppendSync` messages, in channel order. These are the bytes
+ /// to write; every `AppendSync` carries an ack waiter to signal after fsync.
+ pub data: Vec,
+ /// A control message (`Rewrite` / `RewriteSharded` / `RewritePerShard` /
+ /// `Shutdown`) that ended the drain. Handled by the caller AFTER the batch is
+ /// committed — never absorbed into `data`.
+ pub deferred_control: Option,
+}
+
+/// True for the non-data control messages that must break (never join) a batch.
+#[inline]
+fn is_control(msg: &AofMessage) -> bool {
+ matches!(
+ msg,
+ AofMessage::Rewrite(_)
+ | AofMessage::RewriteSharded(_)
+ | AofMessage::RewritePerShard { .. }
+ | AofMessage::Shutdown
+ )
+}
+
+/// Payload byte length of a data message (0 for control / zero-length barrier).
+#[inline]
+fn msg_body_len(msg: &AofMessage) -> usize {
+ match msg {
+ AofMessage::Append { bytes, .. } | AofMessage::AppendSync { bytes, .. } => bytes.len(),
+ _ => 0,
+ }
+}
+
+/// Borrow the payload bytes of a data message (empty for control / a zero-length
+/// H1-BARRIER `AppendSync`). Used by [`commit_group_commit_batch`] and the inline
+/// writer paths to feed the sink in channel order.
+#[inline]
+pub(crate) fn msg_body(msg: &AofMessage) -> &[u8] {
+ match msg {
+ AofMessage::Append { bytes, .. } | AofMessage::AppendSync { bytes, .. } => bytes,
+ _ => &[],
+ }
+}
+
+/// Drain a ready group-commit batch from one message plus a non-blocking puller.
+///
+/// `first` is the message just `recv()`'d (data or control); `try_next` is a
+/// non-blocking channel drain (e.g. `|| rx.try_recv().ok()`).
+///
+/// - `first` is a control message → empty `data`, `deferred_control = Some(first)`
+/// (no batch; the caller handles the control directly).
+/// - otherwise → pull data messages until ONE of: the queue empties · a control
+/// message appears (→ `deferred_control`, drain stops) · `data.len() == max_batch`
+/// · accumulated bytes `>= max_bytes` (whole-message granularity: the crossing
+/// message is included, then the drain stops).
+///
+/// Channel order is preserved exactly and a control message is NEVER placed in
+/// `data`.
+pub fn collect_group_commit_batch(
+ first: AofMessage,
+ mut try_next: impl FnMut() -> Option,
+ max_batch: usize,
+ max_bytes: usize,
+) -> GroupCommitBatch {
+ if is_control(&first) {
+ return GroupCommitBatch {
+ data: Vec::new(),
+ deferred_control: Some(first),
+ };
+ }
+
+ let mut total_bytes = msg_body_len(&first);
+ let mut data = Vec::with_capacity(8);
+ data.push(first);
+ let mut deferred_control = None;
+
+ // Stop conditions are checked at the loop head so the message that crosses a
+ // cap is already included (the cap is a "stop after", not "stop before").
+ while data.len() < max_batch && total_bytes < max_bytes {
+ match try_next() {
+ None => break, // queue drained
+ Some(msg) if is_control(&msg) => {
+ deferred_control = Some(msg);
+ break;
+ }
+ Some(msg) => {
+ total_bytes += msg_body_len(&msg);
+ data.push(msg);
+ }
+ }
+ }
+
+ GroupCommitBatch {
+ data,
+ deferred_control,
+ }
+}
+
+/// Sync, mockable realization of the frozen `` + `flush()+sync_data()`
+/// commit target. Splitting `write_all` from `sync` is what makes the single
+/// per-batch fsync countable in tests (the durability behavior is unchanged).
+pub trait GroupCommitSink {
+ /// Append `buf` to the underlying file. An empty `buf` (a zero-length
+ /// H1-BARRIER `AppendSync`) is a no-op write.
+ fn write_all(&mut self, buf: &[u8]) -> std::io::Result<()>;
+ /// `flush()` then `sync_data()` — the single durability point for the batch.
+ fn sync(&mut self) -> std::io::Result<()>;
+}
+
+/// Result of committing one batch — reported back so the caller can drive its
+/// `write_error` latch and surface failures.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct CommitOutcome {
+ /// Number of `AppendSync` waiters acked `Synced` (0 on any failure, and 0 for
+ /// an `Append`-only batch).
+ pub synced: usize,
+ /// A `write_all` for some message in the batch failed; the latch must engage.
+ pub write_failed: bool,
+ /// The single `flush()+sync_data()` failed; every waiter acked `FsyncFailed`.
+ pub fsync_failed: bool,
+}
+
+/// Verdict applied uniformly to every `AppendSync` waiter in a committed batch.
+#[derive(Clone, Copy)]
+pub(crate) enum BatchAck {
+ Synced,
+ WriteFailed,
+ FsyncFailed,
+}
+
+/// Drain the batch's data, acking every `AppendSync` waiter with `verdict`
+/// (`Append` messages have no waiter). This is the SINGLE source of the
+/// durability ack ordering: it is called ONLY in commit step 3 (after the fsync)
+/// or on a terminal write/fsync failure — never before the fsync result is known,
+/// so `ack_before_fsync` is unreachable by construction. Shared by the sync
+/// [`commit_group_commit_batch`] and the async/framed inline writer paths so all
+/// four writer loops ack identically.
+pub(crate) fn ack_batch(batch: &mut GroupCommitBatch, verdict: BatchAck) -> CommitOutcome {
+ let mut synced = 0usize;
+ for msg in batch.data.drain(..) {
+ if let AofMessage::AppendSync { ack, .. } = msg {
+ let a = match verdict {
+ BatchAck::Synced => {
+ synced += 1;
+ AofAck::Synced
+ }
+ BatchAck::WriteFailed => AofAck::WriteFailed,
+ BatchAck::FsyncFailed => AofAck::FsyncFailed,
+ };
+ // The receiver may be gone (caller dropped it after an F2 timeout);
+ // a failed send is benign — the caller already moved on.
+ let _ = ack.send(a);
+ }
+ }
+ CommitOutcome {
+ synced,
+ write_failed: matches!(verdict, BatchAck::WriteFailed),
+ fsync_failed: matches!(verdict, BatchAck::FsyncFailed),
+ }
+}
+
+/// Commit one batch through a [`GroupCommitSink`] — the durability invariant:
+///
+/// 1. `write_all` every data message's bytes IN ORDER;
+/// 2. if `do_fsync` (true under `Always`): exactly ONE `sync()` AFTER all writes;
+/// 3. THEN ack every `AppendSync` waiter `Synced`.
+///
+/// Reject mappings (one response per §1 reject code):
+/// - a `write_all` error → `batch_write_failed`: every `AppendSync` acked
+/// `WriteFailed`, none `Synced` (`outcome.write_failed = true`). The caller
+/// engages its `write_error` latch — the stream may be torn.
+/// - a `sync` error → `batch_fsync_failed`: every `AppendSync` acked
+/// `FsyncFailed`, none `Synced` (`outcome.fsync_failed = true`).
+/// - `ack_before_fsync` is impossible: acks happen only in step 3, after the
+/// fsync returns.
+///
+/// `do_fsync == false` (everysec/no) writes the batch without a per-batch fsync —
+/// the writer's deadline flush governs durability. Such batches are `Append`-only
+/// (an `AppendSync` is enqueued only under `Always`, see `pool::try_send_append_durable`
+/// / `pool::fsync_barrier`), so `synced == 0`.
+pub fn commit_group_commit_batch(
+ sink: &mut S,
+ batch: &mut GroupCommitBatch,
+ do_fsync: bool,
+) -> CommitOutcome {
+ // Step 1 — write every data message's bytes in channel order.
+ for msg in &batch.data {
+ if sink.write_all(msg_body(msg)).is_err() {
+ return ack_batch(batch, BatchAck::WriteFailed);
+ }
+ }
+ // Step 2 — exactly ONE fsync, AFTER all writes, only under Always.
+ if do_fsync && sink.sync().is_err() {
+ return ack_batch(batch, BatchAck::FsyncFailed);
+ }
+ // Step 3 — ack every AppendSync (only now, after the fsync has returned).
+ ack_batch(batch, BatchAck::Synced)
+}
diff --git a/src/persistence/aof/mod.rs b/src/persistence/aof/mod.rs
index b43b47f23..a75809146 100644
--- a/src/persistence/aof/mod.rs
+++ b/src/persistence/aof/mod.rs
@@ -469,6 +469,10 @@ pub const DEFAULT_AOF_FSYNC_TIMEOUT: Duration = Duration::from_millis(2000);
// ── Submodule decomposition (refactor: aof.rs 4379 lines -> directory module) ──
// Codec (serialize_command/replay_aof) stays in this parent so children reach it
// via `use super::*`. AofWriterPool, writer tasks, and rewrite paths move out.
+/// Group-commit batching seam (coalesce concurrent pending writes into one
+/// fsync under `appendfsync=always`). `pub` so the §4 red suite can pin the pure
+/// seam (collect/commit) against the public API.
+pub mod group_commit;
mod pool;
mod rewrite;
mod writer_task;
diff --git a/src/persistence/aof/writer_task.rs b/src/persistence/aof/writer_task.rs
index 5eac860d1..188b8db4a 100644
--- a/src/persistence/aof/writer_task.rs
+++ b/src/persistence/aof/writer_task.rs
@@ -10,6 +10,58 @@ use super::*;
#[cfg(feature = "runtime-monoio")]
use super::rewrite::{do_rewrite_sharded, do_rewrite_single};
+// Group-commit batching seam (wal-group-commit). All four writer loops drain a
+// ready batch via `collect_group_commit_batch` and ack through `group_commit`'s
+// shared ordering; the sync monoio TopLevel path commits through a
+// `GroupCommitSink`, while the async/framed paths replicate the same invariant
+// inline (one fsync per batch, ack AFTER the fsync) — see group_commit.rs.
+use super::group_commit::{
+ self, AOF_GROUP_COMMIT_MAX_BATCH, AOF_GROUP_COMMIT_MAX_BYTES, BatchAck,
+ collect_group_commit_batch,
+};
+#[cfg(feature = "runtime-monoio")]
+use super::group_commit::{GroupCommitSink, commit_group_commit_batch};
+
+/// A sync [`GroupCommitSink`] over a `std::fs::File` for the monoio writer
+/// loops. `write_all` appends raw bytes (an empty buffer — a zero-length
+/// H1-BARRIER `AppendSync` — is a no-op); `sync` does the single per-batch
+/// `flush()+sync_data()` and records the fsync metric on success.
+///
+/// `fail_sync` mirrors the legacy `MOON_TEST_AOF_FSYNC_FAIL` injection: when set,
+/// `sync()` returns an error so the whole batch resolves `FsyncFailed` (the
+/// client sees `AOF_FSYNC_ERR`, never `+OK`) without touching durable storage.
+#[cfg(feature = "runtime-monoio")]
+struct FileGroupSink<'a> {
+ file: &'a mut std::fs::File,
+ fail_sync: bool,
+}
+
+#[cfg(feature = "runtime-monoio")]
+impl GroupCommitSink for FileGroupSink<'_> {
+ #[inline]
+ fn write_all(&mut self, buf: &[u8]) -> std::io::Result<()> {
+ use std::io::Write;
+ if buf.is_empty() {
+ return Ok(()); // zero-length barrier: fsync+ack only, no on-disk record
+ }
+ self.file.write_all(buf)
+ }
+
+ #[inline]
+ fn sync(&mut self) -> std::io::Result<()> {
+ use std::io::Write;
+ if self.fail_sync {
+ return Err(std::io::Error::other("MOON_TEST_AOF_FSYNC_FAIL"));
+ }
+ let t = Instant::now();
+ let r = self.file.flush().and_then(|_| self.file.sync_data());
+ if r.is_ok() {
+ crate::admin::metrics_setup::record_aof_fsync(t.elapsed().as_micros() as u64);
+ }
+ r
+ }
+}
+
/// Background AOF writer task. Receives commands via mpsc channel and appends them
/// to the AOF file. Handles fsync according to the configured policy.
///
@@ -51,6 +103,21 @@ pub async fn aof_writer_task(
let mut writer = tokio::io::BufWriter::new(file);
#[cfg(feature = "runtime-tokio")]
let mut last_fsync = Instant::now();
+ // Torn-write latch (tokio TopLevel): once a batch write fails partway, the
+ // plain-RESP stream may carry a partial record — never append more bytes nor
+ // claim durability after the tear. Latched for the writer's lifetime; reset
+ // only on a successful rewrite (the rewrite replaces the file with a fresh
+ // one). Mirrors the monoio TopLevel and both per-shard writers, which already
+ // carry this latch (group_commit::CommitOutcome.write_failed ⇒ "latch must
+ // engage").
+ #[cfg(feature = "runtime-tokio")]
+ let mut write_error = false;
+ // Test-only fault injection: when MOON_TEST_AOF_FSYNC_FAIL=1 every AppendSync
+ // batch acks FsyncFailed instead of Synced (read once; zero cost in prod).
+ // Mirrors the monoio TopLevel + per-shard writers so the AOF_FSYNC_ERR wire
+ // path is exercised identically under both runtimes (shards=1 TopLevel).
+ #[cfg(feature = "runtime-tokio")]
+ let fail_fsync_for_test = std::env::var("MOON_TEST_AOF_FSYNC_FAIL").as_deref() == Ok("1");
// Monoio path: multi-part AOF (base RDB + incremental RESP) with sync I/O.
//
@@ -140,97 +207,80 @@ pub async fn aof_writer_task(
let fail_fsync_for_test = std::env::var("MOON_TEST_AOF_FSYNC_FAIL").as_deref() == Ok("1");
loop {
- match rx.recv() {
- // TopLevel writer: legacy v1 disk format is plain RESP. The
- // LSN is ignored — TopLevel is single-shard so per-shard merge
- // by LSN is moot.
- Ok(AofMessage::Append {
- bytes: data,
- lsn: _,
- }) => {
- if write_error {
- continue; // Drop appends after persistent I/O failure
- }
- if let Err(e) = file.write_all(&data) {
- error!(
- "AOF write failed (seq {}): {}. Persistence degraded.",
- manifest.seq, e
- );
- write_error = true;
- continue;
- }
- match fsync {
- FsyncPolicy::Always => {
- let t = Instant::now();
- if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
- error!("AOF sync failed (seq {}, always): {}", manifest.seq, e);
- write_error = true;
- } else {
- crate::admin::metrics_setup::record_aof_fsync(
- t.elapsed().as_micros() as u64,
- );
- }
- }
- FsyncPolicy::EverySec => {
- if last_fsync.elapsed() >= std::time::Duration::from_secs(1) {
- let t = Instant::now();
- if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
- error!(
- "AOF sync failed (seq {}, everysec): {}",
- manifest.seq, e
- );
- // Non-fatal for everysec: retry next interval
- } else {
- crate::admin::metrics_setup::record_aof_fsync(
- t.elapsed().as_micros() as u64,
- );
- last_fsync = Instant::now();
- }
- }
+ // Group commit: block for one message, then opportunistically drain
+ // whatever else is already queued into a bounded batch so a single
+ // fsync makes the whole batch durable (TopLevel = plain RESP bytes).
+ let first = match rx.recv() {
+ Ok(m) => m,
+ Err(_) => {
+ // Channel disconnected — final sync + shut down.
+ if !write_error {
+ if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
+ error!("AOF final sync failed (seq {}): {}", manifest.seq, e);
}
- FsyncPolicy::No => {}
}
+ info!("AOF writer shutting down (monoio, seq {})", manifest.seq);
+ break;
}
- // TopLevel writer (monoio): legacy v1 plain RESP, lsn ignored.
- // AppendSync ALWAYS fsyncs and acks before returning, regardless
- // of the configured policy — that's the durability contract the
- // caller signed up for by choosing AppendSync.
- Ok(AofMessage::AppendSync {
- bytes: data,
- lsn: _,
- ack,
- }) => {
- if write_error {
- let _ = ack.send(AofAck::WriteFailed);
- continue;
- }
- // Test-only: return FsyncFailed immediately without touching disk.
- if fail_fsync_for_test {
- let _ = ack.send(AofAck::FsyncFailed);
- continue;
- }
- if let Err(e) = file.write_all(&data) {
+ };
+ let mut batch = collect_group_commit_batch(
+ first,
+ || rx.try_recv().ok(),
+ AOF_GROUP_COMMIT_MAX_BATCH,
+ AOF_GROUP_COMMIT_MAX_BYTES,
+ );
+
+ // -- commit the data batch (one fsync under Always; deadline under everysec) --
+ if !batch.data.is_empty() {
+ if write_error {
+ // Persistent I/O failure latched: drop appends and fail every
+ // AppendSync waiter — never a false durability claim.
+ let _ = group_commit::ack_batch(&mut batch, BatchAck::WriteFailed);
+ } else {
+ let do_fsync = matches!(fsync, FsyncPolicy::Always);
+ let mut sink = FileGroupSink {
+ file: &mut file,
+ fail_sync: fail_fsync_for_test,
+ };
+ let outcome = commit_group_commit_batch(&mut sink, &mut batch, do_fsync);
+ if outcome.write_failed {
+ // A torn write may leave a partial record — latch so no
+ // further bytes are appended after the tear.
error!(
- "AOF AppendSync write failed (seq {}): {}. Persistence degraded.",
- manifest.seq, e
+ "AOF batch write failed (seq {}). Persistence degraded.",
+ manifest.seq
);
write_error = true;
- let _ = ack.send(AofAck::WriteFailed);
- continue;
}
- let t = Instant::now();
- if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
- error!("AOF AppendSync sync failed (seq {}): {}", manifest.seq, e);
- write_error = true;
- let _ = ack.send(AofAck::FsyncFailed);
- } else {
- crate::admin::metrics_setup::record_aof_fsync(
- t.elapsed().as_micros() as u64
- );
- let _ = ack.send(AofAck::Synced);
+ // EverySec: the batch was written but not per-batch-fsynced
+ // (do_fsync=false; there are no AppendSync waiters under
+ // everysec). Honor the 1s deadline exactly as the old
+ // per-Append path did.
+ if fsync == FsyncPolicy::EverySec
+ && !write_error
+ && last_fsync.elapsed() >= std::time::Duration::from_secs(1)
+ {
+ let t = Instant::now();
+ if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
+ error!("AOF sync failed (seq {}, everysec): {}", manifest.seq, e);
+ // Non-fatal for everysec: retry next interval
+ } else {
+ crate::admin::metrics_setup::record_aof_fsync(
+ t.elapsed().as_micros() as u64
+ );
+ last_fsync = Instant::now();
+ }
}
}
- Ok(AofMessage::Shutdown) | Err(_) => {
+ }
+
+ // -- handle the control message that ended the drain (if any) --
+ // A control message is NEVER absorbed into the batch: the batch above
+ // is already committed before the control message is handled
+ // (batch_straddles_control is structurally impossible).
+ match batch.deferred_control {
+ None => {}
+ Some(AofMessage::Shutdown) => {
if !write_error {
if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
error!("AOF final sync failed (seq {}): {}", manifest.seq, e);
@@ -239,7 +289,7 @@ pub async fn aof_writer_task(
info!("AOF writer shutting down (monoio, seq {})", manifest.seq);
break;
}
- Ok(AofMessage::Rewrite(db)) => {
+ Some(AofMessage::Rewrite(db)) => {
if !write_error {
if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
error!("AOF pre-rewrite sync failed (seq {}): {}", manifest.seq, e);
@@ -254,7 +304,7 @@ pub async fn aof_writer_task(
crate::command::persistence::AOF_REWRITE_IN_PROGRESS
.store(false, std::sync::atomic::Ordering::SeqCst);
}
- Ok(AofMessage::RewriteSharded(shard_dbs)) => {
+ Some(AofMessage::RewriteSharded(shard_dbs)) => {
if !write_error {
if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
error!("AOF pre-rewrite sync failed (seq {}): {}", manifest.seq, e);
@@ -283,11 +333,13 @@ pub async fn aof_writer_task(
// [F6] A TopLevel writer never owns per-shard files; receiving
// RewritePerShard means a routing bug. Self-abort so the
// coordinator's countdown completes and the flag clears.
- Ok(AofMessage::RewritePerShard { coord, .. }) => {
+ Some(AofMessage::RewritePerShard { coord, .. }) => {
warn!("AOF TopLevel writer received RewritePerShard — routing bug; aborting");
coord.mark_failed();
coord.shard_done();
}
+ // collect_group_commit_batch only ever defers a control message.
+ Some(_) => {}
}
}
return;
@@ -309,144 +361,205 @@ pub async fn aof_writer_task(
rx.recv_async(),
) => r,
_ = cancel.cancelled() => {
- let _ = writer.flush().await;
- let _ = writer.get_ref().sync_data().await;
+ // Skip the final sync if the stream is torn — syncing past a
+ // partial record cannot recover it and risks a false durability
+ // signal (mirrors the monoio TopLevel disconnect/shutdown gate).
+ if !write_error {
+ let _ = writer.flush().await;
+ let _ = writer.get_ref().sync_data().await;
+ }
info!("AOF writer cancelled");
break;
}
};
- if let Ok(msg) = recv_result {
- match msg {
- // TopLevel writer (tokio): legacy v1 plain RESP, lsn ignored.
- Ok(AofMessage::Append {
- bytes: data,
- lsn: _,
- }) => {
- if let Err(e) = writer.write_all(&data).await {
- error!("AOF write error: {}", e);
- continue;
+ match recv_result {
+ // Timeout (Elapsed): no message — fall through to the EverySec
+ // deadline check after this block.
+ Err(_) => {}
+ // Channel disconnected — final sync + shut down.
+ Ok(Err(_)) => {
+ if !write_error {
+ let _ = writer.flush().await;
+ let _ = writer.get_ref().sync_data().await;
+ }
+ info!("AOF writer shutting down");
+ break;
+ }
+ Ok(Ok(first)) => {
+ // Group commit: drain whatever else is queued into a bounded
+ // batch so one fsync covers all (TopLevel = plain RESP bytes).
+ let mut batch = collect_group_commit_batch(
+ first,
+ || rx.try_recv().ok(),
+ AOF_GROUP_COMMIT_MAX_BATCH,
+ AOF_GROUP_COMMIT_MAX_BYTES,
+ );
+
+ // -- write the data batch inline (async), then ONE fsync --
+ if !batch.data.is_empty() {
+ if write_error {
+ // Stream already torn — drop appends and fail every
+ // AppendSync waiter (never ack into a corrupt stream).
+ let _ = group_commit::ack_batch(&mut batch, BatchAck::WriteFailed);
+ } else {
+ let mut write_failed = false;
+ for msg in &batch.data {
+ let body = group_commit::msg_body(msg);
+ if body.is_empty() {
+ continue; // zero-length barrier: fsync+ack only
+ }
+ if let Err(e) = writer.write_all(body).await {
+ error!("AOF batch write error: {}", e);
+ write_failed = true;
+ break;
+ }
+ }
+ let do_fsync = matches!(fsync, FsyncPolicy::Always);
+ let verdict = if write_failed {
+ // A torn write may leave a partial record — latch so
+ // no further bytes are appended after the tear.
+ write_error = true;
+ BatchAck::WriteFailed
+ } else if fail_fsync_for_test && do_fsync {
+ // Injected: bytes written, the batch fsync "fails"
+ // → every waiter FsyncFailed (no disk error needed).
+ BatchAck::FsyncFailed
+ } else if do_fsync {
+ let mut fsync_failed = false;
+ if let Err(e) = writer.flush().await {
+ error!("AOF batch flush error: {}", e);
+ fsync_failed = true;
+ } else if let Err(e) = writer.get_ref().sync_data().await {
+ error!("AOF batch sync_data error: {}", e);
+ fsync_failed = true;
+ }
+ if fsync_failed {
+ BatchAck::FsyncFailed
+ } else {
+ BatchAck::Synced
+ }
+ } else {
+ // EverySec/No: batch buffered; the deadline check
+ // after this block fsyncs. An AppendSync is enqueued
+ // ONLY under Always (pool::try_send_append_durable /
+ // fsync_barrier), so this branch acks no waiter.
+ debug_assert!(
+ !batch
+ .data
+ .iter()
+ .any(|m| matches!(m, AofMessage::AppendSync { .. })),
+ "everysec/no batch must contain no AppendSync"
+ );
+ BatchAck::Synced
+ };
+ let _ = group_commit::ack_batch(&mut batch, verdict);
}
- match fsync {
- FsyncPolicy::Always => {
+ }
+
+ // -- handle the control message that ended the drain (if any) --
+ match batch.deferred_control {
+ None => {}
+ Some(AofMessage::Rewrite(db)) => {
+ // Flush current writer before rewrite (skip if torn).
+ if !write_error {
let _ = writer.flush().await;
let _ = writer.get_ref().sync_data().await;
}
- FsyncPolicy::EverySec | FsyncPolicy::No => {
- // EverySec handled by the deadline check after
- // the select!; No does nothing
+
+ match rewrite_aof(db, &aof_path).await {
+ // Rewrite replaced the file with a clean one — the
+ // torn-write latch resets (mirrors monoio TopLevel).
+ Ok(()) => write_error = false,
+ Err(e) => error!("AOF rewrite failed: {}", e),
}
- }
- }
- // AppendSync: write + fsync + ack, regardless of policy.
- Ok(AofMessage::AppendSync {
- bytes: data,
- lsn: _,
- ack,
- }) => {
- if let Err(e) = writer.write_all(&data).await {
- error!("AOF AppendSync write error: {}", e);
- let _ = ack.send(AofAck::WriteFailed);
- continue;
- }
- if let Err(e) = writer.flush().await {
- error!("AOF AppendSync flush error: {}", e);
- let _ = ack.send(AofAck::FsyncFailed);
- continue;
- }
- if let Err(e) = writer.get_ref().sync_data().await {
- error!("AOF AppendSync sync_data error: {}", e);
- let _ = ack.send(AofAck::FsyncFailed);
- continue;
- }
- let _ = ack.send(AofAck::Synced);
- }
- Ok(AofMessage::Rewrite(db)) => {
- // Flush current writer before rewrite
- let _ = writer.flush().await;
- let _ = writer.get_ref().sync_data().await;
+ crate::command::persistence::AOF_REWRITE_IN_PROGRESS
+ .store(false, std::sync::atomic::Ordering::SeqCst);
- if let Err(e) = rewrite_aof(db, &aof_path).await {
- error!("AOF rewrite failed: {}", e);
+ // Reopen file after rewrite (it was replaced)
+ let reopen_result: Result =
+ tokio::fs::OpenOptions::new()
+ .create(true)
+ .append(true)
+ .open(&aof_path)
+ .await;
+ match reopen_result {
+ Ok(f) => {
+ writer = tokio::io::BufWriter::new(f);
+ }
+ Err(e) => {
+ error!("Failed to reopen AOF file after rewrite: {}", e);
+ return;
+ }
+ }
+ // Back-date so the backlog drained right after the
+ // rewrite reaches disk within ~100ms + wake floor,
+ // not a full second later (mirrors the per-shard
+ // writer's post-rewrite back-dating).
+ last_fsync = Instant::now() - std::time::Duration::from_millis(900);
}
- crate::command::persistence::AOF_REWRITE_IN_PROGRESS
- .store(false, std::sync::atomic::Ordering::SeqCst);
-
- // Reopen file after rewrite (it was replaced)
- let reopen_result: Result =
- tokio::fs::OpenOptions::new()
- .create(true)
- .append(true)
- .open(&aof_path)
- .await;
- match reopen_result {
- Ok(f) => {
- writer = tokio::io::BufWriter::new(f);
+ Some(AofMessage::RewriteSharded(shard_dbs)) => {
+ // C4 TopLevel cooperative fold (tokio path):
+ // flush + sync the BufWriter (skip if torn), convert to
+ // std::fs::File for the sync fold (same pattern as tokio
+ // per-shard), then reopen for appending.
+ if !write_error {
+ let _ = writer.flush().await;
+ let _ = writer.get_ref().sync_data().await;
+ }
+ let mut sf = writer.into_inner().into_std().await;
+ match rewrite_aof_sharded_sync(
+ &shard_dbs,
+ &aof_path,
+ &rx,
+ &mut sf,
+ fold_channels.as_ref(),
+ ) {
+ // Fold rewrote aof_path clean — the latch resets.
+ Ok(()) => write_error = false,
+ Err(e) => error!("AOF rewrite (sharded) failed: {}", e),
}
- Err(e) => {
- error!("Failed to reopen AOF file after rewrite: {}", e);
- return;
+ // Drop sf — caller will reopen aof_path below.
+ drop(sf);
+ crate::command::persistence::AOF_REWRITE_IN_PROGRESS
+ .store(false, std::sync::atomic::Ordering::SeqCst);
+ let reopen_result: Result =
+ tokio::fs::OpenOptions::new()
+ .create(true)
+ .append(true)
+ .open(&aof_path)
+ .await;
+ match reopen_result {
+ Ok(f) => writer = tokio::io::BufWriter::new(f),
+ Err(e) => {
+ error!("Failed to reopen AOF after rewrite: {}", e);
+ return;
+ }
}
+ // Back-date so the channel backlog that accumulated
+ // during the blocking fold reaches disk within ~100ms
+ // + wake floor — a SIGKILL shortly after rewrite
+ // completion must not take the tail with it.
+ last_fsync = Instant::now() - std::time::Duration::from_millis(900);
}
- // Back-date so the backlog drained right after the
- // rewrite reaches disk within ~100ms + wake floor,
- // not a full second later (mirrors the per-shard
- // writer's post-rewrite back-dating).
- last_fsync = Instant::now() - std::time::Duration::from_millis(900);
- }
- Ok(AofMessage::RewriteSharded(shard_dbs)) => {
- // C4 TopLevel cooperative fold (tokio path):
- // flush + sync the BufWriter, convert to std::fs::File
- // for the sync fold (same pattern as tokio per-shard),
- // then reopen for appending.
- let _ = writer.flush().await;
- let _ = writer.get_ref().sync_data().await;
- let mut sf = writer.into_inner().into_std().await;
- if let Err(e) = rewrite_aof_sharded_sync(
- &shard_dbs,
- &aof_path,
- &rx,
- &mut sf,
- fold_channels.as_ref(),
- ) {
- error!("AOF rewrite (sharded) failed: {}", e);
+ // [F6] TopLevel writer never owns per-shard files — routing
+ // bug. Self-abort so the countdown completes + flag clears.
+ Some(AofMessage::RewritePerShard { coord, .. }) => {
+ warn!(
+ "AOF TopLevel writer received RewritePerShard — routing bug; aborting"
+ );
+ coord.mark_failed();
+ coord.shard_done();
}
- // Drop sf — caller will reopen aof_path below.
- drop(sf);
- crate::command::persistence::AOF_REWRITE_IN_PROGRESS
- .store(false, std::sync::atomic::Ordering::SeqCst);
- let reopen_result: Result =
- tokio::fs::OpenOptions::new()
- .create(true)
- .append(true)
- .open(&aof_path)
- .await;
- match reopen_result {
- Ok(f) => writer = tokio::io::BufWriter::new(f),
- Err(e) => {
- error!("Failed to reopen AOF after rewrite: {}", e);
- return;
+ Some(AofMessage::Shutdown) => {
+ if !write_error {
+ let _ = writer.flush().await;
+ let _ = writer.get_ref().sync_data().await;
}
+ info!("AOF writer shutting down");
+ break;
}
- // Back-date so the channel backlog that accumulated
- // during the blocking fold reaches disk within ~100ms
- // + wake floor — a SIGKILL shortly after rewrite
- // completion must not take the tail with it.
- last_fsync = Instant::now() - std::time::Duration::from_millis(900);
- }
- // [F6] TopLevel writer never owns per-shard files — routing
- // bug. Self-abort so the countdown completes + flag clears.
- Ok(AofMessage::RewritePerShard { coord, .. }) => {
- warn!(
- "AOF TopLevel writer received RewritePerShard — routing bug; aborting"
- );
- coord.mark_failed();
- coord.shard_done();
- }
- Ok(AofMessage::Shutdown) | Err(_) => {
- let _ = writer.flush().await;
- let _ = writer.get_ref().sync_data().await;
- info!("AOF writer shutting down");
- break;
+ // collect_group_commit_batch only ever defers a control message.
+ Some(_) => {}
}
}
}
@@ -454,8 +567,10 @@ pub async fn aof_writer_task(
// most ~1.2s after it was written (1s deadline + 200ms wake
// floor). tokio's BufWriter holds up to 8KB in userspace — a
// SIGKILL takes that tail with it, so the bound must hold even
- // when the recv arm is saturated with messages.
+ // when the recv arm is saturated with messages. Skip if torn:
+ // syncing past a partial record cannot recover it.
if fsync == FsyncPolicy::EverySec
+ && !write_error
&& last_fsync.elapsed() >= std::time::Duration::from_secs(1)
{
let _ = writer.flush().await;
@@ -641,181 +756,209 @@ pub async fn per_shard_aof_writer_task(
std::time::Duration::from_millis(200),
rx.recv_async(),
) => {
- // On Elapsed (timeout) `r` is Err: skip the match and fall
- // through to the EverySec deadline check after this select!.
- if let Ok(msg) = r {
- match msg {
- // PerShard writer (tokio): per RFC § 2 Rule 1 the on-disk
- // format is `[u64 lsn LE][u32 len LE][RESP bytes]`. Header
- // is written sequentially with the body — both calls land
- // in the same BufWriter so this is one syscall under load.
- Ok(AofMessage::Append { lsn, bytes: data }) => {
- // Latch: stream already torn — drop silently (Append
- // is fire-and-forget; no ack channel to notify).
- if write_error {
- continue;
- }
- #[cfg(test)]
- {
- test_append_ordinal += 1;
- let fail_at = TEST_FAIL_WRITE_AT
- .load(std::sync::atomic::Ordering::Relaxed);
- if fail_at != 0 && fail_at == test_append_ordinal {
- // Reproduce a torn write: header lands, payload
- // "fails". The orphaned header is flushed so the
- // on-disk effect matches the real I/O-error case.
- let mut header = [0u8; 12];
- header[..8].copy_from_slice(&lsn.to_le_bytes());
- header[8..]
- .copy_from_slice(&(data.len() as u32).to_le_bytes());
- let _ = writer.write_all(&header).await;
- let _ = writer.flush().await;
- error!(
- "AOF shard {}: injected torn write after header (test)",
- shard_id
+ // On Elapsed (timeout) `r` is Err: skip and fall through to
+ // the EverySec deadline check after this select!.
+ match r {
+ Err(_) => {}
+ // Channel disconnected — final sync + shut down.
+ Ok(Err(_)) => {
+ let _ = writer.flush().await;
+ let _ = writer.get_ref().sync_data().await;
+ info!("AOF writer shard {} shutting down", shard_id);
+ break;
+ }
+ Ok(Ok(first)) => {
+ // Group commit: drain a bounded batch so ONE fsync
+ // makes all framed records (`[u64 lsn][u32 len][RESP]`)
+ // durable.
+ let mut batch = collect_group_commit_batch(
+ first,
+ || rx.try_recv().ok(),
+ AOF_GROUP_COMMIT_MAX_BATCH,
+ AOF_GROUP_COMMIT_MAX_BYTES,
+ );
+
+ if !batch.data.is_empty() {
+ if write_error {
+ // Stream already torn — drop appends, fail every
+ // AppendSync waiter (no ack into a corrupt stream).
+ let _ = group_commit::ack_batch(
+ &mut batch,
+ BatchAck::WriteFailed,
);
- write_error = true;
- continue;
+ } else {
+ // Write each record framed, header + body, in
+ // order; the single fsync below covers them all.
+ let mut write_failed = false;
+ for msg in &batch.data {
+ let (lsn, data) = match msg {
+ AofMessage::Append { lsn, bytes }
+ | AofMessage::AppendSync { lsn, bytes, .. } => {
+ (*lsn, bytes)
+ }
+ _ => continue,
+ };
+ // H1-BARRIER: a zero-length AppendSync writes
+ // NO record (a len=0 framed header would make
+ // replay reject the file) — fsync+ack only.
+ if data.is_empty() {
+ continue;
+ }
+ #[cfg(test)]
+ {
+ if matches!(msg, AofMessage::Append { .. }) {
+ test_append_ordinal += 1;
+ let fail_at = TEST_FAIL_WRITE_AT
+ .load(std::sync::atomic::Ordering::Relaxed);
+ if fail_at != 0 && fail_at == test_append_ordinal {
+ // Reproduce a torn write: header
+ // lands, payload "fails", latch.
+ let mut header = [0u8; 12];
+ header[..8]
+ .copy_from_slice(&lsn.to_le_bytes());
+ header[8..].copy_from_slice(
+ &(data.len() as u32).to_le_bytes(),
+ );
+ let _ = writer.write_all(&header).await;
+ let _ = writer.flush().await;
+ error!(
+ "AOF shard {}: injected torn write after header (test)",
+ shard_id
+ );
+ write_failed = true;
+ break;
+ }
+ }
+ }
+ let mut header = [0u8; 12];
+ header[..8].copy_from_slice(&lsn.to_le_bytes());
+ header[8..]
+ .copy_from_slice(&(data.len() as u32).to_le_bytes());
+ if let Err(e) = writer.write_all(&header).await {
+ error!(
+ "AOF header write error shard {}: {}",
+ shard_id, e
+ );
+ write_failed = true;
+ break;
+ }
+ if let Err(e) = writer.write_all(data).await {
+ error!("AOF write error shard {}: {}", shard_id, e);
+ write_failed = true;
+ break;
+ }
+ }
+
+ let do_fsync = matches!(fsync, FsyncPolicy::Always);
+ let verdict = if write_failed {
+ // A torn write may leave a partial record —
+ // latch so no further bytes are appended.
+ write_error = true;
+ BatchAck::WriteFailed
+ } else if fail_fsync_for_test && do_fsync {
+ // Injected: bytes written, the batch fsync
+ // "fails" → every waiter FsyncFailed.
+ BatchAck::FsyncFailed
+ } else if do_fsync {
+ let mut ff = false;
+ if let Err(e) = writer.flush().await {
+ error!(
+ "AOF batch flush error shard {}: {}",
+ shard_id, e
+ );
+ ff = true;
+ } else if let Err(e) = writer.get_ref().sync_data().await {
+ error!(
+ "AOF batch sync_data error shard {}: {}",
+ shard_id, e
+ );
+ ff = true;
+ }
+ if ff {
+ BatchAck::FsyncFailed
+ } else {
+ BatchAck::Synced
+ }
+ } else {
+ // EverySec/No: the deadline check fsyncs; no
+ // AppendSync waiters under everysec/no.
+ BatchAck::Synced
+ };
+ let _ = group_commit::ack_batch(&mut batch, verdict);
}
}
- let mut header = [0u8; 12];
- header[..8].copy_from_slice(&lsn.to_le_bytes());
- header[8..].copy_from_slice(&(data.len() as u32).to_le_bytes());
- if let Err(e) = writer.write_all(&header).await {
- error!("AOF header write error shard {}: {}", shard_id, e);
- write_error = true;
- continue;
- }
- if let Err(e) = writer.write_all(&data).await {
- error!("AOF write error shard {}: {}", shard_id, e);
- write_error = true;
- continue;
- }
- if matches!(fsync, FsyncPolicy::Always) {
- let _ = writer.flush().await;
- let _ = writer.get_ref().sync_data().await;
- }
- }
- // AppendSync (tokio + PerShard): framed write + fsync + ack.
- Ok(AofMessage::AppendSync { lsn, bytes: data, ack }) => {
- // Latch: stream already torn — refuse to write more and
- // report failure so the caller does not hang to the F2
- // timeout and does not ack a write into a corrupt stream.
- if write_error {
- let _ = ack.send(AofAck::WriteFailed);
- continue;
- }
- // H1-BARRIER: a zero-length AppendSync is an fsync
- // barrier (pool::fsync_barrier) — fsync + ack only,
- // NO on-disk record. A len=0 framed header would make
- // replay_incr_framed reject the file as corrupt.
- if !data.is_empty() {
- let mut header = [0u8; 12];
- header[..8].copy_from_slice(&lsn.to_le_bytes());
- header[8..].copy_from_slice(&(data.len() as u32).to_le_bytes());
- if let Err(e) = writer.write_all(&header).await {
- error!(
- "AOF AppendSync header write error shard {}: {}",
- shard_id, e
+
+ // -- handle the control message that ended the drain --
+ match batch.deferred_control {
+ None => {}
+ Some(AofMessage::Rewrite(_))
+ | Some(AofMessage::RewriteSharded(_)) => {
+ warn!(
+ "AOF writer shard {}: received Rewrite/RewriteSharded — \
+ not applicable in PerShard layout, dropped.",
+ shard_id
);
- write_error = true;
- let _ = ack.send(AofAck::WriteFailed);
- continue;
}
- if let Err(e) = writer.write_all(&data).await {
- error!(
- "AOF AppendSync write error shard {}: {}",
- shard_id, e
- );
- write_error = true;
- let _ = ack.send(AofAck::WriteFailed);
- continue;
+ // [F6] Per-shard rewrite (tokio): reuse the proven
+ // synchronous fold (`do_rewrite_per_shard`) verbatim.
+ // This writer runs on a DEDICATED std::thread
+ // (block_on_local, main.rs) — not a shared tokio
+ // worker — so the blocking fold cannot starve the
+ // runtime. Flush the BufWriter (its `into_inner` does
+ // NOT flush) so buffered appends are durable in the
+ // OLD incr, convert to `std::fs::File` for the sync
+ // fold, then wrap the (reopened) file back.
+ Some(AofMessage::RewritePerShard {
+ shard_dbs,
+ coord,
+ fold_producer,
+ fold_notifier,
+ }) => {
+ if let Err(e) = writer.flush().await {
+ error!(
+ "F6 tokio per-shard rewrite: shard {} pre-fold flush \
+ failed: {}. Aborting; old generation stays authoritative.",
+ shard_id, e
+ );
+ coord.mark_failed();
+ coord.shard_done();
+ } else {
+ // `into_std().await` waits for in-flight ops
+ // and is infallible; buffer flushed above.
+ let mut sf = writer.into_inner().into_std().await;
+ let res = do_rewrite_per_shard(
+ shard_id, &shard_dbs, &mut sf, &rx, &coord,
+ &fold_producer, &fold_notifier,
+ );
+ // `sf` is left on the committed generation by
+ // the fold's internal barrier: NEW incr on
+ // success, OLD incr on abort/pre-reopen error.
+ // The fold's ShardDoneGuard already did
+ // `shard_done` for every exit, so do NOT
+ // decrement again. Wrap `sf` back either way.
+ writer = tokio::io::BufWriter::new(
+ tokio::fs::File::from_std(sf),
+ );
+ if let Err(e) = res {
+ error!(
+ "F6 tokio per-shard rewrite: shard {} fold failed: {}. \
+ Rewrite aborted by the fold guard; old generation \
+ stays authoritative.",
+ shard_id, e
+ );
+ }
+ }
}
- }
- // Test-only: skip real fsync and return FsyncFailed
- // immediately when the fault-injection env var is set.
- if fail_fsync_for_test {
- let _ = ack.send(AofAck::FsyncFailed);
- continue;
- }
- if let Err(e) = writer.flush().await {
- error!(
- "AOF AppendSync flush error shard {}: {}",
- shard_id, e
- );
- let _ = ack.send(AofAck::FsyncFailed);
- continue;
- }
- if let Err(e) = writer.get_ref().sync_data().await {
- error!(
- "AOF AppendSync sync_data error shard {}: {}",
- shard_id, e
- );
- let _ = ack.send(AofAck::FsyncFailed);
- continue;
- }
- let _ = ack.send(AofAck::Synced);
- }
- Ok(AofMessage::Rewrite(_)) | Ok(AofMessage::RewriteSharded(_)) => {
- warn!(
- "AOF writer shard {}: received Rewrite/RewriteSharded — \
- not applicable in PerShard layout, dropped.",
- shard_id
- );
- }
- // [F6] Per-shard rewrite (tokio): reuse the proven
- // synchronous fold (`do_rewrite_per_shard`) verbatim, so
- // the exactly-once invariant carries over unchanged. This
- // writer runs on a DEDICATED std::thread (block_on_local,
- // main.rs) — not a shared tokio worker — so executing the
- // blocking fold here cannot starve the runtime. We flush
- // the BufWriter (its `into_inner` does NOT flush) so any
- // buffered appends are durable in the OLD incr, convert
- // `tokio::fs::File` -> `std::fs::File` for the sync fold,
- // then wrap the (reopened) file back into the BufWriter.
- Ok(AofMessage::RewritePerShard { shard_dbs, coord, fold_producer, fold_notifier }) => {
- if let Err(e) = writer.flush().await {
- error!(
- "F6 tokio per-shard rewrite: shard {} pre-fold flush \
- failed: {}. Aborting; old generation stays authoritative.",
- shard_id, e
- );
- coord.mark_failed();
- coord.shard_done();
- } else {
- // `into_std().await` waits for in-flight ops and is
- // infallible; the buffer is already flushed above.
- let mut sf = writer.into_inner().into_std().await;
- let res = do_rewrite_per_shard(
- shard_id, &shard_dbs, &mut sf, &rx, &coord,
- &fold_producer, &fold_notifier,
- );
- // `sf` is left pointing at the committed generation
- // by the fold's internal barrier: NEW incr on
- // success, OLD incr on abort (phase-8 rollback) or
- // on a pre-reopen error. The fold's ShardDoneGuard
- // already performed `shard_done` for every exit, so
- // we MUST NOT decrement again here. Wrap `sf` back so
- // the writer stays valid either way.
- writer =
- tokio::io::BufWriter::new(tokio::fs::File::from_std(sf));
- if let Err(e) = res {
- error!(
- "F6 tokio per-shard rewrite: shard {} fold failed: {}. \
- Rewrite aborted by the fold guard; old generation \
- stays authoritative.",
- shard_id, e
- );
+ Some(AofMessage::Shutdown) => {
+ let _ = writer.flush().await;
+ let _ = writer.get_ref().sync_data().await;
+ info!("AOF writer shard {} shutting down", shard_id);
+ break;
}
+ // collect only ever defers a control message.
+ Some(_) => {}
}
}
- Ok(AofMessage::Shutdown) | Err(_) => {
- let _ = writer.flush().await;
- let _ = writer.get_ref().sync_data().await;
- info!("AOF writer shard {} shutting down", shard_id);
- break;
- }
- }
}
}
_ = cancel.cancelled() => {
@@ -948,209 +1091,218 @@ pub async fn per_shard_aof_writer_task(
// Appends arrive after a fold (or when the client stops writing).
// Without a timeout, the writer blocks forever in rx.recv() and
// the 1s fsync window never fires → data loss on kill.
- match rx.recv_timeout(std::time::Duration::from_millis(50)) {
- // AppendSync (monoio + PerShard): framed write + fsync + ack.
- Ok(AofMessage::AppendSync {
- lsn,
- bytes: data,
- ack,
- }) => {
- if write_error {
- let _ = ack.send(AofAck::WriteFailed);
- continue;
- }
- // H1-BARRIER: a zero-length AppendSync is an fsync barrier
- // (pool::fsync_barrier) — fsync + ack only, NO on-disk
- // record. A len=0 framed header would make
- // replay_incr_framed reject the file as corrupt.
- if !data.is_empty() {
- let mut header = [0u8; 12];
- header[..8].copy_from_slice(&lsn.to_le_bytes());
- header[8..].copy_from_slice(&(data.len() as u32).to_le_bytes());
- if let Err(e) = file.write_all(&header) {
- error!(
- "AOF AppendSync header write failed shard {} (seq {}): {}",
- shard_id, manifest.seq, e
- );
- write_error = true;
- let _ = ack.send(AofAck::WriteFailed);
- continue;
- }
- if let Err(e) = file.write_all(&data) {
+ // recv_timeout so the EverySec proactive fsync fires even when no new
+ // Appends arrive after a fold (or when the client stops writing).
+ let first = match rx.recv_timeout(std::time::Duration::from_millis(50)) {
+ Ok(m) => Some(m),
+ // Timeout: no message in the 50ms window. Fall through (None) to
+ // the EverySec proactive fsync below so queued-but-unfsynced
+ // appends are durable within the everysec contract even when idle.
+ Err(flume::RecvTimeoutError::Timeout) => None,
+ Err(flume::RecvTimeoutError::Disconnected) => {
+ if !write_error {
+ if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
error!(
- "AOF AppendSync write failed shard {} (seq {}): {}",
+ "AOF final sync failed shard {} (seq {}): {}",
shard_id, manifest.seq, e
);
- write_error = true;
- let _ = ack.send(AofAck::WriteFailed);
- continue;
}
}
- // Test-only: skip real fsync and return FsyncFailed
- // immediately when the fault-injection env var is set.
- if fail_fsync_for_test {
- let _ = ack.send(AofAck::FsyncFailed);
- continue;
- }
- let t = Instant::now();
- if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
- error!(
- "AOF AppendSync sync failed shard {} (seq {}): {}",
- shard_id, manifest.seq, e
- );
- write_error = true;
- let _ = ack.send(AofAck::FsyncFailed);
- } else {
- crate::admin::metrics_setup::record_aof_fsync(
- t.elapsed().as_micros() as u64
- );
- let _ = ack.send(AofAck::Synced);
- }
+ info!(
+ "AOF writer shard {} shutting down (monoio, seq {}, processed {} appends in {:.3}s)",
+ shard_id,
+ manifest.seq,
+ _dbg_processed,
+ _dbg_start.elapsed().as_secs_f64()
+ );
+ break;
}
- // PerShard writer (monoio): framed `[u64 lsn LE][u32 len LE][RESP]`.
- // See the tokio twin above for format rationale.
- Ok(AofMessage::Append { lsn, bytes: data }) => {
+ };
+
+ if let Some(first) = first {
+ // Group commit: drain a bounded batch so ONE fsync makes all
+ // framed records (`[u64 lsn][u32 len][RESP]`) durable.
+ let mut batch = collect_group_commit_batch(
+ first,
+ || rx.try_recv().ok(),
+ AOF_GROUP_COMMIT_MAX_BATCH,
+ AOF_GROUP_COMMIT_MAX_BYTES,
+ );
+
+ if !batch.data.is_empty() {
if write_error {
- continue;
- }
- _dbg_processed += 1;
- let mut header = [0u8; 12];
- header[..8].copy_from_slice(&lsn.to_le_bytes());
- header[8..].copy_from_slice(&(data.len() as u32).to_le_bytes());
- if let Err(e) = file.write_all(&header) {
- error!(
- "AOF header write failed shard {} (seq {}): {}. Persistence degraded.",
- shard_id, manifest.seq, e
- );
- write_error = true;
- continue;
+ // Stream already torn — drop appends, fail every AppendSync
+ // waiter so callers error instead of acking a corrupt write.
+ let _ = group_commit::ack_batch(&mut batch, BatchAck::WriteFailed);
+ } else {
+ // Write each record framed, header + body, in channel order;
+ // the single fsync below covers them all.
+ let mut write_failed = false;
+ for msg in &batch.data {
+ let (lsn, data) = match msg {
+ AofMessage::Append { lsn, bytes }
+ | AofMessage::AppendSync { lsn, bytes, .. } => (*lsn, bytes),
+ _ => continue,
+ };
+ // H1-BARRIER: a zero-length AppendSync writes NO record
+ // (a len=0 framed header would make replay reject the
+ // file) — fsync + ack only.
+ if data.is_empty() {
+ continue;
+ }
+ _dbg_processed += 1;
+ let mut header = [0u8; 12];
+ header[..8].copy_from_slice(&lsn.to_le_bytes());
+ header[8..].copy_from_slice(&(data.len() as u32).to_le_bytes());
+ if let Err(e) = file.write_all(&header) {
+ error!(
+ "AOF header write failed shard {} (seq {}): {}. Persistence degraded.",
+ shard_id, manifest.seq, e
+ );
+ write_failed = true;
+ break;
+ }
+ if let Err(e) = file.write_all(data) {
+ error!(
+ "AOF write failed shard {} (seq {}): {}. Persistence degraded.",
+ shard_id, manifest.seq, e
+ );
+ write_failed = true;
+ break;
+ }
+ }
+
+ let do_fsync = matches!(fsync, FsyncPolicy::Always);
+ let verdict = if write_failed {
+ // A torn write may leave a partial record — latch so no
+ // further bytes are appended after the tear.
+ write_error = true;
+ BatchAck::WriteFailed
+ } else if fail_fsync_for_test && do_fsync {
+ // Injected: bytes written, the batch fsync "fails".
+ BatchAck::FsyncFailed
+ } else if do_fsync {
+ let t = Instant::now();
+ if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
+ error!(
+ "AOF batch sync failed shard {} (seq {}, always): {}",
+ shard_id, manifest.seq, e
+ );
+ BatchAck::FsyncFailed
+ } else {
+ crate::admin::metrics_setup::record_aof_fsync(
+ t.elapsed().as_micros() as u64,
+ );
+ BatchAck::Synced
+ }
+ } else {
+ // EverySec/No: the proactive fsync below makes the batch
+ // durable; no AppendSync waiters under everysec/no.
+ BatchAck::Synced
+ };
+ let _ = group_commit::ack_batch(&mut batch, verdict);
}
- if let Err(e) = file.write_all(&data) {
- error!(
- "AOF write failed shard {} (seq {}): {}. Persistence degraded.",
- shard_id, manifest.seq, e
+ }
+
+ // -- handle the control message that ended the drain (if any) --
+ match batch.deferred_control {
+ None => {}
+ Some(AofMessage::Rewrite(_)) | Some(AofMessage::RewriteSharded(_)) => {
+ warn!(
+ "AOF writer shard {}: received Rewrite/RewriteSharded — \
+ not applicable in PerShard layout (use per-shard \
+ BGREWRITEAOF), dropped.",
+ shard_id
);
- write_error = true;
- continue;
}
- // Always policy fsyncs inline. EverySec and No policies
- // rely on the proactive fsync check AFTER the match block,
- // which runs after every message OR timeout. Keeping EverySec
- // out of the Append arm prevents it from advancing last_fsync
- // after a fold completes, which would delay the next proactive
- // fsync by a full second and open the EverySec durability window.
- if fsync == FsyncPolicy::Always {
- let t = Instant::now();
- if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
+ // [F6] Per-shard rewrite fan-out (monoio). Fold THIS shard,
+ // then signal the coordinator; the last shard commits the
+ // manifest. On error the old generation stays authoritative
+ // (advance_shard did not commit the seq).
+ Some(AofMessage::RewritePerShard {
+ shard_dbs,
+ coord,
+ fold_producer,
+ fold_notifier,
+ }) => {
+ if let Err(e) = do_rewrite_per_shard(
+ shard_id,
+ &shard_dbs,
+ &mut file,
+ &rx,
+ &coord,
+ &fold_producer,
+ &fold_notifier,
+ ) {
+ // The fold's ShardDoneGuard already marked the rewrite
+ // failed and decremented on this error exit (committing
+ // new_seq with a shard missing its new base would break
+ // recovery), so do NOT decrement again here. `file` is left
+ // on the OLD incr (error exits are pre-reopen).
error!(
- "AOF sync failed shard {} (seq {}, always): {}",
- shard_id, manifest.seq, e
- );
- write_error = true;
- } else {
- crate::admin::metrics_setup::record_aof_fsync(
- t.elapsed().as_micros() as u64
+ "F6 per-shard rewrite: shard {} fold failed: {}. \
+ Rewrite aborted by the fold guard; old generation \
+ stays authoritative.",
+ shard_id, e
);
}
+ // EverySec post-fold drain+fsync: the fold runs synchronously
+ // and does NOT update `last_fsync`. Appends that arrived during
+ // the fold queue in the bounded AOF channel; they land on the NEW
+ // incr but are NOT fsynced until the EverySec timer fires.
+ // Strategy: drain what's currently in the channel and fsync now
+ // (covers appends that landed before this drain), then set
+ // `last_fsync` 900ms in the past so the proactive check below
+ // fires within the NEXT 100ms window (≤150ms total, since the
+ // recv_timeout is 50ms). That second fsync covers any appends
+ // that arrived between the drain and that window close.
+ // Combined, the two fsyncs bound the post-fold EverySec window
+ // to ≤150ms — well within the test's 1500ms kill margin.
+ if !write_error {
+ if let Ok(mut post_drain) =
+ drain_pending_appends_framed(&rx, &mut file, usize::MAX)
+ {
+ if let Err(e) = sync_and_fulfill_drain(
+ &mut post_drain,
+ &mut file,
+ std::path::PathBuf::from(""),
+ ) {
+ error!(
+ "F6 per-shard rewrite: shard {} post-fold fsync \
+ failed: {}. EverySec window open until next Append.",
+ shard_id, e
+ );
+ } else {
+ // Back-date last_fsync by 900ms: the proactive check
+ // (threshold=1s) fires within the next 100ms, covering
+ // any appends that arrived after the drain above.
+ last_fsync =
+ Instant::now() - std::time::Duration::from_millis(900);
+ }
+ }
+ }
}
- }
- Ok(AofMessage::Rewrite(_)) | Ok(AofMessage::RewriteSharded(_)) => {
- warn!(
- "AOF writer shard {}: received Rewrite/RewriteSharded — \
- not applicable in PerShard layout (use per-shard \
- BGREWRITEAOF), dropped.",
- shard_id
- );
- }
- // [F6] Per-shard rewrite fan-out (monoio). Fold THIS shard,
- // then signal the coordinator; the last shard commits the
- // manifest. On error the old generation stays authoritative
- // (advance_shard did not commit the seq).
- Ok(AofMessage::RewritePerShard {
- shard_dbs,
- coord,
- fold_producer,
- fold_notifier,
- }) => {
- if let Err(e) = do_rewrite_per_shard(
- shard_id,
- &shard_dbs,
- &mut file,
- &rx,
- &coord,
- &fold_producer,
- &fold_notifier,
- ) {
- // The fold's ShardDoneGuard already marked the rewrite
- // failed and decremented on this error exit (committing
- // new_seq with a shard missing its new base would break
- // recovery), so do NOT decrement again here. `file` is left
- // on the OLD incr (error exits are pre-reopen).
- error!(
- "F6 per-shard rewrite: shard {} fold failed: {}. \
- Rewrite aborted by the fold guard; old generation \
- stays authoritative.",
- shard_id, e
- );
- }
- // EverySec post-fold drain+fsync: the fold runs synchronously
- // and does NOT update `last_fsync`. Appends that arrived during
- // the fold queue in the bounded AOF channel; they land on the NEW
- // incr but are NOT fsynced until the EverySec timer fires.
- // Strategy: drain what's currently in the channel and fsync now
- // (covers appends that landed before this drain), then set
- // `last_fsync` 900ms in the past so the proactive check below
- // fires within the NEXT 100ms window (≤150ms total, since the
- // recv_timeout is 50ms). That second fsync covers any appends
- // that arrived between the drain and that window close.
- // Combined, the two fsyncs bound the post-fold EverySec window
- // to ≤150ms — well within the test's 1500ms kill margin.
- if !write_error {
- if let Ok(mut post_drain) =
- drain_pending_appends_framed(&rx, &mut file, usize::MAX)
- {
- if let Err(e) = sync_and_fulfill_drain(
- &mut post_drain,
- &mut file,
- std::path::PathBuf::from(""),
- ) {
+ Some(AofMessage::Shutdown) => {
+ if !write_error {
+ if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
error!(
- "F6 per-shard rewrite: shard {} post-fold fsync \
- failed: {}. EverySec window open until next Append.",
- shard_id, e
+ "AOF final sync failed shard {} (seq {}): {}",
+ shard_id, manifest.seq, e
);
- } else {
- // Back-date last_fsync by 900ms: the proactive check
- // (threshold=1s) fires within the next 100ms, covering
- // any appends that arrived after the drain above.
- last_fsync = Instant::now() - std::time::Duration::from_millis(900);
}
}
+ info!(
+ "AOF writer shard {} shutting down (monoio, seq {}, processed {} appends in {:.3}s)",
+ shard_id,
+ manifest.seq,
+ _dbg_processed,
+ _dbg_start.elapsed().as_secs_f64()
+ );
+ break;
}
+ // collect only ever defers a control message.
+ Some(_) => {}
}
- Ok(AofMessage::Shutdown) | Err(flume::RecvTimeoutError::Disconnected) => {
- if !write_error {
- if let Err(e) = file.flush().and_then(|_| file.sync_data()) {
- error!(
- "AOF final sync failed shard {} (seq {}): {}",
- shard_id, manifest.seq, e
- );
- }
- }
- info!(
- "AOF writer shard {} shutting down (monoio, seq {}, processed {} appends in {:.3}s)",
- shard_id,
- manifest.seq,
- _dbg_processed,
- _dbg_start.elapsed().as_secs_f64()
- );
- break;
- }
- // Timeout: no message in the 50ms window. Fall through to
- // the EverySec proactive fsync below so queued-but-unfsynced
- // appends (e.g. after a fold with no new writes) are durable
- // within the everysec contract even if the client goes quiet.
- Err(flume::RecvTimeoutError::Timeout) => {}
}
// EverySec proactive fsync — runs after every loop iteration
// (message processed OR timeout). This is the only path that
diff --git a/tests/wal_group_commit.rs b/tests/wal_group_commit.rs
new file mode 100644
index 000000000..3290eab40
--- /dev/null
+++ b/tests/wal_group_commit.rs
@@ -0,0 +1,655 @@
+//! ADD task `wal-group-commit` §4 TESTS — failing-first (RED) suite for the
+//! group-commit batching seam frozen in §3.
+//!
+//! These are RED until §5 BUILD creates `src/persistence/aof/group_commit.rs` with:
+//! - const `AOF_GROUP_COMMIT_MAX_BATCH` / `AOF_GROUP_COMMIT_MAX_BYTES`
+//! - `struct GroupCommitBatch { data, deferred_control }`
+//! - `fn collect_group_commit_batch(first, try_next, max_batch, max_bytes) -> GroupCommitBatch`
+//! - `trait GroupCommitSink { write_all; sync }` (the sync-capable realization of the frozen
+//! `` + "flush()+sync_data()" — non-behavioral: the durability behavior is unchanged)
+//! - `struct CommitOutcome { synced, write_failed, fsync_failed }`
+//! - `fn commit_group_commit_batch(sink, batch, do_fsync) -> CommitOutcome`
+//! Before build the `use` below is UNRESOLVED → this crate fails to compile. That compile
+//! failure IS the red signal (the other test crates build independently).
+//!
+//! These pin the PURE seam (collect) + the durability invariant (commit: one fsync, ack
+//! AFTER fsync, the 5 reject mappings). The end-to-end durability scenarios
+//! (every_acked_write_survives_crash, rewrite-during-writes) are proven by the integration
+//! crash-matrix (`crash_matrix_per_shard_aof.rs` + a new concurrent-writers SIGKILL test added
+//! at build), NOT here — a unit mock cannot prove on-disk survival across a real kill.
+//!
+//! Running: cargo test --test wal_group_commit
+
+use bytes::Bytes;
+use moon::persistence::aof::group_commit::{
+ AOF_GROUP_COMMIT_MAX_BATCH, AOF_GROUP_COMMIT_MAX_BYTES, CommitOutcome, GroupCommitBatch,
+ GroupCommitSink, collect_group_commit_batch, commit_group_commit_batch,
+};
+use moon::persistence::aof::{AofAck, AofMessage};
+use moon::runtime::channel::{OneshotReceiver, oneshot};
+use std::collections::VecDeque;
+
+// --- helpers ---------------------------------------------------------------
+
+fn append(data: &[u8]) -> AofMessage {
+ AofMessage::Append {
+ lsn: 0,
+ bytes: Bytes::copy_from_slice(data),
+ }
+}
+
+fn append_sync(data: &[u8]) -> (AofMessage, OneshotReceiver) {
+ let (tx, rx) = oneshot::();
+ (
+ AofMessage::AppendSync {
+ lsn: 0,
+ bytes: Bytes::copy_from_slice(data),
+ ack: tx,
+ },
+ rx,
+ )
+}
+
+/// A `GroupCommitSink` that records writes + counts fsyncs, with optional fault
+/// injection. `synced_at_write_count` proves the single fsync happens AFTER all
+/// writes (ack-after-fsync ordering).
+struct CountingSink {
+ writes: Vec>,
+ sync_calls: usize,
+ synced_at_write_count: Option,
+ fail_write_at: Option,
+ fail_sync: bool,
+}
+
+impl CountingSink {
+ fn new() -> Self {
+ CountingSink {
+ writes: Vec::new(),
+ sync_calls: 0,
+ synced_at_write_count: None,
+ fail_write_at: None,
+ fail_sync: false,
+ }
+ }
+}
+
+impl GroupCommitSink for CountingSink {
+ fn write_all(&mut self, buf: &[u8]) -> std::io::Result<()> {
+ if Some(self.writes.len()) == self.fail_write_at {
+ return Err(std::io::Error::other("injected write failure"));
+ }
+ self.writes.push(buf.to_vec());
+ Ok(())
+ }
+
+ fn sync(&mut self) -> std::io::Result<()> {
+ self.sync_calls += 1;
+ self.synced_at_write_count = Some(self.writes.len());
+ if self.fail_sync {
+ return Err(std::io::Error::other("injected sync failure"));
+ }
+ Ok(())
+ }
+}
+
+// --- collect: the pure batching seam --------------------------------------
+
+/// batch_straddles_control — a control message FIRST yields an empty batch with
+/// the control deferred (the writer handles it directly, no batch).
+#[test]
+fn collect_first_control_makes_no_batch() {
+ let batch = collect_group_commit_batch(
+ AofMessage::Shutdown,
+ || None,
+ AOF_GROUP_COMMIT_MAX_BATCH,
+ AOF_GROUP_COMMIT_MAX_BYTES,
+ );
+ assert!(
+ batch.data.is_empty(),
+ "a control message as `first` must not open a data batch"
+ );
+ assert!(
+ matches!(batch.deferred_control, Some(AofMessage::Shutdown)),
+ "the control message is deferred for the caller to handle after commit"
+ );
+}
+
+/// M3 / batch_straddles_control — the drain stops AT a control message,
+/// preserving order, and never pulls data queued after the control.
+#[test]
+fn collect_stops_at_control_preserving_order() {
+ let mut q: VecDeque = VecDeque::new();
+ let (sync_msg, _rx) = append_sync(b"second");
+ q.push_back(sync_msg);
+ q.push_back(AofMessage::Shutdown);
+ q.push_back(append(b"after-control")); // must remain queued
+
+ let batch = collect_group_commit_batch(
+ append(b"first"),
+ || q.pop_front(),
+ AOF_GROUP_COMMIT_MAX_BATCH,
+ AOF_GROUP_COMMIT_MAX_BYTES,
+ );
+
+ assert_eq!(
+ batch.data.len(),
+ 2,
+ "first(Append) + the AppendSync, then STOP at Shutdown"
+ );
+ assert!(matches!(batch.deferred_control, Some(AofMessage::Shutdown)));
+ assert_eq!(
+ q.len(),
+ 1,
+ "the data message after the control must NOT be consumed"
+ );
+}
+
+/// batch_cap_exceeded (count) — the drain stops at max_batch; the rest stay
+/// queued for the next batch (never an unbounded drain).
+#[test]
+fn collect_respects_max_batch() {
+ let mut q: VecDeque = VecDeque::new();
+ for _ in 0..10 {
+ q.push_back(append(b"x"));
+ }
+ let batch = collect_group_commit_batch(
+ append(b"first"),
+ || q.pop_front(),
+ 3,
+ AOF_GROUP_COMMIT_MAX_BYTES,
+ );
+ assert_eq!(
+ batch.data.len(),
+ 3,
+ "max_batch=3 bounds the batch to first + 2 drained"
+ );
+ assert!(batch.deferred_control.is_none());
+ assert_eq!(q.len(), 8, "the remaining 8 stay queued for the next batch");
+}
+
+/// batch_cap_exceeded (bytes) — a soft byte cap stops the drain once the
+/// accumulated payload reaches max_bytes (a message is never split).
+#[test]
+fn collect_respects_max_bytes() {
+ let mut q: VecDeque = VecDeque::new();
+ for _ in 0..10 {
+ q.push_back(append(&[0u8; 100]));
+ }
+ // first=100 → +100=200 (<250) → +100=300 (≥250, include then stop) ⇒ 3 messages.
+ let batch = collect_group_commit_batch(
+ append(&[0u8; 100]),
+ || q.pop_front(),
+ AOF_GROUP_COMMIT_MAX_BATCH,
+ 250,
+ );
+ assert_eq!(
+ batch.data.len(),
+ 3,
+ "byte cap stops the drain once accumulated ≥ max_bytes (whole-message granularity)"
+ );
+}
+
+// --- commit: the durability invariant -------------------------------------
+
+/// M1 — K queued AppendSync writes are made durable by exactly ONE fsync, and
+/// every waiter is acked Synced; the fsync happens AFTER all K writes.
+#[test]
+fn commit_one_fsync_many_acks() {
+ let (m1, rx1) = append_sync(b"a");
+ let (m2, rx2) = append_sync(b"b");
+ let (m3, rx3) = append_sync(b"c");
+ let mut batch = GroupCommitBatch {
+ data: vec![m1, m2, m3],
+ deferred_control: None,
+ };
+ let mut sink = CountingSink::new();
+
+ let outcome = commit_group_commit_batch(&mut sink, &mut batch, true);
+
+ assert_eq!(sink.sync_calls, 1, "exactly ONE fsync for the whole batch");
+ assert_eq!(sink.writes.len(), 3, "all three payloads written");
+ assert_eq!(
+ sink.synced_at_write_count,
+ Some(3),
+ "the fsync runs AFTER all writes (ack-after-fsync ordering)"
+ );
+ assert_eq!(
+ outcome,
+ CommitOutcome {
+ synced: 3,
+ write_failed: false,
+ fsync_failed: false
+ }
+ );
+ assert_eq!(rx1.try_recv(), Ok(AofAck::Synced));
+ assert_eq!(rx2.try_recv(), Ok(AofAck::Synced));
+ assert_eq!(rx3.try_recv(), Ok(AofAck::Synced));
+}
+
+/// batch_write_failed — a write_all failure mid-batch acks WriteFailed and NO
+/// waiter in the batch is acked Synced (no false durability claim).
+#[test]
+fn commit_write_fail_acks_write_failed() {
+ let (m1, rx1) = append_sync(b"a");
+ let (m2, rx2) = append_sync(b"b");
+ let mut batch = GroupCommitBatch {
+ data: vec![m1, m2],
+ deferred_control: None,
+ };
+ let mut sink = CountingSink::new();
+ sink.fail_write_at = Some(1); // the 2nd write fails
+
+ let outcome = commit_group_commit_batch(&mut sink, &mut batch, true);
+
+ assert!(
+ outcome.write_failed,
+ "the batch must report a write failure"
+ );
+ assert_eq!(
+ outcome.synced, 0,
+ "no waiter may be acked Synced on a write failure"
+ );
+ // The ack oneshot is single-use (flume bounded-1): read each receiver ONCE
+ // into a local, then assert on the captured value. (The original suite read
+ // rx2.try_recv() twice, draining it before the second assertion — a
+ // use-after-consume defect that no correct impl could satisfy.)
+ let r1 = rx1.try_recv();
+ let r2 = rx2.try_recv();
+ assert_ne!(r1, Ok(AofAck::Synced));
+ assert_ne!(r2, Ok(AofAck::Synced));
+ assert_eq!(r2, Ok(AofAck::WriteFailed));
+}
+
+/// batch_fsync_failed (+ ack_before_fsync) — when the single fsync fails, ALL
+/// AppendSync waiters are acked FsyncFailed and NONE Synced. That none are
+/// Synced proves no ack was emitted before the fsync result was known.
+#[test]
+fn commit_fsync_fail_acks_fsync_failed() {
+ let (m1, rx1) = append_sync(b"a");
+ let (m2, rx2) = append_sync(b"b");
+ let mut batch = GroupCommitBatch {
+ data: vec![m1, m2],
+ deferred_control: None,
+ };
+ let mut sink = CountingSink::new();
+ sink.fail_sync = true;
+
+ let outcome = commit_group_commit_batch(&mut sink, &mut batch, true);
+
+ assert!(
+ outcome.fsync_failed,
+ "the batch must report an fsync failure"
+ );
+ assert_eq!(
+ outcome.synced, 0,
+ "no waiter may be acked Synced when the fsync fails"
+ );
+ assert_eq!(rx1.try_recv(), Ok(AofAck::FsyncFailed));
+ assert_eq!(rx2.try_recv(), Ok(AofAck::FsyncFailed));
+}
+
+/// M4 everysec_path_unchanged — under do_fsync=false the commit performs NO
+/// fsync; an Append-only batch is written in order and acks nothing.
+#[test]
+fn commit_everysec_no_fsync() {
+ let mut batch = GroupCommitBatch {
+ data: vec![append(b"x"), append(b"y")],
+ deferred_control: None,
+ };
+ let mut sink = CountingSink::new();
+
+ let outcome = commit_group_commit_batch(&mut sink, &mut batch, false);
+
+ assert_eq!(
+ sink.sync_calls, 0,
+ "everysec (do_fsync=false) must not fsync per batch"
+ );
+ assert_eq!(sink.writes.len(), 2, "bytes are still written in order");
+ assert_eq!(
+ outcome.synced, 0,
+ "no AppendSync waiters in an everysec batch"
+ );
+}
+
+/// M2 barrier_property_preserved — a single batch fsync makes preceding
+/// fire-and-forget Append bytes durable too (ordered-channel H1-BARRIER): the
+/// fsync runs after the appends are written, and the trailing AppendSync barrier
+/// is acked Synced.
+#[test]
+fn commit_barrier_covers_preceding_appends() {
+ let (barrier, rx) = append_sync(b""); // zero-length AppendSync barrier
+ let mut batch = GroupCommitBatch {
+ data: vec![append(b"one"), append(b"two"), barrier],
+ deferred_control: None,
+ };
+ let mut sink = CountingSink::new();
+
+ let outcome = commit_group_commit_batch(&mut sink, &mut batch, true);
+
+ assert_eq!(
+ sink.sync_calls, 1,
+ "one fsync covers the appends + the barrier"
+ );
+ assert_eq!(
+ sink.synced_at_write_count,
+ Some(3),
+ "the fsync runs after the two appends AND the barrier are written"
+ );
+ assert_eq!(outcome.synced, 1, "the barrier AppendSync is acked");
+ assert_eq!(rx.try_recv(), Ok(AofAck::Synced));
+}
+
+// ===========================================================================
+// INTEGRATION — real server, SIGKILL, replay (the §4 durability scenarios).
+//
+// A unit mock cannot prove on-disk survival across a real kill, so these spawn
+// the release binary under `appendfsync=always` and drive CONCURRENT durable
+// writers (the cell group commit lives in), then assert every acked write
+// survives a SIGKILL and nothing is double-applied on replay.
+//
+// `#[ignore]` — require the release binary at ./target/release/moon + redis-cli
+// on PATH. Run (monoio default, matches CI release build):
+// cargo build --release
+// cargo test --release --test wal_group_commit -- --ignored
+// Tokio runtime:
+// cargo build --release --no-default-features \
+// --features runtime-tokio,jemalloc,graph,text-index
+// cargo test --release --no-default-features \
+// --features runtime-tokio,jemalloc,graph,text-index \
+// --test wal_group_commit -- --ignored
+// ===========================================================================
+#[cfg(unix)]
+mod integration {
+ use std::io::{BufRead, BufReader, Write};
+ use std::net::TcpStream;
+ use std::process::{Child, Command, Stdio};
+ use std::time::Duration;
+
+ fn unique_port() -> u16 {
+ use std::net::TcpListener;
+ let l = TcpListener::bind("127.0.0.1:0").expect("bind port 0");
+ let p = l.local_addr().expect("local addr").port();
+ drop(l);
+ p
+ }
+
+ fn unique_dir(suffix: &str) -> std::path::PathBuf {
+ let nanos = std::time::SystemTime::now()
+ .duration_since(std::time::UNIX_EPOCH)
+ .map(|d| d.as_nanos())
+ .unwrap_or(0);
+ std::env::temp_dir().join(format!(
+ "moon-wal-gc-{}-{}-{}",
+ std::process::id(),
+ suffix,
+ nanos
+ ))
+ }
+
+ fn start_moon(port: u16, dir: &std::path::Path, shards: u16, fsync: &str) -> Child {
+ Command::new("./target/release/moon")
+ .args([
+ "--port",
+ &port.to_string(),
+ "--shards",
+ &shards.to_string(),
+ "--appendonly",
+ "yes",
+ "--appendfsync",
+ fsync,
+ "--dir",
+ ])
+ .arg(dir)
+ .stdout(std::fs::File::create(dir.join("moon.stdout.log")).expect("create stdout log"))
+ .stderr(std::fs::File::create(dir.join("moon.stderr.log")).expect("create stderr log"))
+ .spawn()
+ .expect("spawn moon — run `cargo build --release` first")
+ }
+
+ fn wait_for_port(port: u16) {
+ for _ in 0..80 {
+ if TcpStream::connect(format!("127.0.0.1:{}", port)).is_ok() {
+ std::thread::sleep(Duration::from_millis(200));
+ return;
+ }
+ std::thread::sleep(Duration::from_millis(100));
+ }
+ panic!("moon did not start within 8s on port {}", port);
+ }
+
+ fn sigkill(child: &mut Child) {
+ let pid = child.id() as i32;
+ // SAFETY: `pid` is this test's own freshly-spawned child; SIGKILL has no
+ // userspace side effects beyond terminating it. We then reap it.
+ unsafe {
+ libc::kill(pid, libc::SIGKILL);
+ }
+ let _ = child.wait();
+ }
+
+ /// Read one RESP reply line and return it trimmed (e.g. ":42", "+OK", "-ERR ...").
+ fn read_reply_line(reader: &mut BufReader) -> Option {
+ let mut line = String::new();
+ match reader.read_line(&mut line) {
+ Ok(0) => None,
+ Ok(_) => Some(line.trim_end_matches("\r\n").to_string()),
+ Err(_) => None,
+ }
+ }
+
+ /// Open a connection and issue `n` sequential `INCR key` commands, each one
+ /// sent only AFTER the previous reply is read (so under `appendfsync=always`
+ /// every counted reply observed an fsync). Returns the count of integer
+ /// replies received — the number of ACKED increments for this key.
+ fn incr_acked(port: u16, key: &str, n: usize) -> usize {
+ let stream = match TcpStream::connect(format!("127.0.0.1:{}", port)) {
+ Ok(s) => s,
+ Err(_) => return 0,
+ };
+ stream.set_read_timeout(Some(Duration::from_secs(10))).ok();
+ let mut writer = stream.try_clone().expect("clone stream");
+ let mut reader = BufReader::new(stream);
+ let cmd = format!("*2\r\n$4\r\nINCR\r\n${}\r\n{}\r\n", key.len(), key);
+ let mut acked = 0usize;
+ for _ in 0..n {
+ if writer.write_all(cmd.as_bytes()).is_err() {
+ break;
+ }
+ if writer.flush().is_err() {
+ break;
+ }
+ match read_reply_line(&mut reader) {
+ Some(l) if l.starts_with(':') => acked += 1,
+ _ => break, // error / disconnect — stop counting at the first non-ack
+ }
+ }
+ acked
+ }
+
+ /// GET an integer counter; returns its value or -1 on miss/parse failure.
+ fn get_int(port: u16, key: &str) -> i64 {
+ let out = Command::new("redis-cli")
+ .args(["-p", &port.to_string(), "GET", key])
+ .stdout(Stdio::piped())
+ .stderr(Stdio::piped())
+ .output()
+ .expect("redis-cli GET");
+ String::from_utf8_lossy(&out.stdout)
+ .trim()
+ .parse()
+ .unwrap_or(-1)
+ }
+
+ /// M2 every_acked_write_survives_crash — N concurrent connections each drive
+ /// M durable INCRs on a distinct counter under `appendfsync=always` (so the
+ /// writer coalesces concurrent AppendSyncs into group-committed fsyncs). All
+ /// threads are joined (every counted INCR is acked ⇒ fsynced) BEFORE the
+ /// SIGKILL, so recovery MUST show each counter == its acked count — every
+ /// acked write durable, none lost, none double-applied (INCR is non-idempotent).
+ fn concurrent_writers_survive_sigkill(shards: u16, tag: &str) {
+ const WRITERS: usize = 16;
+ const PER_WRITER: usize = 60;
+
+ let port = unique_port();
+ let dir = unique_dir(tag);
+ std::fs::create_dir_all(&dir).expect("create test dir");
+
+ let mut child = start_moon(port, &dir, shards, "always");
+ wait_for_port(port);
+
+ // Fan out WRITERS concurrent durable-write loops; collect per-key acked counts.
+ let handles: Vec<_> = (0..WRITERS)
+ .map(|i| {
+ let key = format!("gc:{{{}}}:{}", i, i);
+ std::thread::spawn(move || (key.clone(), incr_acked(port, &key, PER_WRITER)))
+ })
+ .collect();
+ let acked: Vec<(String, usize)> = handles
+ .into_iter()
+ .map(|h| h.join().expect("writer thread"))
+ .collect();
+
+ // Every counted INCR was acked ⇒ fsynced under Always. Kill WITHOUT a
+ // quiescing sleep: the durability contract is that each ack already saw disk.
+ sigkill(&mut child);
+
+ // -- recover --
+ let mut child2 = start_moon(port, &dir, shards, "always");
+ wait_for_port(port);
+
+ let mut wrong: Vec = Vec::new();
+ for (key, want) in &acked {
+ let got = get_int(port, key);
+ if got != *want as i64 {
+ wrong.push(format!("{}: want={} got={}", key, want, got));
+ }
+ }
+ sigkill(&mut child2);
+
+ assert!(
+ wrong.is_empty(),
+ "group-commit durability ({} shards): {} counters wrong after SIGKILL+recovery. \
+ A loss under-counts, a double-apply over-counts. Sample: {:?}",
+ shards,
+ wrong.len(),
+ wrong.iter().take(8).collect::>(),
+ );
+ let _ = std::fs::remove_dir_all(&dir);
+ }
+
+ /// M2 — TopLevel path (shards=1): all durable writes converge on ONE writer
+ /// (the `commit_group_commit_batch` path), maximizing batch coalescing.
+ #[test]
+ #[ignore]
+ fn concurrent_writers_all_acked_survive_sigkill_top_level() {
+ concurrent_writers_survive_sigkill(1, "concurrent-s1");
+ }
+
+ /// M2 — PerShard path (shards=4): writes spread across per-shard framed
+ /// writers; recovery merges. Exercises the production default layout.
+ #[test]
+ #[ignore]
+ fn concurrent_writers_all_acked_survive_sigkill_per_shard() {
+ concurrent_writers_survive_sigkill(4, "concurrent-s4");
+ }
+
+ /// M4 lone_writer_no_added_latency — with exactly ONE writer (C=1) every
+ /// durable write is a batch of 1: it fsyncs immediately and is acked Synced.
+ /// A SIGKILL with no quiescing sleep must leave every acked INCR on disk.
+ #[test]
+ #[ignore]
+ fn lone_writer_fsyncs_immediately() {
+ let port = unique_port();
+ let dir = unique_dir("lone");
+ std::fs::create_dir_all(&dir).expect("create test dir");
+
+ let mut child = start_moon(port, &dir, 1, "always");
+ wait_for_port(port);
+
+ let acked = incr_acked(port, "lone:counter", 200);
+ assert!(acked > 0, "lone writer made no progress");
+ sigkill(&mut child);
+
+ let mut child2 = start_moon(port, &dir, 1, "always");
+ wait_for_port(port);
+ let got = get_int(port, "lone:counter");
+ sigkill(&mut child2);
+
+ assert_eq!(
+ got, acked as i64,
+ "lone (C=1) writer: counter after SIGKILL+recovery = {} (expected {} acked)",
+ got, acked,
+ );
+ let _ = std::fs::remove_dir_all(&dir);
+ }
+
+ /// M3 control_message_breaks_the_batch — BGREWRITEAOF while many durable
+ /// writes are in flight. The Rewrite control must flush the in-progress batch
+ /// (write + fsync + ack) BEFORE it is handled, and post-rewrite replay must
+ /// lose no acked write. Writers are joined (all acked ⇒ durable) before the
+ /// kill, so each counter MUST recover exactly.
+ #[test]
+ #[ignore]
+ fn rewrite_during_active_writes_no_loss() {
+ const WRITERS: usize = 12;
+ const PER_WRITER: usize = 80;
+
+ let port = unique_port();
+ let dir = unique_dir("rewrite");
+ std::fs::create_dir_all(&dir).expect("create test dir");
+
+ let mut child = start_moon(port, &dir, 4, "always");
+ wait_for_port(port);
+
+ // Fire BGREWRITEAOF repeatedly from a side thread while writers run, so a
+ // Rewrite control message lands mid-drain on at least one writer.
+ let rewrite_port = port;
+ let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
+ let stop2 = stop.clone();
+ let rewriter = std::thread::spawn(move || {
+ while !stop2.load(std::sync::atomic::Ordering::Relaxed) {
+ let _ = Command::new("redis-cli")
+ .args(["-p", &rewrite_port.to_string(), "BGREWRITEAOF"])
+ .output();
+ std::thread::sleep(Duration::from_millis(40));
+ }
+ });
+
+ let handles: Vec<_> = (0..WRITERS)
+ .map(|i| {
+ let key = format!("rw:{{{}}}:{}", i, i);
+ std::thread::spawn(move || (key.clone(), incr_acked(port, &key, PER_WRITER)))
+ })
+ .collect();
+ let acked: Vec<(String, usize)> = handles
+ .into_iter()
+ .map(|h| h.join().expect("writer thread"))
+ .collect();
+
+ stop.store(true, std::sync::atomic::Ordering::Relaxed);
+ let _ = rewriter.join();
+
+ sigkill(&mut child);
+
+ let mut child2 = start_moon(port, &dir, 4, "always");
+ wait_for_port(port);
+ let mut wrong: Vec = Vec::new();
+ for (key, want) in &acked {
+ let got = get_int(port, key);
+ if got != *want as i64 {
+ wrong.push(format!("{}: want={} got={}", key, want, got));
+ }
+ }
+ sigkill(&mut child2);
+
+ assert!(
+ wrong.is_empty(),
+ "BGREWRITEAOF during active durable writes: {} counters wrong after recovery. \
+ The control message must flush its batch before the rewrite (no loss, no double-apply). \
+ Sample: {:?}",
+ wrong.len(),
+ wrong.iter().take(8).collect::>(),
+ );
+ let _ = std::fs::remove_dir_all(&dir);
+ }
+}