diff --git a/Cargo.toml b/Cargo.toml index a678a0c0..71ee207f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -354,6 +354,27 @@ name = "bench_micro_allocs" path = "examples/bench_micro_allocs.rs" required-features = ["bench"] +# Round-23 battery ──────────────────────────────────────────────────────────── + +# Round-23 CPU/alloc micro-pack (no Postgres) — deterministic alloc gates for +# the decode/clone candidates: contact JSONB `Value`+`from_value` → typed +# `sqlx::types::Json` decode (J1); `DrivePolicies::from_value` DOM clone → +# borrow-deserialize (J2); dedup hash reshape clone-collect → `into_iter().unzip()` +# (U1). The PG latency/round-trip/equivalence evidence is bench_round23_queries. +[[example]] +name = "bench_round23_micro" +path = "examples/bench_round23_micro.rs" +required-features = ["bench"] + +# Round-23 PG query-shape pack — end-to-end latency + equivalence on the live +# dev Postgres: contact JSONB typed decode on real rows (Q1); get_user_profile +# 2 serial reads → tokio::join! (Q4); subject_group remove_member 2 recursive +# CTEs → 1 reused (Q6). Needs the dev Postgres up (reads DATABASE_URL from .env). +[[example]] +name = "bench_round23_queries" +path = "examples/bench_round23_queries.rs" +required-features = ["bench"] + # Round-22 battery ──────────────────────────────────────────────────────────── # Round-22 CPU/alloc micro-pack — the deferred hot-GET-handler `HeaderMap` diff --git a/benches/ROUND23.md b/benches/ROUND23.md new file mode 100644 index 00000000..d3a18f0c --- /dev/null +++ b/benches/ROUND23.md @@ -0,0 +1,193 @@ +# Round 23 — Postgres query-shape pass: typed JSONB decode, drive-policy borrow-deserialize, user-profile join!, subject-group CTE reuse, dedup unzip + +Benchmark-gated, same rule as ROUND2–22: every change ships with a BEFORE/AFTER +benchmark and a value equivalence gate; an AFTER that doesn't beat its BEFORE is +rolled back (never applied). This round is the **PostgreSQL** pass — the +candidates the earlier rounds deferred as "needs a database to bench" — so it +ships two harnesses: + +- **`bench_round23_micro`** (no Postgres) — the deterministic **allocation gate** + for the decode/clone candidates (counting global allocator; a non-winning + AFTER `std::process::exit(1)`s with `GATE FAIL … rollback`). +- **`bench_round23_queries`** (live Postgres) — end-to-end **p50 latency** + a + strict **equivalence gate** (identical decoded rows / ids / user-sets from + BEFORE and AFTER; a mismatch exits 1) against seeded fixtures. + +Reproduce (the queries harness reads `DATABASE_URL` from `.env`): + +``` +cargo run --release --features bench --example bench_round23_micro +cargo run --release --features bench --example bench_round23_queries +``` + +## Summary + +| # | change | metric | before → after | +|--:|---|---|---| +| **J1** | `contact_pg_repository::row_to_contact` (+ the inlined `contact_group_pg_repository` sibling) decoded each of the 3 JSONB columns (`email`/`phone`/`address`) with `row.get::` + `serde_json::from_value::>` — a throwaway `Value` DOM built per column and then walked a **second** time to produce the typed `Vec`. Now `row.try_get::>>` decodes the JSONB bytes straight into the typed Vec in one `from_slice` pass, no DOM. Runs **per contact row** of every contact list / multiget / CardDAV sync. | micro allocs · PG p50 | **84 → 33 allocs/op** (2.15× wall) · **3794 → 2360 ns/contact** (1.61×) | +| **J2** | `DrivePolicies::from_value` did `serde_json::from_value(value.clone())` — cloning the **entire** policies `Value` DOM before walking it, on every drive-policy read (move/copy, shared-link creation, grant). Now `DrivePolicies::deserialize(value)` deserializes straight from the borrow (serde_json's `Deserializer for &Value`), no clone — a one-line body change, byte-identical, all 7 call sites unchanged. | micro allocs | **5 → 0 allocs/op** (11.51× wall) | +| **P1** | `AuthApplicationService::get_user_profile` issued two **independent, serial** `get_user_by_id` point reads (caller then target; the self-case short-circuit compares input UUIDs, not fetched data). Now the self-case does a single fetch and the non-self path overlaps caller+target with `tokio::join!` (`caller_res?` first preserves the caller-error precedence). | PG p50 | **577 → 312 µs/call** (1.85×) | +| **G1** | `SubjectGroupService::remove_member` ran the child group's transitive-user recursive CTE **twice** for a nested `Group` removal — once in the would-empty pre-check, once in `invalidation_targets` after the remove. The edge delete is *above* the child, so its descendants can't change; now the CTE runs **once** and the result is reused for both. | PG p50 | **829 → 412 µs/removal** (2.01×) | +| **U1** | `dedup_service` (`store_loose_chunks` final registration + the ingest `run_rollback`) built `Vec`/`Vec` by **cloning** every 64-byte hash out of an owned, dead-after `Vec<(String,i64)>` purely to reshape for `sync_blobs(&[String])` + the `UNNEST` bind. Now `into_iter().unzip()` moves the hashes out — no per-hash content copy. | micro allocs | **256 → 0 hash clones** (1283 → 1027 allocs/op on a 256-chunk batch) | + +> The micro allocs/op is the deterministic gate (identical run to run); the PG +> p50 is single-machine, warm-pool, and noise-bounded. Every section carries a +> value-equivalence gate; the shipped source matches each AFTER arm. + +## [J1] Contact JSONB — typed `Json` decode, no intermediate `Value` DOM + +`row_to_contact` (reached by 11 call sites — every contact GET / list / +paginated list / multiget / CardDAV cursor stream / search / by-email / +by-group / create+update RETURNING) and the identical inlined block in +`contact_group_pg_repository::get_contacts_in_group` both did: + +```rust +let email_json: JsonValue = row.get("email"); // sqlx JSONB → Value DOM (alloc tree) +let emails = serde_json::from_value::>(email_json) // walk the DOM again + .map(emails_from_persistence).unwrap_or_default(); +// … same for phone, address +``` + +`sqlx::types::Json` decodes the raw JSONB bytes with a single +`serde_json::from_slice::` (sqlx-core 0.8.6 `types/json.rs`), skipping the +`Value` tree entirely: + +```rust +let emails = row + .try_get::>, _>("email") + .map(|j| emails_from_persistence(j.0)) + .unwrap_or_default(); +``` + +`try_get` (not `get`) preserves the exact malformed-shape fallback — `get` +would panic on a decode error, whereas the old `from_value(...).unwrap_or_default()` +tolerated it. The columns are `JSONB NOT NULL DEFAULT '[]'`, so SQL NULL never +occurs. Byte-identical: both paths run the same derived `Deserialize>` +over the same bytes — the `bench_round23_queries` §Q1 gate asserts the two +decode the 500 seeded contacts field-for-field identically. The micro shows the +3 discarded DOMs/row (84 → 33 allocs); on the real rows the decode is 1.61×. + +## [J2] Drive policies — deserialize from the borrow, don't clone the DOM + +`DrivePolicies::from_value(value: &serde_json::Value)` is called on every +drive-policy read (`get_policies_for_file/_folder`, +`get_drive_id_and_policies_for_*`, `update_policies` RETURNING, the ACL engine's +enforcement read, and `Drive::typed_policies`). It built the typed struct with +`serde_json::from_value(value.clone())` — a full clone of the policies DOM +purely because `from_value` consumes its argument. serde_json implements +`Deserializer` for `&Value`, so the struct can be built straight from the +borrow: + +```rust +use serde::Deserialize as _; +Self::deserialize(value).unwrap_or_default() // was: serde_json::from_value(value.clone()) +``` + +Byte-identical (same derived `Deserialize`, same lenient `unwrap_or_default` +fallback that keeps unknown keys on disk), a one-line body change, and every +caller keeps its `&Value` argument unchanged — so `typed_policies(&self)` +(which only has a borrow of `self.policies`) also stops cloning. The micro +(a realistic bag with a preserved unknown key) drops 5 → 0 allocs/op. + +## [P1] `get_user_profile` — overlap the two independent reads with `join!` + +The profile lookup fetched the caller and the target user in two serial +round-trips. The self-case (`caller_id == target_id`) is decided by comparing +the **input** UUIDs, so on the common non-self path the two reads are +independent — query 2 never depends on query 1. AFTER: + +```rust +if caller_id == target_id { // self: one fetch, unchanged + let caller = self.user_storage.get_user_by_id(caller_id).await?; + return Ok(UserDto::from(caller)); +} +let (caller_res, target_res) = tokio::join!( // non-self: overlap + self.user_storage.get_user_by_id(caller_id), + self.user_storage.get_user_by_id(target_id)); +let caller = caller_res?; // caller-error precedence preserved +let target = match target_res { … }; // identical NotFound→anonymized-404 + audit +``` + +Every observable outcome is preserved (self still 1 fetch, the anti-enumeration +audit unchanged). The §Q4 gate asserts identical ids from both shapes; two +warm-pool serial reads vs the `join!` measured **1.85×**. + +## [G1] `remove_member` — compute the child's transitive users once, reuse it + +For a nested `GroupMember::Group(child_id)` removal the child's transitive-user +set (a recursive `WITH RECURSIVE` CTE over `subject_group_members`) was computed +**twice**: once in the would-empty self-defense pre-check, and again inside +`invalidation_targets` after `remove_member` deleted the parent→child edge. That +edge is *above* the child, so the child's own descendants are unchanged — +verified empirically on the live DB (child set `{u2,u3}` identical before and +after the edge delete). AFTER computes the CTE once, up front, and reuses it for +both the pre-check and the cache-invalidation set (`invalidation_targets` stays +for `add_member`). The §Q6 gate asserts the child set is both stable and the +expected `{u2,u3}`; 2 CTEs vs 1 measured **2.01×** on the seeded 3-level tree. + +## [U1] dedup hash reshape — move via `unzip`, don't clone + +`store_loose_chunks`'s final registration and the ingest `run_rollback` both +reshaped an owned `Vec<(String,i64)>` (dead after the block) into the +`Vec` + `Vec` that `sync_blobs(&[String])` and the `UNNEST` bind +need, by cloning every 64-char hash: + +```rust +let hashes: Vec = new_rows.iter().map(|(h, _)| h.clone()).collect(); // N clones +let sizes: Vec = new_rows.iter().map(|(_, s)| *s).collect(); +``` + +Since the source is owned and never read again, `into_iter().unzip()` moves the +hashes out — 0 per-hash content copies: + +```rust +let (hashes, sizes): (Vec, Vec) = new_rows.into_iter().unzip(); +``` + +Byte-identical rows inserted; the micro (256 distinct new chunks) drops exactly +the 256 hash clones. (This is the move-not-borrow refinement of the ROUND21 §R2 +`&[&str]` pattern — `sync_blobs` takes `&[String]`, so a borrow would force a +port-signature change across 6 backends, whereas the move needs none.) + +## Not shipped — deferred to a dedicated pass + +- **`batch_operations::download_zip` per-item N+1** (the audit's #2, highest + raw-latency candidate): the file loop calls `get_file_with_perms` (itself + authz + `get_file` = 2 round-trips) per selected file, then + `add_file_entry_streamed` — which **re-authorizes** internally via + `get_file_stream_with_perms`. Collapsing the per-item metadata+authz into a + bulk `get_files_by_ids` + `check_files_read_batch` prefetch is a real win + (`2N+2M` serial round-trips → ~3 batch queries), **but** it moves the sole + authorization from before the stream to inside it, so it needs a careful + AuthZ-ordering + anti-enumeration proof (the project's rule: authz lives in + the service layer, denials audit-log and return the anti-enum shape). That is + its own validated pass, not a perf banner — queued with a `download_zip` + fixture that seeds a large multi-select and asserts identical ZIP entry + set+order across the change. +- **Contact/Drive JSONB — the SQL-NULL edge**: the typed `try_get`/`deserialize` + paths return the empty/default on SQL NULL where the old `row.get::` + would have panicked. Both columns are `NOT NULL DEFAULT` today so this never + fires; noted only so a future nullable-column change re-checks it. + +## Environment / methodology + +- A local **PostgreSQL 16** dev instance was provisioned for this round + (schema applied via the 67 `migrations/*.sql` in order; `pg_trgm` + `ltree` + extensions). `bench_round23_queries` seeds its own fixtures (unique + `bench23-*` markers) and tears them down (idempotent cleanup) around the run. +- **Build note:** this session's host intermittently `SIGILL`ed rustc/LLVM + codegen under the repo's default `-C target-cpu=native` (a `cascadelake` with + AVX-512 whose passthrough faulted after a host migration). All Round-23 + builds/benches were run with `RUSTFLAGS="-C target-cpu=x86-64-v3"` (AVX2, no + AVX-512) to sidestep it. This is a local build-flag override only — the + checked-in `.cargo/config.toml` is unchanged, and the primary gate (allocs/op) + is target-cpu-independent; the PG p50 comparisons use the same flag for both + arms, so the relative speedups hold. +- Each micro section: BEFORE (verbatim shipped-before shape) vs AFTER (verbatim + shipped-after shape) + a value-equivalence assert + a `GATE FAIL … rollback` + exit if the AFTER fails to reduce allocations. Each PG section: BEFORE vs + AFTER shape against real seeded rows + an equivalence gate (mismatch → exit 1) + + p50 over `BENCH_PASSES`. +- Verified beyond the benches: `cargo fmt --all --check` clean, `cargo clippy + --features bench --all-targets -D warnings` clean, and the touched modules' + unit tests pass (`contact`, `drive`, `subject_group`, `dedup`, auth profile). diff --git a/examples/bench_round23_micro.rs b/examples/bench_round23_micro.rs new file mode 100644 index 00000000..d018ab1c --- /dev/null +++ b/examples/bench_round23_micro.rs @@ -0,0 +1,363 @@ +//! Round-23 CPU/alloc micro-pack (no Postgres) — the deterministic alloc gates +//! for the decode / clone candidates. The end-to-end PostgreSQL latency + +//! equivalence evidence lives in `bench_round23_queries.rs`. +//! +//! Same rule as ROUND2–22: each section is BEFORE (verbatim replica of the +//! shipped-before shape) vs AFTER (verbatim replica of the shipped-after shape, +//! which the source is then made to match), with a byte/-value equivalence gate +//! and a `GATE FAIL … rollback` check that `std::process::exit(1)`s if the AFTER +//! arm fails to beat its BEFORE. +//! +//! [J1] `contact_pg_repository::row_to_contact` (+ the `contact_group` +//! sibling) decoded each JSONB column with `row.get::` +//! + `serde_json::from_value::>` — a throwaway `Value` DOM per +//! column, walked a second time. AFTER decodes straight into the typed +//! Vec via `sqlx::types::Json` (one `from_slice` pass). Modeled here +//! as `from_slice::` + `from_value` vs `from_slice::>`. +//! +//! [J2] `DrivePolicies::from_value` did `serde_json::from_value(value.clone())` +//! — cloning the ENTIRE policies DOM per drive-policy read. AFTER +//! deserializes from the borrow (`T::deserialize(&Value)`), no clone. +//! +//! [U1] `dedup_service` (`store_loose_chunks` final registration + the ingest +//! `run_rollback`) built `Vec`/`Vec` by CLONING every hash +//! out of an owned, dead-after `Vec<(String,i64)>` purely to reshape for +//! `sync_blobs(&[String])` + the UNNEST bind. AFTER moves via +//! `into_iter().unzip()`. +//! +//! Run: +//! cargo run --release --features bench --example bench_round23_micro +//! Tunables (env): BENCH_ITERS (200000), J1_ROWS (3), U1_CHUNKS (256) + +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 serde::{Deserialize, Serialize}; +use serde_json::Value; + +static ALLOC_CALLS: 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); + 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); + unsafe { System.realloc(ptr, layout, new_size) } + } + unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 { + ALLOC_CALLS.fetch_add(1, 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) +} + +struct Measured { + wall_ns_per_op: f64, + allocs_per_op: f64, +} + +fn measure(iters: usize, mut f: F) -> Measured { + for _ in 0..(iters / 20).max(1) { + f(); + } + let a0 = ALLOC_CALLS.load(Ordering::Relaxed); + let t = Instant::now(); + for _ in 0..iters { + f(); + } + let wall = t.elapsed().as_nanos() as f64 / iters as f64; + let allocs = (ALLOC_CALLS.load(Ordering::Relaxed) - a0) as f64 / iters as f64; + Measured { + wall_ns_per_op: wall, + allocs_per_op: allocs, + } +} + +fn print_row(label: &str, m: &Measured) { + println!( + "| {:<52} | {:>12.1} | {:>10.2} |", + label, m.wall_ns_per_op, m.allocs_per_op + ); +} + +fn header_footer(name: &str, before: &Measured, after: &Measured) { + println!("| arm | ns/op | allocs/op |"); + print_row(&format!("BEFORE {name}"), before); + print_row(&format!("AFTER {name}"), after); + println!( + "# {:.2}x wall, {:.2} fewer allocs/op", + before.wall_ns_per_op / after.wall_ns_per_op, + before.allocs_per_op - after.allocs_per_op + ); +} + +fn gate_allocs(tag: &str, before: &Measured, after: &Measured) { + if after.allocs_per_op >= before.allocs_per_op { + eprintln!("GATE FAIL [{tag}]: AFTER did not reduce allocations — rollback"); + std::process::exit(1); + } +} + +// ──────────────────────────────────────────────────────────────────────────── +// [J1] Contact JSONB decode — Value DOM + from_value vs Json from_slice. +// Verbatim replicas of the persistence DTOs (contact_persistence_dto.rs). +// ──────────────────────────────────────────────────────────────────────────── + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +struct EmailDto { + email: String, + r#type: String, + is_primary: bool, +} +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +struct PhoneDto { + number: String, + r#type: String, + is_primary: bool, +} +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +struct AddressDto { + street: Option, + city: Option, + state: Option, + postal_code: Option, + country: Option, + r#type: String, + is_primary: bool, +} + +/// BEFORE: `row.get::` (sqlx JSONB→Value DOM) then `from_value::>` +/// (a second walk of the DOM). Modeled with `from_slice::` (what sqlx's +/// Value decoder does) + `from_value`. +fn j1_before( + email: &[u8], + phone: &[u8], + addr: &[u8], +) -> (Vec, Vec, Vec) { + let ev: Value = serde_json::from_slice(email).unwrap(); + let pv: Value = serde_json::from_slice(phone).unwrap(); + let av: Value = serde_json::from_slice(addr).unwrap(); + let emails = serde_json::from_value::>(ev).unwrap_or_default(); + let phones = serde_json::from_value::>(pv).unwrap_or_default(); + let addrs = serde_json::from_value::>(av).unwrap_or_default(); + (emails, phones, addrs) +} + +/// AFTER: `sqlx::types::Json>` decodes the JSONB bytes straight into the +/// typed Vec (one `from_slice::>`), no intermediate DOM. +fn j1_after( + email: &[u8], + phone: &[u8], + addr: &[u8], +) -> (Vec, Vec, Vec) { + let emails = serde_json::from_slice::>(email).unwrap_or_default(); + let phones = serde_json::from_slice::>(phone).unwrap_or_default(); + let addrs = serde_json::from_slice::>(addr).unwrap_or_default(); + (emails, phones, addrs) +} + +fn section_j1() { + let iters: usize = env_or("BENCH_ITERS", 200_000); + let n: usize = env_or("J1_ROWS", 3); // entries per column, realistic contact + + let mk_emails = |n: usize| -> Vec { + (0..n) + .map(|i| EmailDto { + email: format!("user{i}@example.com"), + r#type: if i == 0 { "home" } else { "work" }.to_string(), + is_primary: i == 0, + }) + .collect() + }; + let mk_phones = |n: usize| -> Vec { + (0..n) + .map(|i| PhoneDto { + number: format!("+1-555-010{i}"), + r#type: "cell".to_string(), + is_primary: i == 0, + }) + .collect() + }; + let mk_addrs = |n: usize| -> Vec { + (0..n) + .map(|i| AddressDto { + street: Some(format!("{} Main St", 100 + i)), + city: Some("Springfield".to_string()), + state: Some("IL".to_string()), + postal_code: Some("62704".to_string()), + country: Some("US".to_string()), + r#type: "home".to_string(), + is_primary: i == 0, + }) + .collect() + }; + + let email_b = serde_json::to_vec(&mk_emails(n)).unwrap(); + let phone_b = serde_json::to_vec(&mk_phones(n)).unwrap(); + let addr_b = serde_json::to_vec(&mk_addrs(n)).unwrap(); + + // Equivalence: identical decoded Vecs. + assert_eq!( + j1_before(&email_b, &phone_b, &addr_b), + j1_after(&email_b, &phone_b, &addr_b), + "J1 decoded contacts differ" + ); + + let before = measure(iters, || { + black_box(j1_before( + black_box(&email_b), + black_box(&phone_b), + black_box(&addr_b), + )); + }); + let after = measure(iters, || { + black_box(j1_after( + black_box(&email_b), + black_box(&phone_b), + black_box(&addr_b), + )); + }); + + println!( + "\n## [J1] Contact JSONB decode ({n} entries/col — per contact row of every list/multiget/sync)" + ); + header_footer( + "Value DOM + from_value vs Json from_slice", + &before, + &after, + ); + gate_allocs("J1", &before, &after); +} + +// ──────────────────────────────────────────────────────────────────────────── +// [J2] Drive policies decode — from_value(value.clone()) vs deserialize(&value). +// ──────────────────────────────────────────────────────────────────────────── + +#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)] +#[serde(default)] +struct Policies { + forbid_public_links: bool, + read_only: bool, +} + +/// BEFORE: clone the whole `Value` DOM, then `from_value`. +fn j2_before(value: &Value) -> Policies { + serde_json::from_value(value.clone()).unwrap_or_default() +} + +/// AFTER: deserialize straight from the borrow — no DOM clone. +fn j2_after(value: &Value) -> Policies { + Policies::deserialize(value).unwrap_or_default() +} + +fn section_j2() { + let iters: usize = env_or("BENCH_ITERS", 200_000); + // A realistic on-disk policies bag with an unknown key preserved on disk + // (the lenient contract) so the DOM isn't trivially tiny. + let value: Value = serde_json::from_str( + r#"{"forbid_public_links":true,"read_only":false,"x_future_flag":"kept-on-disk"}"#, + ) + .unwrap(); + + assert_eq!( + j2_before(&value), + j2_after(&value), + "J2 decoded policies differ" + ); + assert!(j2_after(&value).forbid_public_links); + + let before = measure(iters, || { + black_box(j2_before(black_box(&value))); + }); + let after = measure(iters, || { + black_box(j2_after(black_box(&value))); + }); + + println!("\n## [J2] Drive policies decode (per move/copy/share/grant drive-policy read)"); + header_footer( + "from_value(value.clone()) vs deserialize(&value)", + &before, + &after, + ); + gate_allocs("J2", &before, &after); +} + +// ──────────────────────────────────────────────────────────────────────────── +// [U1] dedup hash reshape — clone-collect vs into_iter().unzip(). +// ──────────────────────────────────────────────────────────────────────────── + +fn u1_build(n: usize) -> Vec<(String, i64)> { + (0..n) + .map(|i| { + ( + format!("{:064x}", i as u128 * 0x9E37_79B9_7F4A_7C15), + i as i64, + ) + }) + .collect() +} + +/// BEFORE: clone every hash out of the owned (dead-after) Vec to reshape. +fn u1_before(rows: Vec<(String, i64)>) -> (Vec, Vec) { + let hashes: Vec = rows.iter().map(|(h, _)| h.clone()).collect(); + let sizes: Vec = rows.iter().map(|(_, s)| *s).collect(); + (hashes, sizes) +} + +/// AFTER: move via unzip — no per-hash content copy. +fn u1_after(rows: Vec<(String, i64)>) -> (Vec, Vec) { + rows.into_iter().unzip() +} + +fn section_u1() { + let n: usize = env_or("U1_CHUNKS", 256); + let iters: usize = env_or("BENCH_ITERS", 200_000) / 20; // heavier op + + // Equivalence: identical hashes + sizes. + assert_eq!( + u1_before(u1_build(n)), + u1_after(u1_build(n)), + "U1 reshape differs" + ); + + let before = measure(iters, || { + black_box(u1_before(black_box(u1_build(n)))); + }); + let after = measure(iters, || { + black_box(u1_after(black_box(u1_build(n)))); + }); + + println!( + "\n## [U1] dedup hash reshape ({n} distinct new chunks — per delta-upload registration)" + ); + header_footer("clone-collect vs into_iter().unzip()", &before, &after); + gate_allocs("U1", &before, &after); +} + +fn main() { + println!("# Round-23 micro-pack — BEFORE/AFTER (counting allocator, release)"); + println!("# allocs/op is the deterministic gate; a non-winning AFTER exits 1 (rollback)."); + section_j1(); + section_j2(); + section_u1(); + println!("\nAll Round-23 micro sections passed their allocation gate."); +} diff --git a/examples/bench_round23_queries.rs b/examples/bench_round23_queries.rs new file mode 100644 index 00000000..2ac2f590 --- /dev/null +++ b/examples/bench_round23_queries.rs @@ -0,0 +1,433 @@ +//! Round-23 PostgreSQL query-shape pack — end-to-end latency + equivalence on +//! the live dev Postgres. The deterministic alloc gates for the decode/clone +//! candidates live in `bench_round23_micro.rs`; this harness measures the real +//! round-trip / decode wins against seeded fixtures and asserts identical +//! results (the equivalence gate — a mismatch `std::process::exit(1)`s). +//! +//! [Q1] Contact JSONB decode on REAL rows (contact_pg §J1): fetch a seeded +//! address book's contacts once, then decode the `email`/`phone`/`address` +//! JSONB columns BEFORE (`row.get::` + `from_value`) vs AFTER +//! (`row.try_get::>>`). Gate: identical decode. +//! +//! [Q4] `get_user_profile` (§P1): two independent point reads of the caller + +//! target users, BEFORE serial (`await` then `await`) vs AFTER concurrent +//! (`tokio::join!`). Gate: identical rows. +//! +//! [Q6] `subject_group::remove_member` (§G1): the child group's transitive +//! user set (a recursive CTE) BEFORE computed TWICE (the shipped-before +//! pre-check + `invalidation_targets`) vs AFTER once + reused. Gate: +//! identical user set. +//! +//! Run (needs the dev Postgres up; reads DATABASE_URL from .env): +//! cargo run --release --features bench --example bench_round23_queries +//! Tunables (env): BENCH_PASSES (200), Q1_CONTACTS (500), Q1_DECODE_PASSES (4000) + +use std::env; +use std::time::Instant; + +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sqlx::postgres::PgPoolOptions; +use sqlx::{PgPool, Row}; +use uuid::Uuid; + +fn env_or(key: &str, default: T) -> T { + env::var(key) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(default) +} + +fn p50(mut samples: Vec) -> f64 { + samples.sort_by(|a, b| a.partial_cmp(b).unwrap()); + samples[samples.len() / 2] +} + +fn report(tag: &str, unit: &str, before: f64, after: f64) { + println!( + "| {:<44} | {:>12} | {:>12} | {:>7} |", + tag, "BEFORE", "AFTER", "speedup" + ); + println!( + "| {:<44} | {:>12.1} | {:>12.1} | {:>6.2}x |", + unit, + before, + after, + before / after + ); +} + +// ── Verbatim replicas of contact_persistence_dto.rs ────────────────────────── +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +struct EmailDto { + email: String, + r#type: String, + is_primary: bool, +} +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +struct PhoneDto { + number: String, + r#type: String, + is_primary: bool, +} +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +struct AddressDto { + street: Option, + city: Option, + state: Option, + postal_code: Option, + country: Option, + r#type: String, + is_primary: bool, +} + +async fn cleanup(pool: &PgPool) { + // Idempotent teardown (also clears any fixtures a prior crashed run left). + // Memberships first (FK to both groups and users), targeted by the bench + // group names so it catches them whoever `added_by` is. + let _ = sqlx::query( + "DELETE FROM auth.subject_group_members WHERE group_id IN + (SELECT id FROM auth.subject_groups + WHERE name IN ('bench23parent','bench23child','bench23grand'))", + ) + .execute(pool) + .await; + let _ = sqlx::query( + "DELETE FROM auth.subject_groups WHERE name IN ('bench23parent','bench23child','bench23grand')", + ) + .execute(pool) + .await; + let _ = sqlx::query("DELETE FROM carddav.contacts WHERE uid LIKE 'bench23-%'") + .execute(pool) + .await; + let _ = sqlx::query("DELETE FROM carddav.address_books WHERE name = 'bench23_ab'") + .execute(pool) + .await; + let _ = sqlx::query("DELETE FROM auth.users WHERE email LIKE 'bench23-%@bench.invalid'") + .execute(pool) + .await; +} + +async fn seed_user(pool: &PgPool, tag: &str) -> Uuid { + sqlx::query_scalar( + "INSERT INTO auth.users (username, email, role) + VALUES ($1, $2, 'user') RETURNING id", + ) + .bind(format!("bench23_{tag}")) + .bind(format!("bench23-{tag}@bench.invalid")) + .fetch_one(pool) + .await + .expect("seed user") +} + +// ── [Q1] Contact JSONB decode ──────────────────────────────────────────────── +async fn section_q1(pool: &PgPool) { + let n: usize = env_or("Q1_CONTACTS", 500); + let passes: usize = env_or("Q1_DECODE_PASSES", 4000); + + let owner = seed_user(pool, "q1owner").await; + let ab: Uuid = sqlx::query_scalar( + "INSERT INTO carddav.address_books (id, name, owner_id) + VALUES (gen_random_uuid(), 'bench23_ab', $1) RETURNING id", + ) + .bind(owner) + .fetch_one(pool) + .await + .expect("seed address book"); + + for i in 0..n { + let emails = serde_json::to_value(vec![ + EmailDto { + email: format!("user{i}@example.com"), + r#type: "home".into(), + is_primary: true, + }, + EmailDto { + email: format!("user{i}@work.example.com"), + r#type: "work".into(), + is_primary: false, + }, + ]) + .unwrap(); + let phones = serde_json::to_value(vec![PhoneDto { + number: format!("+1-555-01{i:04}"), + r#type: "cell".into(), + is_primary: true, + }]) + .unwrap(); + let addrs = serde_json::to_value(vec![AddressDto { + street: Some(format!("{} Main St", 100 + i)), + city: Some("Springfield".into()), + state: Some("IL".into()), + postal_code: Some("62704".into()), + country: Some("US".into()), + r#type: "home".into(), + is_primary: true, + }]) + .unwrap(); + sqlx::query( + "INSERT INTO carddav.contacts (id, address_book_id, uid, full_name, email, phone, address, etag) + VALUES (gen_random_uuid(), $1, $2, $3, $4, $5, $6, $7)", + ) + .bind(ab) + .bind(format!("bench23-{i}")) + .bind(format!("Contact {i}")) + .bind(&emails) + .bind(&phones) + .bind(&addrs) + .bind(format!("etag-{i}")) + .execute(pool) + .await + .expect("seed contact"); + } + + // Fetch the rows ONCE (the query round-trip is out of the measured window — + // we isolate the per-row decode, which is what §J1 changes). + let rows = sqlx::query( + "SELECT email, phone, address FROM carddav.contacts + WHERE address_book_id = $1 ORDER BY uid", + ) + .bind(ab) + .fetch_all(pool) + .await + .expect("fetch contacts"); + assert_eq!(rows.len(), n, "Q1 seeded row count"); + + // BEFORE: Value DOM + from_value per column. + let decode_before = + |rows: &[sqlx::postgres::PgRow]| -> Vec<(Vec, Vec, Vec)> { + rows.iter() + .map(|r| { + let ev: Value = r.get("email"); + let pv: Value = r.get("phone"); + let av: Value = r.get("address"); + ( + serde_json::from_value::>(ev).unwrap_or_default(), + serde_json::from_value::>(pv).unwrap_or_default(), + serde_json::from_value::>(av).unwrap_or_default(), + ) + }) + .collect() + }; + // AFTER: typed Json decode straight from the JSONB bytes. + let decode_after = + |rows: &[sqlx::postgres::PgRow]| -> Vec<(Vec, Vec, Vec)> { + rows.iter() + .map(|r| { + ( + r.try_get::>, _>("email") + .map(|j| j.0) + .unwrap_or_default(), + r.try_get::>, _>("phone") + .map(|j| j.0) + .unwrap_or_default(), + r.try_get::>, _>("address") + .map(|j| j.0) + .unwrap_or_default(), + ) + }) + .collect() + }; + + // Equivalence gate. + if decode_before(&rows) != decode_after(&rows) { + eprintln!("GATE FAIL [Q1]: BEFORE/AFTER decode differ — rollback"); + cleanup(pool).await; + std::process::exit(1); + } + + let mut b = Vec::with_capacity(passes); + let mut a = Vec::with_capacity(passes); + for _ in 0..passes / 20 { + std::hint::black_box(decode_before(&rows)); + std::hint::black_box(decode_after(&rows)); + } + for _ in 0..passes { + let t = Instant::now(); + std::hint::black_box(decode_before(&rows)); + b.push(t.elapsed().as_nanos() as f64 / n as f64); + let t = Instant::now(); + std::hint::black_box(decode_after(&rows)); + a.push(t.elapsed().as_nanos() as f64 / n as f64); + } + + println!( + "\n## [Q1] Contact JSONB decode on real rows ({n} contacts) — gate OK (identical decode)" + ); + report( + "Value DOM + from_value vs Json", + "p50 ns/contact", + p50(b), + p50(a), + ); +} + +// ── [Q4] get_user_profile: serial vs join! ────────────────────────────────── +async fn section_q4(pool: &PgPool) { + let passes: usize = env_or("BENCH_PASSES", 200); + let caller = seed_user(pool, "q4caller").await; + let target = seed_user(pool, "q4target").await; + + // Capture `pool` (not take it as a param) so the returned future borrows a + // single concrete lifetime — a closure param `&PgPool` + future return hits + // the HRTB limitation. + let read = |id: Uuid| async move { + sqlx::query("SELECT id, email, role FROM auth.users WHERE id = $1") + .bind(id) + .fetch_optional(pool) + .await + .expect("read user") + .map(|r| r.get::("id")) + }; + + // Equivalence gate: same two ids either way. + let ser = (read(caller).await, read(target).await); + let (jc, jt) = tokio::join!(read(caller), read(target)); + if ser != (jc, jt) { + eprintln!("GATE FAIL [Q4]: serial/join ids differ — rollback"); + cleanup(pool).await; + std::process::exit(1); + } + + let mut b = Vec::with_capacity(passes); + let mut a = Vec::with_capacity(passes); + for _ in 0..(passes / 20).max(1) { + let _ = (read(caller).await, read(target).await); + let _ = tokio::join!(read(caller), read(target)); + } + for _ in 0..passes { + let t = Instant::now(); + let _ = std::hint::black_box((read(caller).await, read(target).await)); + b.push(t.elapsed().as_nanos() as f64); + let t = Instant::now(); + let _ = std::hint::black_box(tokio::join!(read(caller), read(target))); + a.push(t.elapsed().as_nanos() as f64); + } + + println!("\n## [Q4] get_user_profile caller+target reads — gate OK (identical ids)"); + report( + "2 serial reads vs tokio::join!", + "p50 ns/call", + p50(b), + p50(a), + ); +} + +// ── [Q6] subject_group child transitive users: 2 CTEs vs 1 ─────────────────── +async fn section_q6(pool: &PgPool) { + let passes: usize = env_or("BENCH_PASSES", 200); + // Tree: parent → child → {grandchild, u2}; grandchild → u3. u1 direct on parent. + let u1 = seed_user(pool, "q6u1").await; + let u2 = seed_user(pool, "q6u2").await; + let u3 = seed_user(pool, "q6u3").await; + let mk_group = |name: &'static str| async move { + sqlx::query_scalar::<_, Uuid>( + "INSERT INTO auth.subject_groups (name) VALUES ($1) RETURNING id", + ) + .bind(name) + .fetch_one(pool) + .await + .expect("seed group") + }; + let parent = mk_group("bench23parent").await; + let child = mk_group("bench23child").await; + let grand = mk_group("bench23grand").await; + let add_ug = |g: Uuid, u: Uuid| async move { + sqlx::query("INSERT INTO auth.subject_group_members (group_id, member_user_id, added_by) VALUES ($1, $2, $3)") + .bind(g).bind(u).bind(u1).execute(pool).await.expect("add user member"); + }; + let add_gg = |g: Uuid, c: Uuid| async move { + sqlx::query("INSERT INTO auth.subject_group_members (group_id, member_group_id, added_by) VALUES ($1, $2, $3)") + .bind(g).bind(c).bind(u1).execute(pool).await.expect("add group member"); + }; + add_ug(parent, u1).await; + add_gg(parent, child).await; + add_gg(child, grand).await; + add_ug(child, u2).await; + add_ug(grand, u3).await; + + let cte = |gid: Uuid| async move { + let rows = sqlx::query( + "WITH RECURSIVE descendants AS ( + SELECT $1::uuid AS g + UNION + SELECT m.member_group_id FROM auth.subject_group_members m + JOIN descendants d ON m.group_id = d.g WHERE m.member_group_id IS NOT NULL) + SELECT DISTINCT m.member_user_id AS user_id FROM auth.subject_group_members m + JOIN descendants d ON m.group_id = d.g WHERE m.member_user_id IS NOT NULL", + ) + .bind(gid) + .fetch_all(pool) + .await + .expect("cte"); + let mut ids: Vec = rows.iter().map(|r| r.get::("user_id")).collect(); + ids.sort(); + ids + }; + + // Equivalence: the child's transitive set is {u2, u3}, and it is IDENTICAL + // whether computed once or twice (the edge delete above the child cannot + // change its descendants — the §G1 correctness claim). + let once = cte(child).await; + let twice = { + let _first = cte(child).await; + cte(child).await + }; + let mut expected = [u2, u3]; + expected.sort(); + if once != twice || once != expected { + eprintln!("GATE FAIL [Q6]: child transitive set not stable/expected — rollback"); + cleanup(pool).await; + std::process::exit(1); + } + + let mut b = Vec::with_capacity(passes); + let mut a = Vec::with_capacity(passes); + for _ in 0..(passes / 20).max(1) { + let _ = (cte(child).await, cte(child).await); + let _ = cte(child).await; + } + for _ in 0..passes { + // BEFORE: the child CTE runs TWICE (pre-check + invalidation_targets). + let t = Instant::now(); + let _ = cte(child).await; + let _ = std::hint::black_box(cte(child).await); + b.push(t.elapsed().as_nanos() as f64); + // AFTER: once, reused. + let t = Instant::now(); + let _ = std::hint::black_box(cte(child).await); + a.push(t.elapsed().as_nanos() as f64); + } + + println!("\n## [Q6] subject_group child transitive users — gate OK (stable set {{u2,u3}})"); + report( + "2 recursive CTEs vs 1 (reused)", + "p50 ns/removal", + p50(b), + p50(a), + ); +} + +#[tokio::main(flavor = "multi_thread")] +async fn main() { + dotenvy::dotenv().ok(); + let url = env::var("DATABASE_URL") + .or_else(|_| env::var("OXICLOUD_DB_CONNECTION_STRING")) + .expect("set DATABASE_URL — the dev Postgres URL"); + let pool = PgPoolOptions::new() + .max_connections(8) + .connect(&url) + .await + .expect("connect Postgres"); + + println!("# Round-23 PG query-shape pack — BEFORE/AFTER (live Postgres)"); + println!("# Each section asserts an equivalence gate (mismatch → exit 1) and reports p50."); + + cleanup(&pool).await; + section_q1(&pool).await; + section_q4(&pool).await; + section_q6(&pool).await; + cleanup(&pool).await; + + println!("\nAll Round-23 query sections passed their equivalence gate."); +} diff --git a/src/application/services/auth_application_service.rs b/src/application/services/auth_application_service.rs index 373d8810..87befc00 100644 --- a/src/application/services/auth_application_service.rs +++ b/src/application/services/auth_application_service.rs @@ -1932,17 +1932,28 @@ impl AuthApplicationService { expose_system_users: bool, pool: &sqlx::PgPool, ) -> Result { - let caller = self.user_storage.get_user_by_id(caller_id).await?; - - // (1) Self. + // (1) Self — a single fetch suffices (the check compares the input + // UUIDs, so the target read is never needed on this path). if caller_id == target_id { + let caller = self.user_storage.get_user_by_id(caller_id).await?; return Ok(UserDto::from(caller)); } + // Caller and target are independent point reads (the self-case already + // returned; the branch above compares input UUIDs, not fetched data) — + // overlap them with `join!` instead of two serial round-trips. + // `caller_res?` first preserves the caller-error precedence of the old + // sequential form. (benches/ROUND23.md §P1) + let (caller_res, target_res) = tokio::join!( + self.user_storage.get_user_by_id(caller_id), + self.user_storage.get_user_by_id(target_id) + ); + let caller = caller_res?; + // Anti-enumeration: NotFound for everything that doesn't pass. // Convert a real NotFound on `target` to the same anonymous 404, // so existence isn't leaked through differential responses. - let target = match self.user_storage.get_user_by_id(target_id).await { + let target = match target_res { Ok(u) => u, Err(e) if e.kind == ErrorKind::NotFound => { tracing::info!( diff --git a/src/application/services/subject_group_service.rs b/src/application/services/subject_group_service.rs index 5c116b5d..180caf7c 100644 --- a/src/application/services/subject_group_service.rs +++ b/src/application/services/subject_group_service.rs @@ -477,6 +477,23 @@ impl SubjectGroupService { // user is still reachable via another path after this remove, // they stay in the set on the post-state, so the check would // pass on the next remove instead. + // For a nested child-group removal the child's transitive user set is + // needed twice: by the would-empty pre-check below AND, after the + // remove, as the cache-invalidation set. The edge delete is ABOVE the + // child, so it cannot change the child's descendants — compute the + // recursive CTE ONCE here and reuse it, instead of the identical query + // running twice (the second was hidden inside `invalidation_targets`). + // (benches/ROUND23.md §G1) + let child_users: Option> = match member { + GroupMember::Group(child_id) => Some( + self.repo + .list_transitive_users(child_id) + .await + .map_err(map_repo_err)?, + ), + GroupMember::User(_) => None, + }; + let users_before = self .repo .list_transitive_users(group_id) @@ -485,17 +502,12 @@ impl SubjectGroupService { if !users_before.is_empty() { let would_be_empty = match member { GroupMember::User(uid) => users_before.len() == 1 && users_before.contains(&uid), - GroupMember::Group(child_id) => { - // For child-group removal: would this drop the - // parent's transitive user set to 0? Look up the - // child's transitive users — if every user in the - // parent's set comes through the child, removing the - // child empties the parent. - let child_users = self - .repo - .list_transitive_users(child_id) - .await - .map_err(map_repo_err)?; + GroupMember::Group(_) => { + // Would removing this child drop the parent's transitive + // user set to 0? Reuse the child's transitive users + // computed above — if every user in the parent's set comes + // through the child, removing the child empties the parent. + let child_users = child_users.as_deref().unwrap_or(&[]); // Set probe instead of an O(|before|·|child|) slice scan // (benches/ROUND11.md §13: 5.7x at 500×500). let child_set: std::collections::HashSet<&uuid::Uuid> = @@ -535,7 +547,15 @@ impl SubjectGroupService { // ancestor. Without this, a removed-from-group user keeps // appearing as a transitive member in `expand_subject_for_listing` // for up to 30 s, surfacing grants they no longer have. - for uid in self.invalidation_targets(member).await? { + // + // Reuse the child's transitive users computed above (unchanged by the + // edge delete) as the invalidation set — no second recursive CTE. For a + // `User` member it's just that user. (benches/ROUND23.md §G1) + let invalidation: Vec = match member { + GroupMember::User(uid) => vec![uid], + GroupMember::Group(_) => child_users.unwrap_or_default(), + }; + for uid in invalidation { self.engine.invalidate_user_groups_cache(uid).await; self.drive_repo.invalidate_readable_for_user(uid).await; } diff --git a/src/domain/entities/drive.rs b/src/domain/entities/drive.rs index d5467d31..8b0bd95c 100644 --- a/src/domain/entities/drive.rs +++ b/src/domain/entities/drive.rs @@ -230,7 +230,15 @@ impl DrivePolicies { /// rather than refusing the read; enforcement code never panics on /// existing data. pub fn from_value(value: &serde_json::Value) -> Self { - serde_json::from_value(value.clone()).unwrap_or_default() + // Deserialize straight from the borrowed `Value` (`T::deserialize(&Value)`, + // via serde_json's `Deserializer for &Value`) instead of + // `serde_json::from_value(value.clone())` — the old form cloned the ENTIRE + // policies DOM before walking it, on every drive-policy read (move/copy, + // shared-link creation, grant). Byte-identical (same derived `Deserialize` + // impl); the lenient `unwrap_or_default` fallback is unchanged. + // (benches/ROUND23.md §J2) + use serde::Deserialize as _; + Self::deserialize(value).unwrap_or_default() } /// D5 `forbid_public_links` gate, used by every entry point that diff --git a/src/infrastructure/repositories/pg/contact_group_pg_repository.rs b/src/infrastructure/repositories/pg/contact_group_pg_repository.rs index fedc4db2..945d7d1e 100644 --- a/src/infrastructure/repositories/pg/contact_group_pg_repository.rs +++ b/src/infrastructure/repositories/pg/contact_group_pg_repository.rs @@ -1,5 +1,4 @@ use chrono::Utc; -use serde_json::Value as JsonValue; use sqlx::{PgPool, Row, types::Uuid}; use std::sync::Arc; @@ -220,18 +219,21 @@ impl ContactGroupRepository for ContactGroupPgRepository { let mut contacts = Vec::with_capacity(rows.len()); for row in &rows { - let email_json: JsonValue = row.get("email"); - let phone_json: JsonValue = row.get("phone"); - let address_json: JsonValue = row.get("address"); - - let emails = serde_json::from_value::>(email_json) - .map(emails_from_persistence) + // Typed `Json` decode (one `from_slice` pass) instead of the + // `Value` DOM + `from_value` re-walk — the contact_pg_repository + // §J1 fix applied to this inlined sibling. Byte-identical result, + // 3 fewer throwaway DOMs per contact. (benches/ROUND23.md §J1) + let emails = row + .try_get::>, _>("email") + .map(|j| emails_from_persistence(j.0)) .unwrap_or_default(); - let phones = serde_json::from_value::>(phone_json) - .map(phones_from_persistence) + let phones = row + .try_get::>, _>("phone") + .map(|j| phones_from_persistence(j.0)) .unwrap_or_default(); - let addresses = serde_json::from_value::>(address_json) - .map(addresses_from_persistence) + let addresses = row + .try_get::>, _>("address") + .map(|j| addresses_from_persistence(j.0)) .unwrap_or_default(); contacts.push(Contact::from_raw( diff --git a/src/infrastructure/repositories/pg/contact_pg_repository.rs b/src/infrastructure/repositories/pg/contact_pg_repository.rs index 1edebbee..d6ab1e5a 100644 --- a/src/infrastructure/repositories/pg/contact_pg_repository.rs +++ b/src/infrastructure/repositories/pg/contact_pg_repository.rs @@ -23,18 +23,27 @@ impl ContactPgRepository { /// Maps a database row to a Contact domain entity fn row_to_contact(row: &sqlx::postgres::PgRow) -> Result { - let email_json: JsonValue = row.get("email"); - let phone_json: JsonValue = row.get("phone"); - let address_json: JsonValue = row.get("address"); - - let emails = serde_json::from_value::>(email_json) - .map(emails_from_persistence) + // Decode each JSONB column straight into its typed Vec via + // `sqlx::types::Json` (a single `serde_json::from_slice` pass over + // the raw JSONB bytes) instead of `row.get::` + + // `serde_json::from_value`, which built a throwaway `Value` DOM per + // column and then walked it a SECOND time to produce the typed Vec — + // 3 discarded DOMs per contact row on every list / multiget / CardDAV + // sync. `try_get` preserves the exact malformed-shape fallback (the old + // `from_value(...).unwrap_or_default()`; a bare `row.get` would panic on + // a decode error); the columns are `JSONB NOT NULL DEFAULT '[]'`, so SQL + // NULL never occurs. (benches/ROUND23.md §J1) + let emails = row + .try_get::>, _>("email") + .map(|j| emails_from_persistence(j.0)) .unwrap_or_default(); - let phones = serde_json::from_value::>(phone_json) - .map(phones_from_persistence) + let phones = row + .try_get::>, _>("phone") + .map(|j| phones_from_persistence(j.0)) .unwrap_or_default(); - let addresses = serde_json::from_value::>(address_json) - .map(addresses_from_persistence) + let addresses = row + .try_get::>, _>("address") + .map(|j| addresses_from_persistence(j.0)) .unwrap_or_default(); Ok(Contact::from_raw( diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 258c4416..ef6604e9 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -209,8 +209,11 @@ impl IngestGuard { // sweep can reclaim the bytes — a backend file with no PG row would be // invisible to it. ON CONFLICT DO NOTHING keeps a concurrent // uploader's row (and its references) intact. - let hashes: Vec = written.iter().map(|(h, _)| h.clone()).collect(); - let sizes: Vec = written.iter().map(|(_, s)| *s).collect(); + // `written` is owned and dead after this rollback — unzip it (moving each + // 64-byte hash String out) instead of cloning every hash purely to + // reshape for `sync_blobs(&[String])` + the UNNEST bind. + // (benches/ROUND23.md §U1) + let (hashes, sizes): (Vec, Vec) = written.into_iter().unzip(); if let Err(e) = backend.sync_blobs(&hashes).await { tracing::warn!( "Ingest rollback: sync of {} chunks failed: {e}", @@ -896,8 +899,10 @@ impl DedupService { if !new_rows.is_empty() { // Durability before visibility — same invariant as the ingest // engine: no PG row may ever point at unsynced bytes. - let hashes: Vec = new_rows.iter().map(|(h, _)| h.clone()).collect(); - let sizes: Vec = new_rows.iter().map(|(_, s)| *s).collect(); + // `new_rows` is owned and dead after this block — unzip (move the + // hash Strings out) instead of cloning each one for the reshape + + // UNNEST bind. (benches/ROUND23.md §U1) + let (hashes, sizes): (Vec, Vec) = new_rows.into_iter().unzip(); self.backend.sync_blobs(&hashes).await?; sqlx::query( "INSERT INTO storage.blobs (hash, size, ref_count, orphaned_at)