perf(durable-streams-rust): per-shard checkpoint triggers — interval knob + retained-WAL size budget

packages/durable-streams-rust/WAL_TUNING.md

@@ -0,0 +1,106 @@
+# WAL write-path tuning — the ideal configuration
+
+**Status:** validated 2026-07-13 on GCP `c4d-standard-64-lssd` (6× Titanium local
+NVMe, raw block), 256 B appends, 8 vCPU server, up to 100k streams. Campaign:
+ds-bench suites `wal-decomp-lane0`, `wal-splitlane`, `wal-sizetrigger`,
+`wal-cpubind` (see ds-bench `results/<suite>/report.md` + `AGENTS.md`).
+**Verdict: the write cardinality cliff is eliminated** — WAL-durable throughput
+is flat from 10k → 100k streams (−5%) at ~26–32× the pre-fix baseline.
+
+## The ladder (what each step bought, @100k streams)
+
+| configuration | ops/s |
+|---|---|
+| pre-fix: streams on network PD, per-stream checkpoint fdatasync storm | 10.4k |
+| + `--wal-checkpoint-syncfs on` (PR #4697) | 13.6k |
+| + stream files on local NVMe (not the boot disk) | 46k |
+| + split-lane layout (WAL shards on their own NVMe devices) | 272k |
+| + size-triggered checkpoint (PR #4704, `--wal-checkpoint-wal-bytes 1GiB`) | 303k |
+| + exclusive pinned cores (Guaranteed QoS + static CPU manager) | 328k |
+| **all of the above stacked** (suite `wal-stacked-1m`) | **383k** |
+| reference: memory durability (no fsync anywhere) | 512k |
+
+The residual ~1.6× gap to memory mode is WAL machinery (staging + double-write),
+not fsync — future work: io_uring segment writer (`wal/segment.rs` seam),
+batched mark-written.
+
+## The ideal configuration
+
+### 1. Hardware / storage layout (the #1 lever)
+
+Use an instance with **multiple physically attached NVMe devices** (GCP: 4th-gen
+`-lssd` types, raw block via `--local-nvme-ssd-block`; do NOT use
+`--ephemeral-storage-local-ssd`, which RAID0-stripes every device into one fsync
+barrier). Then:
+
+- **One device for stream data files** — mount it and point `--data-dir` at it.
+  The per-stream files and the checkpoint's `syncfs` domain live here.
+- **One device per WAL shard** — mount device *j* at `<data-dir>/wal/<i>` (the
+  server opens shard *i* at that path automatically). `--wal-shards` = number of
+  dedicated WAL devices.
+- **Never share a device between WAL and stream data.** Commit `fdatasync` vs
+  checkpoint writeback contention on one queue was worth 5× by itself
+  (55k → 272k). On dedicated lanes the commit-fsync cost is ~zero (checkpoint-off
+  measured *below* an all-lanes-shared no-fsync control).
+- **Never leave stream data on the boot disk / network PD.** On Kubernetes
+  raw-block node pools the default emptyDir sits on the boot PD — this single
+  mistake mismeasures (and misdeploys) WAL mode by 5–26×.
+
+Example (6-device box): device 0 → data root, devices 1–5 → 5 WAL shards:
+
+```
+--data-dir /data/wal/0 --wal-shards 5
+```
+
+### 2. Server flags
+
+```
+--wal-checkpoint-syncfs on            # one syncfs barrier per checkpoint instead of
+                                      # O(N-touched) per-stream fdatasync (PR #4697)
+--wal-checkpoint-wal-bytes 1073741824 # checkpoint a shard when ITS retained WAL
+                                      # exceeds 1 GiB (PR #4704) — checkpoint cost ≈ 0,
+                                      # crash-replay bounded to ≤1 GiB/shard (<1 s NVMe)
+--wal-checkpoint-interval-ms 60000    # fallback timer so an idle shard still recycles
+--wal-shards <number of WAL devices>  # shards = fsync lanes; on a SINGLE shared
+                                      # device keep 2–4 (more only fragments batches)
+--worker-threads <vCPUs>
+```
+
+Leave `--wal-fsync-parallel` at its default (1): fanout parallelizes the
+per-stream fsync loop that syncfs replaces, and measured as a regression.
+
+### 3. CPU binding (+21–24%)
+
+Give the server **exclusive pinned cores**. On Kubernetes: node pool with
+`kubeletConfig.cpuManagerPolicy: static` + a **Guaranteed QoS** pod (every
+container `requests == limits`, server CPU an integer). Measured 356k @10k /
+328k @100k vs 286k/272k on shared cores, same layout and flags. Now that WAL is
+no longer fsync-bound, it scales with cores again — don't starve it.
+
+### 4. What you do NOT need to worry about
+
+- **Read performance while tuning checkpoints.** Reads never touch the WAL and
+  are served zero-copy (`sendfile`) from the data file's page cache, which is
+  written *before* the WAL ack barrier. Checkpoint cadence has zero read-path
+  cost.
+- **Ack latency vs checkpoints.** Acks gate only on the WAL group-commit
+  `fdatasync`; a checkpoint never blocks appends.
+- **Recoverability.** The contract is unchanged by any of these knobs: a WAL
+  segment is recycled only after its records' stream bytes are fsynced into
+  their files **and** the durable-tail map is persisted (`wal/shard.rs`
+  checkpoint ordering; crash-recovery e2e + randomized crash sim cover it).
+  The size trigger only changes *when* that sequence runs.
+
+## Caveats / follow-ups
+
+- Validated to 100k streams, 256 B payloads, single node. The stacked config
+  measured 383k @100k, 244k @500k, and **56k @1M** — a NEW, different wall
+  appears near 1M streams (candidates: one open fd per live stream vs the
+  container nofile ceiling, an ext4 directory with 1M files, stream-map/tails
+  working set, page-cache pressure from 1M dirty files). Profiling pass pending;
+  see `CARDINALITY_1M.md` for the older analysis. Below ~500k streams the
+  configuration above is cliff-free.
+- `--wal-checkpoint-syncfs` and the size trigger are opt-in; flipping defaults
+  (syncfs on for Linux) is a candidate after soak.
+- Larger retained WAL = longer replay: 1 GiB/shard ≈ sub-second on local NVMe,
+  but budget it consciously on slower disks.

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

@@ -300,6 +300,30 @@ fn main() {
                     }
                 }
             }
