diff --git a/Cargo.lock b/Cargo.lock index dde6a649..df91d394 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4131,6 +4131,7 @@ dependencies = [ "fastcdc", "file-rotate", "flate2", + "foldhash 0.2.0", "fs2", "futures", "hex", diff --git a/Cargo.toml b/Cargo.toml index f18f870f..4fcbc6a1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -85,6 +85,10 @@ infer = "0.19" async-compression = { version = "0.4.42", features = ["tokio", "gzip"] } async_zip = { version = "0.0.18", features = ["tokio", "deflate"] } dashmap = "6.2.1" +# Fast, DoS-resistant (per-instance random-seeded) hasher for trusted- and +# attacker-controlled internal maps/sets. Already present transitively via +# hashbrown, so this direct dep adds no new compiled crate (benches/ROUND26.md §G1). +foldhash = "0.2" socket2 = { version = "0.6.4", features = ["all"] } urlencoding = "2.1.3" utoipa = { version = "5.5.0", features = ["axum_extras", "uuid", "chrono"] } @@ -354,6 +358,32 @@ name = "bench_micro_allocs" path = "examples/bench_micro_allocs.rs" required-features = ["bench"] +# Round-26 battery ──────────────────────────────────────────────────────────── + +# Round-26 CPU/alloc micro-pack (no Postgres) — drive-policy JSONB decode through +# a throwaway serde_json::Value DOM → from_slice:: (P1, the §J1 +# pattern applied to the drive-policy path §J2 left behind). +[[example]] +name = "bench_round26_micro" +path = "examples/bench_round26_micro.rs" +required-features = ["bench"] + +# Round-26 disk-I/O pack — CachedBlobBackend redundant per-write create_dir_all +# on warm shards → pre-create the 256 shard dirs at init (D1). (D2, moving the +# eviction unlink off the reactor via spawn_blocking, was tested and REVERTED — +# spawn_blocking dispatch costs more than the fast local unlink; see ROUND26.md.) +[[example]] +name = "bench_round26_diskio" +path = "examples/bench_round26_diskio.rs" +required-features = ["bench"] + +# Round-26 hasher pack — delta-upload have/need hash sets: SipHash → foldhash +# (per-instance random-seeded, DoS-safe for the attacker-controlled hashes) (G1). +[[example]] +name = "bench_round26_hasher" +path = "examples/bench_round26_hasher.rs" +required-features = ["bench"] + # Round-25 battery ──────────────────────────────────────────────────────────── # Round-25 CPU/alloc/RAM micro-pack (no Postgres) — deterministic alloc+bytes diff --git a/benches/ROUND26.md b/benches/ROUND26.md new file mode 100644 index 00000000..5caa67cc --- /dev/null +++ b/benches/ROUND26.md @@ -0,0 +1,156 @@ +# Round 26 — drive-policy JSONB decode (alloc), CachedBlobBackend shard-dir pre-create (disk), delta-upload foldhash (CPU); eviction-unlink off-reactor tested & reverted + +This round drains three high-confidence items from the ROUND25 backlog, each +behind a BEFORE/AFTER benchmark that `std::process::exit(1)`s ("`GATE FAIL … +rollback`") unless AFTER strictly beats BEFORE. A fourth candidate (moving the +cache eviction unlink off the reactor) was **tested and reverted** — the +benchmark refuted it. All three shipped items target the owner's priorities: +allocations, disk-I/O, and CPU. + +Reproduce: + +```bash +RUSTFLAGS="-C target-cpu=x86-64-v3" cargo run --release --features bench --example bench_round26_micro # P1 +RUSTFLAGS="-C target-cpu=x86-64-v3" cargo run --release --features bench --example bench_round26_diskio # D1 +RUSTFLAGS="-C target-cpu=x86-64-v3" cargo run --release --features bench --example bench_round26_hasher # G1 +``` + +--- + +## [P1] Drive-policy reads: throwaway `serde_json::Value` DOM → `from_slice::` (allocations) + +`drive_pg_repository`'s four policy reads (`get_policies_for_file/_folder`, +`get_drive_id_and_policies_for_file/_folder`) fetched `d.policies` as a +`serde_json::Value` and then called `DrivePolicies::from_value(&raw)` +(`Self::deserialize(&Value)`). The `Value` tree — a `Map` + a boxed `String` key ++ a `Value` node per policy field — is built once, walked once, and dropped. This +is the exact throwaway-DOM pattern ROUND23 §J1 removed for contacts; §J2 removed +only the `from_value` *clone*, not the DOM. These reads fire on file/folder +move & copy and on every share/grant creation. + +AFTER fetches through `sqlx::types::Json` — one +`serde_json::from_slice::` over the raw JSONB bytes, no +intermediate DOM — via a shared `policies_from_row` helper that preserves the +lenient `unwrap_or_default` fallback exactly (a malformed bag → all-false, +`try_get(...).unwrap_or_default()`, mirroring §J1). + +| arm | ns/op | allocs/op | bytes/op | +|--------|-------:|----------:|---------:| +| BEFORE | 399.4 | 6.00 | 719 | +| AFTER | 161.0 | 0.00 | 0 | + +**6 → 0 allocs/op, −719 bytes/op, 2.48× wall** — the entire Value DOM removed per +policy read. Gate: AFTER allocs/op strictly lower. Equivalence: the decoded +`DrivePolicies` is asserted identical BEFORE vs AFTER. + +## [D1] `CachedBlobBackend`: pre-create the 256 shard dirs at init, drop the per-write `create_dir_all` (disk-I/O) + +`CachedBlobBackend::initialize` created only `cache_dir`, never the 256 +`{00..ff}` shard dirs (the line-122 comment claimed otherwise). So all three +cache-write sites (`cache_bytes_write_through`, `insert_into_cache`, +`fetch_and_cache`) re-ran `tokio::fs::create_dir_all(parent)` per chunk — a +wasted `mkdirat(EEXIST)` + component stat + blocking-pool dispatch on a shard +that already exists, on every cached-remote write. AFTER creates all 256 shards +once at init (mirroring `LocalBlobBackend::initialize`, reusing its +`HEX_PREFIXES` table) and deletes the three per-write calls; the shard for any +`&hash[..2]` prefix always exists, so the writes just `fs::write`/`fs::copy`. + +Measured on a tmpfs tempdir (`create_dir_all` on an already-existing shard vs the +skip): + +| arm | ns/write | +|--------|---------:| +| BEFORE | 44 801.8 | +| AFTER | 0.3 | + +**~45 µs removed per cache write.** Gate: AFTER ns/write strictly lower. The +on-disk layout is identical; the directory creation simply moved from the hot +path to one-time startup. + +## [G1] Delta-upload have/need hash sets: SipHash → `foldhash::quality::RandomState` (CPU) + +The delta-upload negotiation builds `HashSet`s over up to `max_chunk_count()` +client-supplied 64-hex BLAKE3 hashes per request (`distinct_hashes`, and +`authorize_chunk_download`'s `distinct_seen`). std `HashSet` uses SipHash-1-3 — +DoS-resistant but ~2-4× slower than a modern hash on short keys. AFTER uses +`foldhash::quality::RandomState`, a fast non-cryptographic hasher that **stays +DoS-resistant** because it is per-instance random-seeded — the required property +for these *attacker-controlled* inputs (not `FxHash`/a fixed seed). `foldhash` is +already in the lockfile transitively (via `hashbrown`), so the direct dep adds no +newly-compiled crate. + +Build + membership scan over 40 000 client hashes, p50 over 50 passes: + +| arm | p50 ms (build+scan) | +|-------------------|--------------------:| +| BEFORE (SipHash) | 4.768 | +| AFTER (foldhash) | 2.007 | + +**2.37× wall** on the delta negotiation's hottest set — a bulk sync of a large +file negotiates thousands of chunks. Scales with `max_chunk_count()`. + +Gate: AFTER p50 wall (build set + membership scan over N hashes) strictly lower, +**and** two `RandomState::default()` instances must produce different hashes for +the same key (asserting the random per-instance seed — DoS resistance retained). +The set membership decisions are unchanged, so behaviour is identical. + +--- + +## Tested and reverted + +- **[D2] Move the cache eviction unlink off the reactor via `spawn_blocking`.** + The moka eviction listener unlinks a size-evicted blob with a synchronous + `std::fs::remove_file` inline on the tokio worker that triggered the insert. + The hypothesis: hand it to `spawn_blocking` so the reactor isn't blocked on + `unlink(2)`. The benchmark refutes it on the relevant configuration: + + | arm | ns on reactor / eviction | + |-----------------------------|-------------------------:| + | BEFORE (inline remove_file) | 7 055.7 | + | AFTER (spawn_blocking) | 19 848.1 | + + `CachedBlobBackend` caches on a **local** dir (fast unlink, ~7 µs), and + `spawn_blocking`'s task-dispatch overhead (~20 µs) costs *more* on the reactor + than the inline unlink it replaces — a net loss. The original code comment ("a + quick unlink on the inserting task's thread, off the hot get path") is correct + for the fast-local-cache case. A win would only materialize on genuinely slow + storage (network-backed cache dir), which there is no fixture for here. + **Reverted; kept the inline unlink.** (A "measure before believing" result, like + BASELINE's dropped Task 2.1 / reverted Phase 1.7.) + +## Not shipped — carried forward + +Named in the ROUND25 backlog, still queued (each wants a multi-signature change, +a remote-backend fixture, or a different toolchain): + +- **`format_oc_id_into` buffer** through the NC PROPFIND/REPORT/trashbin emit + loops — a per-row `String` → reused buffer. Threads a buffer through ~4 loop + sites across 3 files and depends on `NextcloudFileIdService`'s instance-id + format; wants its own validated pass so a wrong `oc:id` can't reach a client. +- **S3 read zero-copy forward** (`into_async_read()+ReaderStream` → forward the + SDK `Bytes` frames, Azure-style) — needs a MinIO/stub `ByteStream` fixture. +- **Frontend folder-listing cache** (`getCachedFolder`/`cacheFolder` is dead + code; every navigation refetches with `cache:'no-store'`) — a SvelteKit/Vitest + pass (bandwidth + instant paint on revisits). +- **foldhash for the NC PROPFIND trusted-key maps** (`favorite_ids`, `nc_id`) — + `foldhash::fast` (no random seed needed; server-generated keys). Threads the + hasher type through the emit-loop map builders. +- **Contact create/update `Json` bind** (write-side twin of §J1) and the other + ROUND25 backlog items. + +## Environment / methodology + +- **P1:** counting global allocator (count + bytes), no Postgres. The real + `DrivePolicies` type is imported from the crate; BEFORE replicates the shipped + `serde_json::from_slice::` + `deserialize(&Value)`, AFTER the shipped + `from_slice::`. Value-equivalence asserted; gate on allocs/op. +- **D1:** async wall on a tmpfs `tempfile::tempdir`; BEFORE = `create_dir_all` on + a pre-existing shard, AFTER = the skip. Gate on ns/write. +- **G1:** wall-gated (a hasher swap changes 0 allocations); SipHash vs + `foldhash::quality` build+scan over N random 64-hex hashes; DoS-seed assertion. +- Built with `RUSTFLAGS="-C target-cpu=x86-64-v3"` (the checked-in + `.cargo/config.toml` pins `target-cpu=native`, which `SIGILL`s on this host — + see ROUND23/24; local override only). +- Verified beyond the benches: `cargo fmt --all --check` clean, + `cargo clippy --features bench -- -D warnings` clean, `cargo test --lib + --features bench` green. diff --git a/examples/bench_round26_diskio.rs b/examples/bench_round26_diskio.rs new file mode 100644 index 00000000..f0099c1c --- /dev/null +++ b/examples/bench_round26_diskio.rs @@ -0,0 +1,84 @@ +//! Round-26 disk-I/O pack (no Postgres) — async wall on a tmpfs-backed tempdir. +//! +//! [D1] `CachedBlobBackend::initialize` creates ONLY `cache_dir`, never the 256 +//! `{00..ff}` shard dirs (the line-122 comment claims otherwise), so each +//! of the three cache-write sites re-runs `tokio::fs::create_dir_all(parent)` +//! on the hot path — a wasted `mkdirat(EEXIST)` + component stat + a +//! blocking-pool dispatch per chunk write on cached-remote deployments. +//! AFTER pre-creates the shard dirs at init (mirroring +//! `LocalBlobBackend::initialize`) and drops the per-write call. Gate: +//! AFTER wall (per write) strictly lower than BEFORE (the redundant +//! create_dir_all). +//! +//! [D2] TESTED AND REVERTED — see benches/ROUND26.md. Moving the moka +//! eviction-listener unlink off the reactor via `spawn_blocking` was +//! refuted by the benchmark: on the local cache dir (fast unlink ~7 µs) +//! the `spawn_blocking` dispatch (~20 µs) costs MORE on the reactor than +//! the inline `std::fs::remove_file` it replaces. The original inline +//! unlink ("a quick unlink on the inserting task's thread") is correct +//! for the fast-local-cache case; kept as-is. +//! +//! Run: +//! RUSTFLAGS="-C target-cpu=x86-64-v3" \ +//! cargo run --release --features bench --example bench_round26_diskio +//! Tunables (env): D1_ITERS (20000) + +use std::env; +use std::time::Instant; + +fn env_or(key: &str, default: T) -> T { + env::var(key) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(default) +} + +fn gate(tag: &str, metric: &str, before: f64, after: f64) { + if !(after < before) { + eprintln!("GATE FAIL [{tag}] {metric}: AFTER {after} !< BEFORE {before} — rollback"); + std::process::exit(1); + } +} + +// ── [D1] redundant create_dir_all on a warm shard vs skip ──────────────────── +async fn section_d1() { + let iters: u64 = env_or("D1_ITERS", 20_000); + let dir = tempfile::tempdir().expect("tempdir"); + let shard = dir.path().join("ab"); + // Shard pre-created once (what AFTER's initialize does). + tokio::fs::create_dir_all(&shard).await.unwrap(); + + // warm + let _ = tokio::fs::create_dir_all(&shard).await; + + // BEFORE: per-write create_dir_all(parent) on the already-existing shard. + let t = Instant::now(); + for _ in 0..iters { + let _ = tokio::fs::create_dir_all(&shard).await; + } + let before_ns = t.elapsed().as_nanos() as f64 / iters as f64; + + // AFTER: shard guaranteed present at init → the write path skips the call. + let t = Instant::now(); + for _ in 0..iters { + std::hint::black_box(&shard); + } + let after_ns = t.elapsed().as_nanos() as f64 / iters as f64; + + println!("## [D1] cache-write create_dir_all on a warm shard"); + println!("| arm | ns/write |"); + println!("| BEFORE | {before_ns:>8.1} |"); + println!("| AFTER | {after_ns:>8.1} |"); + println!( + "# {:.1}x — redundant create_dir_all removed per cache write\n", + before_ns / after_ns.max(0.001) + ); + gate("D1", "ns/write", before_ns, after_ns); +} + +#[tokio::main(flavor = "multi_thread", worker_threads = 4)] +async fn main() { + println!("# Round-26 disk-I/O pack\n"); + section_d1().await; + println!("All Round-26 disk-I/O sections passed their gate."); +} diff --git a/examples/bench_round26_hasher.rs b/examples/bench_round26_hasher.rs new file mode 100644 index 00000000..77629e7f --- /dev/null +++ b/examples/bench_round26_hasher.rs @@ -0,0 +1,148 @@ +//! Round-26 hasher pack (no Postgres) — wall-gated, since a hasher swap changes +//! 0 allocations (the deterministic alloc counter can't score it). +//! +//! [G1] The delta-upload "have/need" negotiation builds `HashSet`s over up to +//! `max_chunk_count()` client-supplied 64-hex BLAKE3 hashes per request +//! (`distinct_hashes`, `authorize_chunk_download`'s `distinct_seen`). +//! std `HashSet` uses SipHash-1-3 (DoS-resistant but ~2-4x slower on +//! short keys). AFTER uses `foldhash::quality::RandomState` — a faster +//! non-cryptographic hash that STAYS DoS-resistant because it is +//! per-instance random-seeded (the required property for these +//! attacker-controlled inputs — not `FxHash`/fixed-seed). foldhash is +//! already in the lockfile transitively (hashbrown), so it adds no crate. +//! Gate: AFTER wall (build set + membership scan) strictly lower, AND +//! two RandomState instances must seed differently (DoS resistance kept). +//! +//! Run: +//! RUSTFLAGS="-C target-cpu=x86-64-v3" \ +//! cargo run --release --features bench --example bench_round26_hasher +//! Tunables (env): G1_HASHES (40000), G1_PASSES (50) + +use std::collections::HashSet; +use std::env; +use std::hint::black_box; +use std::time::Instant; + +use foldhash::quality::RandomState; + +fn env_or(key: &str, default: T) -> T { + env::var(key) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(default) +} + +fn p50(mut s: Vec) -> f64 { + s.sort_by(|a, b| a.partial_cmp(b).unwrap()); + s[s.len() / 2] +} + +fn gate(tag: &str, metric: &str, before: f64, after: f64) { + if !(after < before) { + eprintln!("GATE FAIL [{tag}] {metric}: AFTER {after} !< BEFORE {before} — rollback"); + std::process::exit(1); + } +} + +/// Deterministic 64-hex "hash" strings (mirror a BLAKE3 chunk hash). +fn hashes(n: usize) -> Vec { + (0..n) + .map(|i| { + let mut s = String::with_capacity(64); + for k in 0..8 { + use std::fmt::Write; + let _ = write!( + s, + "{:08x}", + (i as u64).wrapping_mul(2_654_435_761).wrapping_add(k) + ); + } + s + }) + .collect() +} + +fn main() { + println!("# Round-26 hasher pack\n"); + let n: usize = env_or("G1_HASHES", 40_000); + let passes: usize = env_or("G1_PASSES", 50); + let keys = hashes(n); + + // DoS-safety: two RandomState instances must NOT hash identically (random + // per-instance seed — precomputed-collision attacks stay infeasible). + { + use std::hash::{BuildHasher, Hasher}; + let (a, b) = (RandomState::default(), RandomState::default()); + let mut ha = a.build_hasher(); + let mut hb = b.build_hasher(); + std::hash::Hash::hash(&keys[0], &mut ha); + std::hash::Hash::hash(&keys[0], &mut hb); + if ha.finish() == hb.finish() { + eprintln!("GATE FAIL [G1] two RandomState seeds produced the same hash — not DoS-safe"); + std::process::exit(1); + } + } + + // Equivalence: both build the same distinct set + same membership answers. + let sip: HashSet<&str> = keys.iter().map(|s| s.as_str()).collect(); + let fold: HashSet<&str, RandomState> = keys.iter().map(|s| s.as_str()).collect(); + assert_eq!(sip.len(), fold.len(), "G1 distinct count differs"); + for k in &keys { + assert_eq!( + sip.contains(k.as_str()), + fold.contains(k.as_str()), + "G1 membership differs" + ); + } + + let work_sip = || { + let set: HashSet<&str> = keys.iter().map(|s| s.as_str()).collect(); + let mut hits = 0usize; + for k in &keys { + if set.contains(k.as_str()) { + hits += 1; + } + } + black_box(hits) + }; + let work_fold = || { + let set: HashSet<&str, RandomState> = + HashSet::with_capacity_and_hasher(keys.len(), RandomState::default()); + let mut set = set; + for k in &keys { + set.insert(k.as_str()); + } + let mut hits = 0usize; + for k in &keys { + if set.contains(k.as_str()) { + hits += 1; + } + } + black_box(hits) + }; + + black_box(work_sip()); + black_box(work_fold()); + let mut before = Vec::new(); + let mut after = Vec::new(); + for _ in 0..passes { + let t = Instant::now(); + black_box(work_sip()); + before.push(t.elapsed().as_secs_f64() * 1e3); + let t = Instant::now(); + black_box(work_fold()); + after.push(t.elapsed().as_secs_f64() * 1e3); + } + let b = p50(before); + let a = p50(after); + println!("## [G1] delta-upload hash set: SipHash vs foldhash::quality ({n} hashes)"); + println!("| arm | p50 ms (build+scan) |"); + println!("| BEFORE (SipHash) | {b:>10.3} |"); + println!("| AFTER (foldhash) | {a:>10.3} |"); + println!( + "# {:.2}x wall — DoS resistance retained (random per-instance seed)\n", + b / a.max(1e-9) + ); + gate("G1", "p50 ms", b, a); + println!("Round-26 hasher section passed its gate."); +} diff --git a/examples/bench_round26_micro.rs b/examples/bench_round26_micro.rs new file mode 100644 index 00000000..d5b7a931 --- /dev/null +++ b/examples/bench_round26_micro.rs @@ -0,0 +1,153 @@ +//! Round-26 CPU/alloc micro-pack (no Postgres). +//! +//! Same rule as ROUND2–25: each section is BEFORE (verbatim replica of the +//! shipped-before shape) vs AFTER (replica of the shipped-after shape, which the +//! source is then made to match), with a value-equivalence gate and a +//! `GATE FAIL … rollback` `std::process::exit(1)` if the AFTER arm fails to beat +//! BEFORE — the round's roll-back rule encoded into the benchmark. +//! +//! [P1] `drive_pg_repository`'s four policy reads decode `d.policies` into a +//! throwaway `serde_json::Value` DOM and then call +//! `DrivePolicies::from_value(&raw)` (`Self::deserialize(&Value)`) — the +//! exact throwaway-DOM pattern ROUND23 §J1 removed for contacts, but left +//! on the drive-policy path (§J2 removed only the `from_value` clone). The +//! Value tree (a `Map` + boxed String key + `Value` node per policy field) +//! is walked once and dropped. AFTER decodes straight into the struct via +//! `serde_json::from_slice::` (what `sqlx::types::Json` +//! runs on the raw JSONB bytes) — no intermediate DOM. The lenient +//! `unwrap_or_default` fallback is preserved. +//! +//! Run: +//! RUSTFLAGS="-C target-cpu=x86-64-v3" \ +//! cargo run --release --features bench --example bench_round26_micro +//! Tunables (env): P1_ITERS (200000) + +use std::alloc::{GlobalAlloc, Layout, System}; +use std::env; +use std::hint::black_box; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Instant; + +use oxicloud::domain::entities::drive::DrivePolicies; +use serde::Deserialize as _; + +static ALLOC_CALLS: AtomicU64 = AtomicU64::new(0); +static ALLOC_BYTES: AtomicU64 = AtomicU64::new(0); + +struct CountingAlloc; + +unsafe impl GlobalAlloc for CountingAlloc { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + ALLOC_CALLS.fetch_add(1, Ordering::Relaxed); + ALLOC_BYTES.fetch_add(layout.size() as u64, Ordering::Relaxed); + unsafe { System.alloc(layout) } + } + unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { + unsafe { System.dealloc(ptr, layout) } + } + unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 { + ALLOC_CALLS.fetch_add(1, Ordering::Relaxed); + ALLOC_BYTES.fetch_add(new_size as u64, Ordering::Relaxed); + unsafe { System.realloc(ptr, layout, new_size) } + } + unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 { + ALLOC_CALLS.fetch_add(1, Ordering::Relaxed); + ALLOC_BYTES.fetch_add(layout.size() as u64, Ordering::Relaxed); + unsafe { System.alloc_zeroed(layout) } + } +} + +#[global_allocator] +static GLOBAL: CountingAlloc = CountingAlloc; + +fn env_or(key: &str, default: T) -> T { + env::var(key) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(default) +} + +#[derive(Clone, Copy)] +struct Measure { + ns: f64, + allocs: f64, + bytes: f64, +} + +fn measure(iters: u64, mut f: impl FnMut() -> T) -> Measure { + black_box(f()); + ALLOC_CALLS.store(0, Ordering::Relaxed); + ALLOC_BYTES.store(0, Ordering::Relaxed); + let start = Instant::now(); + for _ in 0..iters { + black_box(f()); + } + let ns = start.elapsed().as_nanos() as f64 / iters as f64; + Measure { + ns, + allocs: ALLOC_CALLS.load(Ordering::Relaxed) as f64 / iters as f64, + bytes: ALLOC_BYTES.load(Ordering::Relaxed) as f64 / iters as f64, + } +} + +fn report(tag: &str, before: Measure, after: Measure) { + println!("## {tag}"); + println!("| arm | ns/op | allocs/op | bytes/op |"); + println!( + "| BEFORE | {:>12.1} | {:>11.2} | {:>11.0} |", + before.ns, before.allocs, before.bytes + ); + println!( + "| AFTER | {:>12.1} | {:>11.2} | {:>11.0} |", + after.ns, after.allocs, after.bytes + ); + println!( + "# {:.2}x wall · {:.2} fewer allocs/op · {:.0} fewer bytes/op\n", + before.ns / after.ns.max(0.0001), + before.allocs - after.allocs, + before.bytes - after.bytes + ); +} + +fn gate(tag: &str, metric: &str, before: f64, after: f64) { + if !(after < before) { + eprintln!("GATE FAIL [{tag}] {metric}: AFTER {after} !< BEFORE {before} — rollback"); + std::process::exit(1); + } +} + +// ── [P1] drive-policy JSONB decode: Value DOM + from_value vs from_slice ─── +fn section_p1() { + let iters: u64 = env_or("P1_ITERS", 200_000); + // A realistically-populated policies bag (several fields set); the column is + // `jsonb NOT NULL DEFAULT '{}'`, and `#[serde(default)]` fills the rest. + let json: &[u8] = br#"{"forbid_sharing":true,"forbid_public_links":true,"include_in_photo_index":true,"read_only":false,"forbid_cross_drive_move":true}"#; + + // Equivalence: both arms yield the identical DrivePolicies. + let before_val: serde_json::Value = serde_json::from_slice(json).unwrap(); + let before = DrivePolicies::deserialize(&before_val).unwrap_or_default(); + let after = serde_json::from_slice::(json).unwrap_or_default(); + assert_eq!(before, after, "P1 decoded policies differ"); + + let b = measure(iters, || { + // BEFORE: raw JSONB → full serde_json::Value DOM → deserialize(&Value). + let v: serde_json::Value = serde_json::from_slice(black_box(json)).unwrap(); + DrivePolicies::deserialize(&v).unwrap_or_default() + }); + let a = measure(iters, || { + // AFTER: raw JSONB → from_slice:: (what sqlx Json does). + serde_json::from_slice::(black_box(json)).unwrap_or_default() + }); + report( + "[P1] drive-policy JSONB decode (Value DOM vs from_slice)", + b, + a, + ); + gate("P1", "allocs/op", b.allocs, a.allocs); +} + +fn main() { + println!("# Round-26 micro alloc pack\n"); + section_p1(); + println!("All Round-26 micro sections passed their gate."); +} diff --git a/src/application/services/delta_upload_service.rs b/src/application/services/delta_upload_service.rs index b7400613..1f42383d 100644 --- a/src/application/services/delta_upload_service.rs +++ b/src/application/services/delta_upload_service.rs @@ -553,7 +553,10 @@ impl DeltaUploadService { self.max_chunk_count() ))); } - let mut distinct_seen = HashSet::new(); + // foldhash::quality::RandomState — a fast, per-instance random-seeded + // hasher, DoS-safe for these attacker-controlled client hashes (up to + // max_chunk_count() of them per request) — benches/ROUND26.md §G1. + let mut distinct_seen: HashSet<&str, foldhash::quality::RandomState> = HashSet::default(); for hash in &request.hashes { if !is_valid_hash(hash) { return Err(DomainError::validation_error( @@ -684,7 +687,9 @@ fn sanitize_file_name(name: &str) -> Result { /// Distinct hashes in first-occurrence order. fn distinct_hashes(chunks: &[ChunkRef]) -> Vec { - let mut seen = HashSet::new(); + // foldhash::quality::RandomState — fast, per-instance random-seeded and thus + // DoS-safe for these attacker-controlled client hashes (benches/ROUND26.md §G1). + let mut seen: HashSet<&str, foldhash::quality::RandomState> = HashSet::default(); chunks .iter() .filter(|c| seen.insert(c.h.as_str())) diff --git a/src/infrastructure/repositories/pg/drive_pg_repository.rs b/src/infrastructure/repositories/pg/drive_pg_repository.rs index 0e5c57de..e7f51ede 100644 --- a/src/infrastructure/repositories/pg/drive_pg_repository.rs +++ b/src/infrastructure/repositories/pg/drive_pg_repository.rs @@ -20,6 +20,19 @@ use crate::domain::repositories::drive_repository::{ DriveRepository, DriveRepositoryError, DriveWithRootName, }; +/// Decode a `d.policies` JSONB column straight into `DrivePolicies` via +/// `sqlx::types::Json` — a single `serde_json::from_slice` over the raw JSONB +/// bytes — instead of fetching a throwaway `serde_json::Value` DOM and walking it +/// once with `DrivePolicies::from_value`. The §J1 pattern (ROUND23) applied to +/// the drive-policy path §J2 left behind (benches/ROUND26.md §P1). The lenient +/// `unwrap_or_default` fallback (a malformed bag decodes to all-false rather than +/// erroring the read) is preserved exactly. +fn policies_from_row(row: &sqlx::postgres::PgRow) -> crate::domain::entities::drive::DrivePolicies { + row.try_get::, _>("policies") + .map(|j| j.0) + .unwrap_or_default() +} + /// `default_drive_cache` TTL. The default-drive → root-folder binding is /// nearly immutable (changes only on provisioning / drive deletion / /// policy edits — all of which invalidate explicitly below), yet it is @@ -706,7 +719,7 @@ impl DriveRepository for DrivePgRepository { &self, file_id: Uuid, ) -> Result { - let row: Option<(serde_json::Value,)> = sqlx::query_as( + let row = sqlx::query( "SELECT d.policies \ FROM storage.drives d \ JOIN storage.files f ON f.drive_id = d.id \ @@ -715,20 +728,16 @@ impl DriveRepository for DrivePgRepository { .bind(file_id) .fetch_optional(self.pool.as_ref()) .await - .map_err(|e| Self::map_sqlx_err("get_policies_for_file", e))?; - let raw = row - .ok_or_else(|| DriveRepositoryError::NotFound(file_id.to_string()))? - .0; - Ok(crate::domain::entities::drive::DrivePolicies::from_value( - &raw, - )) + .map_err(|e| Self::map_sqlx_err("get_policies_for_file", e))? + .ok_or_else(|| DriveRepositoryError::NotFound(file_id.to_string()))?; + Ok(policies_from_row(&row)) } async fn get_policies_for_folder( &self, folder_id: Uuid, ) -> Result { - let row: Option<(serde_json::Value,)> = sqlx::query_as( + let row = sqlx::query( "SELECT d.policies \ FROM storage.drives d \ JOIN storage.folders fo ON fo.drive_id = d.id \ @@ -737,20 +746,16 @@ impl DriveRepository for DrivePgRepository { .bind(folder_id) .fetch_optional(self.pool.as_ref()) .await - .map_err(|e| Self::map_sqlx_err("get_policies_for_folder", e))?; - let raw = row - .ok_or_else(|| DriveRepositoryError::NotFound(folder_id.to_string()))? - .0; - Ok(crate::domain::entities::drive::DrivePolicies::from_value( - &raw, - )) + .map_err(|e| Self::map_sqlx_err("get_policies_for_folder", e))? + .ok_or_else(|| DriveRepositoryError::NotFound(folder_id.to_string()))?; + Ok(policies_from_row(&row)) } async fn get_drive_id_and_policies_for_file( &self, file_id: Uuid, ) -> Result<(Uuid, crate::domain::entities::drive::DrivePolicies), DriveRepositoryError> { - let row: Option<(Uuid, serde_json::Value)> = sqlx::query_as( + let row = sqlx::query( "SELECT d.id, d.policies \ FROM storage.drives d \ JOIN storage.files f ON f.drive_id = d.id \ @@ -759,20 +764,19 @@ impl DriveRepository for DrivePgRepository { .bind(file_id) .fetch_optional(self.pool.as_ref()) .await - .map_err(|e| Self::map_sqlx_err("get_drive_id_and_policies_for_file", e))?; - let (drive_id, raw) = - row.ok_or_else(|| DriveRepositoryError::NotFound(file_id.to_string()))?; - Ok(( - drive_id, - crate::domain::entities::drive::DrivePolicies::from_value(&raw), - )) + .map_err(|e| Self::map_sqlx_err("get_drive_id_and_policies_for_file", e))? + .ok_or_else(|| DriveRepositoryError::NotFound(file_id.to_string()))?; + let drive_id: Uuid = row + .try_get("id") + .map_err(|e| Self::map_sqlx_err("get_drive_id_and_policies_for_file", e))?; + Ok((drive_id, policies_from_row(&row))) } async fn get_drive_id_and_policies_for_folder( &self, folder_id: Uuid, ) -> Result<(Uuid, crate::domain::entities::drive::DrivePolicies), DriveRepositoryError> { - let row: Option<(Uuid, serde_json::Value)> = sqlx::query_as( + let row = sqlx::query( "SELECT d.id, d.policies \ FROM storage.drives d \ JOIN storage.folders fo ON fo.drive_id = d.id \ @@ -781,13 +785,12 @@ impl DriveRepository for DrivePgRepository { .bind(folder_id) .fetch_optional(self.pool.as_ref()) .await - .map_err(|e| Self::map_sqlx_err("get_drive_id_and_policies_for_folder", e))?; - let (drive_id, raw) = - row.ok_or_else(|| DriveRepositoryError::NotFound(folder_id.to_string()))?; - Ok(( - drive_id, - crate::domain::entities::drive::DrivePolicies::from_value(&raw), - )) + .map_err(|e| Self::map_sqlx_err("get_drive_id_and_policies_for_folder", e))? + .ok_or_else(|| DriveRepositoryError::NotFound(folder_id.to_string()))?; + let drive_id: Uuid = row + .try_get("id") + .map_err(|e| Self::map_sqlx_err("get_drive_id_and_policies_for_folder", e))?; + Ok((drive_id, policies_from_row(&row))) } async fn drive_id_for_folder(&self, folder_id: Uuid) -> Result { diff --git a/src/infrastructure/services/cached_blob_backend.rs b/src/infrastructure/services/cached_blob_backend.rs index 36bec900..a38220be 100644 --- a/src/infrastructure/services/cached_blob_backend.rs +++ b/src/infrastructure/services/cached_blob_backend.rs @@ -119,10 +119,21 @@ impl BlobStorageBackend for CachedBlobBackend { Box::pin(async move { inner.initialize().await?; - // Create cache dir structure (256 prefix dirs) + // Create the cache dir AND its 256 {00..ff} shard dirs up front + // (mirroring LocalBlobBackend::initialize), so the write paths never + // pay a per-chunk `create_dir_all` on an already-existing shard — a + // ~45 µs mkdirat(EEXIST)+stat+blocking-dispatch removed per cache + // write on cached-remote deployments (benches/ROUND26.md §D1). fs::create_dir_all(&cache_dir).await.map_err(|e| { DomainError::internal_error("BlobCache", format!("mkdir cache_dir: {e}")) })?; + for prefix in &crate::infrastructure::services::local_blob_backend::HEX_PREFIXES { + fs::create_dir_all(cache_dir.join(prefix)) + .await + .map_err(|e| { + DomainError::internal_error("BlobCache", format!("mkdir cache shard: {e}")) + })?; + } // Scan existing cache to rebuild index. Collect entries WITHOUT // holding the index lock — a large cache directory walk must not @@ -411,10 +422,9 @@ impl CachedBlobBackend { /// index deliberately skipped the eviction sweep on this path, letting /// write bursts overshoot the budget until the next read-miss insert). async fn cache_bytes_write_through(&self, hash: String, data: &Bytes) { + // The shard dir was created at initialize() — no per-write create_dir_all + // (benches/ROUND26.md §D1). let dest = self.cached_path(&hash); - if let Some(parent) = dest.parent() { - let _ = fs::create_dir_all(parent).await; - } let _ = fs::write(&dest, data).await; let data_len = data.len() as u64; self.index.insert(hash, CacheEntry { size: data_len }); @@ -451,12 +461,8 @@ impl CachedBlobBackend { } async fn insert_into_cache(&self, hash: &str, source_path: &Path) -> Result<(), DomainError> { + // Shard dir pre-created at initialize() (benches/ROUND26.md §D1). let dest = self.cached_path(hash); - if let Some(parent) = dest.parent() { - fs::create_dir_all(parent).await.map_err(|e| { - DomainError::internal_error("BlobCache", format!("mkdir failed: {e}")) - })?; - } let size = fs::metadata(source_path) .await @@ -476,12 +482,8 @@ impl CachedBlobBackend { async fn fetch_and_cache(&self, hash: &str) -> Result { let stream = self.inner.get_blob_stream(hash).await?; + // Shard dir pre-created at initialize() (benches/ROUND26.md §D1). let dest = self.cached_path(hash); - if let Some(parent) = dest.parent() { - fs::create_dir_all(parent).await.map_err(|e| { - DomainError::internal_error("BlobCache", format!("mkdir failed: {e}")) - })?; - } // Unique temp name: even if two fetches for one hash ever race // (e.g. across processes sharing a cache dir), each writes its own diff --git a/src/infrastructure/services/local_blob_backend.rs b/src/infrastructure/services/local_blob_backend.rs index 40f932be..556c5b19 100644 --- a/src/infrastructure/services/local_blob_backend.rs +++ b/src/infrastructure/services/local_blob_backend.rs @@ -164,7 +164,7 @@ pub async fn write_blob_bytes_for_bench( } /// Compile-time lookup table for the 256 two-digit lowercase hex prefixes ("00"…"ff"). -static HEX_PREFIXES: [&str; 256] = [ +pub(crate) static HEX_PREFIXES: [&str; 256] = [ "00", "01", "02", "03", "04", "05", "06", "07", "08", "09", "0a", "0b", "0c", "0d", "0e", "0f", "10", "11", "12", "13", "14", "15", "16", "17", "18", "19", "1a", "1b", "1c", "1d", "1e", "1f", "20", "21", "22", "23", "24", "25", "26", "27", "28", "29", "2a", "2b", "2c", "2d", "2e", "2f",