cubesys: coalesce WAL fsync on commit; stop over-reporting durable seq

commit_txn issued an unconditional fsync per COMMIT, so N concurrent
writers serialized behind N disk syncs (p99 hit the 3s socket timeout
under 8 writers). Add Wal::sync_upto: committers queue on an fsync_gate,
the first one flushes the whole accumulated buffer, and waiters that find
committed_seq past their target return with zero I/O. N commits now cost
~1 fsync with the same durability guarantee.

Also fix a durability over-report: flush_pending stamped committed_seq
from the LIVE seq counter, so sequences taken by appenders that had not
yet buffered their bytes were reported durable. Track max_seq alongside
the pending buffer and advance committed_seq only to what was written.

Wire --wal-fsync-ms / --checkpoint-ms in cube-server (previously
hardcoded to defaults, so the documented knob did nothing).

Both regressions are mutation-verified: each test fails when its bug is
reintroduced.

cubesys/src/bin/cube-server.rs

@@ -362,6 +362,20 @@ fn main() {
     let store_path = arg(&args, "--store").unwrap_or_else(default_store);
     let recovery_log = arg(&args, "--recovery-log").unwrap_or_else(default_recovery);
     let tenant_dir = arg(&args, "--tenant-dir");
+    // Durability tuning, now actually settable from the command line (these
+    // were previously hardcoded to DurabilityConfig::default(), so the
+    // documented --wal-fsync-ms knob had no effect at all).
+    let durability = {
+        let d = DurabilityConfig::default();
+        DurabilityConfig {
+            wal_fsync_ms: arg(&args, "--wal-fsync-ms")
+                .and_then(|v| v.parse().ok())
+                .unwrap_or(d.wal_fsync_ms),
+            checkpoint_ms: arg(&args, "--checkpoint-ms")
+                .and_then(|v| v.parse().ok())
+                .unwrap_or(d.checkpoint_ms),
+        }
+    };
     let (_deny_unknown, allow_anonymous) = (
         args.iter().any(|a| a == "--deny-unknown-tenant"),
         args.iter().any(|a| a == "--allow-anonymous"),
@@ -408,7 +422,7 @@ fn main() {
         Some(dir) => {
             // Real per-tenant isolation: each tenant gets its own durable store
             // under `dir/<tenant>/`.
-            let cfg = TenantConfig::disk(PathBuf::from(dir));
+            let cfg = TenantConfig::disk_with(PathBuf::from(dir), durability);
             (Arc::new(TenantRegistry::with_config(cfg)), true)
         }
         None => {
@@ -419,7 +433,7 @@ fn main() {
                 db: PathBuf::from(&store_path),
                 wal: PathBuf::from(format!("{store_path}.wal")),
                 recovery: PathBuf::from(&recovery_log),
-                durability: DurabilityConfig::default(),
+                durability,
             };
             let reg = TenantRegistry::with_config(cfg.clone());
             let shared = match TenantSession::open(

cubesys/src/store.rs

@@ -98,6 +98,22 @@ pub struct TxnEntry {
     pub value: Option<Vec<u8>>,
 }
 
+/// Buffered-but-not-yet-fsync'd WAL bytes, plus the highest sequence number
+/// contained in them.
+///
+/// Tracking `max_seq` alongside the buffer is what makes `committed_seq`
+/// honest: a flush may only advance `committed_seq` to the highest seq that
+/// was actually in the batch it wrote, NOT to whatever the global `seq`
+/// counter happens to read at fsync time. Concurrent appenders bump `seq`
+/// between the buffer swap and the fsync, and claiming those as durable would
+/// over-report durability (telemetry would say data survived a crash that in
+/// fact never reached the platter).
+#[derive(Default)]
+struct Pending {
+    buf: String,
+    max_seq: u64,
+}
+
 /// The write-ahead log. Append-only NDJSON, group-fsync'd on a timer.
 struct Wal {
     path: PathBuf,
@@ -106,7 +122,13 @@ struct Wal {
     /// Next sequence number to assign.
     seq: AtomicU64,
     /// Buffered NDJSON lines not yet fsync'd (group commit).
-    pending: Mutex<String>,
+    pending: Mutex<Pending>,
+    /// Serializes fsync attempts so concurrent committers COALESCE into one
+    /// fsync instead of each paying for their own. A waiter re-checks
+    /// `committed_seq` after acquiring this gate: if the thread ahead of it
+    /// already flushed past its target seq, it returns without any I/O. This
+    /// is the group-commit win — N concurrent COMMITs cost ~1 fsync.
+    fsync_gate: Mutex<()>,
     /// Highest sequence number known to be fsync'd to disk.
     committed_seq: AtomicU64,
     /// WAL sequence number already folded into the base snapshot. Replay on
@@ -157,7 +179,8 @@ impl Wal {
             path: path.to_path_buf(),
             file: Mutex::new(Some(file)),
             seq: AtomicU64::new(start_seq),
-            pending: Mutex::new(String::new()),
+            pending: Mutex::new(Pending::default()),
+            fsync_gate: Mutex::new(()),
             committed_seq: AtomicU64::new(max_seq.max(cp_seq)),
             cp_seq_wal: AtomicU64::new(cp_seq),
             base_seq: AtomicU64::new(cp_seq),
@@ -177,7 +200,8 @@ impl Wal {
             path: PathBuf::new(),
             file: Mutex::new(None),
             seq: AtomicU64::new(1),
-            pending: Mutex::new(String::new()),
+            pending: Mutex::new(Pending::default()),
+            fsync_gate: Mutex::new(()),
             committed_seq: AtomicU64::new(0),
             cp_seq_wal: AtomicU64::new(0),
             base_seq: AtomicU64::new(0),
@@ -211,11 +235,13 @@ impl Wal {
         self.dirty.lock().unwrap().clear();
     }
 
-    /// Append a transaction. In memory mode this is a no-op.
-    fn append(&self, op: WalOp, coord: Czyx, value: Vec<u8>) {
+    /// Append a transaction. In memory mode this is a no-op. Returns the
+    /// sequence number assigned (0 in memory mode) so a caller that needs
+    /// synchronous durability can wait for exactly that seq.
+    fn append(&self, op: WalOp, coord: Czyx, value: Vec<u8>) -> u64 {
         let has_file = self.file.lock().unwrap().is_some();
         if !has_file {
-            return;
+            return 0;
         }
         let s = self.seq.fetch_add(1, Ordering::SeqCst) + 1;
         self.mark_dirty(coord);
@@ -227,17 +253,21 @@ impl Wal {
             batch: Vec::new(),
         });
         let mut p = self.pending.lock().unwrap();
-        p.push_str(&line);
-        p.push('\n');
+        p.buf.push_str(&line);
+        p.buf.push('\n');
+        if s > p.max_seq {
+            p.max_seq = s;
+        }
+        s
     }
 
     /// Append a transaction commit. In memory mode this is a no-op. The whole
     /// batch lands as ONE WAL line (`op:"txn"`), so a crash mid-txn replays as
-    /// a single all-or-nothing unit.
-    fn append_txn(&self, batch: &[TxnEntry]) {
+    /// a single all-or-nothing unit. Returns the assigned sequence number.
+    fn append_txn(&self, batch: &[TxnEntry]) -> u64 {
         let has_file = self.file.lock().unwrap().is_some();
         if !has_file {
-            return;
+            return 0;
         }
         let s = self.seq.fetch_add(1, Ordering::SeqCst) + 1;
         for e in batch {
@@ -251,28 +281,64 @@ impl Wal {
             batch: batch.to_vec(),
         });
         let mut p = self.pending.lock().unwrap();
-        p.push_str(&line);
-        p.push('\n');
+        p.buf.push_str(&line);
+        p.buf.push('\n');
+        if s > p.max_seq {
+            p.max_seq = s;
+        }
+        s
     }
 
     /// Flush buffered lines to disk and fsync them (group commit).
+    ///
+    /// `committed_seq` advances ONLY to the highest seq actually present in the
+    /// bytes this call wrote — never to the live `seq` counter, which
+    /// concurrent appenders may have already pushed past our batch.
     fn flush_pending(&self) {
-        let bytes = {
+        let _gate = self.fsync_gate.lock().unwrap();
+        self.flush_locked();
+    }
+
+    /// Flush the current buffer. Caller must hold `fsync_gate`.
+    fn flush_locked(&self) {
+        let (bytes, batch_max) = {
             let mut p = self.pending.lock().unwrap();
-            if p.is_empty() {
+            if p.buf.is_empty() {
                 return;
             }
-            std::mem::take(&mut *p)
+            let taken = std::mem::take(&mut *p);
+            (taken.buf, taken.max_seq)
         };
         let mut fopt = self.file.lock().unwrap();
         if let Some(f) = fopt.as_mut() {
             if f.write_all(bytes.as_bytes()).is_ok() && f.flush().is_ok() && f.sync_all().is_ok() {
-                self.committed_seq
-                    .store(self.seq.load(Ordering::SeqCst), Ordering::SeqCst);
+                self.committed_seq.fetch_max(batch_max, Ordering::SeqCst);
             }
         }
     }
 
+    /// Make sequence `target` durable, COALESCING with any concurrent flush.
+    ///
+    /// This is the batch-fsync core. Instead of every COMMIT issuing its own
+    /// `sync_all` (which serialized N writers behind N fsyncs and produced the
+    /// multi-second p99 stalls under 8 concurrent writers), committers queue on
+    /// `fsync_gate`. Whoever gets in first flushes the WHOLE accumulated
+    /// buffer — including the lines appended by everyone still waiting. Those
+    /// waiters then observe `committed_seq >= target` and return with zero
+    /// I/O. N concurrent commits therefore cost ~1 fsync, not N, while every
+    /// caller still gets a true durability guarantee before returning.
+    fn sync_upto(&self, target: u64) {
+        if target == 0 || self.committed_seq.load(Ordering::SeqCst) >= target {
+            return;
+        }
+        let _gate = self.fsync_gate.lock().unwrap();
+        // Re-check: the holder ahead of us may already have flushed past us.
+        if self.committed_seq.load(Ordering::SeqCst) >= target {
+            return;
+        }
+        self.flush_locked();
+    }
+
     /// Replay entries with `seq > cp_seq` into `store`. Returns the number of
     /// entries applied (0 means the checkpoint alone was sufficient).
     fn replay_after(&self, cp_seq: u64, store: &mut CubeStore<HashBackend>) -> u64 {
@@ -565,9 +631,11 @@ impl ConcurrentStore {
         f(&mut g)
     }
 
-    /// Durability-log a put whose bytes were written via [`with_mut`].
+    /// Durability-log a put whose bytes were written via [`with_mut`]. The
+    /// append is buffered; the group-commit thread fsyncs it within
+    /// `wal_fsync_ms`. Callers needing an immediate guarantee use `commit_txn`.
     pub fn log_put(&self, key: Czyx, value: Vec<u8>) {
-        self.wal.append(WalOp::Put, key, value);
+        let _seq = self.wal.append(WalOp::Put, key, value);
     }
 
     /// Apply a transaction batch atomically: under ONE store write lock, apply
@@ -577,6 +645,11 @@ impl ConcurrentStore {
     /// The WAL is fsync'd before returning so a `COMMIT` is durable the moment
     /// the caller gets control back (not merely "eventually" via the group
     /// thread). This is what makes `commit` a real transaction boundary.
+    ///
+    /// The fsync is COALESCED via [`Wal::sync_upto`]: concurrent committers
+    /// share one fsync rather than each paying for their own, so the durability
+    /// guarantee is unchanged while N writers no longer serialize behind N
+    /// disk syncs.
     pub fn commit_txn(&self, batch: &[TxnEntry]) {
         let mut g = self.inner.write().unwrap();
         for e in batch {
@@ -586,8 +659,8 @@ impl ConcurrentStore {
             }
         }
         drop(g); // release the write lock before touching the WAL
-        self.wal.append_txn(batch);
-        self.wal.flush_pending(); // synchronous fsync: COMMIT == durable
+        let seq = self.wal.append_txn(batch);
+        self.wal.sync_upto(seq); // coalesced fsync: COMMIT == durable
     }
 
     /// Store a code cell at `path` (the path->code bridge), durability-logged.
@@ -1139,6 +1212,106 @@ mod tests {
         cleanup(&db);
     }
 
+    /// Regression: `committed_seq` must never exceed what was actually fsync'd.
+    ///
+    /// The old `flush_pending` stamped `committed_seq = seq.load()` — the LIVE
+    /// counter. `Wal::append` bumps `seq` (fetch_add) and pushes its bytes into
+    /// `pending` under a *separate* lock, so there is a real window where `seq`
+    /// has advanced but the bytes are not yet buffered. A flush landing in that
+    /// window would mark the absent seq durable.
+    ///
+    /// This test drives that window deterministically rather than relying on
+    /// thread timing: it advances `seq` exactly as `append`'s `fetch_add` does,
+    /// WITHOUT buffering bytes, then flushes. `wal_durable_seq()` must not
+    /// claim the phantom sequence.
+    #[test]
+    fn durable_seq_never_over_reports() {
+        let db = tmp("durseq");
+        let wal_p = format!("{db}.wal");
+        let rec = format!("{db}.rec");
+        cleanup(&db);
+        let s = ConcurrentStore::open(&db, &wal_p, &rec, DurabilityConfig::default()).unwrap();
+
+        // One real, fully-buffered write, then flush it.
+        s.put_raw(Czyx::new(9, 0, 0, 1), vec![1]);
+        s.wal.flush_pending();
+        let after_real = s.wal_durable_seq();
+
+        // Buffer a real write (takes seq N and pushes its bytes)...
+        s.put_raw(Czyx::new(9, 0, 0, 2), vec![2]);
+        // ...then reproduce the window: a concurrent appender takes seq N+1 but
+        // has not yet pushed its bytes. It must NOT be counted as durable.
+        let phantom = s.wal.seq.fetch_add(1, Ordering::SeqCst) + 1;
+        s.wal.flush_pending();
+
+        let claimed = s.wal_durable_seq();
+        let on_disk = fs::read_to_string(&wal_p)
+            .unwrap_or_default()
+            .lines()
+            .filter_map(decode_wal)
+            .map(|e| e.seq)
+            .max()
+            .unwrap_or(0);
+
+        assert!(
+            claimed <= on_disk,
+            "durability over-reported: claimed durable seq {claimed} but only {on_disk} is on disk \
+             (phantom seq {phantom} was never written)"
+        );
+        assert!(
+            claimed > after_real,
+            "flush failed to advance durability at all: {claimed} vs {after_real}"
+        );
+        s.shutdown();
+        cleanup(&db);
+    }
+
+    /// Concurrent commits must COALESCE into shared fsyncs while EVERY commit
+    /// still returns genuinely durable. This crashes the store without any
+    /// checkpoint, so recovery is pure WAL replay: if the coalescing logic ever
+    /// lets a committer return before its own bytes are fsync'd, a coord goes
+    /// missing here.
+    #[test]
+    fn concurrent_commits_are_durable_and_coalesced() {
+        let db = tmp("coalesce");
+        let wal = format!("{db}.wal");
+        let rec = format!("{db}.rec");
+        cleanup(&db);
+        {
+            let s = Arc::new(
+                ConcurrentStore::open(&db, &wal, &rec, DurabilityConfig::default()).unwrap(),
+            );
+            let mut hs = Vec::new();
+            for t in 0..8u32 {
+                let s = s.clone();
+                hs.push(thread::spawn(move || {
+                    for i in 0..25u32 {
+                        let x = (t * 25 + i) as u8;
+                        s.commit_txn(&[TxnEntry {
+                            coord: Czyx::new(8, 0, 0, x),
+                            value: Some(vec![x]),
+                        }]);
+                    }
+                }));
+            }
+            for h in hs {
+                h.join().unwrap();
+            }
+            // Crash WITHOUT checkpoint: every commit must still be recoverable
+            // purely from the fsync'd WAL.
+        }
+        let s = ConcurrentStore::open(&db, &wal, &rec, DurabilityConfig::default()).unwrap();
+        let keys = s.keys();
+        for x in 0..200u8 {
+            assert!(
+                keys.contains(&Czyx::new(8, 0, 0, x)),
+                "commit at x={x} was lost — coalesced fsync broke the durability guarantee"
+            );
+        }
+        s.shutdown();
+        cleanup(&db);
+    }
+
     #[test]
     fn concurrent_writes() {
         let s = Arc::new(ConcurrentStore::memory());

cubesys/src/tenant.rs

@@ -110,6 +110,16 @@ impl TenantConfig {
             durability: DurabilityConfig::default(),
         }
     }
+
+    /// Disk-backed tenant config at `store_dir` with explicit durability
+    /// tuning, so the daemon's `--wal-fsync-ms` / `--checkpoint-ms` flags
+    /// actually reach each per-tenant store's WAL.
+    pub fn disk_with(store_dir: impl Into<PathBuf>, durability: DurabilityConfig) -> Self {
+        TenantConfig::Disk {
+            store_dir: store_dir.into(),
+            durability,
+        }
+    }
 }
 
 /// A client's asserted identity, declared via `HELLO <tenant> <owner_local>`