+            // Checkpoint time trigger: per-shard cadence in ms (default 3000).
+            "--wal-checkpoint-interval-ms" => {
+                let v = val(args.next(), "--wal-checkpoint-interval-ms");
+                match v.parse::<u64>() {
+                    Ok(ms) if ms >= 1 => wal::shard::set_checkpoint_interval_ms(ms),
+                    _ => {
+                        eprintln!("--wal-checkpoint-interval-ms must be a positive integer (milliseconds)");
+                        std::process::exit(2);
+                    }
+                }
+            }
+            // Checkpoint size trigger: checkpoint a shard as soon as its retained
+            // WAL exceeds this many bytes (0 = disabled). An explicit replay-time
+            // budget that also self-staggers shards by their own write rates.
+            "--wal-checkpoint-wal-bytes" => {
+                let v = val(args.next(), "--wal-checkpoint-wal-bytes");
+                match v.parse::<u64>() {
+                    Ok(bytes) => wal::shard::set_checkpoint_wal_bytes(bytes),
+                    _ => {
+                        eprintln!("--wal-checkpoint-wal-bytes must be a non-negative integer (bytes)");
+                        std::process::exit(2);
+                    }
+                }
+            }
             other => {
                 eprintln!("unknown argument: {other}");
                 std::process::exit(2);
@@ -474,43 +498,68 @@ fn main() {
     });
 }
 
-/// How often the checkpoint ticker drives each shard's `checkpoint` (spec §7).
-/// A sane v1 constant: frequent enough that the WAL doesn't grow unbounded on a
-/// busy server, infrequent enough that the batched per-stream `fdatasync`s stay
-/// amortized. Checkpoint is non-blocking w.r.t. acks, so this is purely the
-/// WAL-recycle / per-stream-durability cadence (tunable is follow-up #9).
-const CHECKPOINT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(3);
+/// How often the checkpoint ticker POLLS its triggers. The actual checkpoint
+/// cadence is per-shard and knob-driven (see `wal::shard::checkpoint_interval_ms`
+/// / `checkpoint_wal_bytes`); this is just the trigger-evaluation resolution.
+/// 250 ms keeps the size trigger responsive (a shard writing 1 GB/s overshoots a
+/// 1 GiB budget by ≤ 250 MB) at negligible poll cost (two atomic loads/shard).
+const CHECKPOINT_POLL: std::time::Duration = std::time::Duration::from_millis(250);
 
