perf(authz): cache resource owner lookups in PgAclEngine
The owner short-circuit in PgAclEngine::check ran a PK query (SELECT user_id FROM storage.folders/files WHERE id=$1) on every authorization check of a folder/file — the common case, since users mostly act on their own resources. Memoise it in an owner_cache (moka, TTL 300s, 100k cap). The owner column is immutable, so this is safe: the cache maps resource -> real owner and can never grant a non-owner access (a different caller's owner==uid test fails against the cached owner and falls through to grants); a hard-deleted resource that briefly resolves to its former owner simply fails later at execution with NotFound. The per-check sql_queries counter now increments only on a miss. Removes 1 DB query + 1 pool-connection acquisition per owner check. Magnitude is deployment-specific (query latency x whether the pool is contended); see benches/ACL-OWNER-CACHE.md. Also adds two DB perf-investigation harnesses, gated behind the `bench` feature (need the dev Postgres; zero prod impact): - examples/bench_db_pool.rs + benches/DB-POOL.md — pool size vs tail latency - examples/bench_owner_cache.rs + benches/ACL-OWNER-CACHE.md — owner query vs cache Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
DioCrafts
@@ -0,0 +1,151 @@
|
||||
//! DB connection-pool tail-latency benchmark.
|
||||
//!
|
||||
//! Isolates the variable under test — `max_connections` — from the HTTP/auth
|
||||
//! stack. Builds a real `sqlx` Postgres pool of size P and drives it with `C`
|
||||
//! concurrent workers, each looping `SELECT pg_sleep($query_ms)` (a query of
|
||||
//! known duration). The measured per-request latency is **acquire-wait + query**
|
||||
//! — exactly the pool-exhaustion mechanism: when in-flight queries exceed P, the
|
||||
//! surplus queues on `acquire()`, inflating p95/p99.
|
||||
//!
|
||||
//! `pg_sleep` is a faithful stand-in for "a query that occupies a connection for
|
||||
//! T ms" — real listing/auth queries take a few ms each. We hold P constant per
|
||||
//! run and sweep it, so the *shape* of tail-latency-vs-pool-size is what matters.
|
||||
//!
|
||||
//! Run (needs the dev Postgres up; reads DATABASE_URL from .env):
|
||||
//! cargo run --release --features bench --example bench_db_pool
|
||||
//! Tunables (env): BENCH_CONCURRENCY (default 96), BENCH_QUERY_MS (3),
|
||||
//! BENCH_SECONDS (4), BENCH_POOL_SIZES ("10,20,40,70").
|
||||
|
||||
use std::env;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use sqlx::postgres::PgPoolOptions;
|
||||
|
||||
fn env_or<T: std::str::FromStr>(key: &str, default: T) -> T {
|
||||
env::var(key)
|
||||
.ok()
|
||||
.and_then(|v| v.parse().ok())
|
||||
.unwrap_or(default)
|
||||
}
|
||||
|
||||
#[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 (or OXICLOUD_DB_CONNECTION_STRING) — the dev Postgres URL");
|
||||
|
||||
let concurrency: usize = env_or("BENCH_CONCURRENCY", 96);
|
||||
let query_ms: u64 = env_or("BENCH_QUERY_MS", 3);
|
||||
let secs: u64 = env_or("BENCH_SECONDS", 4);
|
||||
let pool_sizes: Vec<u32> = env::var("BENCH_POOL_SIZES")
|
||||
.ok()
|
||||
.map(|s| s.split(',').filter_map(|x| x.trim().parse().ok()).collect())
|
||||
.unwrap_or_else(|| vec![10, 20, 40, 70]);
|
||||
|
||||
println!("\n###########################################################");
|
||||
println!("# DB pool tail-latency benchmark");
|
||||
println!("# concurrency (in-flight requests): {concurrency}");
|
||||
println!("# query duration: pg_sleep({query_ms} ms) window: {secs}s/pool");
|
||||
println!("# latency = acquire-wait + query (the pool-queue effect)");
|
||||
println!("###########################################################\n");
|
||||
println!(
|
||||
"| {:>4} | {:>9} | {:>10} | {:>8} | {:>8} | {:>8} | {:>9} | {:>6} |",
|
||||
"pool", "requests", "req/s", "p50 ms", "p95 ms", "p99 ms", "max ms", "errors"
|
||||
);
|
||||
println!(
|
||||
"|{:-<6}|{:-<11}|{:-<12}|{:-<10}|{:-<10}|{:-<10}|{:-<11}|{:-<8}|",
|
||||
"", "", "", "", "", "", "", ""
|
||||
);
|
||||
|
||||
let qsec = query_ms as f64 / 1000.0;
|
||||
|
||||
for &pool_size in &pool_sizes {
|
||||
let pool = PgPoolOptions::new()
|
||||
.max_connections(pool_size)
|
||||
.min_connections(pool_size) // pre-warm so we don't time connection setup
|
||||
.acquire_timeout(Duration::from_secs(10)) // matches prod connect_timeout default
|
||||
.connect(&url)
|
||||
.await
|
||||
.unwrap_or_else(|e| panic!("connect pool={pool_size}: {e}"));
|
||||
|
||||
// Warm-up burst (discarded).
|
||||
{
|
||||
let mut warm = Vec::new();
|
||||
for _ in 0..concurrency {
|
||||
let pool = pool.clone();
|
||||
warm.push(tokio::spawn(async move {
|
||||
let _ = sqlx::query("SELECT pg_sleep($1)")
|
||||
.bind(qsec)
|
||||
.execute(&pool)
|
||||
.await;
|
||||
}));
|
||||
}
|
||||
for h in warm {
|
||||
let _ = h.await;
|
||||
}
|
||||
}
|
||||
|
||||
let start = Instant::now();
|
||||
let deadline = start + Duration::from_secs(secs);
|
||||
let mut handles = Vec::with_capacity(concurrency);
|
||||
for _ in 0..concurrency {
|
||||
let pool = pool.clone();
|
||||
handles.push(tokio::spawn(async move {
|
||||
let mut lats_us: Vec<u32> = Vec::with_capacity(8192);
|
||||
let mut errors: u64 = 0;
|
||||
while Instant::now() < deadline {
|
||||
let t = Instant::now();
|
||||
match sqlx::query("SELECT pg_sleep($1)")
|
||||
.bind(qsec)
|
||||
.execute(&pool)
|
||||
.await
|
||||
{
|
||||
Ok(_) => lats_us.push(t.elapsed().as_micros() as u32),
|
||||
Err(_) => errors += 1,
|
||||
}
|
||||
}
|
||||
(lats_us, errors)
|
||||
}));
|
||||
}
|
||||
|
||||
let mut all: Vec<u32> = Vec::new();
|
||||
let mut errors: u64 = 0;
|
||||
for h in handles {
|
||||
let (l, e) = h.await.expect("join worker");
|
||||
all.extend(l);
|
||||
errors += e;
|
||||
}
|
||||
let elapsed = start.elapsed().as_secs_f64();
|
||||
pool.close().await;
|
||||
|
||||
all.sort_unstable();
|
||||
let n = all.len();
|
||||
let pct = |q: f64| -> f64 {
|
||||
if n == 0 {
|
||||
return 0.0;
|
||||
}
|
||||
let idx = ((q / 100.0) * (n as f64 - 1.0)).round() as usize;
|
||||
all[idx.min(n - 1)] as f64 / 1000.0
|
||||
};
|
||||
let tput = n as f64 / elapsed;
|
||||
|
||||
println!(
|
||||
"| {:>4} | {:>9} | {:>10.0} | {:>8.2} | {:>8.2} | {:>8.2} | {:>9.2} | {:>6} |",
|
||||
pool_size,
|
||||
n,
|
||||
tput,
|
||||
pct(50.0),
|
||||
pct(95.0),
|
||||
pct(99.0),
|
||||
pct(100.0),
|
||||
errors,
|
||||
);
|
||||
}
|
||||
|
||||
println!(
|
||||
"\nNote: pg_sleep models query DURATION (connection occupancy), not CPU.\n\
|
||||
At fixed concurrency, raising the pool cuts queue-wait until pool ≈ concurrency,\n\
|
||||
then plateaus — the tail-latency shape that tells you the right size for your load.\n"
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,196 @@
|
||||
//! ACL owner-cache benchmark.
|
||||
//!
|
||||
//! Models the owner short-circuit in `PgAclEngine::check` (the common case: a
|
||||
//! user touching their own files). Before the cache, every authorization check
|
||||
//! on a folder/file ran one PK query `SELECT user_id FROM storage.folders WHERE
|
||||
//! id=$1` — a DB round-trip that also occupies a pool connection. After, repeat
|
||||
//! checks for the same resource hit an in-memory moka cache (owner is immutable).
|
||||
//!
|
||||
//! This isolates that exact query vs a moka hit, under concurrency C, against the
|
||||
//! real dev Postgres. It shows both the latency win and — by NOT touching the
|
||||
//! pool — the relief it gives the connection pool (ties into the pool benchmark).
|
||||
//!
|
||||
//! Run (needs the dev Postgres up with at least one folder; reads DATABASE_URL):
|
||||
//! cargo run --release --features bench --example bench_owner_cache
|
||||
//! Tunables: BENCH_CONCURRENCY (64), BENCH_POOL_SIZE (20 = prod default),
|
||||
//! BENCH_SECONDS (4).
|
||||
|
||||
use std::env;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use moka::future::Cache;
|
||||
use sqlx::Row;
|
||||
use sqlx::postgres::PgPoolOptions;
|
||||
use uuid::Uuid;
|
||||
|
||||
fn env_or<T: std::str::FromStr>(key: &str, default: T) -> T {
|
||||
env::var(key)
|
||||
.ok()
|
||||
.and_then(|v| v.parse().ok())
|
||||
.unwrap_or(default)
|
||||
}
|
||||
|
||||
struct Stats {
|
||||
reqs: u64,
|
||||
tput: f64,
|
||||
p50_us: f64,
|
||||
p95_us: f64,
|
||||
p99_us: f64,
|
||||
max_us: f64,
|
||||
}
|
||||
|
||||
fn summarize(mut lats_ns: Vec<u64>, elapsed: f64) -> Stats {
|
||||
lats_ns.sort_unstable();
|
||||
let n = lats_ns.len();
|
||||
let pct = |q: f64| -> f64 {
|
||||
if n == 0 {
|
||||
return 0.0;
|
||||
}
|
||||
let idx = ((q / 100.0) * (n as f64 - 1.0)).round() as usize;
|
||||
lats_ns[idx.min(n - 1)] as f64 / 1000.0
|
||||
};
|
||||
Stats {
|
||||
reqs: n as u64,
|
||||
tput: n as f64 / elapsed,
|
||||
p50_us: pct(50.0),
|
||||
p95_us: pct(95.0),
|
||||
p99_us: pct(99.0),
|
||||
max_us: pct(100.0),
|
||||
}
|
||||
}
|
||||
|
||||
#[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 concurrency: usize = env_or("BENCH_CONCURRENCY", 64);
|
||||
let pool_size: u32 = env_or("BENCH_POOL_SIZE", 20); // production default
|
||||
let secs: u64 = env_or("BENCH_SECONDS", 4);
|
||||
|
||||
let pool = PgPoolOptions::new()
|
||||
.max_connections(pool_size)
|
||||
.min_connections(pool_size)
|
||||
.acquire_timeout(Duration::from_secs(10))
|
||||
.connect(&url)
|
||||
.await
|
||||
.expect("connect dev Postgres");
|
||||
|
||||
// A real folder + its owner to check against.
|
||||
let Some(row) = sqlx::query("SELECT id, user_id FROM storage.folders LIMIT 1")
|
||||
.fetch_optional(&pool)
|
||||
.await
|
||||
.expect("query folder")
|
||||
else {
|
||||
eprintln!("No folders in the dev DB — seed some first (`just load-seed`).");
|
||||
return;
|
||||
};
|
||||
let folder_id: Uuid = row.get("id");
|
||||
let owner_id: Uuid = row.get("user_id");
|
||||
|
||||
println!("\n###########################################################");
|
||||
println!("# ACL owner-cache benchmark (owner short-circuit path)");
|
||||
println!("# concurrency: {concurrency} pool: {pool_size} window: {secs}s/mode");
|
||||
println!("# folder {folder_id} owner {owner_id}");
|
||||
println!("###########################################################\n");
|
||||
|
||||
// ── BEFORE: one PK owner query per check (hits DB + pool) ──────────────
|
||||
let uncached = {
|
||||
let start = Instant::now();
|
||||
let deadline = start + Duration::from_secs(secs);
|
||||
let mut handles = Vec::with_capacity(concurrency);
|
||||
for _ in 0..concurrency {
|
||||
let pool = pool.clone();
|
||||
handles.push(tokio::spawn(async move {
|
||||
let mut lats = Vec::with_capacity(16384);
|
||||
while Instant::now() < deadline {
|
||||
let t = Instant::now();
|
||||
let _: Uuid =
|
||||
sqlx::query_scalar("SELECT user_id FROM storage.folders WHERE id = $1")
|
||||
.bind(folder_id)
|
||||
.fetch_one(&pool)
|
||||
.await
|
||||
.expect("owner query");
|
||||
lats.push(t.elapsed().as_nanos() as u64);
|
||||
}
|
||||
lats
|
||||
}));
|
||||
}
|
||||
let mut all = Vec::new();
|
||||
for h in handles {
|
||||
all.extend(h.await.expect("join"));
|
||||
}
|
||||
let s = summarize(all, start.elapsed().as_secs_f64());
|
||||
(s.reqs, s) // reqs == DB queries
|
||||
};
|
||||
|
||||
// ── AFTER: moka hit per check (no DB, no pool) ────────────────────────
|
||||
let cache: Cache<Uuid, Uuid> = Cache::builder()
|
||||
.max_capacity(100_000)
|
||||
.time_to_live(Duration::from_secs(300))
|
||||
.build();
|
||||
cache.insert(folder_id, owner_id).await; // 1 warm-up "query"
|
||||
let cached = {
|
||||
let start = Instant::now();
|
||||
let deadline = start + Duration::from_secs(secs);
|
||||
let mut handles = Vec::with_capacity(concurrency);
|
||||
for _ in 0..concurrency {
|
||||
let cache = cache.clone();
|
||||
handles.push(tokio::spawn(async move {
|
||||
let mut lats = Vec::with_capacity(65536);
|
||||
while Instant::now() < deadline {
|
||||
let t = Instant::now();
|
||||
let owner = cache.get(&folder_id).await.expect("cache hit");
|
||||
std::hint::black_box(owner);
|
||||
lats.push(t.elapsed().as_nanos() as u64);
|
||||
}
|
||||
lats
|
||||
}));
|
||||
}
|
||||
let mut all = Vec::new();
|
||||
for h in handles {
|
||||
all.extend(h.await.expect("join"));
|
||||
}
|
||||
summarize(all, start.elapsed().as_secs_f64())
|
||||
};
|
||||
pool.close().await;
|
||||
|
||||
let (uncached_queries, a) = uncached;
|
||||
println!(
|
||||
"| {:<16} | {:>10} | {:>11} | {:>9} | {:>9} | {:>9} | {:>10} |",
|
||||
"mode", "DB queries", "checks/s", "p50 µs", "p95 µs", "p99 µs", "max µs"
|
||||
);
|
||||
println!(
|
||||
"|{:-<18}|{:-<12}|{:-<13}|{:-<11}|{:-<11}|{:-<11}|{:-<12}|",
|
||||
"", "", "", "", "", "", ""
|
||||
);
|
||||
println!(
|
||||
"| {:<16} | {:>10} | {:>11.0} | {:>9.1} | {:>9.1} | {:>9.1} | {:>10.1} |",
|
||||
"uncached (before)", uncached_queries, a.tput, a.p50_us, a.p95_us, a.p99_us, a.max_us
|
||||
);
|
||||
println!(
|
||||
"| {:<16} | {:>10} | {:>11.0} | {:>9.3} | {:>9.3} | {:>9.3} | {:>10.3} |",
|
||||
"cached (after)",
|
||||
1,
|
||||
cached.tput,
|
||||
cached.p50_us,
|
||||
cached.p95_us,
|
||||
cached.p99_us,
|
||||
cached.max_us
|
||||
);
|
||||
|
||||
println!(
|
||||
"\nPer authorized action on an owned resource, the cache removes 1 DB query\n\
|
||||
+ 1 pool-connection occupancy, turning a {:.0} µs round-trip into a {:.3} µs\n\
|
||||
memory hit ({:.0}× lower p99). Over a network/loaded DB the absolute saving is\n\
|
||||
larger; the freed connections directly relieve the pool (see DB-POOL.md).\n",
|
||||
a.p99_us,
|
||||
cached.p99_us,
|
||||
if cached.p99_us > 0.0 {
|
||||
a.p99_us / cached.p99_us
|
||||
} else {
|
||||
0.0
|
||||
}
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user