diff --git a/packages/durable-streams-rust/Dockerfile b/packages/durable-streams-rust/Dockerfile deleted file mode 100644 index 0fb792ec68..0000000000 --- a/packages/durable-streams-rust/Dockerfile +++ /dev/null @@ -1,28 +0,0 @@ -# syntax=docker/dockerfile:1 -# -# Built multi-arch (linux/amd64 + linux/arm64) natively per arch by -# .github/workflows/docker_multiarch_image.yml — the build context is the repo -# root and this Dockerfile is `packages/durable-streams-rust/Dockerfile`. - -# ---- build stage: compile the release binary (glibc, matches the runtime) ---- -FROM rust:1-bookworm AS build -WORKDIR /app -# Copy only what the build needs (no target/, no npm/) so we don't depend on a -# .dockerignore at the shared repo root. -COPY packages/durable-streams-rust/Cargo.toml packages/durable-streams-rust/Cargo.lock ./ -COPY packages/durable-streams-rust/src ./src -# Default features only (no `tier`/`telemetry`) — minimal image, matching the -# conformance matrix. To ship S3 tiering, add `--features tier` here AND -# `ca-certificates` to the runtime stage. -RUN cargo build --release --locked - -# ---- runtime stage: distroless (glibc cc), no shell / package manager ---- -FROM gcr.io/distroless/cc-debian12 AS runtime -COPY --from=build /app/target/release/durable-streams-server /usr/local/bin/durable-streams-server -# Protocol default port (PROTOCOL.md §13.1); override with `--port`. -EXPOSE 4437 -ENTRYPOINT ["/usr/local/bin/durable-streams-server"] -# Bind all interfaces by default (the binary defaults to 127.0.0.1, unreachable -# from outside the container). Override by passing your own args. For persistence, -# mount a volume and add `--data-dir /your/path` (default is an ephemeral tmp dir). -CMD ["--host", "0.0.0.0"] diff --git a/packages/durable-streams-rust/WAL_TUNING.md b/packages/durable-streams-rust/WAL_TUNING.md new file mode 100644 index 0000000000..f442ecff4e --- /dev/null +++ b/packages/durable-streams-rust/WAL_TUNING.md @@ -0,0 +1,106 @@ +# 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 | +| + `--wal-checkpoint-syncfs on` (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 | + +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×. + +Example (6-device box): device 0 → data root, devices 1–5 → 5 WAL shards: + +``` +--data-dir /data/wal/0 --wal-shards 5 +``` + +### 2. Server flags + +``` +--wal-checkpoint-syncfs on # one syncfs barrier per checkpoint instead of + # O(N-touched) per-stream fdatasync (PR #4697) +--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) +--worker-threads +``` + +Leave `--wal-fsync-parallel` at its default (1): fanout parallelizes the +per-stream fsync loop that syncfs replaces, and measured as a regression. + +### 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 + +- Validated to 100k streams, 256 B payloads, single node. The stacked config + measured 383k @100k, 244k @500k, and **56k @1M** — a NEW, different wall + appears near 1M streams (candidates: one open fd per live stream vs the + container nofile ceiling, an ext4 directory with 1M files, stream-map/tails + working set, page-cache pressure from 1M dirty files). Profiling pass pending; + see `CARDINALITY_1M.md` for the older analysis. Below ~500k streams the + configuration above is cliff-free. +- `--wal-checkpoint-syncfs` and the size trigger are opt-in; flipping defaults + (syncfs on for Linux) 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/main.rs b/packages/durable-streams-rust/src/main.rs index 3bbb50e45e..19b341ffe3 100644 --- a/packages/durable-streams-rust/src/main.rs +++ b/packages/durable-streams-rust/src/main.rs @@ -300,6 +300,30 @@ fn main() { } } } + // 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); + } + } + } + // 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); @@ -474,43 +498,68 @@ 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(); + 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() { + if let Ok(i) = done { + 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 { if let Err(e) = shard.checkpoint().await { eprintln!("WAL checkpoint failed for shard {:?}: {e}", shard.dir()); } + i }); } - while wave.join_next().await.is_some() {} } }); } diff --git a/packages/durable-streams-rust/src/wal/shard.rs b/packages/durable-streams-rust/src/wal/shard.rs index df195225d1..4d94591b8e 100644 --- a/packages/durable-streams-rust/src/wal/shard.rs +++ b/packages/durable-streams-rust/src/wal/shard.rs @@ -423,6 +423,33 @@ fn checkpoint_syncfs() -> bool { CHECKPOINT_SYNCFS.load(Ordering::Relaxed) } +/// Checkpoint cadence knobs (cardinality-cliff follow-up). The checkpoint's only +/// job is bounding retained-WAL size (= crash-replay time): acks never gate on it +/// (`shard.rs` "disk-bounded safety valve") and reads never touch the WAL, so +/// firing it less often is free except for disk space and replay budget. 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`