-/// Spawn the per-shard checkpoint ticker (spec §7). One `tokio::time::interval`
-/// driver that, each tick, runs every shard's `checkpoint` (each: batched
-/// `fdatasync` of its touched per-stream files → persist `checkpoint_lsn` →
-/// recycle WAL segments below it). A checkpoint error is logged, not fatal — a
-/// failed/lagging checkpoint only delays WAL recycling (the disk-bounded safety
-/// valve, spec §7), never blocks appends.
+/// Spawn the per-shard checkpoint driver (spec §7). Each poll tick, a shard is
+/// checkpointed iff (a) its retained WAL exceeds `--wal-checkpoint-wal-bytes`
+/// (size trigger, 0 = off), or (b) `--wal-checkpoint-interval-ms` has elapsed
+/// since ITS last checkpoint (time trigger, default 3000 = the historical 3 s
+/// cadence). Due shards checkpoint CONCURRENTLY (each is one spawn_blocking:
+/// capture + fsync/syncfs of touched stream files → persist tails/checkpoint_lsn
+/// → recycle); a serial walk would queue every shard's fsync behind one
+/// device's. Because each shard's clock restarts when IT finishes, shards drift
+/// apart naturally instead of storming in a synchronized wave — and with the
+/// size trigger they self-schedule by their own write rates. A checkpoint error
+/// is logged, not fatal — a failed/lagging checkpoint only delays WAL recycling
+/// (the disk-bounded safety valve, spec §7), never blocks appends. A shard that
+/// is still checkpointing is never re-fired (the in-flight set guards it), so a
+/// checkpoint that takes longer than the interval degrades to back-to-back
+/// checkpoints for that shard only.
 fn spawn_checkpoint_ticker(walset: Arc<wal::walset::WalSet>) {
     tokio::spawn(async move {
-        let mut ticker = tokio::time::interval(CHECKPOINT_INTERVAL);
-        // Skip the immediate first tick — there is nothing to checkpoint at boot.
+        let interval =
+            std::time::Duration::from_millis(wal::shard::checkpoint_interval_ms());
+        let wal_bytes = wal::shard::checkpoint_wal_bytes();
+        let n = walset.shards().len();
+        let mut last_done: Vec<std::time::Instant> = vec![std::time::Instant::now(); n];
+        let mut in_flight: Vec<bool> = vec![false; n];
+        let mut wave: tokio::task::JoinSet<usize> = tokio::task::JoinSet::new();
+        let mut ticker = tokio::time::interval(CHECKPOINT_POLL.min(interval));
         ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+        // Skip the immediate first tick — there is nothing to checkpoint at boot.
         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() {
+            // Reap finished checkpoints (non-blocking) and restart their clocks.
+            while let Some(done) = wave.try_join_next() {
+                if let Ok(i) = done {
+                    in_flight[i] = false;
+                    last_done[i] = std::time::Instant::now();
+                }
+            }
+            for (i, shard) in walset.shards().iter().enumerate() {
+                if in_flight[i] {
+                    continue;
+                }
+                let size_due = wal_bytes > 0 && shard.wal_size_bytes() >= wal_bytes;
+                let time_due = last_done[i].elapsed() >= interval;
+                if !(size_due || time_due) {
+                    continue;
+                }
+                in_flight[i] = true;
                 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());
                     }
+                    i
                 });
             }
-            while wave.join_next().await.is_some() {}
         }
     });
 }

packages/durable-streams-rust/src/wal/shard.rs

@@ -423,6 +423,33 @@ fn checkpoint_syncfs() -> bool {
     CHECKPOINT_SYNCFS.load(Ordering::Relaxed)
 }
 
+/// Checkpoint cadence knobs (cardinality-cliff follow-up). The checkpoint's only
+/// job is bounding retained-WAL size (= crash-replay time): acks never gate on it
+/// (`shard.rs` "disk-bounded safety valve") and reads never touch the WAL, so
+/// firing it less often is free except for disk space and replay budget. Two
+/// triggers, whichever comes first per shard:
+///   * time: `--wal-checkpoint-interval-ms` (default 3000 — the historical 3 s
+///     wave cadence, now per-shard).
+///   * size: `--wal-checkpoint-wal-bytes` (default 0 = disabled) — checkpoint a
+///     shard as soon as its retained WAL exceeds this many bytes. This turns the
+///     hardcoded timer into an explicit replay-time budget (e.g. 1 GiB ≈ <1 s of
+///     replay on NVMe) and lets shards self-stagger by their own write rates
+///     instead of storming together on a shared tick.
+static CHECKPOINT_INTERVAL_MS: AtomicU64 = AtomicU64::new(3_000);
+pub fn set_checkpoint_interval_ms(ms: u64) {
+    CHECKPOINT_INTERVAL_MS.store(ms.max(1), Ordering::Relaxed);
+}
+pub fn checkpoint_interval_ms() -> u64 {
+    CHECKPOINT_INTERVAL_MS.load(Ordering::Relaxed)
+}
+static CHECKPOINT_WAL_BYTES: AtomicU64 = AtomicU64::new(0);
+pub fn set_checkpoint_wal_bytes(bytes: u64) {
+    CHECKPOINT_WAL_BYTES.store(bytes, Ordering::Relaxed);
+}
+pub fn checkpoint_wal_bytes() -> u64 {
+    CHECKPOINT_WAL_BYTES.load(Ordering::Relaxed)
+}
+
 /// Name of the per-shard durable-tail map: `<shard_dir>/tails` (task 11b). A
 /// CUMULATIVE `stream_id durable_tail` line map (plain decimal text, one stream
 /// per line). At checkpoint, each touched stream's current logical `Shared.tail`