perf: 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: BEFORE/AFTER with a value-equivalence gate and rollback-on-regression. Two harnesses — bench_round23_micro (no Postgres; deterministic allocation gate) and bench_round23_queries (live Postgres; p50 latency + strict equivalence gate against seeded fixtures). See benches/ROUND23.md. - J1: contact_pg_repository::row_to_contact (+ the inlined contact_group sibling) decode the 3 JSONB columns via sqlx::types::Json<T> (one from_slice pass) instead of row.get::<serde_json::Value> + from_value (a throwaway Value DOM per column, walked a second time). Per contact row of every list / multiget / CardDAV sync. Micro 84 -> 33 allocs/op (2.15x); PG 3794 -> 2360 ns/contact (1.61x) on 500 real rows. - J2: DrivePolicies::from_value deserializes straight from the borrow (T::deserialize(&Value)) instead of from_value(value.clone()) — dropping the full-DOM clone on every drive-policy read (move/copy, share, grant); one-line body change, all 7 callers unchanged. Micro 5 -> 0 allocs/op (11.51x). - P1: get_user_profile overlaps the two independent caller+target reads with tokio::join! (self-case still a single fetch; caller-error precedence preserved via caller_res? first) instead of two serial round-trips. PG 577 -> 312 us/call (1.85x). - G1: subject_group remove_member computes the child's transitive-user recursive CTE once and reuses it for both the would-empty pre-check and the cache invalidation, instead of running the identical CTE twice (the edge delete is above the child, so its descendants can't change). PG 829 -> 412 us/removal (2.01x). - U1: dedup_service (store_loose_chunks final registration + the ingest run_rollback) reshapes the owned, dead-after Vec<(String,i64)> via into_iter().unzip() instead of cloning every 64-byte hash for the sync_blobs(&[String]) + UNNEST bind. Micro 256 -> 0 hash clones. Verified: cargo clippy --features bench --all-targets -D warnings clean, cargo fmt --all --check clean, cargo test --lib --features bench = 529 passed / 0 failed. The PG benches run against a local PostgreSQL 16 (schema applied from migrations/); every equivalence gate passes. The download_zip per-item N+1 (the audit's highest raw-latency candidate) is deferred to a dedicated pass: its fix moves the sole authorization inside the stream call, so it needs an AuthZ-ordering + anti-enumeration proof, not a perf banner. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DKyQ4AnYtgp1JtjzweyMeo
This commit is contained in:
@@ -1932,17 +1932,28 @@ impl AuthApplicationService {
|
||||
expose_system_users: bool,
|
||||
pool: &sqlx::PgPool,
|
||||
) -> Result<UserDto, DomainError> {
|
||||
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!(
|
||||
|
||||
@@ -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<Vec<uuid::Uuid>> = 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<uuid::Uuid> = 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;
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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::<Vec<EmailPersistenceDto>>(email_json)
|
||||
.map(emails_from_persistence)
|
||||
// Typed `Json<T>` 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::<sqlx::types::Json<Vec<EmailPersistenceDto>>, _>("email")
|
||||
.map(|j| emails_from_persistence(j.0))
|
||||
.unwrap_or_default();
|
||||
let phones = serde_json::from_value::<Vec<PhonePersistenceDto>>(phone_json)
|
||||
.map(phones_from_persistence)
|
||||
let phones = row
|
||||
.try_get::<sqlx::types::Json<Vec<PhonePersistenceDto>>, _>("phone")
|
||||
.map(|j| phones_from_persistence(j.0))
|
||||
.unwrap_or_default();
|
||||
let addresses = serde_json::from_value::<Vec<AddressPersistenceDto>>(address_json)
|
||||
.map(addresses_from_persistence)
|
||||
let addresses = row
|
||||
.try_get::<sqlx::types::Json<Vec<AddressPersistenceDto>>, _>("address")
|
||||
.map(|j| addresses_from_persistence(j.0))
|
||||
.unwrap_or_default();
|
||||
|
||||
contacts.push(Contact::from_raw(
|
||||
|
||||
@@ -23,18 +23,27 @@ impl ContactPgRepository {
|
||||
|
||||
/// Maps a database row to a Contact domain entity
|
||||
fn row_to_contact(row: &sqlx::postgres::PgRow) -> Result<Contact, DomainError> {
|
||||
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::<Vec<EmailPersistenceDto>>(email_json)
|
||||
.map(emails_from_persistence)
|
||||
// Decode each JSONB column straight into its typed Vec via
|
||||
// `sqlx::types::Json<T>` (a single `serde_json::from_slice` pass over
|
||||
// the raw JSONB bytes) instead of `row.get::<serde_json::Value>` +
|
||||
// `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::<sqlx::types::Json<Vec<EmailPersistenceDto>>, _>("email")
|
||||
.map(|j| emails_from_persistence(j.0))
|
||||
.unwrap_or_default();
|
||||
let phones = serde_json::from_value::<Vec<PhonePersistenceDto>>(phone_json)
|
||||
.map(phones_from_persistence)
|
||||
let phones = row
|
||||
.try_get::<sqlx::types::Json<Vec<PhonePersistenceDto>>, _>("phone")
|
||||
.map(|j| phones_from_persistence(j.0))
|
||||
.unwrap_or_default();
|
||||
let addresses = serde_json::from_value::<Vec<AddressPersistenceDto>>(address_json)
|
||||
.map(addresses_from_persistence)
|
||||
let addresses = row
|
||||
.try_get::<sqlx::types::Json<Vec<AddressPersistenceDto>>, _>("address")
|
||||
.map(|j| addresses_from_persistence(j.0))
|
||||
.unwrap_or_default();
|
||||
|
||||
Ok(Contact::from_raw(
|
||||
|
||||
@@ -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<String> = written.iter().map(|(h, _)| h.clone()).collect();
|
||||
let sizes: Vec<i64> = 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<String>, Vec<i64>) = 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<String> = new_rows.iter().map(|(h, _)| h.clone()).collect();
|
||||
let sizes: Vec<i64> = 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<String>, Vec<i64>) = new_rows.into_iter().unzip();
|
||||
self.backend.sync_blobs(&hashes).await?;
|
||||
sqlx::query(
|
||||
"INSERT INTO storage.blobs (hash, size, ref_count, orphaned_at)
|
||||
|
||||
Reference in New Issue
Block a user