feat(durable-streams-rust): replicated durability — quorum-replication acks via Raft (openraft)

.changeset/delete-ack-durability.md

@@ -0,0 +1,10 @@
+---
+"@electric-ax/durable-streams-server-rust": patch
+---
+
+fix: an acked DELETE is now durable before the 204 — the data-file and
+sidecar unlinks (plus a parent-directory fsync) previously ran on a
+detached background task, so a crash right after the ack resurrected the
+stream with all its data on the next boot. Soft deletes (fork-referenced
+streams) likewise persist the flag before acking. Found by the
+crash/fault simulation (seed 20387; details in `CRASH_SIM_FINDINGS.md`).

.changeset/replicated-durability.md

@@ -0,0 +1,14 @@
+---
+"@electric-ax/durable-streams-server-rust": minor
+---
+
+feat: `--durability replicated` — Kafka-style durability via quorum
+replication (openraft) instead of local fsync. An append acks once a quorum
+of replicas has committed and applied it; no fsync anywhere on the hot path.
+Any node accepts writes (forward-to-leader with read-your-writes), every node
+serves reads, log-first apply keeps replicas byte-identical through
+fail-over. Includes restart-rejoin (durable vote + manifest snapshots with
+mesh byte-fetch), live membership change (`/_repl/add-learner`,
+`/_repl/change-membership`), `REPL_STATS`/`/_repl/status` observability, and
+a 3-node deploy kit (`deploy/replicated/`: local cluster, docker compose,
+k8s, smoke incl. leader-kill and restart-rejoin).

.changeset/wal-1m-cardinality.md

@@ -0,0 +1,9 @@
+---
+"@electric-ax/durable-streams-server-rust": patch
+---
+
+perf: fix the high-stream-cardinality (200k–1M streams) write cliff — O(1) WAL
+checkpoint drain, checkpoint fully off the async runtime with concurrent shards
+and a resident tails map, meta sidecar flush moved from per-append to the
+checkpoint, single registry lookup per append. 1M streams now sustains 1.11M
+ops/s on 16 vCPU (was 862k, with 405 ms worst-case latency now 150 ms).

.changeset/wal-multi-segment-recovery.md

@@ -0,0 +1,12 @@
+---
+"@electric-ax/durable-streams-server-rust": patch
+---
+
+fix: WAL recovery no longer loses acked data when the WAL spans multiple
+segments. Boot re-preallocated the first segment unconditionally, so a sealed
+(exactly-packed) `1.wal` grew a zero tail that replay mis-read as the end of
+the durable log — dropping every later segment's acked records and truncating
+the per-stream files to the stale frontier — and a checkpoint-recycled `1.wal`
+was recreated empty, making replay recover nothing. Boot now opens existing
+segments non-destructively. Found by the new seeded crash/fault simulation
+(`src/wal/sim_tests.rs`, findings in `CRASH_SIM_FINDINGS.md`).

.changeset/wal-quiet-stream-torn-tail.md

@@ -0,0 +1,13 @@
+---
+"@electric-ax/durable-streams-server-rust": patch
+---
+
+fix: recovery now truncates a torn, never-acked tail on streams with no
+surviving WAL record and no checkpoint tails entry (e.g. a stream created
+after the last checkpoint whose only in-flight append was torn by power
+loss) — previously the torn fragment became reader-visible. The `.meta`
+sidecar persists a `durable_tail` proof (riding along on existing fsynced
+meta writes; no new hot-path fsyncs) and recovery seeds every stream's
+durable frontier from it. Old sidecars keep the previous behavior until
+their next natural meta write. Found by the crash/fault simulation
+(seed 20230; details in `CRASH_SIM_FINDINGS.md`).

docs/superpowers/specs/2026-07-02-wal-crash-simulation-design.md

