perf(wal): fix the high-stream-cardinality write cliff (1M streams → 1.11M ops/s @ 16 vCPU)

.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).

packages/durable-streams-rust/CARDINALITY_1M.md

@@ -0,0 +1,54 @@
+# 1M-stream cardinality fixes — findings + results (2026-07-02)
+
+Follow-up to `WRITE_BOTTLENECKS_1M.md` (bottleneck #2: stream cardinality) and
+`CONTENTION_INVESTIGATION.md`. Server commit: `662b0c845` on
+`perf/combined-t1a-t1c-t2a`. **Outcome: 1M streams reaches 1,114,644 ops/s on a
+16 vCPU `c4d-standard-16-lssd` (ladder unsaturated), and the 500k→1M degradation at
+equal load is −17% (was a cliff).**
+
+## Root causes (evidence-first, local Linux repro at 20k→400k streams)
+
+The key mechanism: **at high cardinality, ops/stream/checkpoint-interval drops below
+1, so every "amortized once-per-stream-per-interval" cost becomes a per-op cost.**
+
+| #   | cost                                                                                                                                                                                                                                  | evidence                                                                                                           | fix                                                                                                                                                                                                                                                  |
+| --- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| 1   | Checkpoint drain did O(touched) capture (`shared.read()` + Arc clone per stream) **while holding the shard `dirty` mutex** — the same mutex every append's epoch-transition takes; at 400k streams every append is a transition       | `WAL_CKPT drain_us` 25–140 ms/tick; `dirty_wait_load` 0.01→0.30 cores as streams 20k→400k; p99 ≈ max drain         | O(1) critical section: take Vec + bump epoch under the lock, capture after release                                                                                                                                                                   |
+| 2   | Checkpoint capture / cumulative tails-file re-read+re-sort+rewrite / recycle ran **on async runtime threads**, serially across shards                                                                                                 | `WAL_CKPT` capture 31 ms + tails 24 ms per tick per shard on runtime threads                                       | whole checkpoint body in one `spawn_blocking`; tails map memory-resident; shards checkpoint concurrently (`JoinSet`)                                                                                                                                 |
+| 3   | **Per-append meta sidecar flush**: with inter-append gap > the 100 ms debounce (always, at high cardinality) every producer append did JSON + `File::create(.meta.tmp)` + `rename` → all workers spin on the **data-dir inode rwsem** | perf: `osq_lock`+`rwsem_spin_on_owner` under `write_meta_sync` = **38–46% of ALL server CPU at every cardinality** | WAL-staged appends only mark `meta_dirty`; checkpoint writes sidecars for drained streams after recycle (memory-mode keeps the debounced flush). Producer/access staleness bound: 100 ms debounce → checkpoint cadence (contract already allows lag) |
+| 4   | Two registry lookups per append (`handle_append` metric label + `_inner`) — 2× SipHash + cold DashMap walk at 1M keys                                                                                                                 | code inspection                                                                                                    | `_inner` returns `is_json`                                                                                                                                                                                                                           |
+
+## Local A/B (contention-repro-linux.sh, 6 srv cores, conn=256, shards=6)
+
+| streams | before        | after      | p99          |
+| ------- | ------------- | ---------- | ------------ |
+| 20k     | ~43–46k ops/s | **80.4k**  | 41 → 7.7 ms  |
+| 200k    | 32.3k         | **50.6k**  | 53 → 17.9 ms |
+| 400k    | 16.0k         | **36–44k** | 144 → ~28 ms |
+
+Correctness: 95 crate tests + 326 conformance tests pass. New telemetry: `WAL_CKPT`
+per-shard checkpoint phase line (`--wal-stats`), and the repro script grew
+`--tmpfs` / `--wal-stats` knobs + WAL_CKPT summarizing.
+
+## Remote validation (GKE, 16 vCPU, pool client, 256 B, batch 1)
+
+Suite `ds-bench/suites/run-durable-cpu16-1m-card.json`, image
+`durable-streams:combined-card@sha256:d74840bd…`; full detail + caveats in
+`ds-bench/results/run-durable-cpu16-1m-card/FINDINGS.md`.
+
+| streams | pods | ops/s         | p50 / p99 / max                                              |
+| ------- | ---- | ------------- | ------------------------------------------------------------ |
+| 1M      | 32   | 898,582       | 3.5 / 30.5 / **149.6 ms** (baseline 862k, 3.3 / 32 / 405 ms) |
+| 1M      | 64   | **1,114,644** | 3.4 / 60.3 / 211 ms — still climbing (+21%/+16 pods)         |
+| 500k    | 48   | 1,110,268     | 3.1 / 42.7 / 145 ms                                          |
+
+## Open follow-ups
+
+1. True 16 vCPU ceiling at 1M (ladder past 64 pods) and a 32 vCPU 1M run.
+2. **Per-shard producer-state journal**: sidecar writes/s ≈ ops/s at full cardinality
+   (off the hot path now, but it stretches checkpoint cadence — locally ~1.6 s/shard
+   meta phase at 400k — and bounds producer-state staleness). One cumulative
+   per-shard file per tick, recovery overlays producers by max(epoch, seq).
+3. Read+write mix at 1M streams (everything here is write-only).
+4. `--wal-stats` cell on NVMe (`run-durable-cpu16-1m-card-stats.json`, never ran) to
+   confirm checkpoint fsync/meta phase behavior at 1M on real disks.

packages/durable-streams-rust/scripts/contention-repro-linux.sh

@@ -16,6 +16,7 @@
 #   scripts/contention-repro-linux.sh [--shards N] [--connections C]
 #       [--streams S] [--duration D] [--warmup W] [--payload P] [--batch B]
 #       [--srv-cpus 0-5] [--cli-cpus 6-9] [--label NAME] [--no-build]
+#       [--tmpfs SIZE] [--wal-stats 0|1]
 set -euo pipefail
 
 SHARDS=1
@@ -29,6 +30,8 @@ SRV_CPUS="0-5"
 CLI_CPUS="6-9"
 LABEL=""
 DO_BUILD=1
+TMPFS_SIZE=2g   # at high stream cardinality each non-empty file pins >=1 page: size ~ streams*4k + data
+WAL_STATS=1
 
 SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 CRATE_DIR="$(cd "$SCRIPT_DIR/.." && pwd)"
@@ -56,6 +59,8 @@ while [ $# -gt 0 ]; do
     --cli-cpus) CLI_CPUS="$2"; shift 2;;
     --label) LABEL="$2"; shift 2;;
     --no-build) DO_BUILD=0; shift;;
