perf(durable-streams-rust): --stream-lanes — shard stream data files across per-device lane dirs

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

@@ -311,6 +311,20 @@ fn main() {
                     }
                 }
             }
+            // Stream data lanes: hash stream files across streams/<0..N>/ subdirs,
+            // one per (intended) device, so checkpoint writeback spreads over N
+            // devices with N parallel syncfs barriers (the ~1M-stream wall fix).
+            // A LAYOUT choice: must match the on-disk layout across restarts.
+            "--stream-lanes" => {
+                let v = val(args.next(), "--stream-lanes");
+                match v.parse::<usize>() {
+                    Ok(n) if n >= 1 => store::set_stream_lanes(n),
+                    _ => {
+                        eprintln!("--stream-lanes must be a positive integer");
+                        std::process::exit(2);
+                    }
+                }
+            }
             // Checkpoint size trigger: checkpoint a shard as soon as its retained
             // WAL exceeds this many bytes (0 = disabled). An explicit replay-time
             // budget that also self-staggers shards by their own write rates.

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

@@ -186,6 +186,89 @@ pub(crate) fn syncfs_barrier(_file: &File) -> std::io::Result<()> {
     ))
 }
 
+/// Stream-lane count (`--stream-lanes`, default 1 = the flat `streams/` layout).
+/// With N > 1, stream data files are hashed across `streams/<0..N>/` subdirs so
+/// each lane can be mounted on its OWN device: the checkpoint's dirty-file
+/// writeback (the ~1M-stream wall — one `syncfs` measured at 60–74 s when every
+/// stream shared one device, wal-1m-diag 2026-07-13) spreads over N devices and
+/// runs N barriers in parallel, and no single ext4 directory holds every stream.
+/// Must be set BEFORE `Store::open` and match the on-disk layout across restarts
+/// (same N or files won't be found — a layout choice, not a runtime tunable).
+static STREAM_LANES: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(1);
+pub fn set_stream_lanes(n: usize) {
+    STREAM_LANES.store(n.max(1), Ordering::Relaxed);
+}
+pub fn stream_lanes() -> usize {
+    STREAM_LANES.load(Ordering::Relaxed)
+}
+
+/// Stable lane for a stream data-file name (FNV-1a; the fname embeds the stream
+/// id, so this is fixed for the stream's lifetime and recomputable anywhere).
+fn lane_of(fname: &str) -> usize {
+    let lanes = stream_lanes();
+    if lanes <= 1 {
+        return 0;
+    }
+    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
+    for b in fname.as_bytes() {
+        h ^= *b as u64;
+        h = h.wrapping_mul(0x1000_0000_01b3);
+    }
+    (h % lanes as u64) as usize
+}
+
+/// Directory of lane `lane`: the flat `streams/` when lanes == 1 (byte-identical
+/// to the historical layout), else `streams/<lane>/`.
+fn lane_dir(data_dir: &std::path::Path, lane: usize) -> PathBuf {
+    let root = data_dir.join("streams");
+    if stream_lanes() <= 1 {
+        root
+    } else {
+        root.join(lane.to_string())
+    }
+}
+
+/// Open directory fds, one per stream lane, registered at `Store::open` — the
+/// checkpoint's syncfs must barrier EVERY lane's filesystem (touched files can
+/// live on any lane), and a dir fd is a valid syncfs target. Replaced (not
+/// appended) per open so tests that build many stores target the latest layout.
+static LANE_SYNC_FDS: StdMutex<Option<Arc<Vec<File>>>> = StdMutex::new(None);
+
+/// Checkpoint durability barrier across all stream lanes: one `syncfs` per lane,
+/// parallelized (each is a full device writeback and the lanes are independent
+/// devices in the intended deployment). Falls back to a single barrier on
+/// `fallback`'s fs when no lane registry exists (e.g. shard-only unit tests).
+pub(crate) fn syncfs_stream_lanes(fallback: &File) -> std::io::Result<()> {
+    let fds = LANE_SYNC_FDS.lock().unwrap().clone();
+    match fds {
+        Some(fds) if !fds.is_empty() => {
+            if fds.len() == 1 {
+                return syncfs_barrier(&fds[0]);
+            }
+            std::thread::scope(|s| {
+                let handles: Vec<_> = fds
+                    .iter()
+                    .map(|f| s.spawn(move || syncfs_barrier(f)))
+                    .collect();
+                let mut first_err = None;
+                for h in handles {
+                    if let Err(e) = h
+                        .join()
+                        .unwrap_or_else(|_| Err(std::io::Error::other("syncfs thread panicked")))
+                    {
+                        first_err.get_or_insert(e);
+                    }
+                }
+                match first_err {
+                    None => Ok(()),
+                    Some(e) => Err(e),
+                }
+            })
+        }
+        _ => syncfs_barrier(fallback),
+    }
+}
+
 pub struct StreamState {
     pub id: u64,
     pub path: String,
@@ -472,6 +555,17 @@ impl Store {
     ) -> std::io::Result<Self> {
         let streams_dir = data_dir.join("streams");
         std::fs::create_dir_all(&streams_dir)?;
+        // Create every stream-lane dir and register their dir fds for the
+        // checkpoint's per-lane syncfs barrier (see `syncfs_stream_lanes`).
+        {
+            let mut lane_fds = Vec::with_capacity(stream_lanes());
+            for lane in 0..stream_lanes() {
+                let d = lane_dir(&data_dir, lane);
+                std::fs::create_dir_all(&d)?;
+                lane_fds.push(File::open(&d)?);
+            }
+            *LANE_SYNC_FDS.lock().unwrap() = Some(Arc::new(lane_fds));
+        }
         // Stream data can be sensitive; keep the data dir owner-only (best-effort).
         #[cfg(unix)]
         {
@@ -511,10 +605,16 @@ impl Store {
     /// everything else. Orphan files (crash between create and meta write) are
     /// discarded.
     fn recover(&self, streams_dir: &std::path::Path) -> std::io::Result<()> {
+        let _ = streams_dir; // root; per-lane dirs derived below (lane 0 == root when lanes == 1)
         let mut metas: HashMap<String, (Meta, PathBuf)> = HashMap::new();
         let mut data_files: Vec<PathBuf> = Vec::new();
-        for entry in std::fs::read_dir(streams_dir)? {
-            let p = entry?.path();
+        let mut entries: Vec<PathBuf> = Vec::new();
+        for lane in 0..stream_lanes() {
+            for entry in std::fs::read_dir(lane_dir(&self.data_dir, lane))? {
+                entries.push(entry?.path());
+            }
+        }
+        for p in entries {
             let name = p.file_name().and_then(|n| n.to_str()).unwrap_or("");
             if name.ends_with(".meta.tmp") {
                 let _ = std::fs::remove_file(&p);
@@ -859,7 +959,7 @@ impl Store {
         }
         let id = self.next_id.fetch_add(1, Ordering::Relaxed);
         let fname = format!("{}~{}", encode_path(path), id);
-        let file_path = self.data_dir.join("streams").join(fname);
+        let file_path = lane_dir(&self.data_dir, lane_of(&fname)).join(fname);
         let file = Arc::new(
             OpenOptions::new()
                 .create(true)

packages/durable-streams-rust/src/wal/e2e_tests.rs

@@ -952,6 +952,68 @@ async fn e2e_recycled_first_segment_acked_records_survive_crash() {
     let _ = std::fs::remove_dir_all(&dir);
 }
 
+/// `--stream-lanes N`: stream data files hash across `streams/<0..N>/` subdirs
+/// (one per device in the intended deployment — the ~1M-stream writeback-wall
+/// fix). Crash recovery must find every file in its lane dir, and the
+/// checkpoint's per-lane syncfs must preserve durability-before-recycle exactly
+/// as the single-lane layout does. Guarded by DurabilityGuard (serialized) since
+/// stream-lanes is process-global state; reset to 1 before releasing the guard.
+#[tokio::test]
+async fn e2e_stream_lanes_recover_acked_records() {
+    let _guard = DurabilityGuard::wal();
+    crate::store::set_stream_lanes(3);
+    crate::wal::shard::set_checkpoint_syncfs(true);
+    const SEG: u64 = 4096;
+    let dir = tmp("stream-lanes");
+
+    let h = Harness::boot_with_segment_size(&dir, Some(1), 1, SEG).unwrap();
+    // Enough streams that the FNV lane hash populates more than one lane.
+    let names: Vec<String> = (0..12).map(|i| format!("lane-s{i}")).collect();
+    for n in &names {
+        create_stream(&h.store, n, OCTET).await;
+    }
+    let mut expected: std::collections::HashMap<String, Vec<u8>> = Default::default();
+    for round in 0..40usize {
+        for n in &names {
+            let rec = format!("{n}-r{round:03}|").into_bytes();
+            append_acked(&h.store, n, OCTET, &rec).await;
+            expected.entry(n.clone()).or_default().extend_from_slice(&rec);
+        }
+    }
+    // Checkpoint (per-lane syncfs + recycle), then more acked appends on top.
+    h.walset.shards()[0].checkpoint().await.unwrap();
+    for n in &names {
+        let rec = format!("{n}-post|").into_bytes();
+        append_acked(&h.store, n, OCTET, &rec).await;
+        expected.entry(n.clone()).or_default().extend_from_slice(&rec);
+    }
+
+    h.crash();
+
+    let h2 = Harness::boot_with_segment_size(&dir, None, 1, SEG).unwrap();
+    // Layout sanity: files actually spread across lane subdirs.
+    let lanes_used = (0..3)
+        .filter(|l| {
+            std::fs::read_dir(dir.join("streams").join(l.to_string()))
+                .map(|d| d.flatten().next().is_some())
+                .unwrap_or(false)
+        })
+        .count();
+    assert!(lanes_used >= 2, "expected streams spread over lanes, got {lanes_used}");
+    for n in &names {
+        let got = stream_file_bytes(&h2.store, n);
+        assert_eq!(
+            &got,
+            expected.get(n).unwrap(),
+            "stream {n} recovers byte-identical across lanes"
+        );
+    }
+    h2.crash();
+    crate::wal::shard::set_checkpoint_syncfs(false);
+    crate::store::set_stream_lanes(1);
+    let _ = std::fs::remove_dir_all(&dir);
+}
+
 /// Cardinality-cliff #1: with `--wal-checkpoint-syncfs on`, the checkpoint makes
 /// touched per-stream files durable via ONE `syncfs()` barrier instead of the
 /// per-stream `fdatasync` loop. This must preserve the durability-before-recycle

packages/durable-streams-rust/src/wal/shard.rs

@@ -919,7 +919,10 @@ impl Shard {
                 // dirty page on it — all touched files become durable at once. Still
                 // strictly before persist_durable_tails/recycle, so the
                 // durability-before-recycle ordering is unchanged.
-                crate::store::syncfs_barrier(&touched[0].2)?;
+                // One syncfs per stream lane (touched files may span lanes on
+                // multi-device layouts); falls back to touched[0]'s fs when no
+                // lane registry exists (shard-only unit tests).
+                crate::store::syncfs_stream_lanes(&touched[0].2)?;
             } else if fanout <= 1 {
                 for (_, _, f) in &touched {
                     crate::store::barrier_fsync(f)?;