perf(durable-streams-rust): eliminate the write cardinality cliff — syncfs checkpoints, per-shard triggers, stream lanes

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

@@ -896,6 +896,9 @@ async fn handle_append(store: Arc<Store>, req: Req, path: String) -> Resp {
 
 async fn handle_append_inner(store: Arc<Store>, req: Req, path: String) -> (Resp, AppendOutcome, bool) {
     use AppendOutcome::*;
+    // Load-telemetry probe: bumps the in-flight gauge and records service time on
+    // drop (covers every return path). No-op unless `--server-stats` is on.
+    let _probe = crate::srvstats::AppendProbe::start();
     let st = match store.get(&path) {
         Some(s) => s,
         None => return (text_response(404, "stream not found"), Conflict, false),
@@ -950,8 +953,10 @@ async fn handle_append_inner(store: Arc<Store>, req: Req, path: String) -> (Resp
     // Serialize per stream: producer validation + write + state update under one
     // lock. Time the wait separately — lock contention is a key bottleneck.
     let lock_t0 = crate::telemetry::Timer::start();
+    let srv_lock_t0 = std::time::Instant::now();
     let mut ap = st.appender.lock().await;
     crate::telemetry::record_append_lock_wait(lock_t0.elapsed_secs());
+    crate::srvstats::record_applock_wait(srv_lock_t0.elapsed());
 
     // Closed checks (precedence: closed → seq regression → gap).
     {
@@ -1083,6 +1088,16 @@ async fn handle_append_inner(store: Arc<Store>, req: Req, path: String) -> (Resp
             Err(_) => ret!(text_response(500, "write failed"), Conflict),
         }
     }
+    // Does this append change state the memory-mode sidecar must persist? Captured
+    // BEFORE `seq_header` is consumed below. Producer/seq updates are idempotency
+    // state; a TTL stream's sliding `last_access` must survive restart (mirrors the
+    // read path, which marks dirty only for TTL streams). A plain append to a non-TTL stream changes
+    // only `durable_tail`/`last_access`. In BOTH modes the durable tail is carried
+    // elsewhere (memory: re-derived from the data-file length on restart; wal: the
+    // checkpoint's per-shard `tails` map), and `last_access` only gates TTL — so a
+    // plain non-TTL append needs no sidecar flush at all (cardinality-cliff #1).
+    let meta_persist_needed =
+        producer.is_some() || seq_header.is_some() || st.config.ttl_seconds.is_some();
     {
         let mut s = st.shared.write().unwrap();
         if let Some(p) = &producer {
@@ -1130,7 +1145,9 @@ async fn handle_append_inner(store: Arc<Store>, req: Req, path: String) -> (Resp
 
     // Wait for durability off the lock before exposing the bytes.
     if let Some(lsn) = staged_lsn {
+        let dur_t0 = std::time::Instant::now();
         wait_durable_lsn(&store, &st, lsn).await;
+        crate::srvstats::record_durwait(dur_t0.elapsed());
     }
 
     // Durable now (wal) / page-cache written (memory): expose the new bytes to
@@ -1165,12 +1182,43 @@ async fn handle_append_inner(store: Arc<Store>, req: Req, path: String) -> (Resp
         // 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 {
+        //
+        // GATED (cardinality-cliff #1): only mark when the append changed state
+        // the sidecar must persist — producer/seq idempotency or a sliding TTL.
+        // A plain append still gets its fdatasync AND its `durable_tail` recorded
+        // in the checkpoint's per-shard `tails` map (register_dirty + the
+        // unconditional `persist_durable_tails`, independent of this flag) — and
+        // that map, not the sidecar, is the authoritative durable-tail proof
+        // recovery reconciles against (see wal/shard.rs step 3a, wal/recovery.rs).
+        // `last_access` only gates TTL. So a plain non-TTL append needs no sidecar
+        // rewrite here — dropping it removes the O(touched) `write_meta_sync` calls
+        // that dominate the checkpoint's meta phase at high stream cardinality.
+        // `wal_meta_gate()` (default on) can be turned off to restore always-mark
+        // for a same-binary A/B of the gate.
+        if meta_persist_needed || !crate::store::wal_meta_gate() {
+            st.meta_dirty.store(true, std::sync::atomic::Ordering::Release);
+        }
+    } else if meta_persist_needed {
         // No WAL record staged (memory durability): no checkpoint will flush
         // the sidecar — queue it for the store-level periodic sweeper. Same
         // batched treatment the wal branch above gets from the checkpoint: no
         // per-stream timer task, no per-append sidecar rewrite (#4691).
+        //
+        // GATED (cardinality-cliff fix): only queue when the append actually
+        // changed state the sidecar must persist — producer/seq idempotency or a
+        // sliding TTL (see `meta_persist_needed`). A plain append to a non-TTL
+        // stream changes only `durable_tail`/`last_access`, and memory-mode
+        // recovery reads NEITHER (the tail is re-derived from the data-file
+        // length in `Store::new_with_tier`; `last_access` only gates TTL expiry,
+        // which these streams don't have). Skipping the queue for that common
+        // case removes the per-append sidecar rewrite whose cost stops amortizing
+        // at high stream cardinality (CARDINALITY_CLIFF_CAUSES.md #1).
+        //
+        // `mem_meta_gate()` (default on) can be turned off to restore always-queue
+        // for a same-binary A/B of the fix.
+        store.mark_meta_dirty(&st);
+    } else if !crate::store::mem_meta_gate() {
+        // A/B baseline (gate off): old behavior — queue every memory-mode append.
         store.mark_meta_dirty(&st);
     }
     if !wire.is_empty() {
@@ -2298,5 +2346,49 @@ mod memory_mode_tests {
 
         let _ = std::fs::remove_dir_all(&dir);
     }
+
+    /// Cardinality-cliff fix (#1): a PLAIN append (no producer/seq, non-TTL
+    /// stream) in memory mode must NOT queue a sidecar flush. The tail is
+    /// recovered from the data-file length and `last_access` only gates TTL, so
+    /// the per-append sidecar rewrite is pure overhead whose cost stops
+    /// amortizing at high stream cardinality. Contrast
+    /// `memory_append_defers_sidecar_to_store_sweep`, which uses a producer
+    /// append and therefore still marks dirty.
+    #[tokio::test]
+    async fn memory_plain_append_skips_sidecar_flush() {
+        let _guard = crate::handlers::test_support::DurabilityGuard::memory();
+        let dir = tmp("mem-plain-noflush");
+        let store =
+            Arc::new(Store::new_with_tier(dir.clone(), TierConfig::default()).unwrap());
+
+        let resp = handle(Arc::clone(&store), put_req("m/p", "application/octet-stream")).await;
+        assert!((200..300).contains(&resp.status), "create: {}", resp.status);
+        // Drain anything the create queued so we measure only the append's effect.
+        let store2 = Arc::clone(&store);
+        let _ = tokio::task::spawn_blocking(move || store2.sweep_meta_once())
+            .await
+            .unwrap();
+
+        // Plain append: no producer headers, non-TTL stream.
+        let resp = handle(
+            Arc::clone(&store),
+            post_req("m/p", "application/octet-stream", b"payload"),
+        )
+        .await;
+        assert!((200..300).contains(&resp.status), "append: {}", resp.status);
+
+        let flushed = tokio::task::spawn_blocking({
+            let store = Arc::clone(&store);
+            move || store.sweep_meta_once()
+        })
+        .await
+        .unwrap();
+        assert_eq!(
+            flushed, 0,
+            "a plain non-TTL memory-mode append must not queue a sidecar flush"
+        );
+
+        let _ = std::fs::remove_dir_all(&dir);
+    }
 }
 

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

@@ -3,6 +3,7 @@ mod blobstore;
 mod engine_raw;
 mod handlers;
 mod http1;
+mod srvstats;
 #[cfg(target_os = "linux")]
 mod sse_reactor;
 mod store;
@@ -137,6 +138,7 @@ fn main() {
     // append path). Dependency-free — the measurement vehicle for the contention
     // investigation, independent of the heavy `telemetry` OTLP feature.
     let mut wal_stats_secs: Option<u64> = None;
+    let mut server_stats_secs: Option<u64> = None;
     let mut args = std::env::args().skip(1);
     while let Some(a) = args.next() {
         match a.as_str() {
@@ -238,6 +240,66 @@ fn main() {
                     }
                 }
             }
+            // Dev/benchmark toggles for the cardinality-cliff investigation.
+            // `--meta-sweep-disable` drops the memory-mode sidecar write (makes
+            // the sidecar permanently stale — bench-only); `--meta-sweep-stats`
+            // logs a META_SWEEP line per tick.
+            "--meta-sweep-disable" => store::set_meta_sweep_disable(true),
+            "--meta-sweep-stats" => store::set_meta_sweep_stats(true),
+            // Periodic SRV_STATS line (both modes): cpu_cores / inflight / service
+            // + appender-lock + durability wait — bottleneck analysis.
+            "--server-stats" => {
+                let n: u64 = parse_val(args.next(), "--server-stats");
+                if n == 0 {
+                    eprintln!("--server-stats must be ≥ 1 (seconds)");
+                    std::process::exit(2);
+                }
+                server_stats_secs = Some(n);
+            }
+            "--wal-meta-gate" => {
+                let v = val(args.next(), "--wal-meta-gate");
+                match v.as_str() {
+                    "on" => store::set_wal_meta_gate(true),
+                    "off" => store::set_wal_meta_gate(false),
+                    _ => {
+                        eprintln!("--wal-meta-gate must be on|off");
+                        std::process::exit(2);
+                    }
+                }
+            }
+            "--mem-meta-gate" => {
+                let v = val(args.next(), "--mem-meta-gate");
+                match v.as_str() {
+                    "on" => store::set_mem_meta_gate(true),
+                    "off" => store::set_mem_meta_gate(false),
+                    _ => {
+                        eprintln!("--mem-meta-gate must be on|off");
+                        std::process::exit(2);
+                    }
+                }
+            }
+            // Checkpoint fdatasync fan-out (H4). `1` = serial baseline.
+            "--wal-fsync-parallel" => {
+                let n: u64 = parse_val(args.next(), "--wal-fsync-parallel");
+                if n == 0 {
+                    eprintln!("--wal-fsync-parallel must be ≥ 1");
+                    std::process::exit(2);
+                }
+                wal::shard::set_fsync_fanout(n);
+            }
+            // Checkpoint durability via ONE syncfs() barrier instead of the
+            // O(N_touched) per-stream fdatasync loop (cardinality-cliff #1). Linux-only.
+            "--wal-checkpoint-syncfs" => {
+                let v = val(args.next(), "--wal-checkpoint-syncfs");
+                match v.as_str() {
+                    "on" => wal::shard::set_checkpoint_syncfs(true),
+                    "off" => wal::shard::set_checkpoint_syncfs(false),
+                    _ => {
+                        eprintln!("--wal-checkpoint-syncfs must be on|off");
+                        std::process::exit(2);
+                    }
+                }
+            }
             other => {
                 eprintln!("unknown argument: {other}");
                 std::process::exit(2);
@@ -303,6 +365,11 @@ fn main() {
         // touches here (its append path flushes via the checkpoint instead).
         spawn_meta_sweeper(Arc::clone(&store));
 
+        // Server load telemetry (both modes) for bottleneck analysis.
+        if let Some(secs) = server_stats_secs {
+            srvstats::spawn(secs);
+        }
+
         // ---- WAL wiring (Wal mode only) ----
         //
         // Skipped entirely in `--durability memory` mode — no WAL is opened,

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

@@ -0,0 +1,140 @@
+//! Lightweight, always-cheap server-side load telemetry for bottleneck analysis
+//! (cardinality-cliff performance work). Answers the core question — is the server
+//! CPU-bound, fsync/durability-bound, or lock-bound? — for BOTH wal and memory
+//! modes, which the WAL-only `--wal-stats` counters cannot.
+//!
+//! Enabled by `--server-stats N` (seconds). Off by default and gated by
+//! `STATS_ON`, so the hot-path instrumentation is a single relaxed load + branch
+//! when disabled. Each tick prints a `SRV_STATS` line and resets the interval
+//! accumulators.
+//!
+//! Fields:
+//! - `cpu_cores`   process CPU utilization in cores (utime+stime delta / wall);
+//!                 ≈ the cgroup cpu quota ⇒ CPU-bound. Linux only (`-1` elsewhere).
+//! - `appends_s`   acked appends/sec over the interval.
+//! - `inflight`    in-flight append handlers sampled at tick time (queue depth).
+//! - `svc_us`      mean append handler wall time (service time).
+//! - `applock_us`  mean time waiting to acquire the per-stream appender lock.
+//! - `durwait_us`  mean time in `wait_durable_lsn` (WAL fsync wait; ~0 in memory).
+
+use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering};
+use std::time::Instant;
+
+static STATS_ON: AtomicBool = AtomicBool::new(false);
+static APPENDS: AtomicU64 = AtomicU64::new(0);
+static INFLIGHT: AtomicI64 = AtomicI64::new(0);
+static SVC_US: AtomicU64 = AtomicU64::new(0);
+static APPLOCK_US: AtomicU64 = AtomicU64::new(0);
+static DURWAIT_US: AtomicU64 = AtomicU64::new(0);
+
+pub fn set_enabled(v: bool) {
+    STATS_ON.store(v, Ordering::Relaxed);
+}
+#[inline]
+pub fn enabled() -> bool {
+    STATS_ON.load(Ordering::Relaxed)
+}
+
+/// RAII probe for one append handler: bumps the in-flight gauge on creation and,
+/// on drop (covering every early return), records the service time and counts the
+/// append. Create it once `enabled()` is true.
+pub struct AppendProbe {
+    start: Instant,
+}
+impl AppendProbe {
+    #[inline]
+    pub fn start() -> Option<Self> {
+        if !enabled() {
+            return None;
+        }
+        INFLIGHT.fetch_add(1, Ordering::Relaxed);
+        Some(Self { start: Instant::now() })
+    }
+}
+impl Drop for AppendProbe {
+    fn drop(&mut self) {
+        INFLIGHT.fetch_sub(1, Ordering::Relaxed);
+        SVC_US.fetch_add(self.start.elapsed().as_micros() as u64, Ordering::Relaxed);
+        APPENDS.fetch_add(1, Ordering::Relaxed);
+    }
+}
+
+#[inline]
+pub fn record_applock_wait(d: std::time::Duration) {
+    if enabled() {
+        APPLOCK_US.fetch_add(d.as_micros() as u64, Ordering::Relaxed);
+    }
+}
+#[inline]
+pub fn record_durwait(d: std::time::Duration) {
+    if enabled() {
+        DURWAIT_US.fetch_add(d.as_micros() as u64, Ordering::Relaxed);
+    }
+}
+
+/// Read process CPU time (utime+stime) in seconds from `/proc/self/stat`. Linux
+/// only; `None` elsewhere (macOS dev boxes — use a Linux container to profile).
+#[cfg(target_os = "linux")]
+fn cpu_secs() -> Option<f64> {
+    let s = std::fs::read_to_string("/proc/self/stat").ok()?;
+    // comm (field 2) may contain spaces/parens — parse after the final ')'.
+    let rest = &s[s.rfind(')')? + 1..];
+    let f: Vec<&str> = rest.split_whitespace().collect();
+    // After ')': [state, ppid, ...]; utime is field 14 ⇒ index 11, stime ⇒ 12.
+    let utime: f64 = f.get(11)?.parse().ok()?;
+    let stime: f64 = f.get(12)?.parse().ok()?;
+    let hz = 100.0; // _SC_CLK_TCK is 100 on all our targets.
+    Some((utime + stime) / hz)
+}
+#[cfg(not(target_os = "linux"))]
+fn cpu_secs() -> Option<f64> {
+    None
+}
+
+/// Spawn the periodic printer. Each tick emits one `SRV_STATS` line and resets the
+/// interval accumulators.
+pub fn spawn(secs: u64) {
+    set_enabled(true);
+    tokio::spawn(async move {
+        let mut ticker = tokio::time::interval(std::time::Duration::from_secs(secs));
+        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+        ticker.tick().await;
+        let mut last_cpu = cpu_secs();
+        let mut last = Instant::now();
+        loop {
+            ticker.tick().await;
+            let now = Instant::now();
+            let wall = now.duration_since(last).as_secs_f64();
+            last = now;
+
+            let appends = APPENDS.swap(0, Ordering::Relaxed);
+            let svc = SVC_US.swap(0, Ordering::Relaxed);
+            let applock = APPLOCK_US.swap(0, Ordering::Relaxed);
+            let durwait = DURWAIT_US.swap(0, Ordering::Relaxed);
+            let inflight = INFLIGHT.load(Ordering::Relaxed);
+
+            let cpu_cores = match (cpu_secs(), last_cpu) {
+                (Some(c), Some(p)) if wall > 0.0 => {
+                    last_cpu = Some(c);
+                    (c - p) / wall
+                }
+                (Some(c), _) => {
+                    last_cpu = Some(c);
+                    -1.0
+                }
+                _ => -1.0,
+            };
+
+            let n = appends.max(1) as f64;
+            eprintln!(
+                "SRV_STATS cpu_cores={:.2} appends_s={:.0} inflight={} svc_us={:.0} applock_us={:.1} durwait_us={:.1}",
+                cpu_cores,
+                appends as f64 / wall.max(0.001),
+                inflight,
+                svc as f64 / n,
+                applock as f64 / n,
+                durwait as f64 / n,
+            );
+        }
+    });
+}