From ec84fbc7258fcaeb5279e34b801c3715b701931c Mon Sep 17 00:00:00 2001 From: CUBELinux-2 Date: Tue, 11 Aug 2026 18:09:29 -0400 Subject: [PATCH] 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 | 18 ++- cubesys/src/store.rs | 217 +++++++++++++++++++++++++++++---- cubesys/src/tenant.rs | 10 ++ 3 files changed, 221 insertions(+), 24 deletions(-) diff --git a/cubesys/src/bin/cube-server.rs b/cubesys/src/bin/cube-server.rs index 439184a..fb389bf 100644 --- a/cubesys/src/bin/cube-server.rs +++ b/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//`. - 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( diff --git a/cubesys/src/store.rs b/cubesys/src/store.rs index c9f874e..c6a7ff6 100644 --- a/cubesys/src/store.rs +++ b/cubesys/src/store.rs @@ -98,6 +98,22 @@ pub struct TxnEntry { pub value: Option>, } +/// 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, + pending: Mutex, + /// 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) { + /// 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) -> 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) -> 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) { - 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()); diff --git a/cubesys/src/tenant.rs b/cubesys/src/tenant.rs index ce264ab..9b71f3a 100644 --- a/cubesys/src/tenant.rs +++ b/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, durability: DurabilityConfig) -> Self { + TenantConfig::Disk { + store_dir: store_dir.into(), + durability, + } + } } /// A client's asserted identity, declared via `HELLO `