diff --git a/.changeset/wal-cardinality-cliff.md b/.changeset/wal-cardinality-cliff.md new file mode 100644 index 0000000000..d38c7b7956 --- /dev/null +++ b/.changeset/wal-cardinality-cliff.md @@ -0,0 +1,13 @@ +--- +"@electric-ax/durable-streams-server-rust": patch +--- + +Eliminate the WAL write cardinality cliff (10.4k → 383k appends/s @100k streams; 212k @1M). + +- Checkpoint durability now uses one `syncfs` barrier per stream lane on Linux (was O(touched-streams) per-file `fdatasync` — the barrier storm that collapsed throughput at high stream counts). +- New `--wal-checkpoint-interval-ms` (per-shard time trigger, default 3000) and `--wal-checkpoint-wal-bytes` (retained-WAL size budget, 0 = off): checkpoint cadence is an explicit crash-replay budget and shards self-stagger instead of storming together. +- New `--stream-lanes N` (default 1 = unchanged layout): hash stream data files across `streams/<0..N>/` dirs, one per device, spreading checkpoint writeback over N devices with N parallel barriers. The lane count is persisted and validated on open. +- New `--server-stats N` telemetry (`SRV_STATS`: cpu / inflight / service / lock / durability-wait per interval) — the dependency-free bottleneck diagnostics used to find all of the above. Memory-mode plain appends no longer queue redundant sidecar flushes. +- Removed dead/diagnostic flags: `--wal-fsync-parallel`, `--wal-meta-gate`, `--mem-meta-gate`, `--meta-sweep-disable`, `--meta-sweep-stats`, `--tier local` / `--tier-local-dir` (tier is `off|s3`). `--durability memory` combined with `--tier` is now rejected at startup. WAL records without `PAYLOAD_CHECKSUMMED` decode as torn (no released writer ever emitted them). + +Deployment guidance (device layout, CPU pinning, checkpoint budgets): `WAL_TUNING.md`. diff --git a/packages/durable-streams-rust/README.md b/packages/durable-streams-rust/README.md index 45b5ee5683..49a005ee38 100644 --- a/packages/durable-streams-rust/README.md +++ b/packages/durable-streams-rust/README.md @@ -87,29 +87,28 @@ durable, single-node server on `127.0.0.1:4437` with its data dir under `$TMPDIR **Cold-storage tier** — off by default; see [Tiered storage](#tiered-storage-cold-offload). With `--tier off` the server is byte-identical to a single-file deployment. -| Flag | Default | Description | -| --------------------------------------------- | ---------- | ---------------------------------------------------------------------------------------- | -| `--tier` | `off` | `off` \| `local` (sealed segments to a local dir) \| `s3` (S3-compatible object storage) | -| `--tier-local-dir` | — | (`tier=local`) directory for sealed segments | -| `--tier-endpoint` | — | (`tier=s3`) S3 endpoint URL | -| `--tier-region` | — | (`tier=s3`) region | -| `--tier-bucket` | — | (`tier=s3`) bucket name | -| `--tier-key-prefix` | — | object-key prefix for sealed segments | -| `--tier-segment-bytes` | `8 MiB` | sealed-segment size (fixed-size, CDN-friendly) | -| `--tier-compact-bytes` | `64 MiB` | small-segment compaction threshold | -| `--tier-path-style` / `--tier-virtual-hosted` | path-style | S3 addressing style | -| `--tier-allow-http` | off | allow plain HTTP to the S3 endpoint (e.g. a local MinIO) | +| Flag | Default | Description | +| --------------------------------------------- | ---------- | -------------------------------------------------------- | +| `--tier` | `off` | `off` \| `s3` (S3-compatible object storage) | +| `--tier-endpoint` | — | (`tier=s3`) S3 endpoint URL | +| `--tier-region` | — | (`tier=s3`) region | +| `--tier-bucket` | — | (`tier=s3`) bucket name | +| `--tier-key-prefix` | — | object-key prefix for sealed segments | +| `--tier-segment-bytes` | `8 MiB` | sealed-segment size (fixed-size, CDN-friendly) | +| `--tier-compact-bytes` | `64 MiB` | small-segment compaction threshold | +| `--tier-path-style` / `--tier-virtual-hosted` | path-style | S3 addressing style | +| `--tier-allow-http` | off | allow plain HTTP to the S3 endpoint (e.g. a local MinIO) | S3 credentials come from the **environment**, never flags: `DS_S3_ACCESS_KEY_ID` / `DS_S3_SECRET_ACCESS_KEY` (or the standard `AWS_ACCESS_KEY_ID` / `AWS_SECRET_ACCESS_KEY`). ### Choosing a configuration -| Your situation | Use | -| --------------------------------------------------- | ----------------------------------------------------------------------------------------------------------- | -| **Bounded local disk with long history** | `--tier s3` (or `local`) — seal cold segments to object storage; the recent tail stays local and zero-copy. | -| **High fan-out p99 across many streams on Linux** | try `--tail-cache-bytes 65536` (off by default there because `sendfile` already covers it). | -| **Experimenting on macOS and writes take ~2–10 ms** | expected — see [macOS write latency](#macos-write-latency-f_fullfsync) below. | +| Your situation | Use | +| --------------------------------------------------- | ---------------------------------------------------------------------------------------------- | +| **Bounded local disk with long history** | `--tier s3` — seal cold segments to object storage; the recent tail stays local and zero-copy. | +| **High fan-out p99 across many streams on Linux** | try `--tail-cache-bytes 65536` (off by default there because `sendfile` already covers it). | +| **Experimenting on macOS and writes take ~2–10 ms** | expected — see [macOS write latency](#macos-write-latency-f_fullfsync) below. | ### macOS write latency (`F_FULLFSYNC`) @@ -179,7 +178,7 @@ Opt-in (`--tier`, off by default). Because streams are append-only and immutable - **S3-compatible, not AWS-only.** The `s3` backend works with any S3-compatible endpoint — Cloudflare R2, Fly/Tigris, MinIO, Backblaze B2 — via a configurable endpoint + path-style addressing. It's built on the `object_store` crate behind the **`tier` Cargo feature** (so a default build pulls no object-storage dependencies): `cargo build --release --features tier`. A `local` backend (sealed segments to a directory) needs no feature/deps and is handy for testing. - **How it stays correct.** Each stream keeps a manifest of its sealed segments. The lifecycle is seal → upload → `head`-verify → durably flip the manifest entry `local → remote` → only then unlink the staged chunk file (safe even under an in-flight read — Unix keeps an open fd readable after unlink). A read resolves each requested offset against the manifest — local ranges (live file or sealed chunk file) keep the zero-copy `sendfile` path; remote ranges are fetched by range-GET and streamed in. JSON seals always land on a value boundary (never inside a string). Durability is unchanged: an append still acks only after the local fsync — offload is strictly post-durability. Fully-sealed ranges are stamped `Cache-Control: immutable` for long-lived CDN caching, and remote objects are GC'd ref-count-aware with forks. The live data file's redundant sealed prefix is reclaimed by **compaction**: once it exceeds `--tier-compact-bytes` (default 64 MiB), the live file is rewritten to hold only the hot tail, under the appender lock and crash-safe via a `pending_compaction` intent. In-flight reads drain off the old fd, so reads stay lock-free — this is why compaction is used rather than `fallocate` hole-punching, which raced those reads. Set `--tier-compact-bytes 0` to disable. -- **Flags:** `--tier {off|local|s3}`, `--tier-segment-bytes`, `--tier-compact-bytes` (live-file compaction threshold; `0` disables), `--tier-key-prefix`, `--tier-local-dir` (local), `--tier-endpoint` / `--tier-region` / `--tier-bucket`, `--tier-path-style` / `--tier-virtual-hosted`, `--tier-allow-http`. S3 credentials come from `DS_S3_ACCESS_KEY_ID` / `DS_S3_SECRET_ACCESS_KEY` (with `AWS_*` fallback), env only. +- **Flags:** `--tier {off|s3}`, `--tier-segment-bytes`, `--tier-compact-bytes` (live-file compaction threshold; `0` disables), `--tier-key-prefix`, `--tier-endpoint` / `--tier-region` / `--tier-bucket`, `--tier-path-style` / `--tier-virtual-hosted`, `--tier-allow-http`. S3 credentials come from `DS_S3_ACCESS_KEY_ID` / `DS_S3_SECRET_ACCESS_KEY` (with `AWS_*` fallback), env only. Conformance passes with tiering on as well as off (verified manually with a small `--tier-segment-bytes` so streams seal and offload mid-suite; catch-up reads, ETag/304, closed-stream EOF, and forks are all served correctly from cold). Note this is a manual check, not a CI gate: the `rust-conformance` matrix in `.github/workflows/ci.yml` builds the default feature set (no `--features tier`) and never passes `--tier`, so the tier code path is not exercised in CI today. ## Observability (OpenTelemetry) diff --git a/packages/durable-streams-rust/WAL_TUNING.md b/packages/durable-streams-rust/WAL_TUNING.md new file mode 100644 index 0000000000..cecd158a42 --- /dev/null +++ b/packages/durable-streams-rust/WAL_TUNING.md @@ -0,0 +1,120 @@ +# WAL write-path tuning — the ideal configuration + +**Status:** validated 2026-07-13 on GCP `c4d-standard-64-lssd` (6× Titanium local +NVMe, raw block), 256 B appends, 8 vCPU server, up to 100k streams. Campaign: +ds-bench suites `wal-decomp-lane0`, `wal-splitlane`, `wal-sizetrigger`, +`wal-cpubind` (see ds-bench `results//report.md` + `AGENTS.md`). +**Verdict: the write cardinality cliff is eliminated** — WAL-durable throughput +is flat from 10k → 100k streams (−5%) at ~26–32× the pre-fix baseline. + +## The ladder (what each step bought, @100k streams) + +| configuration | ops/s | +| ------------------------------------------------------------------------- | -------- | +| pre-fix: streams on network PD, per-stream checkpoint fdatasync storm | 10.4k | +| + syncfs checkpoint barrier (PR #4697) | 13.6k | +| + stream files on local NVMe (not the boot disk) | 46k | +| + split-lane layout (WAL shards on their own NVMe devices) | 272k | +| + size-triggered checkpoint (PR #4704, `--wal-checkpoint-wal-bytes 1GiB`) | 303k | +| + exclusive pinned cores (Guaranteed QoS + static CPU manager) | 328k | +| **all of the above stacked** (suite `wal-stacked-1m`) | **383k** | +| reference: memory durability (no fsync anywhere) | 512k | + +At extreme cardinality the stacked config on ONE data lane hits a second wall — +checkpoint writeback (~40× metadata amplification of small appends) saturates +the single data device (`syncfs` = 60–74 s at 1M streams): 244k @500k, 56–68k +@1M. `--stream-lanes 3` (suite `wal-streamlanes-1m`, 3 data lanes + 3 WAL +lanes) breaks it: **374k @100k, 285k @500k, 212k @1M** (`syncfs` 5.7–11 s). + +The residual ~1.6× gap to memory mode is WAL machinery (staging + double-write), +not fsync — future work: io_uring segment writer (`wal/segment.rs` seam), +batched mark-written. + +## The ideal configuration + +### 1. Hardware / storage layout (the #1 lever) + +Use an instance with **multiple physically attached NVMe devices** (GCP: 4th-gen +`-lssd` types, raw block via `--local-nvme-ssd-block`; do NOT use +`--ephemeral-storage-local-ssd`, which RAID0-stripes every device into one fsync +barrier). Then: + +- **One device for stream data files** — mount it and point `--data-dir` at it. + The per-stream files and the checkpoint's `syncfs` domain live here. +- **One device per WAL shard** — mount device _j_ at `/wal/` (the + server opens shard _i_ at that path automatically). `--wal-shards` = number of + dedicated WAL devices. +- **Never share a device between WAL and stream data.** Commit `fdatasync` vs + checkpoint writeback contention on one queue was worth 5× by itself + (55k → 272k). On dedicated lanes the commit-fsync cost is ~zero (checkpoint-off + measured _below_ an all-lanes-shared no-fsync control). +- **Never leave stream data on the boot disk / network PD.** On Kubernetes + raw-block node pools the default emptyDir sits on the boot PD — this single + mistake mismeasures (and misdeploys) WAL mode by 5–26×. + +Split the box between data lanes and WAL lanes **by cardinality**: + +- ≤ ~100k streams: 1 data lane is enough — e.g. device 0 → data root, + devices 1–5 → 5 WAL shards: `--data-dir /data/wal/0 --wal-shards 5` +- ≥ ~500k streams: checkpoint writeback dominates; give data more lanes — e.g. + device 0 → data root (stream lane 0), devices 1–2 → stream lanes 1–2 + (mounted at `/streams/1`, `/2`), devices 3–5 → 3 WAL shards: + `--data-dir /data/wal/0 --wal-shards 3 --stream-lanes 3` + (`--stream-lanes` is a LAYOUT choice like the shard count: it must match the + on-disk layout across restarts.) + +### 2. Server flags + +``` +--wal-checkpoint-wal-bytes 1073741824 # checkpoint a shard when ITS retained WAL + # exceeds 1 GiB (PR #4704) — checkpoint cost ≈ 0, + # crash-replay bounded to ≤1 GiB/shard (<1 s NVMe) +--wal-checkpoint-interval-ms 60000 # fallback timer so an idle shard still recycles +--wal-shards # shards = fsync lanes; on a SINGLE shared + # device keep 2–4 (more only fragments batches) +--stream-lanes # hash stream files across per-device dirs + # (PR #4705); layout choice, default 1 +--worker-threads +``` + +The syncfs checkpoint barrier (PR #4697) is unconditional on Linux — one +`syncfs` per stream lane instead of O(N-touched) per-stream `fdatasync`. +Non-Linux uses the serial per-file loop (no `syncfs` there). + +### 3. CPU binding (+21–24%) + +Give the server **exclusive pinned cores**. On Kubernetes: node pool with +`kubeletConfig.cpuManagerPolicy: static` + a **Guaranteed QoS** pod (every +container `requests == limits`, server CPU an integer). Measured 356k @10k / +328k @100k vs 286k/272k on shared cores, same layout and flags. Now that WAL is +no longer fsync-bound, it scales with cores again — don't starve it. + +### 4. What you do NOT need to worry about + +- **Read performance while tuning checkpoints.** Reads never touch the WAL and + are served zero-copy (`sendfile`) from the data file's page cache, which is + written _before_ the WAL ack barrier. Checkpoint cadence has zero read-path + cost. +- **Ack latency vs checkpoints.** Acks gate only on the WAL group-commit + `fdatasync`; a checkpoint never blocks appends. +- **Recoverability.** The contract is unchanged by any of these knobs: a WAL + segment is recycled only after its records' stream bytes are fsynced into + their files **and** the durable-tail map is persisted (`wal/shard.rs` + checkpoint ordering; crash-recovery e2e + randomized crash sim cover it). + The size trigger only changes _when_ that sequence runs. + +## Caveats / follow-ups + +- The 1M-stream writeback wall is diagnosed and broken (`--stream-lanes`, PR + #4705: 68k → 212k @1M). The residual slope (374k @100k → 212k @1M on 3 data + lanes) is per-file writeback amplification against total data-lane + capacity — add data lanes, or see #4695 (log-structured store) for the + structural end-state. +- fd ceiling: the server holds one fd per live stream — 1,005,724 fds at 1M + streams = 96% of the default 1,048,576 limit. Not the throughput wall, but a + hard scale ceiling just above 1M: raise LimitNOFILE, or see #4706 (lazy fd + management). +- The checkpoint size trigger is opt-in (`--wal-checkpoint-wal-bytes 0` default); + making ~1 GiB the default is a candidate after soak. +- Larger retained WAL = longer replay: 1 GiB/shard ≈ sub-second on local NVMe, + but budget it consciously on slower disks. diff --git a/packages/durable-streams-rust/src/handlers.rs b/packages/durable-streams-rust/src/handlers.rs index a4d2c74e04..29d1dc974e 100644 --- a/packages/durable-streams-rust/src/handlers.rs +++ b/packages/durable-streams-rust/src/handlers.rs @@ -44,7 +44,7 @@ pub enum DurabilityMode { #[default] Wal, /// No WAL, no fsync: ack on the page-cache write. Durability comes from replication - /// (future). Linux-only (binary appends use zero-copy socket→file). + /// (future). Memory, } @@ -896,6 +896,9 @@ async fn handle_append(store: Arc, req: Req, path: String) -> Resp { async fn handle_append_inner(store: Arc, req: Req, path: String) -> (Resp, AppendOutcome, bool) { use AppendOutcome::*; + // Load-telemetry probe: bumps the in-flight gauge and records service time on + // drop (covers every return path). No-op unless `--server-stats` is on. + let _probe = crate::srvstats::AppendProbe::start(); let st = match store.get(&path) { Some(s) => s, None => return (text_response(404, "stream not found"), Conflict, false), @@ -950,8 +953,10 @@ async fn handle_append_inner(store: Arc, req: Req, path: String) -> (Resp // Serialize per stream: producer validation + write + state update under one // lock. Time the wait separately — lock contention is a key bottleneck. let lock_t0 = crate::telemetry::Timer::start(); + let srv_lock_t0 = std::time::Instant::now(); let mut ap = st.appender.lock().await; crate::telemetry::record_append_lock_wait(lock_t0.elapsed_secs()); + crate::srvstats::record_applock_wait(srv_lock_t0.elapsed()); // Closed checks (precedence: closed → seq regression → gap). { @@ -1083,6 +1088,16 @@ async fn handle_append_inner(store: Arc, req: Req, path: String) -> (Resp Err(_) => ret!(text_response(500, "write failed"), Conflict), } } + // Does this append change state the memory-mode sidecar must persist? Captured + // BEFORE `seq_header` is consumed below. Producer/seq updates are idempotency + // state; a TTL stream's sliding `last_access` must survive restart (mirrors the + // read path, which marks dirty only for TTL streams). A plain append to a non-TTL stream changes + // only `durable_tail`/`last_access`. In BOTH modes the durable tail is carried + // elsewhere (memory: re-derived from the data-file length on restart; wal: the + // checkpoint's per-shard `tails` map), and `last_access` only gates TTL — so a + // plain non-TTL append needs no sidecar flush at all (cardinality-cliff #1). + let meta_persist_needed = + producer.is_some() || seq_header.is_some() || st.config.ttl_seconds.is_some(); { let mut s = st.shared.write().unwrap(); if let Some(p) = &producer { @@ -1130,7 +1145,9 @@ async fn handle_append_inner(store: Arc, req: Req, path: String) -> (Resp // Wait for durability off the lock before exposing the bytes. if let Some(lsn) = staged_lsn { + let dur_t0 = std::time::Instant::now(); wait_durable_lsn(&store, &st, lsn).await; + crate::srvstats::record_durwait(dur_t0.elapsed()); } // Durable now (wal) / page-cache written (memory): expose the new bytes to @@ -1165,12 +1182,34 @@ async fn handle_append_inner(store: Arc, req: Req, path: String) -> (Resp // saturation) plus a timer task OFF the per-append path. Producer/access // updates are already documented as a non-durable, lagging flush; the lag // bound moves from the 100 ms debounce to the checkpoint cadence. - st.meta_dirty.store(true, std::sync::atomic::Ordering::Release); - } else { + // + // GATED (cardinality-cliff #1): only mark when the append changed state + // the sidecar must persist — producer/seq idempotency or a sliding TTL. + // A plain append still gets its fdatasync AND its `durable_tail` recorded + // in the checkpoint's per-shard `tails` map (register_dirty + the + // unconditional `persist_durable_tails`, independent of this flag) — and + // that map, not the sidecar, is the authoritative durable-tail proof + // recovery reconciles against (see wal/shard.rs step 3a, wal/recovery.rs). + // `last_access` only gates TTL. So a plain non-TTL append needs no sidecar + // rewrite here — dropping it removes the O(touched) `write_meta_sync` calls + // that dominate the checkpoint's meta phase at high stream cardinality. + if meta_persist_needed { + st.meta_dirty.store(true, std::sync::atomic::Ordering::Release); + } + } else if meta_persist_needed { // 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). + // + // Only queued when the append actually changed state the sidecar must + // persist — producer/seq idempotency or a sliding TTL. A plain append to + // a non-TTL stream changes only `durable_tail`/`last_access`, and + // memory-mode recovery reads NEITHER (the tail is re-derived from the + // data-file length in `Store::new_with_tier`; `last_access` only gates + // TTL expiry, which these streams don't have). Skipping the queue for + // that common case removes the per-append sidecar rewrite whose cost + // stops amortizing at high stream cardinality. store.mark_meta_dirty(&st); } if !wire.is_empty() { @@ -1597,7 +1636,19 @@ async fn handle_long_poll( } } _ = tokio::time::sleep(deadline.saturating_duration_since(Instant::now())) => { + // Deadline hit — but re-check the tail EXACTLY like the + // closed-channel arm above. Returning a timeout that advertises + // the fresh tail as `Stream-Next-Offset` while NOT delivering + // the bytes behind it silently SKIPS data: an append whose + // durable-tail publish lands inside the deadline window gets + // jumped over (the client adopts the header offset from every + // response, including empty 204s) and is never delivered. + // Invariant: a long-poll response never advances the client's + // offset beyond the bytes it actually delivered. let t = st.tail(); + if t.bytes > from { + return long_poll_data(&st, from, t, client_cursor, true, cache_hit).await; + } return long_poll_timeout(t.bytes, cursor, t.closed); } } @@ -2298,5 +2349,128 @@ mod memory_mode_tests { let _ = std::fs::remove_dir_all(&dir); } + + /// Cardinality-cliff fix (#1): a PLAIN append (no producer/seq, non-TTL + /// stream) in memory mode must NOT queue a sidecar flush. The tail is + /// recovered from the data-file length and `last_access` only gates TTL, so + /// the per-append sidecar rewrite is pure overhead whose cost stops + /// amortizing at high stream cardinality. Contrast + /// `memory_append_defers_sidecar_to_store_sweep`, which uses a producer + /// append and therefore still marks dirty. + #[tokio::test] + async fn memory_plain_append_skips_sidecar_flush() { + let _guard = crate::handlers::test_support::DurabilityGuard::memory(); + let dir = tmp("mem-plain-noflush"); + let store = + Arc::new(Store::new_with_tier(dir.clone(), TierConfig::default()).unwrap()); + + let resp = handle(Arc::clone(&store), put_req("m/p", "application/octet-stream")).await; + assert!((200..300).contains(&resp.status), "create: {}", resp.status); + // Drain anything the create queued so we measure only the append's effect. + let store2 = Arc::clone(&store); + let _ = tokio::task::spawn_blocking(move || store2.sweep_meta_once()) + .await + .unwrap(); + + // Plain append: no producer headers, non-TTL stream. + let resp = handle( + Arc::clone(&store), + post_req("m/p", "application/octet-stream", b"payload"), + ) + .await; + assert!((200..300).contains(&resp.status), "append: {}", resp.status); + + let flushed = tokio::task::spawn_blocking({ + let store = Arc::clone(&store); + move || store.sweep_meta_once() + }) + .await + .unwrap(); + assert_eq!( + flushed, 0, + "a plain non-TTL memory-mode append must not queue a sidecar flush" + ); + + let _ = std::fs::remove_dir_all(&dir); + } + + /// Long-poll deadline/data race (conformance flake root cause): a long-poll + /// response must NEVER advance `Stream-Next-Offset` past the client's `from` + /// without delivering the bytes in between — otherwise an append whose + /// durable-tail publish lands inside the deadline window is advertised but + /// not sent, and the client skips it forever. This drives many iterations of + /// an append racing a long-poll whose deadline is aligned with the append + /// (the conformance suite's exact shape, tightened) and asserts the + /// invariant on every response. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn long_poll_timeout_never_skips_observed_data() { + let _guard = crate::handlers::test_support::DurabilityGuard::memory(); + crate::handlers::set_long_poll_timeout(20); + let dir = tmp("lp-race"); + let store = Arc::new(Store::new_with_tier(dir.clone(), TierConfig::default()).unwrap()); + + let resp = handle(Arc::clone(&store), put_req("lp/r", "application/octet-stream")).await; + assert!((200..300).contains(&resp.status), "create: {}", resp.status); + + let next_offset = |r: &crate::api::Resp| -> u64 { + r.headers + .iter() + .find(|(k, _)| *k == "stream-next-offset") + .map(|(_, v)| v.rsplit('_').next().unwrap().parse().unwrap()) + .unwrap_or(0) + }; + let mut from: u64 = 0; + for i in 0..200u32 { + // Long-poll from the current offset... + let lp = tokio::spawn(handle( + Arc::clone(&store), + Req { + method: Method::Get, + path: "lp/r".to_string(), + query: Some(format!("live=long-poll&offset={:016}_{:016}", 0, from)), + headers: vec![], + body: Bytes::new(), + }, + )); + // ...and race an append onto the deadline (spread over the window). + tokio::time::sleep(std::time::Duration::from_millis(15 + (i % 10) as u64)) + .await; + let ar = handle( + Arc::clone(&store), + post_req("lp/r", "application/octet-stream", b"x"), + ) + .await; + assert!((200..300).contains(&ar.status), "append: {}", ar.status); + let r = lp.await.unwrap(); + let next = next_offset(&r); + let has_body = !matches!(r.body, crate::api::Body::Empty); + assert!( + has_body || next <= from, + "iteration {i}: empty long-poll response advanced the offset \ + {from} -> {next} without delivering data (status {})", + r.status + ); + // Catch up for the next round. + from = from.max(next); + if !has_body && next == from { + // ensure we don't fall behind the appends + let t_resp = handle( + Arc::clone(&store), + Req { + method: Method::Get, + path: "lp/r".to_string(), + query: Some(format!("offset={:016}_{:016}", 0, from)), + headers: vec![], + body: Bytes::new(), + }, + ) + .await; + from = from.max(next_offset(&t_resp)); + } + } + + crate::handlers::set_long_poll_timeout(30_000); + 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 721240c4ec..839098308f 100644 --- a/packages/durable-streams-rust/src/main.rs +++ b/packages/durable-streams-rust/src/main.rs @@ -3,6 +3,7 @@ mod blobstore; mod engine_raw; mod handlers; mod http1; +mod srvstats; #[cfg(target_os = "linux")] mod sse_reactor; mod store; @@ -137,6 +138,7 @@ fn main() { // append path). Dependency-free — the measurement vehicle for the contention // investigation, independent of the heavy `telemetry` OTLP feature. let mut wal_stats_secs: Option = None; + let mut server_stats_secs: Option = None; let mut args = std::env::args().skip(1); while let Some(a) = args.next() { match a.as_str() { @@ -166,10 +168,9 @@ fn main() { let v = val(args.next(), "--tier"); tier.kind = match v.as_str() { "off" => tier::TierKind::Off, - "local" => tier::TierKind::Local, "s3" => tier::TierKind::S3, _ => { - eprintln!("--tier must be off|local|s3"); + eprintln!("--tier must be off|s3"); std::process::exit(2); } }; @@ -193,9 +194,6 @@ fn main() { "--tier-allow-http" => { tier.allow_http = true; } - "--tier-local-dir" => { - tier.local_dir = Some(val(args.next(), "--tier-local-dir").into()); - } "--wal-shards" => { let n: usize = parse_val(args.next(), "--wal-shards"); if n == 0 { @@ -238,6 +236,54 @@ fn main() { } } } + // Periodic SRV_STATS line (both modes): cpu_cores / inflight / service + // + appender-lock + durability wait — bottleneck analysis. + "--server-stats" => { + let n: u64 = parse_val(args.next(), "--server-stats"); + if n == 0 { + eprintln!("--server-stats must be ≥ 1 (seconds)"); + std::process::exit(2); + } + server_stats_secs = Some(n); + } + // Checkpoint time trigger: per-shard cadence in ms (default 3000). + "--wal-checkpoint-interval-ms" => { + let v = val(args.next(), "--wal-checkpoint-interval-ms"); + match v.parse::() { + Ok(ms) if ms >= 1 => wal::shard::set_checkpoint_interval_ms(ms), + _ => { + eprintln!("--wal-checkpoint-interval-ms must be a positive integer (milliseconds)"); + std::process::exit(2); + } + } + } + // 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::() { + 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. + "--wal-checkpoint-wal-bytes" => { + let v = val(args.next(), "--wal-checkpoint-wal-bytes"); + match v.parse::() { + Ok(bytes) => wal::shard::set_checkpoint_wal_bytes(bytes), + _ => { + eprintln!("--wal-checkpoint-wal-bytes must be a non-negative integer (bytes)"); + std::process::exit(2); + } + } + } other => { eprintln!("unknown argument: {other}"); std::process::exit(2); @@ -250,6 +296,13 @@ fn main() { // tail-cache-off — those belonged to the removed zero-copy path); the only // gate is refusing to silently ignore a WAL left by a previous wal run. if handlers::durability() == handlers::DurabilityMode::Memory { + // Memory mode acks before anything is fsynced, so pairing it with a cold + // tier would offload un-fsynced (loseable) data as if it were durable — + // a combination with no coherent durability story. Refuse it. + if tier.kind != tier::TierKind::Off { + eprintln!("error: --durability memory cannot be combined with --tier (memory-mode acks are not durable; tiering presumes durable segments)"); + std::process::exit(2); + } // Fail fast on a WAL left by a previous `--durability wal` run: memory mode // never opens/replays it, so starting here would silently ignore those // records (and drop any not yet folded into the per-stream files). Refuse @@ -303,6 +356,11 @@ fn main() { // touches here (its append path flushes via the checkpoint instead). spawn_meta_sweeper(Arc::clone(&store)); + // Server load telemetry (both modes) for bottleneck analysis. + if let Some(secs) = server_stats_secs { + srvstats::spawn(secs); + } + // ---- WAL wiring (Wal mode only) ---- // // Skipped entirely in `--durability memory` mode — no WAL is opened, @@ -407,43 +465,82 @@ fn main() { }); } -/// How often the checkpoint ticker drives each shard's `checkpoint` (spec §7). -/// A sane v1 constant: frequent enough that the WAL doesn't grow unbounded on a -/// busy server, infrequent enough that the batched per-stream `fdatasync`s stay -/// amortized. Checkpoint is non-blocking w.r.t. acks, so this is purely the -/// WAL-recycle / per-stream-durability cadence (tunable is follow-up #9). -const CHECKPOINT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(3); +/// How often the checkpoint ticker POLLS its triggers. The actual checkpoint +/// cadence is per-shard and knob-driven (see `wal::shard::checkpoint_interval_ms` +/// / `checkpoint_wal_bytes`); this is just the trigger-evaluation resolution. +/// 250 ms keeps the size trigger responsive (a shard writing 1 GB/s overshoots a +/// 1 GiB budget by ≤ 250 MB) at negligible poll cost (two atomic loads/shard). +const CHECKPOINT_POLL: std::time::Duration = std::time::Duration::from_millis(250); -/// Spawn the per-shard checkpoint ticker (spec §7). One `tokio::time::interval` -/// driver that, each tick, runs every shard's `checkpoint` (each: batched -/// `fdatasync` of its touched per-stream files → persist `checkpoint_lsn` → -/// recycle WAL segments below it). A checkpoint error is logged, not fatal — a -/// failed/lagging checkpoint only delays WAL recycling (the disk-bounded safety -/// valve, spec §7), never blocks appends. +/// Spawn the per-shard checkpoint driver (spec §7). Each poll tick, a shard is +/// checkpointed iff (a) its retained WAL exceeds `--wal-checkpoint-wal-bytes` +/// (size trigger, 0 = off), or (b) `--wal-checkpoint-interval-ms` has elapsed +/// since ITS last checkpoint (time trigger, default 3000 = the historical 3 s +/// cadence). Due shards checkpoint CONCURRENTLY (each is one spawn_blocking: +/// capture + fsync/syncfs of touched stream files → persist tails/checkpoint_lsn +/// → recycle); a serial walk would queue every shard's fsync behind one +/// device's. Because each shard's clock restarts when IT finishes, shards drift +/// apart naturally instead of storming in a synchronized wave — and with the +/// size trigger they self-schedule by their own write rates. A checkpoint error +/// is logged, not fatal — a failed/lagging checkpoint only delays WAL recycling +/// (the disk-bounded safety valve, spec §7), never blocks appends. A shard that +/// is still checkpointing is never re-fired (the in-flight set guards it), so a +/// checkpoint that takes longer than the interval degrades to back-to-back +/// checkpoints for that shard only. fn spawn_checkpoint_ticker(walset: Arc) { tokio::spawn(async move { - let mut ticker = tokio::time::interval(CHECKPOINT_INTERVAL); - // Skip the immediate first tick — there is nothing to checkpoint at boot. + let interval = + std::time::Duration::from_millis(wal::shard::checkpoint_interval_ms()); + let wal_bytes = wal::shard::checkpoint_wal_bytes(); + let n = walset.shards().len(); + let mut last_done: Vec = vec![std::time::Instant::now(); n]; + let mut in_flight: Vec = vec![false; n]; + let mut wave: tokio::task::JoinSet = tokio::task::JoinSet::new(); + // task-id → shard index, so a PANICKED checkpoint task (JoinError carries + // no payload) still clears its shard's in-flight guard — otherwise one + // panic would silence that shard's checkpoints forever (unbounded WAL). + let mut task_shard: std::collections::HashMap = + std::collections::HashMap::new(); + let mut ticker = tokio::time::interval(CHECKPOINT_POLL.min(interval)); ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + // Skip the immediate first tick — there is nothing to checkpoint at boot. ticker.tick().await; loop { ticker.tick().await; - // All shards checkpoint CONCURRENTLY. Each checkpoint is one - // spawn_blocking task (capture + per-stream fdatasyncs + tails/ckpt - // persist + recycle), so a serial walk makes every per-stream fsync - // across the whole server queue behind a single shard's — at high - // stream cardinality that serialization is what stretches the - // checkpoint wave (and on real disks wastes the device's parallelism). - let mut wave = tokio::task::JoinSet::new(); - for shard in walset.shards() { + // Reap finished checkpoints (non-blocking) and restart their clocks. + while let Some(done) = wave.try_join_next() { + let i = match &done { + Ok(i) => Some(*i), + Err(e) => { + eprintln!("WAL checkpoint task failed: {e}"); + task_shard.get(&e.id()).copied() + } + }; + if let Some(i) = i { + task_shard.retain(|_, v| *v != i); + in_flight[i] = false; + last_done[i] = std::time::Instant::now(); + } + } + for (i, shard) in walset.shards().iter().enumerate() { + if in_flight[i] { + continue; + } + let size_due = wal_bytes > 0 && shard.wal_size_bytes() >= wal_bytes; + let time_due = last_done[i].elapsed() >= interval; + if !(size_due || time_due) { + continue; + } + in_flight[i] = true; let shard = Arc::clone(shard); - wave.spawn(async move { + let handle = wave.spawn(async move { if let Err(e) = shard.checkpoint().await { eprintln!("WAL checkpoint failed for shard {:?}: {e}", shard.dir()); } + i }); + task_shard.insert(handle.id(), i); } - while wave.join_next().await.is_some() {} } }); } diff --git a/packages/durable-streams-rust/src/srvstats.rs b/packages/durable-streams-rust/src/srvstats.rs new file mode 100644 index 0000000000..7014e8da59 --- /dev/null +++ b/packages/durable-streams-rust/src/srvstats.rs @@ -0,0 +1,140 @@ +//! Lightweight, always-cheap server-side load telemetry for bottleneck analysis +//! (cardinality-cliff performance work). Answers the core question — is the server +//! CPU-bound, fsync/durability-bound, or lock-bound? — for BOTH wal and memory +//! modes, which the WAL-only `--wal-stats` counters cannot. +//! +//! Enabled by `--server-stats N` (seconds). Off by default and gated by +//! `STATS_ON`, so the hot-path instrumentation is a single relaxed load + branch +//! when disabled. Each tick prints a `SRV_STATS` line and resets the interval +//! accumulators. +//! +//! Fields: +//! - `cpu_cores` process CPU utilization in cores (utime+stime delta / wall); +//! ≈ the cgroup cpu quota ⇒ CPU-bound. Linux only (`-1` elsewhere). +//! - `appends_s` acked appends/sec over the interval. +//! - `inflight` in-flight append handlers sampled at tick time (queue depth). +//! - `svc_us` mean append handler wall time (service time). +//! - `applock_us` mean time waiting to acquire the per-stream appender lock. +//! - `durwait_us` mean time in `wait_durable_lsn` (WAL fsync wait; ~0 in memory). + +use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering}; +use std::time::Instant; + +static STATS_ON: AtomicBool = AtomicBool::new(false); +static APPENDS: AtomicU64 = AtomicU64::new(0); +static INFLIGHT: AtomicI64 = AtomicI64::new(0); +static SVC_US: AtomicU64 = AtomicU64::new(0); +static APPLOCK_US: AtomicU64 = AtomicU64::new(0); +static DURWAIT_US: AtomicU64 = AtomicU64::new(0); + +pub fn set_enabled(v: bool) { + STATS_ON.store(v, Ordering::Relaxed); +} +#[inline] +pub fn enabled() -> bool { + STATS_ON.load(Ordering::Relaxed) +} + +/// RAII probe for one append handler: bumps the in-flight gauge on creation and, +/// on drop (covering every early return), records the service time and counts the +/// append. Create it once `enabled()` is true. +pub struct AppendProbe { + start: Instant, +} +impl AppendProbe { + #[inline] + pub fn start() -> Option { + if !enabled() { + return None; + } + INFLIGHT.fetch_add(1, Ordering::Relaxed); + Some(Self { start: Instant::now() }) + } +} +impl Drop for AppendProbe { + fn drop(&mut self) { + INFLIGHT.fetch_sub(1, Ordering::Relaxed); + SVC_US.fetch_add(self.start.elapsed().as_micros() as u64, Ordering::Relaxed); + APPENDS.fetch_add(1, Ordering::Relaxed); + } +} + +#[inline] +pub fn record_applock_wait(d: std::time::Duration) { + if enabled() { + APPLOCK_US.fetch_add(d.as_micros() as u64, Ordering::Relaxed); + } +} +#[inline] +pub fn record_durwait(d: std::time::Duration) { + if enabled() { + DURWAIT_US.fetch_add(d.as_micros() as u64, Ordering::Relaxed); + } +} + +/// Read process CPU time (utime+stime) in seconds from `/proc/self/stat`. Linux +/// only; `None` elsewhere (macOS dev boxes — use a Linux container to profile). +#[cfg(target_os = "linux")] +fn cpu_secs() -> Option { + let s = std::fs::read_to_string("/proc/self/stat").ok()?; + // comm (field 2) may contain spaces/parens — parse after the final ')'. + let rest = &s[s.rfind(')')? + 1..]; + let f: Vec<&str> = rest.split_whitespace().collect(); + // After ')': [state, ppid, ...]; utime is field 14 ⇒ index 11, stime ⇒ 12. + let utime: f64 = f.get(11)?.parse().ok()?; + let stime: f64 = f.get(12)?.parse().ok()?; + let hz = 100.0; // _SC_CLK_TCK is 100 on all our targets. + Some((utime + stime) / hz) +} +#[cfg(not(target_os = "linux"))] +fn cpu_secs() -> Option { + None +} + +/// Spawn the periodic printer. Each tick emits one `SRV_STATS` line and resets the +/// interval accumulators. +pub fn spawn(secs: u64) { + set_enabled(true); + tokio::spawn(async move { + let mut ticker = tokio::time::interval(std::time::Duration::from_secs(secs)); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + ticker.tick().await; + let mut last_cpu = cpu_secs(); + let mut last = Instant::now(); + loop { + ticker.tick().await; + let now = Instant::now(); + let wall = now.duration_since(last).as_secs_f64(); + last = now; + + let appends = APPENDS.swap(0, Ordering::Relaxed); + let svc = SVC_US.swap(0, Ordering::Relaxed); + let applock = APPLOCK_US.swap(0, Ordering::Relaxed); + let durwait = DURWAIT_US.swap(0, Ordering::Relaxed); + let inflight = INFLIGHT.load(Ordering::Relaxed); + + let cpu_cores = match (cpu_secs(), last_cpu) { + (Some(c), Some(p)) if wall > 0.0 => { + last_cpu = Some(c); + (c - p) / wall + } + (Some(c), _) => { + last_cpu = Some(c); + -1.0 + } + _ => -1.0, + }; + + let n = appends.max(1) as f64; + eprintln!( + "SRV_STATS cpu_cores={:.2} appends_s={:.0} inflight={} svc_us={:.0} applock_us={:.1} durwait_us={:.1}", + cpu_cores, + appends as f64 / wall.max(0.001), + inflight, + svc as f64 / n, + applock as f64 / n, + durwait as f64 / n, + ); + } + }); +} diff --git a/packages/durable-streams-rust/src/store.rs b/packages/durable-streams-rust/src/store.rs index 1b69a5ab30..70cabde507 100644 --- a/packages/durable-streams-rust/src/store.rs +++ b/packages/durable-streams-rust/src/store.rs @@ -161,6 +161,114 @@ pub(crate) fn barrier_fsync(file: &File) -> std::io::Result<()> { } } +/// A single filesystem-wide durability barrier (Linux `syncfs`). Flushes ALL dirty +/// data+metadata on the filesystem that `file` lives on — used by the checkpoint's +/// `--wal-checkpoint-syncfs` path to make every touched per-stream file durable with +/// ONE syscall instead of `O(N_touched)` `fdatasync`s (cardinality-cliff #1). `file` +/// can be any open fd on the target fs (the checkpoint passes a touched stream file). +/// Linux-only; the caller gates on `cfg!(target_os = "linux")`, so the non-Linux stub +/// (which errors) is never reached in practice — it exists only so the crate compiles +/// on macOS. +#[cfg(target_os = "linux")] +pub(crate) fn syncfs_barrier(file: &File) -> std::io::Result<()> { + // SAFETY: `fd` is a valid open descriptor for the lifetime of `file`. + if unsafe { libc::syncfs(file.as_raw_fd()) } == 0 { + Ok(()) + } else { + Err(std::io::Error::last_os_error()) + } +} +#[cfg(not(target_os = "linux"))] +pub(crate) fn syncfs_barrier(_file: &File) -> std::io::Result<()> { + Err(std::io::Error::new( + std::io::ErrorKind::Unsupported, + "syncfs is Linux-only", + )) +} + +/// 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//`. +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>>> = 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, @@ -405,6 +513,61 @@ impl Store { ) -> std::io::Result { let streams_dir = data_dir.join("streams"); std::fs::create_dir_all(&streams_dir)?; + // Persist + validate the stream-lane count (mirrors the WAL shard count's + // persisted-N contract): opening a laned layout with a different + // `--stream-lanes` would make every existing stream silently invisible + // (recovery walks the wrong dirs). Refuse loudly instead. + { + let marker = streams_dir.join(".lanes"); + match std::fs::read_to_string(&marker) { + Ok(txt) => { + let on_disk: usize = txt.trim().parse().map_err(|_| { + std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("corrupt stream-lane marker {}", marker.display()), + ) + })?; + if on_disk != stream_lanes() { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!( + "--stream-lanes {} does not match this data dir's on-disk layout ({} lanes, recorded in {}). The lane count is a layout choice and must match across restarts.", + stream_lanes(), + on_disk, + marker.display() + ), + )); + } + } + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + // Legacy pre-marker dirs are all lanes == 1: refuse enabling + // lanes over an existing flat layout (its files would vanish). + if stream_lanes() > 1 + && std::fs::read_dir(&streams_dir)? + .flatten() + .any(|e| e.path().is_file()) + { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "--stream-lanes > 1 over an existing flat streams/ layout; this data dir was created with 1 lane", + )); + } + std::fs::write(&marker, format!("{}\n", stream_lanes()))?; + } + Err(e) => return Err(e), + } + } + // 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)] { @@ -444,11 +607,21 @@ 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 = HashMap::new(); let mut data_files: Vec = Vec::new(); - for entry in std::fs::read_dir(streams_dir)? { - let p = entry?.path(); + let mut entries: Vec = 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 == ".lanes" { + // stream-lane layout marker, not stream data + continue; + } if name.ends_with(".meta.tmp") { let _ = std::fs::remove_file(&p); } else if name.ends_with(".compact.tmp") { @@ -792,7 +965,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) diff --git a/packages/durable-streams-rust/src/tier.rs b/packages/durable-streams-rust/src/tier.rs index 64328515ae..b7a08f3fd1 100644 --- a/packages/durable-streams-rust/src/tier.rs +++ b/packages/durable-streams-rust/src/tier.rs @@ -173,6 +173,9 @@ pub enum TierKind { Off, /// Local-filesystem BlobStore (a chunk directory) — for testing offload /// without S3. + /// Sealed segments to a local directory. Test-only backend (no CLI + /// surface): exercises the seal/offload/compaction machinery without S3. + #[allow(dead_code)] Local, /// S3-compatible object storage (feature `tier`). S3, diff --git a/packages/durable-streams-rust/src/wal/codec.rs b/packages/durable-streams-rust/src/wal/codec.rs index 0408f29335..6fa8517dc1 100644 --- a/packages/durable-streams-rust/src/wal/codec.rs +++ b/packages/durable-streams-rust/src/wal/codec.rs @@ -24,11 +24,8 @@ //! trivially true even when a crash left a valid header over a zeroed, //! never-fully-written payload. Every WAL record is written by the buffered path, //! which has the payload in userspace and always sets `PAYLOAD_CHECKSUMMED`, so -//! such a torn record now fails decode. The old `--zero-copy` splice relay (the -//! only writer that left the flag clear) has been removed, so Bug #1 is **fully -//! closed** — there is no longer any unchecksummed WAL record in production. The -//! `PAYLOAD_CHECKSUMMED` flag and the flag-clear decode branch are retained only -//! for on-disk format stability. +//! such a torn record now fails decode. Every WAL record is checksummed; a +//! record with the flag clear is treated as torn/corrupt (ends the durable log). //! //! The header CRC also doubles as a torn-header detector: a partially-written //! header almost always fails the CRC, and an all-zero (`fallocate`'d, @@ -41,10 +38,8 @@ pub const HEADER_LEN: usize = 38; /// `flags` bit 0: the record's `payload_crc` field is a valid crc32c over the -/// payload and MUST be verified on decode. Every WAL writer (`encode_into`) sets -/// this — the old zero-copy splice relay that left it clear has been removed, so -/// all production WAL records are checksummed. The flag-clear branch in -/// [`decode_at`] is retained only for on-disk format stability. +/// payload. Every WAL writer (`encode_into`) sets it, and decode REQUIRES it — +/// a flag-clear header is treated as torn/corrupt. pub const PAYLOAD_CHECKSUMMED: u8 = 0b0000_0001; /// Record kind discriminant. The numeric values are the on-disk encoding and @@ -229,12 +224,12 @@ pub fn decode_at(seg: &[u8], off: usize) -> Decoded { return Decoded::Torn; } - // Bug #1 fix: if the writer checksummed the payload (buffered path), verify - // it. A torn/zeroed payload under a valid header (the fallocate'd-tail case) - // fails here and is rejected as Torn. Zero-copy records leave the flag clear - // and keep the "bytes present = complete" behavior (documented residual). - if flags & PAYLOAD_CHECKSUMMED != 0 - && crc32c::crc32c(&seg[payload_off..payload_off + len]) != payload_crc + // Bug #1 fix: verify the payload CRC. A torn/zeroed payload under a valid + // header (the fallocate'd-tail case) fails here and is rejected as Torn. + // Every writer sets PAYLOAD_CHECKSUMMED; a clear flag marks the record + // torn/corrupt (no writer has ever legitimately omitted the checksum here). + if flags & PAYLOAD_CHECKSUMMED == 0 + || crc32c::crc32c(&seg[payload_off..payload_off + len]) != payload_crc { return Decoded::Torn; } @@ -315,31 +310,6 @@ mod tests { } } - // Pure decode-behavior test (NOT a production case): a header with - // PAYLOAD_CHECKSUMMED clear opts out of the payload CRC, so a torn/zeroed - // payload tail decodes as `Record` (the legacy "bytes present = complete" - // branch). No WAL writer emits flag-clear records any more — the zero-copy - // splice relay that used to has been removed — so this branch is unreachable - // in production and Bug #1 is fully closed. The test pins the decode - // semantics only, for on-disk format stability. - #[test] - fn flag_clear_payload_skips_crc_validation_decode_only() { - let len: usize = 8192; - let prefix: usize = 4096; - let mut seg = Vec::new(); - // flags=0, payload_crc=0 — a hand-built flag-clear header (no production - // writer emits this). - encode_header_into(&mut seg, 1, RecordKind::Append, 42, 0, len as u32, 0, 0); - seg.extend(std::iter::repeat(0xAB).take(prefix)); - seg.extend(std::iter::repeat(0u8).take(len - prefix)); - seg.extend(std::iter::repeat(0u8).take(1 << 20)); - - assert!( - matches!(decode_at(&seg, 0), Decoded::Record { .. }), - "a flag-clear (PAYLOAD_CHECKSUMMED unset) header skips payload-CRC \ - validation — decode-only behavior, unreachable in production" - ); - } #[test] fn encode_decode_roundtrip_and_torn() { diff --git a/packages/durable-streams-rust/src/wal/e2e_tests.rs b/packages/durable-streams-rust/src/wal/e2e_tests.rs index e2e3819f3e..8b0c1511aa 100644 --- a/packages/durable-streams-rust/src/wal/e2e_tests.rs +++ b/packages/durable-streams-rust/src/wal/e2e_tests.rs @@ -952,6 +952,122 @@ 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); + 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 = (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> = 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(); + // Layout-mismatch guard: reopening this 3-lane dir with a different lane + // count must be REFUSED (persisted `.lanes` marker) — a silent mismatch + // would make every existing stream invisible. + crate::store::set_stream_lanes(2); + let err = Store::new_with_tier(dir.clone(), TierConfig::default()) + .err() + .expect("opening a 3-lane layout with --stream-lanes 2 must fail"); + assert!( + err.to_string().contains("stream-lanes"), + "mismatch error should name the knob: {err}" + ); + 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 +/// guarantee: after a checkpoint recycles the WAL, acked records (both those below +/// the checkpoint floor, made durable by `syncfs`, and those appended after) must +/// still recover byte-identically. On Linux this exercises the real `syncfs` path; +/// on other targets the code falls back to the per-stream loop (still correct). +#[tokio::test] +async fn e2e_checkpoint_syncfs_recovers_acked_records() { + let _guard = DurabilityGuard::wal(); + const SEG: u64 = 4096; + let dir = tmp("syncfs-ckpt"); + + let h = Harness::boot_with_segment_size(&dir, Some(1), 1, SEG).unwrap(); + create_stream(&h.store, "s", OCTET).await; + + let mut expected = Vec::new(); + // Enough records + small segments to force at least one roll, so the checkpoint + // actually recycles a fully-below-floor segment (relying on the syncfs'd file). + for i in 0..400usize { + let rec = format!("syncfs-{i:04}|").into_bytes(); + append_acked(&h.store, "s", OCTET, &rec).await; + expected.extend_from_slice(&rec); + } + // Force the checkpoint → syncfs barrier → recycle. + h.walset.shards()[0].checkpoint().await.unwrap(); + // More acked appends after the checkpoint (live segment). + for i in 400..500usize { + let rec = format!("syncfs-{i:04}|").into_bytes(); + append_acked(&h.store, "s", OCTET, &rec).await; + expected.extend_from_slice(&rec); + } + + h.crash(); + + let h2 = Harness::boot_with_segment_size(&dir, None, 1, SEG).unwrap(); + let got = stream_file_bytes(&h2.store, "s"); + assert_eq!( + got, expected, + "syncfs-checkpoint acked records recover byte-identical (durability-before-recycle held)" + ); + h2.crash(); + let _ = std::fs::remove_dir_all(&dir); +} + // =========================================================================== // (7c) WAL-QUIET stream: torn unacked tail truncated via the sidecar proof // =========================================================================== diff --git a/packages/durable-streams-rust/src/wal/shard.rs b/packages/durable-streams-rust/src/wal/shard.rs index 765d61dbf1..eefbf3208d 100644 --- a/packages/durable-streams-rust/src/wal/shard.rs +++ b/packages/durable-streams-rust/src/wal/shard.rs @@ -379,6 +379,36 @@ pub struct Shard { /// recyclable. const CHECKPOINT_FILE: &str = "checkpoint"; +/// Checkpoint cadence knobs. The checkpoint's only job is bounding retained-WAL +/// size (= crash-replay time): acks never gate on it (the disk-bounded safety +/// valve) and reads never touch the WAL, so firing it less often is free except +/// for disk space and replay budget. (Durability strategy is not a knob: on +/// Linux, checkpoint step 2 issues one `syncfs` per stream lane; elsewhere it +/// runs the serial per-file `fdatasync` loop. Ordering is identical: the barrier +/// completes before `persist_durable_tails`/recycle.) Two triggers, whichever +/// comes first per shard: +/// * time: `--wal-checkpoint-interval-ms` (default 3000 — the historical 3 s +/// wave cadence, now per-shard). +/// * size: `--wal-checkpoint-wal-bytes` (default 0 = disabled) — checkpoint a +/// shard as soon as its retained WAL exceeds this many bytes. This turns the +/// hardcoded timer into an explicit replay-time budget (e.g. 1 GiB ≈ <1 s of +/// replay on NVMe) and lets shards self-stagger by their own write rates +/// instead of storming together on a shared tick. +static CHECKPOINT_INTERVAL_MS: AtomicU64 = AtomicU64::new(3_000); +pub fn set_checkpoint_interval_ms(ms: u64) { + CHECKPOINT_INTERVAL_MS.store(ms.max(1), Ordering::Relaxed); +} +pub fn checkpoint_interval_ms() -> u64 { + CHECKPOINT_INTERVAL_MS.load(Ordering::Relaxed) +} +static CHECKPOINT_WAL_BYTES: AtomicU64 = AtomicU64::new(0); +pub fn set_checkpoint_wal_bytes(bytes: u64) { + CHECKPOINT_WAL_BYTES.store(bytes, Ordering::Relaxed); +} +pub fn checkpoint_wal_bytes() -> u64 { + CHECKPOINT_WAL_BYTES.load(Ordering::Relaxed) +} + /// Name of the per-shard durable-tail map: `/tails` (task 11b). A /// CUMULATIVE `stream_id durable_tail` line map (plain decimal text, one stream /// per line). At checkpoint, each touched stream's current logical `Shared.tail` @@ -833,9 +863,23 @@ impl Shard { let n_touched = touched.len(); let t_capture = t_start.elapsed(); - // 2. fdatasync each touched per-stream file. - for (_, _, f) in &touched { - crate::store::barrier_fsync(f)?; + // 2. Make every touched per-stream file durable, strictly BEFORE + // persist_durable_tails/recycle (the durability-before-recycle + // ordering). Two modes: + // * syncfs (Linux default): one filesystem-wide barrier per stream + // lane — O(lanes) syscalls instead of O(N_touched) fdatasyncs + // (the cardinality cliff). Lanes are independent devices in the + // intended layout, so the barriers run in parallel. + // * per-file fdatasync loop: the non-Linux (or --wal-checkpoint- + // syncfs off) fallback. Serial on purpose: parallel fan-out + // regressed in every controlled test (it steals device budget + // from the commit path). + if cfg!(target_os = "linux") && n_touched > 0 { + crate::store::syncfs_stream_lanes(&touched[0].2)?; + } else { + for (_, _, f) in &touched { + crate::store::barrier_fsync(f)?; + } } let t_fsync = t_start.elapsed();