diff --git a/packages/durable-streams-rust/Cargo.lock b/packages/durable-streams-rust/Cargo.lock index f448af9c2d..02a42ac6b7 100644 --- a/packages/durable-streams-rust/Cargo.lock +++ b/packages/durable-streams-rust/Cargo.lock @@ -200,6 +200,7 @@ dependencies = [ "crc32c", "dashmap", "httparse", + "io-uring", "libc", "object_store", "opentelemetry", @@ -675,6 +676,17 @@ dependencies = [ "hashbrown 0.17.1", ] +[[package]] +name = "io-uring" +version = "0.7.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9080b15e63775b9a2ac7dca720f7050a8b955e092ea0f6020a4a80f69998cdc0" +dependencies = [ + "bitflags", + "cfg-if", + "libc", +] + [[package]] name = "ipnet" version = "2.12.0" diff --git a/packages/durable-streams-rust/Cargo.toml b/packages/durable-streams-rust/Cargo.toml index 654dbd6298..f6ed5aa7b7 100644 --- a/packages/durable-streams-rust/Cargo.toml +++ b/packages/durable-streams-rust/Cargo.toml @@ -78,3 +78,7 @@ codegen-units = 1 # server the per-connection isolation is worth more than abort's small # binary-size / codegen win. +[target.'cfg(target_os = "linux")'.dependencies] +# io_uring write paths (--io-uring on): WAL segment writes/fsync + stream +# data-file appends. Runtime-probed, syscall fallback when unavailable. +io-uring = "0.7" diff --git a/packages/durable-streams-rust/README.md b/packages/durable-streams-rust/README.md index 45b5ee5683..2a2fe0c98d 100644 --- a/packages/durable-streams-rust/README.md +++ b/packages/durable-streams-rust/README.md @@ -89,8 +89,7 @@ durable, single-node server on `127.0.0.1:4437` with its data dir under `$TMPDIR | 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` | `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 | @@ -107,7 +106,7 @@ S3 credentials come from the **environment**, never flags: `DS_S3_ACCESS_KEY_ID` | 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. | +| **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. | @@ -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..e61a73ccca --- /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..35ed051a2f 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, } @@ -735,7 +735,18 @@ async fn wait_durable_lsn(store: &Arc, st: &Arc, lsn: u64) { /// awaited in `wait_durable_lsn`. fn write_wire(st: &StreamState, ap: &mut Appender, wire: &Bytes) -> std::io::Result { use std::io::Write; - (&*ap.file).write_all(wire)?; + // io_uring path (--io-uring on, Linux): the data-file append goes through + // the calling thread's ring (O_APPEND semantics preserved via offset -1). + // This is BOTH modes' persistence write — memory mode's entire disk path. + if crate::uring::enabled() { + #[cfg(target_os = "linux")] + { + use std::os::fd::AsRawFd; + crate::uring::append_all(ap.file.as_raw_fd(), wire)?; + } + } else { + (&*ap.file).write_all(wire)?; + } ap.written += wire.len() as u64; let tail = { let mut s = st.shared.write().unwrap(); @@ -896,6 +907,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 +964,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 +1099,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 +1156,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 +1193,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() { @@ -2298,5 +2348,49 @@ 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); + } } diff --git a/packages/durable-streams-rust/src/main.rs b/packages/durable-streams-rust/src/main.rs index 721240c4ec..65069ba8ab 100644 --- a/packages/durable-streams-rust/src/main.rs +++ b/packages/durable-streams-rust/src/main.rs @@ -3,11 +3,13 @@ mod blobstore; mod engine_raw; mod handlers; mod http1; +mod srvstats; #[cfg(target_os = "linux")] mod sse_reactor; mod store; mod telemetry; mod tier; +mod uring; mod wal; use std::net::SocketAddr; @@ -137,6 +139,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 +169,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 +195,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 +237,68 @@ 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); + } + // io_uring write paths (WAL segment I/O + stream data-file appends, + // both durability modes). Linux-only; probed at first use with a + // transparent syscall fallback. Default off. + "--io-uring" => { + let v = val(args.next(), "--io-uring"); + match v.as_str() { + "on" => uring::set_requested(true), + "off" => uring::set_requested(false), + _ => { + eprintln!("--io-uring must be on|off"); + std::process::exit(2); + } + } + } + // 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 +311,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 +371,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 +480,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..aee1a396e2 --- /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/uring.rs b/packages/durable-streams-rust/src/uring.rs new file mode 100644 index 0000000000..dfc8fbb956 --- /dev/null +++ b/packages/durable-streams-rust/src/uring.rs @@ -0,0 +1,169 @@ +//! io_uring-backed file I/O for the write paths (`--io-uring on`, default off). +//! +//! Scope: the two hot write sites share this module — +//! * WAL segment I/O (`wal/segment.rs` `SegmentWriter`): appender-side +//! positioned record writes + the committer's group-commit `fdatasync`. +//! * Stream data-file appends (`handlers::write_wire`), which BOTH durability +//! modes execute on every append (memory mode's entire persistence path). +//! +//! Design: one **thread-local ring** per OS thread (rings are not `Sync`; the +//! committer is a dedicated thread and tokio workers each get their own), plain +//! submit-and-wait per op. That converts `pwrite`/`write`/`fdatasync` syscalls +//! into `io_uring_enter` round-trips — roughly syscall-neutral per op, but it +//! keeps the door open for linked WRITE→FSYNC chains and SQPOLL without touching +//! call sites again, and measures the uring path end-to-end. +//! +//! Fallback ladder (always correct, never fails closed): +//! * flag off (default) → caller uses its normal syscalls. +//! * non-Linux → flag is accepted but this module reports `enabled() == false`. +//! * Linux without io_uring (kernel/seccomp) → one-time probe fails → +//! `enabled() == false`, callers fall back to syscalls. + +use std::sync::atomic::{AtomicBool, Ordering}; + +static REQUESTED: AtomicBool = AtomicBool::new(false); + +pub fn set_requested(on: bool) { + REQUESTED.store(on, Ordering::Relaxed); +} + +/// True iff `--io-uring on` AND the platform actually supports it (probed once). +#[inline] +pub fn enabled() -> bool { + REQUESTED.load(Ordering::Relaxed) && probe_ok() +} + +#[cfg(target_os = "linux")] +fn probe_ok() -> bool { + use std::sync::OnceLock; + static OK: OnceLock = OnceLock::new(); + *OK.get_or_init(|| match io_uring::IoUring::new(8) { + Ok(_) => true, + Err(e) => { + eprintln!("io_uring unavailable ({e}); --io-uring falling back to syscalls"); + false + } + }) +} +#[cfg(not(target_os = "linux"))] +fn probe_ok() -> bool { + false +} + +#[cfg(target_os = "linux")] +mod linux { + use std::cell::RefCell; + use std::io; + use std::os::fd::RawFd; + + use io_uring::{opcode, types, IoUring}; + + // Ring depth 64: a single synchronous op per call needs 1, but leaves room + // for future linked chains/batches on the same ring. + thread_local! { + static RING: RefCell> = const { RefCell::new(None) }; + } + + fn with_ring(f: impl FnOnce(&mut IoUring) -> io::Result) -> io::Result { + RING.with(|r| { + let mut r = r.borrow_mut(); + if r.is_none() { + *r = Some(IoUring::new(64)?); + } + f(r.as_mut().unwrap()) + }) + } + + /// Submit one SQE and wait for its CQE, returning the raw kernel result. + fn submit_one(ring: &mut IoUring, sqe: io_uring::squeue::Entry) -> io::Result { + // SAFETY: the buffers referenced by `sqe` outlive the synchronous wait + // below (callers pass slices that live across this call). + unsafe { + ring.submission() + .push(&sqe) + .map_err(|_| io::Error::other("io_uring SQ full"))?; + } + ring.submit_and_wait(1)?; + let cqe = ring + .completion() + .next() + .ok_or_else(|| io::Error::other("io_uring missing CQE"))?; + Ok(cqe.result()) + } + + /// Positioned write of the whole slice (loops on short writes). + pub fn pwrite_all(fd: RawFd, mut off: u64, mut buf: &[u8]) -> io::Result<()> { + while !buf.is_empty() { + let n = with_ring(|ring| { + let sqe = opcode::Write::new(types::Fd(fd), buf.as_ptr(), buf.len() as u32) + .offset(off) + .build(); + submit_one(ring, sqe) + })?; + if n < 0 { + return Err(io::Error::from_raw_os_error(-n)); + } + if n == 0 { + return Err(io::Error::new(io::ErrorKind::WriteZero, "uring write returned 0")); + } + off += n as u64; + buf = &buf[n as usize..]; + } + Ok(()) + } + + /// Append the whole slice to an `O_APPEND` fd (offset −1 = use file + /// position; io_uring honours `O_APPEND` semantics). + pub fn append_all(fd: RawFd, mut buf: &[u8]) -> io::Result<()> { + while !buf.is_empty() { + let n = with_ring(|ring| { + let sqe = opcode::Write::new(types::Fd(fd), buf.as_ptr(), buf.len() as u32) + .offset(u64::MAX) // -1: append/current position + .build(); + submit_one(ring, sqe) + })?; + if n < 0 { + return Err(io::Error::from_raw_os_error(-n)); + } + if n == 0 { + return Err(io::Error::new(io::ErrorKind::WriteZero, "uring write returned 0")); + } + buf = &buf[n as usize..]; + } + Ok(()) + } + + /// `fdatasync` via io_uring (`IORING_FSYNC_DATASYNC`). + pub fn fdatasync(fd: RawFd) -> io::Result<()> { + let n = with_ring(|ring| { + let sqe = opcode::Fsync::new(types::Fd(fd)) + .flags(types::FsyncFlags::DATASYNC) + .build(); + submit_one(ring, sqe) + })?; + if n < 0 { + return Err(io::Error::from_raw_os_error(-n)); + } + Ok(()) + } +} + +#[cfg(target_os = "linux")] +pub use linux::{append_all, fdatasync, pwrite_all}; + +// Non-Linux stubs: never called (enabled() is false), exist so the crate compiles. +#[cfg(not(target_os = "linux"))] +#[allow(dead_code)] +pub fn pwrite_all(_fd: i32, _off: u64, _buf: &[u8]) -> std::io::Result<()> { + unreachable!("io_uring path is Linux-only and gated by enabled()") +} +#[cfg(not(target_os = "linux"))] +#[allow(dead_code)] +pub fn append_all(_fd: i32, _buf: &[u8]) -> std::io::Result<()> { + unreachable!("io_uring path is Linux-only and gated by enabled()") +} +#[cfg(not(target_os = "linux"))] +#[allow(dead_code)] +pub fn fdatasync(_fd: i32) -> std::io::Result<()> { + unreachable!("io_uring path is Linux-only and gated by enabled()") +} 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..c28e59f85d 100644 --- a/packages/durable-streams-rust/src/wal/e2e_tests.rs +++ b/packages/durable-streams-rust/src/wal/e2e_tests.rs @@ -952,6 +952,158 @@ async fn e2e_recycled_first_segment_acked_records_survive_crash() { let _ = std::fs::remove_dir_all(&dir); } +/// `--io-uring on`: the WAL segment writes/fsync and the stream data-file +/// appends go through io_uring (Linux; transparent syscall fallback elsewhere, +/// so this test is meaningful on Linux and a fallback-path check on macOS). +/// Byte-identical crash recovery must hold — the ring changes the submission +/// mechanism, not the durability contract. +#[tokio::test] +async fn e2e_io_uring_wal_recovers_acked_records() { + let _guard = DurabilityGuard::wal(); + crate::uring::set_requested(true); + const SEG: u64 = 4096; + let dir = tmp("io-uring"); + + let h = Harness::boot_with_segment_size(&dir, Some(1), 1, SEG).unwrap(); + create_stream(&h.store, "u", OCTET).await; + let mut expected = Vec::new(); + for i in 0..400usize { + let rec = format!("uring-{i:04}|").into_bytes(); + append_acked(&h.store, "u", OCTET, &rec).await; + expected.extend_from_slice(&rec); + } + h.walset.shards()[0].checkpoint().await.unwrap(); + for i in 400..500usize { + let rec = format!("uring-{i:04}|").into_bytes(); + append_acked(&h.store, "u", 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, "u"); + assert_eq!(got, expected, "io_uring WAL path recovers byte-identical"); + h2.crash(); + crate::uring::set_requested(false); + 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/segment.rs b/packages/durable-streams-rust/src/wal/segment.rs index bd74249960..a74ab0bcd3 100644 --- a/packages/durable-streams-rust/src/wal/segment.rs +++ b/packages/durable-streams-rust/src/wal/segment.rs @@ -179,6 +179,11 @@ fn macos_full_fsync(fd: libc::c_int) -> io::Result<()> { impl SegmentWriter for FileSegment { fn write_at(&self, off: u64, bytes: &[u8]) -> io::Result<()> { let fd = self.file.as_raw_fd(); + // io_uring path (--io-uring on, Linux): same positioned-write semantics + // through the ring; falls back to pwrite below when off/unavailable. + if crate::uring::enabled() { + return crate::uring::pwrite_all(fd, off, bytes); + } let mut written: usize = 0; // pwrite may return a short count; loop until the whole slice lands. while written < bytes.len() { @@ -209,6 +214,10 @@ impl SegmentWriter for FileSegment { fn fdatasync(&self) -> io::Result<()> { let fd = self.file.as_raw_fd(); + // io_uring path: IORING_FSYNC_DATASYNC, same durability contract. + if crate::uring::enabled() { + return crate::uring::fdatasync(fd); + } // Mirror `store::barrier_fsync`: macOS has no fdatasync, so use // F_FULLFSYNC for a true flush-to-platter (power-loss durable); Linux // uses fdatasync. 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();