@@ -0,0 +1,110 @@
+# WAL crash/recovery randomized simulation — design
+
+**Date:** 2026-07-02
+**Goal:** find correctness issues in the durable-streams Rust server's recovery path under
+network failures (client drops), disk failures (power-loss page-cache loss, torn writes),
+and arbitrary crash points — via seeded, reproducible randomized simulation.
+
+## Why this shape
+
+The existing coverage (`src/wal/e2e_tests.rs`, `src/wal/recovery.rs` unit tests) is
+deterministic: each test hand-picks one crash scenario. A randomized simulator explores the
+scenario space — interleavings of creates/appends/closes/forks/checkpoints, crash points,
+and fault combinations — that nobody thought to write a test for.
+
+Approaches considered:
+
+1. **In-process randomized crash simulation (chosen).** A `#[cfg(test)]` module that drives
+   the real handler path (`handlers::handle`), simulates crashes exactly like the existing
+   e2e `Harness` (stop committers, drop store + WalSet, keep the data dir), injects disk
+   faults constrained to the documented fault model, and re-runs the real boot sequence
+   (`WalSet::open` → sidecar pass → `wal::recovery::recover` → `reset_after_recovery`).
+   Deterministic per seed; can consult internal state (durable LSN, checkpoint tails) to
+   keep injected faults inside the fault model.
+2. Black-box process-level fuzzer (kill -9 the real binary over HTTP). Higher socket-layer
+   fidelity, but kill -9 preserves the page cache, so it cannot simulate power loss — the
+   most interesting failure class here — and reproduction is flaky.
+3. cargo-fuzz on the WAL codec. Narrow; the codec already has CRC framing + unit tests.
+
+## Fault model (what the simulator may break)
+
+Grounded in ARCHITECTURE.md + recovery.rs:
+
+- **Process crash:** committers stop, process state vanishes, all file bytes written so far
+  survive (page cache == disk from the test's point of view).
+- **Power loss / disk failure**, applied on top of a crash:
+  - _Per-stream data files_ are fsynced only at checkpoint (and at recovery repair). Any
+    byte beyond a stream's last checkpointed durable tail may be lost or garbage. The
+    simulator may truncate, zero, or scribble the region past that floor.
+  - _WAL segments_ are fdatasync'd up to the durable LSN (that is what releases acks).
+    Bytes of records with `lsn > durable_lsn` at crash time may be torn: the simulator
+    scans the segment with the real codec, finds the byte offset where staged-but-unacked
+    records start, and corrupts/zeroes a random suffix from a random point at or past it.
+  - `.meta` sidecars: create/close fsync them; the lazy tail flush does not. The simulator
+    does not corrupt sidecars in v1 (identity durability is a separate contract).
+- **Network failure:** a client connection dropping mid-append is modeled by aborting the
+  spawned append task at a random await point (tokio cancellation), and by crashing with
+  appends still in flight. Such appends are **maybe-applied**: the oracle accepts their
+  presence or absence, but never a torn fragment of them.
+
+## Workload generator
+
+Seeded xorshift PRNG (no new dependencies; `rand` stays out of the dependency tree — the
+existing crate has zero dev-deps and tests use std only). Per step, weighted choice of:
+
+- create stream (octet or JSON content type)
+- append a self-describing record (`<stream>#<seq>|` for octet; `{"s":..,"i":..}` for
+  JSON) via the real POST path — sizes varied, occasionally multi-KB
+- append with cancellation: spawn, then abort after a random yield count (maybe-applied)
+- close a stream (real POST close path)
+- fork a stream at a random offset ≤ parent tail (exercises `file_base > 0` recovery)
+- delete a stream (recovery must not resurrect it)
+- checkpoint a random shard (`shard.checkpoint()`), then refresh that shard's per-stream
+  durable-tail floors from `read_durable_tails()` (governs data-file fault legality)
+- read a random range via the real GET path and check it against the oracle
+
+## Crash / fault / recover cycle
+
+Each seed runs G generations (default 4). Per generation: run K workload steps, then
+crash (with 0..3 appends deliberately still in flight), then with independent
+probabilities inject data-file faults and WAL-suffix faults as bounded above, then boot
+the real recovery sequence and run the oracle. Subsequent generations continue the
+workload on the recovered store — recovery-of-recovery bugs (stale `appender.written`,
+tail/watch mismatches) only show up this way.
+
+## Oracle (checked after every recovery, and on every read)
+
+Per stream, the oracle tracks: records issued (unique tokens, in order), ack status of
+each (acked / maybe / rejected), closed status, expected file_base.
+
+1. **No loss:** the recovered data file's bytes parse as a concatenation of whole issued
+   records, in issue order, containing **every acked record** — i.e. an ordered
+   subsequence of issued records ⊇ acked records. (Maybe-applied records may appear or
+   not; nothing else may.)
+2. **No torn record:** the parse consumes the file exactly — no trailing fragment, no
+   interior garbage. For JSON streams every recovered record re-parses as JSON.
+3. **Tail consistency:** `Shared.tail == file_base + file_len`, `durable_tail == tail`,
+   and the watch channel publishes the same tail.
+4. **Closed-ness durability:** a stream whose close was acked recovers `closed == true`
+   (position may lawfully shrink only under power-loss faults, and never below the
+   checkpointed floor).
+5. **No resurrection:** a deleted stream does not reappear after recovery.
+6. **Read correctness:** GETs (catch-up path) return exactly the oracle's bytes for the
+   requested range.
+7. **Recovery never panics** and never errors on in-model faults.
+
+On violation: print the seed, generation, step trace tail, and the diff — everything
+needed to replay deterministically.
+
+## Placement & running
+
+- New module `src/wal/sim_tests.rs` (`#[cfg(test)]`, registered in `wal/mod.rs`), reusing
+  the `Harness` boot/crash idiom from `e2e_tests.rs` and `DurabilityGuard::wal()`.
+- CI-friendly default: a small fixed seed set (fast, deterministic).
+- Long-run mode via env: `DS_SIM_SEEDS=<n>` and `DS_SIM_STEPS=<k>` scale the exploration;
+  a wrapper invocation runs thousands of seeds locally to hunt for issues.
+
+## Out of scope (v1)
+
+Tiering/offload faults (S3), `.meta` corruption (byzantine tier), memory-mode recovery,
+multi-process concurrency, and the HTTP socket layer itself. Each can be layered on later.

packages/durable-streams-rust/.gitignore

@@ -8,3 +8,5 @@
 # the cross-server suite may still drop these here when run against this checkout:
 /bench/
 benchmark-results.json
+# deploy/replicated/local-cluster.sh state (data dirs, logs, pids)
+/.local-cluster/

packages/durable-streams-rust/ARCHITECTURE.md

@@ -35,10 +35,10 @@ flowchart LR
     W2 --> W3["encode_wire<br/>(JSON flatten, append delimiter)"]
     W3 --> W4[["per-stream appender mutex"]]
     W4 --> W5["write_all → data file<br/>(lands in page cache)"]
-    W5 --> W6["update tail + resident cache"]
-    W6 --> W8["publish tail<br/>(watch channel)"]
-    W8 --> W7["group-commit fsync<br/>(WAL shard committer)"]
-    W7 --> W9["204 / 200 — only after durable"]
+    W5 --> W6["advance writer tail<br/>+ stage into WAL shard"]
+    W6 --> W7["group-commit fsync<br/>(WAL shard committer)"]
+    W7 --> W8["publish durable tail + resident cache<br/>(watch channel)"]
+    W8 --> W9["204 / 200 — only after durable"]
   end
 
   subgraph READ["READ  ·  GET"]
@@ -72,10 +72,10 @@ The dotted edges are the only coupling between writers and readers: publishing t
 1. **Parse idempotency headers** — `Producer-Id` / `Producer-Epoch` / `Stream-Seq`. Duplicate `(producer, epoch, seq)` is acknowledged without re-appending (exactly-once producers).
 2. **`encode_wire`** — turn the request body into the contiguous wire representation. In JSON mode this flattens arrays and appends the `,` delimiter so the on-disk bytes are already a valid stream fragment.
 3. **Acquire the per-stream appender mutex** (`AsyncMutex<Appender>`). This is the _only_ serialization point, and it's per-stream — different streams never contend.
-4. **`write_wire`** — `write_all` the bytes to the data file (they land in the OS page cache immediately), advance the tail under a short `RwLock` write, update the **resident tail cache**, then **publish the new tail** on the `watch` channel. Note the order: the cache is populated _before_ the wake, so a woken subscriber reliably hits it.
-5. **Durability** — in `wal` mode the append is staged into the stream's assigned WAL shard, and the response returns only after the shard's group-commit committer `fdatasync`s the segment covering this record. See [Durability](#durability) below. In `wal` mode everything above and the entire read path are identical regardless of workload. In `memory` mode binary appends take a separate socket→file splice path (zero-copy `splice(2)`, no WAL, no `fdatasync`) that bypasses `handle_append` / `encode_wire` via the engine zero-copy intercept and acks immediately after the page-cache write.
+4. **`write_wire`** — `write_all` the bytes to the data file (they land in the OS page cache immediately) and advance the _writer_ tail (`Shared.tail`) under a short `RwLock` write. The reader-observable tail does **not** move yet.
+5. **Durability, then visibility** — in `wal` mode the append is staged into the stream's assigned WAL shard (under the appender mutex, so per-stream LSN order matches byte order) and the handler awaits the shard's group-commit `fdatasync`. Only after the record is durable does `publish_durable_tail` advance the **reader-observable `durable_tail`** (monotonically), populate the **resident tail cache**, and **publish on the `watch` channel** — cache before wake, so a woken subscriber reliably hits it. Then the 2xx is returned. See [Durability](#durability) below. In `memory` mode there is no WAL: the page-cache write is the ack, and binary appends take a separate socket→file splice path (zero-copy `splice(2)`) that bypasses `handle_append` / `encode_wire` via the engine zero-copy intercept.
 
-Visibility vs durability are deliberately decoupled: the bytes are in the page cache (and the tail is published) before durability resolves, so a live reader sees data with minimal latency, while the _appender_ doesn't get its 2xx until the data is durable.
+Visibility is gated on durability (PROTOCOL.md §4.1): a live reader never observes bytes (or an EOF) that a crash could roll back. The writer tail runs ahead in `Shared.tail`; readers see `durable_tail`, which follows it as group commits resolve.
 
 ## Durability
 
@@ -105,7 +105,7 @@ Every append is written to the per-stream data file (page cache, no hot fsync 
 
 Per-stream files are `fdatasync`'d off the ack path at a periodic **checkpoint**, after which the bounded WAL is recycled. On boot, recovery replays the WAL from its oldest retained segment, reconciles each stream's durable tail (torn-tail repair via truncation + `fdatasync`), then resets the WAL for fresh appends.
 
-The invariant: **visibility is never gated on durability**. Bytes land in the page cache and the tail is published before the WAL `fdatasync`, so live readers see an append at memory latency while the appender's acknowledgement waits for the WAL commit.
+The invariant: **readers only ever observe durable bytes** (PROTOCOL.md §4.1). Bytes land in the page cache immediately, but the reader-observable `durable_tail` (and the `watch` wake) advances only after the WAL `fdatasync` covering the record — the same barrier that releases the appender's acknowledgement. A crash therefore never rolls back anything a reader has seen.
 
 ### `memory` mode
 
@@ -140,8 +140,8 @@ flowchart TB
   end
 
   A3 ==> PC[("OS page cache<br/>— the hot tier")]
-  A3 ==> RC["resident tail cache<br/>(last chunk, configurable via --tail-cache-bytes,<br/>in heap)"]
-  A3 -.-> TW[/"tail watch channel<br/>(one notify per append — before fsync)"/]
+  A4 ==> RC["resident tail cache<br/>(last chunk, configurable via --tail-cache-bytes,<br/>in heap)"]
+  A4 -.-> TW[/"tail watch channel<br/>(one notify per append — after the group commit)"/]
 
   subgraph FAN["② Fan-out"]
     direction TB
@@ -159,7 +159,7 @@ The techniques, each with its mechanism and payoff:
 1. **Contiguous wire-byte storage.** The file _is_ the response. Reads are byte ranges with no reframing and no per-message copy — and this is what makes `sendfile` zero-copy possible at all.
 2. **Group-commit coalesced fsync.** The durability contract ("return after fsync") is the expensive part of an append. Concurrent appenders share a single in-flight barrier fsync, so throughput scales with the _batch size_ per fsync rather than one fsync per message. This is why unbatched appends hit ~30k/s where a per-append-fsync server (Node) does ~130/s.
 3. **Per-stream single writer, lock-free reads.** One async mutex orders a stream's appends; there is no global lock (streams live in a `DashMap`). Reads take a brief tail snapshot and do positioned reads — they never block the writer and never wait on each other.
-4. **Pre-fsync visibility.** Appended bytes are in the page cache and the tail is published before the fsync resolves, so live readers see data at memory latency; only the appender's acknowledgement waits for durability.
+4. **Durable-gated visibility, group-commit-amortized.** The reader-observable tail is published only after the record's group-commit fsync (readers never see bytes a crash could roll back — PROTOCOL.md §4.1). Because N concurrent appends share one barrier fsync, the visibility latency cost is one amortized group commit, not one fsync per append.
 5. **`watch`-channel wakeups.** Live readers park on a per-stream `watch`; an append is one `send_replace` that wakes all of them. No polling loop, no timer churn.
 6. **Resident tail cache (fan-out de-duplication).** Without it, N caught-up SSE/long-poll subscribers each re-read (and re-encode) the _same_ just-appended bytes — N× duplicated work that grows with audience size. The cache keeps the last chunk in the heap so all N share one read (and SSE encodes once per subscriber off that shared buffer). For small hot reads it's also fewer syscalls than `sendfile` and skips the read-offload pool hop.
 7. **Zero-copy egress.** `FileRange` reads are served with `sendfile` (page cache → socket, no userspace copy → ~5× less CPU per byte than a buffered copy). The **`--read-offload`** strategy keeps a cold backfill's disk fault off the async workers so one slow read can't stall unrelated requests.