perf(durable-streams-rust): batch memory-mode meta sidecar flushes into a periodic sweep

.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.

.github/workflows/auto-approve-durable-streams.yml

@@ -24,6 +24,7 @@ jobs:
         run: |
           scopes=(
             "packages/durable-streams-rust/"
+            ".changeset/"
           )
 
           in_scope=true

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

@@ -1167,9 +1167,11 @@ async fn handle_append_inner(store: Arc<Store>, 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<Store>, 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);
+    }
 }
 

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<wal::walset::WalSet>) {
     });
 }
 
+/// 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<Store>) {
+    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() {

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<Arc<crate::wal::walset::WalSet>>,
+    /// 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<Vec<Arc<StreamState>>>,
 }
 
 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<Self>) {
-        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<StreamState>) {
+        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<Arc<StreamState>> =
+            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<StreamState> {
+        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);
+    }
+}