+    --tmpfs) TMPFS_SIZE="$2"; shift 2;;
+    --wal-stats) WAL_STATS="$2"; shift 2;;
     *) echo "unknown arg: $1" >&2; exit 2;;
   esac
 done
@@ -92,11 +97,11 @@ docker rm -f "$SRV" >/dev/null 2>&1 || true
 docker run -d --name "$SRV" --network "$NET" \
   --cpuset-cpus="$SRV_CPUS" \
   -v "$TARGET_VOL":/target:ro \
-  --tmpfs /data:rw,size=2g \
+  --tmpfs /data:rw,size="$TMPFS_SIZE" \
   debian:bookworm-slim \
   /target/release/durable-streams-server \
     --host 0.0.0.0 --port 4437 --data-dir /data \
-    --durability wal --wal-shards "$SHARDS" --wal-stats 1 >/dev/null
+    --durability wal --wal-shards "$SHARDS" --wal-stats "$WAL_STATS" >/dev/null
 
 for _ in $(seq 1 100); do
   docker logs "$SRV" 2>&1 | grep -q "listening on" && break
@@ -144,4 +149,11 @@ print(f"  fsync_per_s        = {avg('fsync/s'):,.0f}   batch_avg = {avg('batch_a
 print(f"  inner_wait_us      = {avg('inner_wait_us'):.2f}   inner_wait_load = {avg('inner_wait_load'):.2f} cores")
 print(f"  dirty_wait_us      = {avg('dirty_wait_us'):.2f}   dirty_wait_load = {avg('dirty_wait_load'):.2f} cores")
 print(f"  waiters_woken_avg  = {avg('waiters_woken_avg'):.1f}")
+ck=[{k:float(v) for k,v in pat.findall(l)} for l in open(srv_log,errors="ignore") if "WAL_CKPT" in l]
+if ck:
+    n=len(ck)
+    m=lambda k:max(r.get(k,0.0) for r in ck)
+    a=lambda k:sum(r.get(k,0.0) for r in ck)/n
+    print(f"  ckpt (n={n}) touched avg/max      = {a('touched'):,.0f} / {m('touched'):,.0f}   tails_entries max = {m('tails_entries'):,.0f}   meta avg = {a('meta'):,.0f}")
+    print(f"  ckpt us avg/max: capture={a('capture_us'):,.0f}/{m('capture_us'):,.0f} fsync={a('fsync_us'):,.0f}/{m('fsync_us'):,.0f} tails={a('tails_us'):,.0f}/{m('tails_us'):,.0f} rest={a('rest_us'):,.0f}/{m('rest_us'):,.0f} meta={a('meta_us'):,.0f}/{m('meta_us'):,.0f} total={a('total_us'):,.0f}/{m('total_us'):,.0f}")
 PY

packages/durable-streams-rust/src/handlers.rs

@@ -886,25 +886,26 @@ impl AppendOutcome {
 
 async fn handle_append(store: Arc<Store>, req: Req, path: String) -> Resp {
     let t0 = crate::telemetry::Timer::start();
-    // is_json is needed for the metric label even on the not-found path, where we
-    // don't have a stream; default to false there.
-    let is_json = store.get(&path).map(|s| s.is_json).unwrap_or(false);
-    let (resp, outcome) = handle_append_inner(store, req, path).await;
+    // is_json comes back from the inner handler (false on the not-found path) so
+    // the metric label doesn't cost a SECOND registry lookup per append — at high
+    // stream cardinality each lookup is a cold walk of a million-key map.
+    let (resp, outcome, is_json) = handle_append_inner(store, req, path).await;
     crate::telemetry::record_append(t0.elapsed_secs(), outcome.label(), is_json);
     resp
 }
 
-async fn handle_append_inner(store: Arc<Store>, req: Req, path: String) -> (Resp, AppendOutcome) {
+async fn handle_append_inner(store: Arc<Store>, req: Req, path: String) -> (Resp, AppendOutcome, bool) {
     use AppendOutcome::*;
+    let st = match store.get(&path) {
+        Some(s) => s,
+        None => return (text_response(404, "stream not found"), Conflict, false),
+    };
+    let is_json = st.is_json;
     macro_rules! ret {
         ($resp:expr, $oc:expr) => {
-            return ($resp, $oc)
+            return ($resp, $oc, is_json)
         };
     }
-    let st = match store.get(&path) {
-        Some(s) => s,
-        None => ret!(text_response(404, "stream not found"), Conflict),
-    };
     if st.shared.read().unwrap().soft_deleted {
         ret!(gone(), Conflict);
     }
@@ -1156,7 +1157,18 @@ async fn handle_append_inner(store: Arc<Store>, req: Req, path: String) -> (Resp
         st.tail_tx.send_replace(Tail { bytes: tail, closed: true });
         #[cfg(target_os = "linux")]
         crate::sse_reactor::wake_stream(&st);
+    } else if staged_lsn.is_some() {
+        // WAL mode: the stream is in its shard's dirty set (register_dirty ran
+        // during staging), so the ~3 s checkpoint will write the sidecar for us —
+        // just mark it. This keeps the meta `File::create`+`rename` (and its
+        // parent-directory rwsem, measured at ~40% of server CPU under write
+        // saturation) plus a timer task OFF the per-append path. Producer/access
+        // updates are already documented as a non-durable, lagging flush; the lag
+        // bound moves from the 100 ms debounce to the checkpoint cadence.
+        st.meta_dirty.store(true, std::sync::atomic::Ordering::Release);
     } else {
+        // No WAL record staged (memory durability): no checkpoint will flush the
+        // sidecar — keep the debounced flush.
         st.schedule_meta_flush();
     }
     if !wire.is_empty() {
@@ -1178,7 +1190,7 @@ async fn handle_append_inner(store: Arc<Store>, req: Req, path: String) -> (Resp
     if tail.closed {
         b = b.hs(H_CLOSED, "true");
     }
-    (b.body(empty()), Accept)
+    (b.body(empty()), Accept, is_json)
 }
 
 fn closed_conflict(tail: u64) -> Resp {

packages/durable-streams-rust/src/main.rs

@@ -433,11 +433,22 @@ fn spawn_checkpoint_ticker(walset: Arc<wal::walset::WalSet>) {
         ticker.tick().await;
         loop {
             ticker.tick().await;
+            // All shards checkpoint CONCURRENTLY. Each checkpoint is one
+            // spawn_blocking task (capture + per-stream fdatasyncs + tails/ckpt
+            // persist + recycle), so a serial walk makes every per-stream fsync
+            // across the whole server queue behind a single shard's — at high
+            // stream cardinality that serialization is what stretches the
+            // checkpoint wave (and on real disks wastes the device's parallelism).
+            let mut wave = tokio::task::JoinSet::new();
             for shard in walset.shards() {
-                if let Err(e) = shard.checkpoint().await {
-                    eprintln!("WAL checkpoint failed for shard {:?}: {e}", shard.dir());
-                }
+                let shard = Arc::clone(shard);
+                wave.spawn(async move {
+                    if let Err(e) = shard.checkpoint().await {
+                        eprintln!("WAL checkpoint failed for shard {:?}: {e}", shard.dir());
+                    }
+                });
             }
+            while wave.join_next().await.is_some() {}
         }
     });
 }