diff --git a/.changeset/memory-mode-meta-sweep.md b/.changeset/memory-mode-meta-sweep.md new file mode 100644 index 0000000000..b0eb5d6074 --- /dev/null +++ b/.changeset/memory-mode-meta-sweep.md @@ -0,0 +1,7 @@ +--- +"@electric-ax/durable-streams-server-rust": patch +--- + +Memory-mode CPU fix: batch meta sidecar flushes into a periodic sweep (#4691). + +`--durability memory` appends no longer schedule a per-stream debounced sidecar flush (a timer task + full sidecar rewrite per stream per 100 ms — ~5x wal-mode CPU at high stream cardinality under low per-stream rates). Appends and TTL read touches now only mark the stream dirty in a store-level set; a single 1 s sweeper flushes all dirty sidecars in one pass, mirroring the batched checkpoint treatment wal mode got in the write-path overhaul. The sidecar's producer/access state remains a non-durable lagging flush; its lag bound moves from 100 ms to the 1 s sweep cadence. Durable flush-on-close/delete paths are unchanged, and a pending flush can no longer resurrect the sidecar of a hard-deleted stream. diff --git a/.github/workflows/auto-approve-durable-streams.yml b/.github/workflows/auto-approve-durable-streams.yml index b1e29aa221..717888a6fc 100644 --- a/.github/workflows/auto-approve-durable-streams.yml +++ b/.github/workflows/auto-approve-durable-streams.yml @@ -24,6 +24,7 @@ jobs: run: | scopes=( "packages/durable-streams-rust/" + ".changeset/" ) in_scope=true diff --git a/packages/durable-streams-rust/src/handlers.rs b/packages/durable-streams-rust/src/handlers.rs index 43d9775f52..a4d2c74e04 100644 --- a/packages/durable-streams-rust/src/handlers.rs +++ b/packages/durable-streams-rust/src/handlers.rs @@ -1167,9 +1167,11 @@ async fn handle_append_inner(store: Arc, req: Req, path: String) -> (Resp // 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(); + // 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). + store.mark_meta_dirty(&st); } if !wire.is_empty() { maybe_seal_bg(&store, &st); @@ -1416,7 +1418,7 @@ async fn handle_read(store: Arc, req: Req, path: String) -> Resp { // non-TTL streams to keep their read path lock-free. if st.config.ttl_seconds.is_some() { st.touch(); - st.schedule_meta_flush(); // sliding TTL must survive restarts + store.mark_meta_dirty(&st); // sliding TTL must survive restarts } let q = match parse_query(req.query.as_deref()) { Ok(q) => q, @@ -2241,5 +2243,60 @@ mod memory_mode_tests { let _ = std::fs::remove_dir_all(&dir); } + + /// #4691: a memory-mode append must NOT flush the meta sidecar via a + /// per-stream debounced timer (100 ms sleep + spawn_blocking per stream — + /// ~5x wal-mode CPU at high stream cardinality). It only marks the stream + /// dirty; the store-level periodic sweeper writes the sidecar in batch, + /// mirroring wal mode's checkpoint treatment from #4675. + #[tokio::test] + async fn memory_append_defers_sidecar_to_store_sweep() { + let _guard = crate::handlers::test_support::DurabilityGuard::memory(); + let dir = tmp("mem-sweep"); + let store = Arc::new( + Store::new_with_tier(dir.clone(), TierConfig::default()).unwrap(), + ); + + let resp = handle(Arc::clone(&store), put_req("m/s", "application/octet-stream")).await; + assert!((200..300).contains(&resp.status), "create: {}", resp.status); + + // Append with a producer so the pending sidecar change is observable. + let mut req = post_req("m/s", "application/octet-stream", b"payload"); + req.headers.push(("producer-id".into(), "p1".into())); + req.headers.push(("producer-epoch".into(), "1".into())); + req.headers.push(("producer-seq".into(), "0".into())); + let resp = handle(Arc::clone(&store), req).await; + assert!((200..300).contains(&resp.status), "append: {}", resp.status); + + // Well past the old 100 ms debounce: the sidecar must still be + // unwritten — no per-append timer task may exist anymore. + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + let st = store.get("m/s").unwrap(); + let meta: crate::store::Meta = + serde_json::from_slice(&std::fs::read(crate::store::meta_path(&st.file_path)).unwrap()) + .unwrap(); + assert!( + !meta.producers.contains_key("p1"), + "sidecar was flushed per-append (debounce timer still active)" + ); + + // The batched store sweep is what persists it. + let flushed = tokio::task::spawn_blocking({ + let store = Arc::clone(&store); + move || store.sweep_meta_once() + }) + .await + .unwrap(); + assert_eq!(flushed, 1, "the appended stream is swept"); + let meta: crate::store::Meta = + serde_json::from_slice(&std::fs::read(crate::store::meta_path(&st.file_path)).unwrap()) + .unwrap(); + assert!( + meta.producers.contains_key("p1"), + "sweep must persist the pending producer state" + ); + + let _ = std::fs::remove_dir_all(&dir); + } } diff --git a/packages/durable-streams-rust/src/main.rs b/packages/durable-streams-rust/src/main.rs index 5e838b1fe8..721240c4ec 100644 --- a/packages/durable-streams-rust/src/main.rs +++ b/packages/durable-streams-rust/src/main.rs @@ -296,6 +296,12 @@ fn main() { let store = Arc::new( Store::new_with_tier(data_dir.clone(), tier.clone()).expect("failed to init store"), ); + // Batched meta-sidecar sweeper (#4691): flushes every stream queued by + // `Store::mark_meta_dirty` (memory-mode appends, TTL read touches) in + // one pass per tick, replacing the per-stream 100 ms debounce timer. + // Spawned in BOTH durability modes — wal mode still queues TTL read + // touches here (its append path flushes via the checkpoint instead). + spawn_meta_sweeper(Arc::clone(&store)); // ---- WAL wiring (Wal mode only) ---- // @@ -442,6 +448,33 @@ fn spawn_checkpoint_ticker(walset: Arc) { }); } +/// How often the meta sweeper flushes dirty sidecars (#4691). The sidecar's +/// producer/access state is a non-durable, lagging flush by contract; 1 s keeps +/// the lag tighter than the wal checkpoint's 3 s cadence while still batching +/// away the per-stream timer + per-append rewrite the 100 ms debounce cost. +const META_SWEEP_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1); + +/// Spawn the store-level meta-sidecar sweeper: each tick drains the +/// `mark_meta_dirty` queue and writes every still-dirty stream's sidecar in one +/// `spawn_blocking` task (vs one timer task + one blocking task PER STREAM per +/// 100 ms under the old debounce — the ~5x memory-mode CPU overhead of #4691). +fn spawn_meta_sweeper(store: Arc) { + tokio::spawn(async move { + let mut ticker = tokio::time::interval(META_SWEEP_INTERVAL); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + // Skip the immediate first tick — nothing can be dirty at boot. + ticker.tick().await; + loop { + ticker.tick().await; + if store.meta_sweep.lock().unwrap().is_empty() { + continue; + } + let s = Arc::clone(&store); + let _ = tokio::task::spawn_blocking(move || s.sweep_meta_once()).await; + } + }); +} + /// Resolve on SIGINT (Ctrl-C) or SIGTERM (systemd/Kubernetes stop). On non-Unix, /// only Ctrl-C. async fn shutdown_signal() { diff --git a/packages/durable-streams-rust/src/store.rs b/packages/durable-streams-rust/src/store.rs index 8a99261309..1b69a5ab30 100644 --- a/packages/durable-streams-rust/src/store.rs +++ b/packages/durable-streams-rust/src/store.rs @@ -382,6 +382,11 @@ pub struct Store { /// the `WalSet`, runs WAL recovery, and `set`s it here — all before serving. /// The hot-path read (`store.wal.get()`) is lock-free. pub wal: std::sync::OnceLock>, + /// Streams with a pending non-durable sidecar flush (memory-mode appends, + /// TTL read touches), drained in batch by the periodic meta sweeper + /// (`sweep_meta_once`). The `meta_dirty` CAS in `mark_meta_dirty` keeps + /// each stream in here at most once per sweep cycle (#4691). + pub meta_sweep: StdMutex>>, } pub enum CreateResult { @@ -421,6 +426,7 @@ impl Store { tier_config, blobstore, wal: std::sync::OnceLock::new(), + meta_sweep: StdMutex::new(Vec::new()), }; store.recover(&streams_dir)?; Ok(store) @@ -1147,22 +1153,53 @@ pub(crate) fn fsync_parent_dir(path: &std::path::Path) -> std::io::Result<()> { } } -impl StreamState { - /// Schedule a debounced, non-durable meta flush (producer/access updates). - pub fn schedule_meta_flush(self: &Arc) { - if self +impl Store { + /// Queue a non-durable meta sidecar flush (producer/access updates) for the + /// next periodic sweep (#4691). Replaces the per-stream 100 ms debounce + /// timer, which cost a tokio timer task + `spawn_blocking` + full sidecar + /// rewrite per stream per 100 ms — at high stream cardinality ~5x wal-mode + /// CPU for the same load. The lag bound moves from the 100 ms debounce to + /// the sweep cadence, exactly the trade wal mode made in #4675. + /// + /// The `meta_dirty` CAS dedupes: while a flush is pending the stream sits + /// in `meta_sweep` at most once. The wal-mode append path never calls this + /// — it stores `meta_dirty` directly and the shard checkpoint flushes the + /// sidecar (see `handle_append_inner`); if that path already set the flag, + /// the checkpoint owns the flush and the CAS failing here avoids a + /// duplicate write. + pub fn mark_meta_dirty(&self, st: &Arc) { + if st .meta_dirty .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) - .is_err() + .is_ok() { - return; // flush already scheduled + self.meta_sweep.lock().unwrap().push(Arc::clone(st)); } - let st = self.clone(); - tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(100)).await; - st.meta_dirty.store(false, Ordering::Release); - let _ = tokio::task::spawn_blocking(move || write_meta_sync(&st, false)).await; - }); + } + + /// Drain the pending sweep set and write each still-dirty stream's sidecar + /// (non-durable), returning how many were written. Blocking file I/O — + /// call from a blocking context. Errors are ignored exactly like the + /// debounced flush ignored them (the sidecar is non-durable by contract). + pub fn sweep_meta_once(&self) -> usize { + let drained: Vec> = + std::mem::take(&mut *self.meta_sweep.lock().unwrap()); + let mut n = 0; + for st in drained { + // A hard-deleted stream's files are already unlinked — flushing + // would resurrect its sidecar. Same `Arc` identity check as + // delete's `remove_if`. (Soft-deleted streams stay in the map and + // must flush: the sidecar records the `soft_deleted` flag.) + let live = self + .streams + .get(&st.path) + .is_some_and(|cur| Arc::ptr_eq(cur.value(), &st)); + if live && st.meta_dirty.swap(false, Ordering::AcqRel) { + let _ = write_meta_sync(&st, false); + n += 1; + } + } + n } } @@ -2177,3 +2214,100 @@ mod tier_tests { let _ = std::fs::remove_dir_all(&dir); } } + +// ---------------- batched meta sweep tests (#4691) ---------------- + +#[cfg(test)] +mod meta_sweep_tests { + use super::*; + use crate::tier::TierConfig; + + fn tmp_dir(tag: &str) -> PathBuf { + let p = std::env::temp_dir().join(format!( + "ds-meta-sweep-{tag}-{}-{}", + std::process::id(), + SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos() + )); + let _ = std::fs::remove_dir_all(&p); + p + } + + fn octet_cfg() -> StreamConfig { + StreamConfig { + content_type: "application/octet-stream".into(), + ttl_seconds: None, + expires_at: None, + expires_at_raw: None, + create_closed: false, + forked_from: None, + fork_offset_raw: None, + fork_sub_offset: None, + } + } + + fn create(store: &Store, path: &str) -> Arc { + match store.create(path, octet_cfg(), None, 0).unwrap() { + CreateResult::Created(s) => s, + _ => panic!("create failed"), + } + } + + fn disk_meta(st: &StreamState) -> Meta { + serde_json::from_slice(&std::fs::read(meta_path(&st.file_path)).unwrap()).unwrap() + } + + /// Marking is idempotent while a flush is pending (one sweep entry per + /// stream per cycle), the sweep persists the pending state, and a clean + /// sweep is a no-op. + #[tokio::test] + async fn mark_dedupes_and_sweep_flushes() { + let dir = tmp_dir("flush"); + let store = Store::new_with_tier(dir.clone(), TierConfig::default()).unwrap(); + let st = create(&store, "s"); + + st.shared + .write() + .unwrap() + .producers + .insert("p1".into(), ProducerState { epoch: 1, last_seq: 3 }); + store.mark_meta_dirty(&st); + store.mark_meta_dirty(&st); // second mark while pending: deduped + + assert!( + !disk_meta(&st).producers.contains_key("p1"), + "marking alone must not write the sidecar" + ); + assert_eq!(store.sweep_meta_once(), 1, "one dirty stream, one flush"); + let meta = disk_meta(&st); + let p = meta.producers.get("p1").expect("sweep persists the pending producer state"); + assert_eq!((p.epoch, p.last_seq), (1, 3)); + assert!( + !st.meta_dirty.load(Ordering::Acquire), + "sweep clears the dirty flag" + ); + assert_eq!(store.sweep_meta_once(), 0, "nothing left to sweep"); + + let _ = std::fs::remove_dir_all(&dir); + } + + /// A stream hard-deleted after being marked dirty must NOT get its sidecar + /// resurrected by a later sweep (the file unlinks already happened). + #[tokio::test] + async fn sweep_skips_hard_deleted_stream() { + let dir = tmp_dir("deleted"); + let store = Store::new_with_tier(dir.clone(), TierConfig::default()).unwrap(); + let st = create(&store, "s"); + + store.mark_meta_dirty(&st); + store.delete_or_soft_delete_durable(&st).unwrap(); + assert!(!meta_path(&st.file_path).exists(), "hard delete unlinked the sidecar"); + + assert_eq!(store.sweep_meta_once(), 0, "deleted stream is skipped"); + assert!( + !meta_path(&st.file_path).exists(), + "sweep must not resurrect a deleted stream's sidecar" + ); + + let _ = std::fs::remove_dir_all(&dir); + } +}