From d44031f4187b285dad9b367c114f017c7858df9b Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Tue, 3 Nov 2026 11:15:49 -0500 Subject: [PATCH] perf(relay): cut writer-pool round trips past the 48k relay ceiling (#8846) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Five changes that take load off the writer pool and the relay's connection bookkeeping. On a single-relay load rig, the clean ceiling moves from 40k to about 48k WebSocket clients. In the A/B, connect p99 at 48k fell from 4.1 s (abort) to 1.65 s. 1. **Ban/mute restriction cache with a 1 s max age.** HTTP API requests, publishes and ephemeral writes read `(community, pubkey)` restriction state through one cache. An entry is served for 1 s from when its read began, matching the open-channel cache's bound. Socket admission, moderation-command actor checks and the `/buzz/v1` ban recheck still read fresh. An admitted socket outlives the 1 s window, so a banned client that reconnects inside it would otherwise keep a socket open indefinitely. A ban drops the target's entry, and each owned agent's entry, on every pod as part of the live-disconnect fan-out, best-effort. Timeouts, untimeouts and unbans drop it on the pod that handles them. 2. **Fewer round trips per event read.** Replica reads skip the per-read `statement_timeout` tighten, because the reader pool installs the deadline at connect. They keep the savepoint, so a failed or cancelled read leaves the session usable and releases its locks. That takes them from 4 round trips to 3. Writer reads send `BEGIN` and the tighten as one message, which also takes them from 4 to 3. 3. **Concurrent open-channel cache misses share one read.** The first miss in a community takes a per-community gate and reads; the others wait for it and use its result. A waiter whose leader failed reads once on its own and does not queue again. A waiter also stops waiting after the cache's 1 s max age and reads on its own, because the leader's set would be stale by then anyway. So a leader whose read hangs stalls only its own request. 4. **`has_admitted_connection` is one map lookup.** A `(community, pubkey) → admitted count` index is maintained under each socket's pubkey lock in `mark_admitted`, `set_authenticated_identity` and `deregister`. It replaces a scan of every connection on each disconnect. 5. **Host → community cache with a 1 s max age.** Only successful lookups are cached; unmapped hosts and errors go to the database every time. ## What got simpler - The ban handler's separate pod-local cache drop is gone; the disconnect every ban already runs covers it on every pod. - One write-restriction gate instead of two. `enforce_cached_write_restriction`, `enforce_relay_membership_ephemeral` and the `cached_ban` parameter chain through `enforce_membership` / `admit_member` / `deny_banned` are deleted. - The ephemeral path's separate 30 s ban cache is gone, so the relay has one staleness rule for restriction state instead of two. ## Notes - Bans reach the restriction cache on every pod. Every ban path (signed command, report action, relay admin) closes the target's and its agents' sockets clusterwide, and each pod drops those entries in the same step. A dropped Redis publish, or a restriction read already in flight when the drop lands (it can refill the old row), falls back to the 1 s bound. Timeouts on other pods also rely on the 1 s bound, and so do unbans and untimeouts, which fail closed. - On the load rig the restriction-cache hit rate was 28%, because HTTP agents poll every 5 s. A longer TTL with Redis invalidation would raise it, but that is a separate change. ## Load test Single relay pinned to 8 cores, native Postgres 16, a seeded fixture with 2,000 open channels, and a 6-shard generator. Steady ramp with 120 s holds; a step aborts at p99 > 3 s. Each step adds 10% HTTP agents. | clients | base connect p99 | this PR connect p99 | base relay cores | this PR relay cores | |---|---|---|---|---| | 32k | 34 ms | 31 ms | 1.71 | 1.66 | | 40k | 45 ms | 26 ms | 2.33 | 2.15 | | 48k | 4,107 ms (abort) | 1,652 ms | 4.7 | 3.0 | | 56k | n/a | 4,535 ms (abort) | n/a | 4.0 | Open-channel re-reads over the run fell from 72k to 1.6k. The tenant cache hit rate was 94%. 48k is this branch's knee. A later run of the same build aborted at 48k: connect p99 2.2 s, but 29% subscribe errors from the concurrent-handler cap. With a 10 s restriction max age, it passed 48k, at connect p99 1.56 s, 60% hit rate and 10% fewer relay cores, and both arms stopped at 56k. ## Testing - `cargo fmt --check`, `cargo clippy -p buzz-relay -p buzz-db --all-targets -D warnings` - `cargo test -p buzz-db --lib` and `-p buzz-relay --lib`, run in parallel. The only failures were 8 DB-seeding tests that need a migrated default database; the Postgres lane covers them. Commits after that run changed the replica savepoint, the `/buzz/v1` recheck revert and some test edits. - Postgres lane at head: `scripts/postgres-test-run.sh -p buzz-relay -p buzz-db --lib --bins --tests`. All 1,183 tests passed. - Mutants. Each one must turn its new test red: - waiters skipping the gate - a no-op index removal - the tenant cache ignoring age - the restriction cache ignoring age - a waiter waiting on the gate with no limit - a disconnect that keeps the cached restriction row - replica reads dropping the savepoint buzz-review-completed 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Luke Tornquist Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz> Co-authored-by: Claude Opus 5.5 --- Justfile | 5 + crates/buzz-db/src/runtime/mod.rs | 2 +- crates/buzz-db/src/store/event.rs | 161 +++++- crates/buzz-relay/src/api/admin/direct.rs | 2 +- crates/buzz-relay/src/api/admin/mod.rs | 10 +- crates/buzz-relay/src/api/bridge.rs | 34 +- .../buzz-relay/src/api/invites/claim_auth.rs | 2 +- crates/buzz-relay/src/api/mod.rs | 36 +- crates/buzz-relay/src/api/nip05.rs | 2 +- .../src/api/typing_postgres_tests.rs | 18 +- crates/buzz-relay/src/audio/handler.rs | 2 +- crates/buzz-relay/src/handlers/auth.rs | 7 +- crates/buzz-relay/src/handlers/event.rs | 89 ++- crates/buzz-relay/src/handlers/ingest.rs | 40 +- .../src/handlers/moderation_commands.rs | 1 - crates/buzz-relay/src/handlers/relay_admin.rs | 4 +- crates/buzz-relay/src/nip11.rs | 8 +- crates/buzz-relay/src/nip_fi_http.rs | 2 +- .../src/nip_fi_resolver/tests/router_tests.rs | 3 + crates/buzz-relay/src/nip_fi_shadow.rs | 4 +- crates/buzz-relay/src/router.rs | 2 +- crates/buzz-relay/src/state.rs | 509 ++++++++++++++++-- crates/buzz-relay/src/tenant.rs | 63 +++ scripts/run-tests.sh | 7 + 24 files changed, 768 insertions(+), 245 deletions(-) diff --git a/Justfile b/Justfile index c260b05b66..eb6c3020a3 100644 --- a/Justfile +++ b/Justfile @@ -637,6 +637,11 @@ test-unit: + test(/^api::nip_fi::/)' # Invitation drain/cutoff boundary contracts are infra-free. cargo nextest run -p buzz-relay --lib -E 'test(/^invite_token::/) + test(/^config::invite_cutoff_tests::/)' + # Relay read caches: the admitted-connection index, open-set + # single-flight, tenant host cache and the ban disconnect's + # restriction-cache drop. Infra-free; the database URL + # refuses at once. Keep scripts/run-tests.sh in step. + cargo nextest run -p buzz-relay --lib -E 'test(=state::tests::admitted_index_follows_every_key_change) + test(=state::tests::concurrent_open_channel_misses_share_the_in_flight_read) + test(=tenant::tests::app_state_resolver_serves_fresh_cached_hosts_only) + test(=state::tests::disconnect_drops_the_cached_restriction_row)' # Retain partially consumed mesh receives across housekeeping ticks. cargo nextest run -p buzz-relay --lib -E 'test(=mesh_boot::tests::demo_echo_retains_pending_receive_across_housekeeping_ticks)' # boot_lifecycle spawns the real relay binary and asserts its startup diff --git a/crates/buzz-db/src/runtime/mod.rs b/crates/buzz-db/src/runtime/mod.rs index 78b0b85e8e..cc78e66615 100644 --- a/crates/buzz-db/src/runtime/mod.rs +++ b/crates/buzz-db/src/runtime/mod.rs @@ -355,7 +355,7 @@ impl ReadSession { pub async fn query_events(&mut self, q: &EventQuery) -> Result> { let degraded = match &mut self.inner { ReadSessionInner::Replica { tx, writer } => { - match event::query_events_on(tx, q).await { + match event::query_events_on(tx, q, event::ReadDeadline::Session).await { Ok(rows) => return Ok(rows), // Surface a cancel instead of re-running it on the writer; see // `DbError::is_statement_cancelled` for the trade-off. diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 55e21b5ec8..52296a2087 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -552,7 +552,7 @@ pub(crate) async fn query_events_with_operation( return answer; } let mut conn = crate::observability::acquire_writer(pool, operation).await?; - query_events_on(&mut conn, q).await + query_events_on(&mut conn, q, ReadDeadline::PerRead).await } /// [`query_events`] on a specific session — the replica-routing path runs @@ -561,13 +561,14 @@ pub(crate) async fn query_events_with_operation( pub(crate) async fn query_events_on( conn: &mut sqlx::PgConnection, q: &EventQuery, + deadline: ReadDeadline, ) -> Result> { if let Some(answer) = answer_without_db(q) { return answer; } let mut qb = build_query_events_sql(q); - let rows = fetch_with_read_deadline(conn, &mut qb, !has_tag_values(q)).await?; + let rows = fetch_with_read_deadline(conn, &mut qb, !has_tag_values(q), deadline).await?; let mut out = Vec::with_capacity(rows.len()); for row in rows { @@ -877,6 +878,14 @@ fn push_mention_join(b: &mut QueryBuilder, q: &EventQuery, p_hex } } +/// [`READ_STATEMENT_TIMEOUT`] as a literal, so [`BEGIN_WITH_READ_DEADLINE_SQL`] +/// can splice it into a static statement. +macro_rules! read_statement_timeout { + () => { + "20s" + }; +} + /// Server-side deadline for every generic event read ([`query_events_on`] and /// [`count_events_on`]: REQ history, `/query`, `/count`, aux closures), kept /// under the desktop's 30 s request timeout so a runaway query is cancelled @@ -884,7 +893,7 @@ fn push_mention_join(b: &mut QueryBuilder, q: &EventQuery, p_hex /// effective `statement_timeout`: a shorter operator or caller cap is kept, /// and a disabled (`0`) or longer one is lowered to 20 s. The limit is per /// statement, not per request. -pub const READ_STATEMENT_TIMEOUT: &str = "20s"; +pub const READ_STATEMENT_TIMEOUT: &str = read_statement_timeout!(); /// Lower `statement_timeout` to `$1` unless it is already tighter; `$2` /// selects transaction-local (`true`) or session (`false`) scope. @@ -893,6 +902,31 @@ pub(crate) const TIGHTEN_STATEMENT_TIMEOUT_SQL: &str = "SELECT set_config('state OR current_setting('statement_timeout')::interval > $1::text::interval \ THEN $1::text ELSE current_setting('statement_timeout') END, $2)"; +/// `BEGIN` plus the [`TIGHTEN_STATEMENT_TIMEOUT_SQL`] tighten to +/// [`READ_STATEMENT_TIMEOUT`], transaction-local, as one simple-query message: +/// one round trip instead of two. +const BEGIN_WITH_READ_DEADLINE_SQL: &str = concat!( + "BEGIN; SELECT set_config('statement_timeout', \ + CASE WHEN current_setting('statement_timeout')::interval = interval '0' \ + OR current_setting('statement_timeout')::interval > interval '", + read_statement_timeout!(), + "' THEN '", + read_statement_timeout!(), + "' ELSE current_setting('statement_timeout') END, true)" +); + +/// How an event read runs under [`READ_STATEMENT_TIMEOUT`]. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum ReadDeadline { + /// The session already runs under it: every reader-pool connection + /// tightens its session `statement_timeout` once, at connect (see + /// `Db::read_pool_options`), so the read skips the tighten. + Session, + /// Tighten it for this read only, as writer sessions run under a longer + /// timeout of their own. + PerRead, +} + /// Lower `conn`'s session `statement_timeout` to [`READ_STATEMENT_TIMEOUT`] /// unless it is already tighter. For pools whose every statement is a read: /// run it once per new connection (`after_connect`), so no statement on that @@ -909,21 +943,37 @@ pub async fn tighten_session_read_statement_timeout( } /// Run an event read under [`READ_STATEMENT_TIMEOUT`] without changing the -/// deadline for anything else on `conn`. `begin()` opens a transaction on an -/// autocommit writer connection or a savepoint inside the replica's read -/// transaction; rolling it back (the read has no writes to keep) reverts the -/// transaction-local `set_config` in both cases. +/// deadline for anything else on `conn`. Inside an open transaction the read +/// takes a savepoint, so a failed or cancelled read leaves the transaction +/// usable and the read's locks are released when it ends. With +/// [`ReadDeadline::PerRead`] the savepoint tightens the deadline, and an +/// autocommit connection opens a transaction whose `BEGIN` carries the +/// transaction-local tighten. Rolling either back (the read has no writes to +/// keep) reverts the tighten. async fn fetch_with_read_deadline( conn: &mut sqlx::PgConnection, qb: &mut QueryBuilder, persistent: bool, + deadline: ReadDeadline, ) -> Result> { - let mut tx = sqlx::Connection::begin(&mut *conn).await?; - sqlx::query(TIGHTEN_STATEMENT_TIMEOUT_SQL) - .bind(READ_STATEMENT_TIMEOUT) - .bind(true) - .execute(&mut *tx) - .await?; + use sqlx::Connection as _; + let mut tx = match (conn.is_in_transaction(), deadline) { + (false, ReadDeadline::Session) => { + return Ok(qb.build().persistent(persistent).fetch_all(conn).await?); + } + (false, ReadDeadline::PerRead) => conn.begin_with(BEGIN_WITH_READ_DEADLINE_SQL).await?, + (true, _) => { + let mut tx = conn.begin().await?; + if deadline == ReadDeadline::PerRead { + sqlx::query(TIGHTEN_STATEMENT_TIMEOUT_SQL) + .bind(READ_STATEMENT_TIMEOUT) + .bind(true) + .execute(&mut *tx) + .await?; + } + tx + } + }; let rows = qb .build() .persistent(persistent) @@ -1106,13 +1156,17 @@ pub async fn count_events(pool: &PgPool, q: &EventQuery) -> Result { crate::observability::WriterOperation::SubscriptionHistory, ) .await?; - count_events_on(&mut conn, q).await + count_events_on(&mut conn, q, ReadDeadline::PerRead).await } /// [`count_events`] on a specific session — the replica-routing path runs /// the count on the exact reader connection whose heartbeat observation /// proved its predicate. -pub(crate) async fn count_events_on(conn: &mut sqlx::PgConnection, q: &EventQuery) -> Result { +pub(crate) async fn count_events_on( + conn: &mut sqlx::PgConnection, + q: &EventQuery, + deadline: ReadDeadline, +) -> Result { // Empty list means "match nothing" — return 0 immediately. if q.kinds.as_deref().is_some_and(|k| k.is_empty()) { return Ok(0); @@ -1130,7 +1184,7 @@ pub(crate) async fn count_events_on(conn: &mut sqlx::PgConnection, q: &EventQuer let mut qb = count_events_sql(q); // Same transaction-local deadline as `query_events_on`, so a COUNT cannot // stall indefinitely either. - let row = fetch_with_read_deadline(conn, &mut qb, !has_tag_values(q)) + let row = fetch_with_read_deadline(conn, &mut qb, !has_tag_values(q), deadline) .await? .into_iter() .next() @@ -2394,7 +2448,9 @@ impl Db { .await { crate::RouteDecision::Replica(mut tx, _entry, reason) => { - match crate::event::query_events_on(&mut tx, q).await { + match crate::event::query_events_on(&mut tx, q, crate::event::ReadDeadline::Session) + .await + { Ok(events) => { Self::record_route(path, "replica", reason); Ok(events) @@ -2456,7 +2512,9 @@ impl Db { .await { crate::RouteDecision::Replica(mut tx, _entry, reason) => { - match crate::event::query_events_on(&mut tx, q).await { + match crate::event::query_events_on(&mut tx, q, crate::event::ReadDeadline::Session) + .await + { Ok(events) => { Self::record_route(path, "replica", reason); Ok(events) @@ -2520,7 +2578,9 @@ impl Db { .await { crate::RouteDecision::Replica(mut tx, _entry, reason) => { - match crate::event::count_events_on(&mut tx, q).await { + match crate::event::count_events_on(&mut tx, q, crate::event::ReadDeadline::Session) + .await + { Ok(count) => { Self::record_route(path, "replica", reason); Ok(count) @@ -3046,15 +3106,21 @@ mod postgres_tests { q.kinds = Some(vec![39000]); q.tag_values = vec![("t".to_owned(), vec!["for-you".to_owned()])]; for _ in 0..6 { - query_events_on(&mut conn, &q).await.expect("label query"); - count_events_on(&mut conn, &q).await.expect("label count"); + query_events_on(&mut conn, &q, ReadDeadline::PerRead) + .await + .expect("label query"); + count_events_on(&mut conn, &q, ReadDeadline::PerRead) + .await + .expect("label count"); } assert_eq!(prepared_like(&mut conn, "%tags @> ANY(%").await, 0); // Control: the same query without a tag match stays a cached statement, // so the check above can see one. q.tag_values.clear(); - query_events_on(&mut conn, &q).await.expect("plain query"); + query_events_on(&mut conn, &q, ReadDeadline::PerRead) + .await + .expect("plain query"); assert!(prepared_like(&mut conn, "%FROM events%").await > 0); } @@ -3062,7 +3128,7 @@ mod postgres_tests { /// the value restored on `conn` afterwards. async fn deadline_inside_and_after(conn: &mut sqlx::PgConnection) -> (String, String) { let mut qb = QueryBuilder::new("SELECT current_setting('statement_timeout')"); - let rows = fetch_with_read_deadline(conn, &mut qb, true) + let rows = fetch_with_read_deadline(conn, &mut qb, true, ReadDeadline::PerRead) .await .expect("deadline read"); let inside: String = rows[0].get(0); @@ -3259,7 +3325,7 @@ mod postgres_tests { .expect("set session timeout"); // Autocommit connection (writer path). let mut qb = QueryBuilder::new("SELECT 1/0"); - let err = fetch_with_read_deadline(&mut conn, &mut qb, true).await; + let err = fetch_with_read_deadline(&mut conn, &mut qb, true, ReadDeadline::PerRead).await; assert!(err.is_err(), "division by zero must fail"); let after: String = sqlx::query_scalar("SHOW statement_timeout") .fetch_one(&mut *conn) @@ -3271,7 +3337,7 @@ mod postgres_tests { // the failed savepoint must roll back and leave the outer tx usable. let mut tx = sqlx::Connection::begin(&mut *conn).await.expect("begin"); let mut qb = QueryBuilder::new("SELECT 1/0"); - let err = fetch_with_read_deadline(&mut tx, &mut qb, true).await; + let err = fetch_with_read_deadline(&mut tx, &mut qb, true, ReadDeadline::PerRead).await; assert!(err.is_err(), "division by zero must fail"); let after: String = sqlx::query_scalar("SHOW statement_timeout") .fetch_one(&mut *tx) @@ -3285,6 +3351,47 @@ mod postgres_tests { .expect("reset"); } + /// Inside a replica read session's transaction, `ReadDeadline::Session` + /// skips the tighten (the reader pool installs the deadline at connect) + /// but still reads under a savepoint, so the read's locks end with it and + /// a later DDL on the replica is not blocked by an idle session. + /// `PerRead` in the same transaction still tightens. + /// Mutation: tighten for `Session`, or drop its savepoint → RED. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn session_deadline_reads_skip_the_tighten_and_release_locks() { + let pool = setup_pool().await; + let mut tx = pool.begin().await.expect("begin session"); + sqlx::query("SET LOCAL statement_timeout = '1min'") + .execute(&mut *tx) + .await + .expect("set session timeout"); + let mut qb = QueryBuilder::new( + "SELECT current_setting('statement_timeout'), (SELECT count(*) FROM events)", + ); + let rows = fetch_with_read_deadline(&mut tx, &mut qb, true, ReadDeadline::Session) + .await + .expect("session read"); + let inside: String = rows[0].get(0); + assert_eq!(inside, "1min", "Session must not tighten"); + let held: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_locks \ + WHERE pid = pg_backend_pid() AND relation = 'events'::regclass", + ) + .fetch_one(&mut *tx) + .await + .expect("count locks"); + assert_eq!(held, 0, "a Session read must release its locks"); + + let mut qb = QueryBuilder::new("SELECT current_setting('statement_timeout')"); + let rows = fetch_with_read_deadline(&mut tx, &mut qb, true, ReadDeadline::PerRead) + .await + .expect("per-read read"); + let inside: String = rows[0].get(0); + assert_eq!(inside, READ_STATEMENT_TIMEOUT, "PerRead tightens"); + tx.rollback().await.expect("rollback"); + } + /// Pins the dispatch: a plain (non-`#e`) read — the agent `#p` + `#h` + /// `since` REQ shape that ran away on bb-public — through the production /// `query_events_on` is cancelled at the 20 s deadline even with the @@ -3313,7 +3420,7 @@ mod postgres_tests { let started = std::time::Instant::now(); let result = tokio::time::timeout( std::time::Duration::from_secs(40), - query_events_on(&mut conn, &q), + query_events_on(&mut conn, &q, ReadDeadline::PerRead), ) .await .expect("deadline must fire before the 40 s guard"); @@ -3353,7 +3460,7 @@ mod postgres_tests { let started = std::time::Instant::now(); let result = tokio::time::timeout( std::time::Duration::from_secs(40), - count_events_on(&mut conn, &q), + count_events_on(&mut conn, &q, ReadDeadline::PerRead), ) .await .expect("deadline must fire before the 40 s guard"); diff --git a/crates/buzz-relay/src/api/admin/direct.rs b/crates/buzz-relay/src/api/admin/direct.rs index b262c7827c..a490b75610 100644 --- a/crates/buzz-relay/src/api/admin/direct.rs +++ b/crates/buzz-relay/src/api/admin/direct.rs @@ -118,7 +118,7 @@ async fn direct_action( let target = decode_hex_pubkey(&target_hex)?; let host = buzz_core::tenant::validate_community_host(&q.community_host) .map_err(|msg| ApiError::bad_request("invalid_community_host", msg))?; - let tenant = match crate::tenant::bind_community(&state.db, &host).await { + let tenant = match crate::tenant::bind_community(&*state, &host).await { Ok(t) => t, Err(crate::tenant::BindError::UnmappedHost) => { return Err(ApiError::bad_request( diff --git a/crates/buzz-relay/src/api/admin/mod.rs b/crates/buzz-relay/src/api/admin/mod.rs index 84a59e917b..f6ee03af3c 100644 --- a/crates/buzz-relay/src/api/admin/mod.rs +++ b/crates/buzz-relay/src/api/admin/mod.rs @@ -350,7 +350,7 @@ async fn feedback_attachment( // Resolve the tenant from server-owned feedback provenance, then assert the // resolved row still agrees with the feedback FK. Client input never names // a community, host, object key, extension, or upstream URL. - let tenant = crate::tenant::bind_community(&state.db, community_host) + let tenant = crate::tenant::bind_community(&*state, community_host) .await .map_err(|_| ApiError::not_found())?; if tenant.community().as_uuid() != &community_id { @@ -505,7 +505,7 @@ async fn resolve_report( }; // Bind tenant from server-owned report provenance (never from client input). - let tenant = crate::tenant::bind_community(&state.db, &report_detail.report.community_host) + let tenant = crate::tenant::bind_community(&*state, &report_detail.report.community_host) .await .map_err(|_| ApiError::internal())?; @@ -689,7 +689,7 @@ async fn reopen_report( .ok_or_else(ApiError::not_found)?; // Bind tenant from server-owned report provenance (never from client input). - let tenant = crate::tenant::bind_community(&state.db, &report_detail.report.community_host) + let tenant = crate::tenant::bind_community(&*state, &report_detail.report.community_host) .await .map_err(|_| ApiError::internal())?; @@ -766,7 +766,7 @@ async fn cancel_report( .ok_or_else(ApiError::not_found)?; // Bind tenant from server-owned report provenance (never from client input). - let tenant = crate::tenant::bind_community(&state.db, &report_detail.report.community_host) + let tenant = crate::tenant::bind_community(&*state, &report_detail.report.community_host) .await .map_err(|_| ApiError::internal())?; @@ -1165,7 +1165,7 @@ async fn community_for_host( state: &crate::state::AppState, host: &str, ) -> Result { - match crate::tenant::bind_community(&state.db, host).await { + match crate::tenant::bind_community(state, host).await { Ok(tenant) => Ok(tenant.community()), Err(crate::tenant::BindError::UnmappedHost) => Err(ApiError::bad_request( "unknown_community_host", diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 1d75835ff1..3c6320e044 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -1213,30 +1213,16 @@ async fn submit_event_authed( let is_typing = kind_u32 == buzz_core::kind::KIND_TYPING_INDICATOR; // Enforce relay membership (with NIP-OA fallback via x-auth-tag header). - // Typing pulses arrive every few seconds per working agent, so their ban - // check uses the same short-TTL cache as their write-restriction check. let auth_tag = super::relay_members::extract_auth_tag_header(headers); - let membership = if is_typing { - super::relay_members::enforce_relay_membership_ephemeral( - state, - tenant, - &pubkey_bytes, - auth_tag, - signed_auth_created_at, - token_owner, - ) - .await - } else { - super::relay_members::enforce_relay_membership( - state, - tenant, - &pubkey_bytes, - auth_tag, - signed_auth_created_at, - token_owner, - ) - .await - }; + let membership = super::relay_members::enforce_relay_membership( + state, + tenant, + &pubkey_bytes, + auth_tag, + signed_auth_created_at, + token_owner, + ) + .await; let nip_oa_owner = match membership { Ok(owner) => owner.or_else(|| { if !state.config.require_relay_membership { @@ -2677,7 +2663,7 @@ pub async fn workflow_webhook( .get(axum::http::header::HOST) .and_then(|v| v.to_str().ok()) .unwrap_or(""); - let tenant = crate::tenant::bind_community(&state.db, raw_host) + let tenant = crate::tenant::bind_community(&*state, raw_host) .await .map_err(|_| not_found("workflow not found"))?; let community_id = tenant.community(); diff --git a/crates/buzz-relay/src/api/invites/claim_auth.rs b/crates/buzz-relay/src/api/invites/claim_auth.rs index 756b2ba848..a85973e06a 100644 --- a/crates/buzz-relay/src/api/invites/claim_auth.rs +++ b/crates/buzz-relay/src/api/invites/claim_auth.rs @@ -24,7 +24,7 @@ pub(super) async fn authenticate( .get(axum::http::header::HOST) .and_then(|v| v.to_str().ok()) .unwrap_or(""); - let tenant = crate::tenant::bind_community(&state.db, raw_host) + let tenant = crate::tenant::bind_community(&**state, raw_host) .await .map_err(|_| { api_error( diff --git a/crates/buzz-relay/src/api/mod.rs b/crates/buzz-relay/src/api/mod.rs index d9059d03ae..eb37c91814 100644 --- a/crates/buzz-relay/src/api/mod.rs +++ b/crates/buzz-relay/src/api/mod.rs @@ -332,31 +332,6 @@ pub mod relay_members { auth_tag_header, signed_auth_created_at, token_owner, - false, - ) - .await - } - - /// [`enforce_relay_membership`] for HTTP typing pulses: the ban check - /// reads through the 30-second ephemeral-path restriction cache, the same - /// staleness bound the pulse's own write-restriction check already has. - /// Every other HTTP request reads ban state fresh. - pub async fn enforce_relay_membership_ephemeral( - state: &AppState, - tenant: &TenantContext, - pubkey_bytes: &[u8], - auth_tag_header: Option<&str>, - signed_auth_created_at: Option, - token_owner: Option, - ) -> Result, MembershipRefusal> { - enforce_membership( - state, - tenant, - pubkey_bytes, - auth_tag_header, - signed_auth_created_at, - token_owner, - true, ) .await } @@ -369,7 +344,6 @@ pub mod relay_members { auth_tag_header: Option<&str>, signed_auth_created_at: Option, token_owner: Option, - cached_ban: bool, ) -> Result, MembershipRefusal> { admit_member( state, @@ -378,7 +352,6 @@ pub mod relay_members { auth_tag_header, signed_auth_created_at, token_owner, - cached_ban, ) .await .map_err(|refusal| match token_owner { @@ -397,7 +370,6 @@ pub mod relay_members { auth_tag_header: Option<&str>, signed_auth_created_at: Option, token_owner: Option, - cached_ban: bool, ) -> Result, (StatusCode, Json)> { let community = tenant.community(); let owner = match check_relay_membership( @@ -433,7 +405,6 @@ pub mod relay_members { auth_tag_header, signed_auth_created_at, token_owner, - cached_ban, ) .await?; if let Some(token_owner) = token_owner { @@ -458,7 +429,9 @@ pub mod relay_members { /// Refuse a community-banned principal (own ban or its agent owner's) on /// HTTP. Bans only: a timeout blocks writes, and HTTP writes reach the - /// ingest gate, while reads stay allowed. Fails closed with 503. + /// ingest gate, while reads stay allowed. Each request's grant ends with + /// the request, so this reads through the restriction cache (see + /// [`AppState::restriction_state_cached`]). Fails closed with 503. async fn deny_banned( state: &AppState, community: CommunityId, @@ -466,7 +439,6 @@ pub mod relay_members { auth_tag_header: Option<&str>, signed_auth_created_at: Option, token_owner: Option, - cached: bool, ) -> Result<(), (StatusCode, Json)> { let Ok(pubkey) = nostr::PublicKey::from_slice(pubkey_bytes) else { return Err(super::internal_error("invalid pubkey for ban check")); @@ -478,7 +450,7 @@ pub mod relay_members { auth_tag_header, signed_auth_created_at, token_owner, - cached, + true, ) .await { diff --git a/crates/buzz-relay/src/api/nip05.rs b/crates/buzz-relay/src/api/nip05.rs index 2424f3b0e5..74ec3ad48d 100644 --- a/crates/buzz-relay/src/api/nip05.rs +++ b/crates/buzz-relay/src/api/nip05.rs @@ -36,7 +36,7 @@ pub async fn nostr_nip05( .unwrap_or(""); let json = match ( params.name, - crate::tenant::bind_community(&state.db, raw_host).await, + crate::tenant::bind_community(&*state, raw_host).await, ) { (Some(n), Ok(tenant)) => { let name = n.to_lowercase(); diff --git a/crates/buzz-relay/src/api/typing_postgres_tests.rs b/crates/buzz-relay/src/api/typing_postgres_tests.rs index 6c7c7b1982..4f618876e4 100644 --- a/crates/buzz-relay/src/api/typing_postgres_tests.rs +++ b/crates/buzz-relay/src/api/typing_postgres_tests.rs @@ -163,7 +163,7 @@ async fn timed_out_member_typing_is_rejected() { #[ignore = "requires Postgres and Redis"] async fn typing_uses_cached_restriction_row() { let f = Fixture::new().await; - f.state.restriction_cache.insert( + f.state.seed_restriction( (f.community, f.member.public_key().to_bytes().to_vec()), buzz_db::moderation::RestrictionState { banned: true, @@ -196,7 +196,7 @@ async fn typing_ban_gate_uses_cached_restriction_row() { ) .await .unwrap(); - f.state.restriction_cache.insert( + f.state.seed_restriction( (f.community, member), buzz_db::moderation::RestrictionState { banned: false, @@ -209,14 +209,15 @@ async fn typing_ban_gate_uses_cached_restriction_row() { assert_eq!(status, axum::http::StatusCode::OK, "{body}"); } -/// Only typing uses the cached ban gate: a persistent post reads ban state -/// fresh, so a stale cached ban does not refuse a member Postgres cleared. -/// Mutation: use the cached ban gate for every kind → RED. +/// A persistent post takes the same cached ban gate as typing: a fresh +/// cached ban refuses a member Postgres never restricted. +/// Mutation: read ban state fresh in the HTTP membership gate and in ingest +/// → RED. #[tokio::test] #[ignore = "requires Postgres and Redis"] -async fn persistent_post_ban_gate_reads_fresh() { +async fn persistent_post_ban_gate_uses_cached_restriction_row() { let f = Fixture::new().await; - f.state.restriction_cache.insert( + f.state.seed_restriction( (f.community, f.member.public_key().to_bytes().to_vec()), buzz_db::moderation::RestrictionState { banned: true, @@ -228,8 +229,7 @@ async fn persistent_post_ban_gate_reads_fresh() { .sign_with_keys(&f.member) .unwrap(); let (status, body) = f.post(&f.member, &message).await; - assert_eq!(status, axum::http::StatusCode::OK, "{body}"); - assert_eq!(body["accepted"], true, "{body}"); + assert_eq!(status, axum::http::StatusCode::FORBIDDEN, "{body}"); } /// HTTP typing reads the cached serving state, like the WebSocket path. diff --git a/crates/buzz-relay/src/audio/handler.rs b/crates/buzz-relay/src/audio/handler.rs index 2cbb6b5935..809bb44dae 100644 --- a/crates/buzz-relay/src/audio/handler.rs +++ b/crates/buzz-relay/src/audio/handler.rs @@ -124,7 +124,7 @@ pub async fn ws_audio_handler( .get(axum::http::header::HOST) .and_then(|v| v.to_str().ok()) .unwrap_or(""); - let tenant = match crate::tenant::bind_community(&state.db, raw_host).await { + let tenant = match crate::tenant::bind_community(&*state, raw_host).await { Ok(ctx) => ctx, Err(_) => { return ( diff --git a/crates/buzz-relay/src/handlers/auth.rs b/crates/buzz-relay/src/handlers/auth.rs index b2bd26efe7..626d114dfd 100644 --- a/crates/buzz-relay/src/handlers/auth.rs +++ b/crates/buzz-relay/src/handlers/auth.rs @@ -63,9 +63,10 @@ fn classify_relay_membership( /// on the principal blocks it directly; if the principal is clear, a ban on /// its proven owner (extracted from the self-proving auth tag) blocks it too. /// -/// `cached` reads the restriction rows through the 30-second ephemeral-path -/// cache ([`AppState::restriction_state_cached`]); only per-pulse ephemeral -/// traffic sets it. Socket auth and every other HTTP request read fresh. +/// `cached` reads the restriction rows through +/// [`AppState::restriction_state_cached`], for per-request gates whose grant +/// ends with the request (HTTP requests, ephemeral pulses). Socket admission +/// reads fresh: an admitted socket outlives the cache's staleness bound. pub(crate) async fn community_ban_outcome( state: &AppState, community: buzz_core::CommunityId, diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 76ffc21f7f..ea696e5a85 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -715,19 +715,15 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc return Err(IngestError::Rejected(format!("invalid: {e}"))), Err(_) => return Err(IngestError::Internal("error: internal error".into())), } - // Same short-TTL caches as the WebSocket ephemeral path: this runs per - // typing pulse, and persistent ingest keeps the uncached durable fence. - super::ingest::enforce_cached_write_restriction( - state, - tenant, - KIND_TYPING_INDICATOR, - &auth_pubkey, - ) - .await?; + // Same restriction gate and serving-fence cache as the WebSocket + // ephemeral path: this runs per typing pulse. + super::ingest::enforce_write_restriction(state, tenant, KIND_TYPING_INDICATOR, &auth_pubkey) + .await?; super::ingest::map_serving_fence_state( state.is_serving_active_cached(tenant.community()).await, )?; @@ -3870,7 +3861,7 @@ mod tests { let (state, tenant, user) = unrestricted_fixture().await; let key = (tenant.community(), user.public_key().to_bytes().to_vec()); - state.restriction_cache.insert( + state.seed_restriction( key.clone(), buzz_db::moderation::RestrictionState { banned: true, @@ -3883,7 +3874,7 @@ mod tests { "a cached ban must refuse the ephemeral event; got {frame}" ); - state.restriction_cache.insert( + state.seed_restriction( key, buzz_db::moderation::RestrictionState { banned: false, @@ -3924,7 +3915,8 @@ mod tests { assert_eq!( state .restriction_cache - .get(&(tenant.community(), user.public_key().to_bytes().to_vec())), + .get(&(tenant.community(), user.public_key().to_bytes().to_vec())) + .map(|cached| cached.state), Some(buzz_db::moderation::RestrictionState::default()) ); } @@ -3935,42 +3927,45 @@ mod tests { .expect("sign text note") } - /// Stored writes, and NIP-43 leave, skip the cached restriction row: - /// a cached ban on a sender Postgres has never restricted refuses - /// neither. The leave reaches ingest's later membership check, past - /// its uncached restriction check. - /// Mutation: make ingest call `enforce_cached_write_restriction`, or - /// drop the 28936 carve-out from the `handle_event` gate → RED. + /// Stored writes read the restriction through the cache, and only + /// while the entry is younger than `RESTRICTION_CACHE_MAX_AGE`: a + /// fresh cached ban on a sender Postgres has never restricted refuses + /// a stored write, and the same row past that age is reread and + /// admits it. + /// Mutation: make ingest read the restriction uncached, or drop the + /// max-age check from `restriction_state_cached` → RED. #[tokio::test] #[ignore = "requires Postgres — runs in postgres-ci nextest lane"] - async fn stored_writes_ignore_cached_ban() { + async fn stored_writes_honor_cached_ban_until_it_ages_out() { let (state, tenant, user) = unrestricted_fixture().await; - state.restriction_cache.insert( - (tenant.community(), user.public_key().to_bytes().to_vec()), - buzz_db::moderation::RestrictionState { - banned: true, - muted_until: None, - }, - ); + let key = (tenant.community(), user.public_key().to_bytes().to_vec()); + let banned = buzz_db::moderation::RestrictionState { + banned: true, + muted_until: None, + }; let scopes = buzz_auth::Scope::all_non_admin(); + state.seed_restriction(key.clone(), banned.clone()); let frame = ok_frame_scoped(&state, &tenant, &user, text_note(&user), scopes.clone()).await; assert!( - frame.contains(",true,"), - "a stored write must read the restriction from Postgres; got {frame}" + frame.contains("blocked: you are banned"), + "a fresh cached ban must refuse a stored write; got {frame}" ); - let leave = nostr::EventBuilder::new( - nostr::Kind::Custom(buzz_core::kind::KIND_NIP43_LEAVE_REQUEST as u16), - "", - ) - .sign_with_keys(&user) - .expect("sign leave"); - let frame = ok_frame_scoped(&state, &tenant, &user, leave, scopes).await; + state.restriction_cache.insert( + key, + crate::state::CachedRestriction { + state: banned, + read_at: std::time::Instant::now() + .checked_sub(crate::state::RESTRICTION_CACHE_MAX_AGE) + .expect("instant past max age"), + }, + ); + let frame = ok_frame_scoped(&state, &tenant, &user, text_note(&user), scopes).await; assert!( - frame.contains("relay membership is not enabled"), - "NIP-43 leave must skip the cached gate and pass ingest's uncached one; got {frame}" + frame.contains(",true,"), + "an aged-out cached ban must be reread from Postgres; got {frame}" ); } diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 9cae758646..b2031936ab 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -2374,39 +2374,16 @@ fn write_restriction_denial( /// by its owner's (`users.agent_owner_pubkey`). A ban is normally enforced by /// the auth seam and the live disconnect; this gate is the durable backstop /// for a missed disconnect and for HTTP writes. A timeout has no auth-seam -/// presence, so this is where it is enforced. Fails closed on a DB error. +/// presence, so this is where it is enforced. It runs per write, so it reads +/// through [`AppState::restriction_state_cached`]: a restriction change takes +/// effect within [`crate::state::RESTRICTION_CACHE_MAX_AGE`]. The verdict is +/// still taken at `now`, so a cached timeout lifts on time. Fails closed on a +/// DB error. pub(crate) async fn enforce_write_restriction( state: &AppState, tenant: &TenantContext, kind: u32, pubkey: &nostr::PublicKey, -) -> Result<(), IngestError> { - #[cfg(test)] - crate::nip_fi_test_hooks::before_write_ban_check(tenant.community()).await; - let restriction = state - .db - .moderation_restriction_state(tenant.community(), pubkey.as_bytes()) - .await - .map_err(|e| { - IngestError::ban( - BanRefusal::Unavailable, - IngestError::Internal(format!( - "error: internal error checking restriction state: {e}" - )), - ) - })?; - write_restriction_denial(kind, &restriction, Utc::now()).map_or(Ok(()), Err) -} - -/// [`enforce_write_restriction`] over a 30-second cached restriction row, for -/// the WebSocket ephemeral and observer paths. Those run per typing pulse and -/// observer frame; persistent ingest keeps the uncached read. The verdict is -/// still taken at `now`, so a cached timeout lifts on time. -pub(crate) async fn enforce_cached_write_restriction( - state: &AppState, - tenant: &TenantContext, - kind: u32, - pubkey: &nostr::PublicKey, ) -> Result<(), IngestError> { #[cfg(test)] crate::nip_fi_test_hooks::before_write_ban_check(tenant.community()).await; @@ -7558,6 +7535,11 @@ mod postgres_tests { .unban_community_member(tenant.community(), &owner_bytes, &owner_bytes) .await .expect("unban owner"); + // The unban command clears the owner's entry; the agent's entry + // carries the owner's ban until it ages out + // (`RESTRICTION_CACHE_MAX_AGE`). + state.invalidate_restriction_cache(tenant.community(), &owner_bytes); + state.invalidate_restriction_cache(tenant.community(), &agent.public_key().to_bytes()); enforce_write_restriction(&state, &tenant, KIND_REACTION, &agent.public_key()) .await .expect("unbanning the owner restores the agent"); @@ -7605,6 +7587,8 @@ mod postgres_tests { .unban_community_member(tenant.community(), &owner_bytes, &owner_bytes) .await .expect("unban owner"); + // As the unban command does on its own pod. + state.invalidate_restriction_cache(tenant.community(), &owner_bytes); crate::api::relay_members::enforce_relay_membership( &state, &tenant, diff --git a/crates/buzz-relay/src/handlers/moderation_commands.rs b/crates/buzz-relay/src/handlers/moderation_commands.rs index a5a3587b1a..c51c247647 100644 --- a/crates/buzz-relay/src/handlers/moderation_commands.rs +++ b/crates/buzz-relay/src/handlers/moderation_commands.rs @@ -199,7 +199,6 @@ async fn handle_ban( ) .await .map_err(|e| error(format!("database error: {e}")))?; - state.invalidate_restriction_cache(tenant.community(), &target); // Live enforcement: close open sessions for the banned principal now — // this pod's sockets synchronously (fenced to this community) and every diff --git a/crates/buzz-relay/src/handlers/relay_admin.rs b/crates/buzz-relay/src/handlers/relay_admin.rs index 83c9138e1f..94d38fae30 100644 --- a/crates/buzz-relay/src/handlers/relay_admin.rs +++ b/crates/buzz-relay/src/handlers/relay_admin.rs @@ -1477,7 +1477,7 @@ mod postgres_tests { vec![Tag::public_key(target.public_key())], ), ] { - state.restriction_cache.insert( + state.seed_restriction( key.clone(), buzz_db::moderation::RestrictionState::default(), ); @@ -1487,7 +1487,7 @@ mod postgres_tests { .await .unwrap_or_else(|e| panic!("kind {} must apply: {e:?}", event.kind.as_u16())); assert_eq!( - state.restriction_cache.get(&key), + state.restriction_cache.get(&key).map(|cached| cached.state), None, "kind {} must drop the cached row", event.kind.as_u16() diff --git a/crates/buzz-relay/src/nip11.rs b/crates/buzz-relay/src/nip11.rs index 0f880fb5e5..9c55c150b9 100644 --- a/crates/buzz-relay/src/nip11.rs +++ b/crates/buzz-relay/src/nip11.rs @@ -398,7 +398,7 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st if state.config.artifact_incremental_enabled { info.artifacts["incremental"] = crate::api::artifact_sync::capability(); } - if let Ok(tenant) = crate::tenant::bind_community(&state.db, raw_host).await { + if let Ok(tenant) = crate::tenant::bind_community(state, raw_host).await { if let Some(profile) = crate::verified_profiles::discovery(state, &tenant).await { if let Some(fi) = info.federated_identity.as_mut() { fi["public_profile"] = profile; @@ -423,7 +423,7 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st })); } let tenant_host = if state.config.push_enabled { - crate::tenant::bind_community(&state.db, raw_host) + crate::tenant::bind_community(state, raw_host) .await .ok() .map(|tenant| tenant.host().to_owned()) @@ -455,9 +455,7 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st /// `None` (no `icon` field): NIP-11 is intentionally served to unmapped hosts /// too, and an icon lookup failure must not break that. async fn workspace_icon_for_host(state: &crate::state::AppState, raw_host: &str) -> Option { - let tenant = crate::tenant::bind_community(&state.db, raw_host) - .await - .ok()?; + let tenant = crate::tenant::bind_community(state, raw_host).await.ok()?; state .db .get_community_icon(tenant.community()) diff --git a/crates/buzz-relay/src/nip_fi_http.rs b/crates/buzz-relay/src/nip_fi_http.rs index 6ff8a653ea..1578050ef8 100644 --- a/crates/buzz-relay/src/nip_fi_http.rs +++ b/crates/buzz-relay/src/nip_fi_http.rs @@ -460,7 +460,7 @@ async fn observe_owner( .get(axum::http::header::HOST) .and_then(|v| v.to_str().ok()) .unwrap_or(""); - let tenant = match crate::tenant::bind_community(&state.db, host).await { + let tenant = match crate::tenant::bind_community(state, host).await { Ok(tenant) => tenant, Err(crate::tenant::BindError::UnmappedHost) => return None, Err(crate::tenant::BindError::Lookup(_)) => { diff --git a/crates/buzz-relay/src/nip_fi_resolver/tests/router_tests.rs b/crates/buzz-relay/src/nip_fi_resolver/tests/router_tests.rs index eaa8974b01..c92731e003 100644 --- a/crates/buzz-relay/src/nip_fi_resolver/tests/router_tests.rs +++ b/crates/buzz-relay/src/nip_fi_resolver/tests/router_tests.rs @@ -702,6 +702,9 @@ mod postgres_tests { if let Some((start, at)) = crossing { self.clock.set(start); *self.advance_on_acquire.lock().unwrap() = Some(at); + // The clock moves when the tenant lookup acquires a + // connection, so the lookup must miss the host cache. + self.pod.state.tenant_host_cache.invalidate_all(); } outcomes.push( self.pod diff --git a/crates/buzz-relay/src/nip_fi_shadow.rs b/crates/buzz-relay/src/nip_fi_shadow.rs index ca98b5d14e..b87e7dd4a3 100644 --- a/crates/buzz-relay/src/nip_fi_shadow.rs +++ b/crates/buzz-relay/src/nip_fi_shadow.rs @@ -318,9 +318,7 @@ pub(crate) async fn bind_tenant( .get(axum::http::header::HOST) .and_then(|v| v.to_str().ok()) .unwrap_or(""); - let tenant = crate::tenant::bind_community(&state.db, raw_host) - .await - .ok(); + let tenant = crate::tenant::bind_community(state, raw_host).await.ok(); if tenant.is_none() { observe_unbound(state, headers).await; } diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index adc19fe506..6800ca7686 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -735,7 +735,7 @@ async fn nip11_or_ws_handler( // // The NIP-FI upgrade gate runs above (before bind_community) so denied // upgrades skip the tenant lookup; the `Ok(ws)` backstop runs below. - let tenant = match crate::tenant::bind_community(&state.db, raw_host).await { + let tenant = match crate::tenant::bind_community(&*state, raw_host).await { Ok(ctx) => ctx, Err(_) => { // Generic rejection: do not distinguish "unmapped" from "lookup diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 5c4c55e108..70148f6f95 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -693,6 +693,12 @@ where /// Tracks active Nostr WebSocket connections and provides message routing by connection ID. pub struct ConnectionManager { connections: DashMap, + /// Admitted connections per (community, authenticated pubkey), so a + /// disconnect can ask whether the key is still online without scanning + /// every connection. Kept in step with each entry's `admitted` flag and + /// `authenticated_pubkey`, which change only under that pubkey's write + /// lock; a key is removed when its count reaches zero. + admitted_by_pubkey: DashMap<(CommunityId, Vec), usize>, /// Sticky drain flag set by [`Self::drain_all`]. Registrations that land /// after the drain snapshot self-signal, so no upgrade-vs-shutdown /// interleaving can produce a connection that misses the restart close. @@ -704,6 +710,7 @@ impl ConnectionManager { pub fn new() -> Self { Self { connections: DashMap::new(), + admitted_by_pubkey: DashMap::new(), draining: AtomicBool::new(false), } } @@ -765,7 +772,34 @@ impl ConnectionManager { /// Removes a connection from the registry. pub fn deregister(&self, conn_id: Uuid) { - self.connections.remove(&conn_id); + if let Some((_, entry)) = self.connections.remove(&conn_id) { + let pubkey = entry + .authenticated_pubkey + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if entry.admitted.load(Ordering::Acquire) { + self.unindex_admitted(entry.community_id, pubkey.as_deref()); + } + } + } + + fn index_admitted(&self, community_id: CommunityId, pubkey: Option<&[u8]>) { + if let Some(pubkey) = pubkey { + *self + .admitted_by_pubkey + .entry((community_id, pubkey.to_vec())) + .or_insert(0) += 1; + } + } + + fn unindex_admitted(&self, community_id: CommunityId, pubkey: Option<&[u8]>) { + if let Some(pubkey) = pubkey { + self.admitted_by_pubkey + .remove_if_mut(&(community_id, pubkey.to_vec()), |_, count| { + *count = count.saturating_sub(1); + *count == 0 + }); + } } /// Record the authenticated pubkey for a connection after NIP-42 succeeds. @@ -788,6 +822,10 @@ impl ConnectionManager { if let Ok(mut session_slot) = entry.nip_fi.write() { *session_slot = nip_fi; } + if entry.admitted.load(Ordering::Acquire) { + self.unindex_admitted(entry.community_id, slot.as_deref()); + self.index_admitted(entry.community_id, Some(&pubkey_bytes)); + } *slot = Some(pubkey_bytes); } } @@ -804,21 +842,21 @@ impl ConnectionManager { /// Mark `conn_id` admitted, once AUTH has succeeded. pub fn mark_admitted(&self, conn_id: Uuid) { if let Some(entry) = self.connections.get(&conn_id) { - entry.admitted.store(true, Ordering::Release); + let pubkey = entry + .authenticated_pubkey + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if !entry.admitted.swap(true, Ordering::AcqRel) { + self.index_admitted(entry.community_id, pubkey.as_deref()); + } } } /// Whether `pubkey_bytes` has an admitted connection in one community on /// this pod. Sockets still mid-admission do not count. pub fn has_admitted_connection(&self, community_id: CommunityId, pubkey_bytes: &[u8]) -> bool { - self.connections.iter().any(|entry| { - entry.community_id == community_id - && entry.admitted.load(Ordering::Acquire) - && entry - .authenticated_pubkey - .read() - .is_ok_and(|value| value.as_deref() == Some(pubkey_bytes)) - }) + self.admitted_by_pubkey + .contains_key(&(community_id, pubkey_bytes.to_vec())) } /// Return live connection IDs authenticated as `pubkey_bytes` in one community. @@ -1205,6 +1243,23 @@ impl Default for ConnectionManager { /// here. pub const OPEN_CHANNELS_MAX_AGE: std::time::Duration = std::time::Duration::from_secs(1); +/// How long a pod serves a cached ban/timeout row after the database read +/// that produced it began (see [`AppState::restriction_cache`]). +pub const RESTRICTION_CACHE_MAX_AGE: std::time::Duration = std::time::Duration::from_secs(1); + +/// How long a pod serves a cached host → community mapping after the +/// database read that produced it began (see [`AppState::tenant_host_cache`]). +pub const TENANT_HOST_CACHE_MAX_AGE: std::time::Duration = std::time::Duration::from_secs(1); + +/// A cached restriction row and when the read that produced it began. +#[derive(Debug, Clone)] +pub struct CachedRestriction { + /// The restriction row as read. + pub state: buzz_db::moderation::RestrictionState, + /// When the database read that produced it began. + pub read_at: Instant, +} + /// A community's listed open channel IDs and when the read that listed them /// began. #[derive(Debug)] @@ -1305,6 +1360,10 @@ pub struct AppState { /// the community's accessible channels, and used only for /// [`OPEN_CHANNELS_MAX_AGE`] after its database read began. pub open_channels_cache: Arc>>, + /// Per-community gate held while a pod reads a community's open set, so + /// concurrent misses wait for that read instead of each issuing their own + /// (see [`AppState::get_accessible_channel_ids_cached`]). + pub open_channels_reads: Arc>>>, /// Per-community channel visibility string, used to gate the private-channel fan-out /// access check so open channels stay zero-cost. Invalidated on a flip. pub channel_visibility_cache: Arc>, @@ -1394,18 +1453,26 @@ pub struct AppState { /// uncached read as its durable write fence; on the ephemeral path archive /// also disconnects the community's live sockets. pub serving_active_cache: Arc>, - /// Ephemeral-path cache of the raw ban/timeout row, not the verdict, so a - /// timeout still lifts the moment `muted_until` passes. (`banned` is - /// computed at read time, so an expiring ban can outlive its expiry by up - /// to the TTL.) Key: (community, pubkey bytes). The 30s TTL is the only - /// staleness bound: on every pod and for every restriction writer, the - /// ephemeral path sees a ban/timeout change within 30s. As a best effort, - /// signed moderation commands (9040–9043) also drop the target's own entry - /// on the pod that handles them; owned agents' entries, other pods, and the - /// admin/report paths just age out. Persistent ingest stays uncached. + /// Cache of the raw ban/timeout row, not the verdict, so a timeout still + /// lifts the moment `muted_until` passes. (`banned` is computed at read + /// time, so an expiring ban can outlive its expiry by up to + /// [`RESTRICTION_CACHE_MAX_AGE`].) Key: (community, pubkey bytes). An entry + /// is served only for [`RESTRICTION_CACHE_MAX_AGE`] after its read began, + /// which is the only staleness bound: on every pod and for every + /// restriction writer, a cached reader sees a ban/timeout change within + /// that long. As a best effort, signed moderation commands (9040–9043) + /// also drop the target's own entry on the pod that handles them; owned + /// agents' entries, other pods, and the admin/report paths just age out. + /// Socket admission and moderation-command actors read fresh: a grant that + /// outlives the bound must not rest on a cached verdict. #[allow(clippy::type_complexity)] - pub restriction_cache: - Arc), buzz_db::moderation::RestrictionState>>, + pub restriction_cache: Arc), CachedRestriction>>, + /// Normalized request host → community, from the row-zero host binding + /// ([`crate::tenant::bind_community`]). Only mapped hosts are cached, and + /// an entry is served only for [`TENANT_HOST_CACHE_MAX_AGE`] after its read + /// began: a host remapped, archived or deleted elsewhere resolves to its + /// old community for at most that long. + pub tenant_host_cache: Arc>, /// Runtime conformance tracer. Production binds [`crate::conformance::NoopTracer`] /// (zero cost). Conformance tests bind [`crate::conformance::JsonlTracer`] to @@ -1626,6 +1693,12 @@ impl AppState { .time_to_live(std::time::Duration::from_secs(10)) .build(), ), + open_channels_reads: Arc::new( + moka::sync::Cache::builder() + .max_capacity(10_000) + .time_to_idle(std::time::Duration::from_secs(60)) + .build(), + ), channel_visibility_cache: Arc::new( moka::sync::Cache::builder() .max_capacity(10_000) @@ -1683,10 +1756,18 @@ impl AppState { .time_to_live(std::time::Duration::from_secs(10)) .build(), ), + // Readers use an entry only for `RESTRICTION_CACHE_MAX_AGE`; the + // TTL just reclaims memory. restriction_cache: Arc::new( moka::sync::Cache::builder() .max_capacity(100_000) - .time_to_live(std::time::Duration::from_secs(30)) + .time_to_live(std::time::Duration::from_secs(10)) + .build(), + ), + tenant_host_cache: Arc::new( + moka::sync::Cache::builder() + .max_capacity(10_000) + .time_to_live(std::time::Duration::from_secs(10)) .build(), ), // Default to NoopTracer: production builds pay zero cost. @@ -1811,8 +1892,10 @@ impl AppState { Ok(result) } - /// `moderation_restriction_state` with a 30-second cache, for the - /// ephemeral path only. Errors are not cached. + /// `moderation_restriction_state` served from [`Self::restriction_cache`] + /// for [`RESTRICTION_CACHE_MAX_AGE`] after the read that produced it + /// began. For per-request gates (HTTP requests, publishes, ephemeral + /// pulses) whose grant ends with the request. Errors are not cached. pub async fn restriction_state_cached( &self, community_id: CommunityId, @@ -1820,21 +1903,32 @@ impl AppState { ) -> Result { let key = (community_id, pubkey.to_vec()); if let Some(cached) = self.restriction_cache.get(&key) { - metrics::counter!("buzz_restriction_cache_hits_total").increment(1); - return Ok(cached); + if cached.read_at.elapsed() < RESTRICTION_CACHE_MAX_AGE { + metrics::counter!("buzz_restriction_cache_hits_total").increment(1); + return Ok(cached.state); + } } metrics::counter!("buzz_restriction_cache_misses_total").increment(1); - let result = self + let read_at = Instant::now(); + let state = self .db .moderation_restriction_state(community_id, pubkey) .await?; - self.restriction_cache.insert(key, result.clone()); - Ok(result) + self.restriction_cache.insert( + key, + CachedRestriction { + state: state.clone(), + read_at, + }, + ); + Ok(state) } - /// Best-effort drop of a pubkey's cached restriction row after a signed - /// ban, unban, timeout or untimeout. This pod and this key only; the - /// 30-second TTL is the actual bound everywhere else. + /// Best-effort drop of a pubkey's cached restriction row on this pod. + /// A ban drops it on every pod, for the target and its agents, through + /// [`Self::disconnect_pubkey_local`]; an unban, timeout or untimeout only + /// on the pod that handles it. A read in flight at the drop can refill + /// the old row; [`RESTRICTION_CACHE_MAX_AGE`] bounds that and the rest. pub fn invalidate_restriction_cache(&self, community_id: CommunityId, pubkey: &[u8]) { self.restriction_cache .invalidate(&(community_id, pubkey.to_vec())); @@ -2004,6 +2098,11 @@ impl AppState { reason: &str, unowned_only: bool, ) -> usize { + // A ban's disconnect runs here on every pod, for the target and each + // agent it owns, so the pod's next HTTP request or write rereads the + // ban. A read already in flight can still refill the old row, and the + // fan-out is best-effort; `RESTRICTION_CACHE_MAX_AGE` bounds both. + self.invalidate_restriction_cache(community, pubkey); self.conn_manager .disconnect_pubkey(community, pubkey, event_id, reason, unowned_only) + self @@ -2359,30 +2458,78 @@ impl AppState { ids } }; - let open = match self.open_channels_cache.get(&community_id) { - Some(open) if open.read_at.elapsed() < OPEN_CHANNELS_MAX_AGE => { - metrics::counter!("buzz_open_channels_cache_hits_total").increment(1); - open - } - cached => { - let reason = if cached.is_some() { - "expired" - } else { - "absent" - }; - metrics::counter!("buzz_open_channels_cache_misses_total", "reason" => reason) - .increment(1); - let read_at = Instant::now(); - let ids = self.db.get_listed_open_channel_ids(community_id).await?; - let open = Arc::new(OpenChannels { ids, read_at }); - self.open_channels_cache.insert(community_id, open.clone()); - open - } - }; + let open = self.open_channels(community_id).await?; buzz_db::channel_members::merge_channel_ids(&mut ids, &open.ids); Ok(ids) } + /// The community's open set from [`Self::open_channels_cache`] while it + /// is younger than [`OPEN_CHANNELS_MAX_AGE`], else from a fresh read. + /// + /// Concurrent misses share one read: the first takes the community's + /// gate in [`Self::open_channels_reads`] and reads; the rest wait for it + /// to finish and use its set. A waiter whose leader failed or was + /// cancelled reads on its own rather than queueing behind another leader. + /// A waiter also stops waiting after [`OPEN_CHANNELS_MAX_AGE`] and reads + /// on its own: the leader's set would be stale on arrival by then, so a + /// leader whose read hangs stalls only its own request. + async fn open_channels( + &self, + community_id: CommunityId, + ) -> Result, buzz_db::DbError> { + let fresh = || { + self.open_channels_cache + .get(&community_id) + .filter(|open| open.read_at.elapsed() < OPEN_CHANNELS_MAX_AGE) + }; + if let Some(open) = fresh() { + metrics::counter!("buzz_open_channels_cache_hits_total").increment(1); + return Ok(open); + } + let gate = self + .open_channels_reads + .get_with(community_id, || Arc::new(tokio::sync::Mutex::new(()))); + let _leader = match gate.try_lock() { + Ok(guard) => Some(guard), + Err(_) => { + drop(tokio::time::timeout(OPEN_CHANNELS_MAX_AGE, gate.lock()).await); + if let Some(open) = fresh() { + metrics::counter!("buzz_open_channels_cache_hits_total").increment(1); + return Ok(open); + } + None + } + }; + let reason = if self.open_channels_cache.contains_key(&community_id) { + "expired" + } else { + "absent" + }; + metrics::counter!("buzz_open_channels_cache_misses_total", "reason" => reason).increment(1); + let read_at = Instant::now(); + let ids = self.db.get_listed_open_channel_ids(community_id).await?; + let open = Arc::new(OpenChannels { ids, read_at }); + self.open_channels_cache.insert(community_id, open.clone()); + Ok(open) + } + + /// Seeds a fresh restriction-cache entry for `key` without touching the + /// database. + #[cfg(test)] + pub(crate) fn seed_restriction( + &self, + key: (CommunityId, Vec), + state: buzz_db::moderation::RestrictionState, + ) { + self.restriction_cache.insert( + key, + CachedRestriction { + state, + read_at: Instant::now(), + }, + ); + } + /// Seeds both accessible-channel caches so `pubkey` reads exactly /// `channels`, with no open channels, without touching the database. #[cfg(test)] @@ -3370,6 +3517,140 @@ pub(crate) mod tests { use super::*; use buzz_core::channel::{ChannelType, ChannelVisibility}; + /// Open-set reads park on a lock on `channels`, where the test counts + /// them; waiters are counted as they arrive by the clones of the + /// community's gate they hold. A leader's eight waiters share its + /// set. When the leader is cancelled, its orphans each pass the gate + /// and read, none holding it across its own read. When the leader's + /// read hangs, its waiters stop waiting after + /// [`OPEN_CHANNELS_MAX_AGE`] and each read on their own. + /// Mutations: drop the waiter path (distinct sets in the first + /// phase), hold the gate across a waiter's own read (the gate never + /// frees in the second), or wait on the gate with no limit (the third + /// never gets past one read) → RED. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn open_set_misses_share_one_read_and_orphans_read_concurrently() { + let database_url = crate::test_support::database_url(); + let state = test_state_with_database_url_and_pool_options( + &database_url, + sqlx::postgres::PgPoolOptions::new().max_connections(16), + ) + .await; + let admin = sqlx::PgPool::connect(&database_url).await.unwrap(); + let community = state + .db + .ensure_configured_community(&format!("open-set-sf-{}.local", Uuid::new_v4())) + .await + .unwrap() + .id; + let parked = || async { + sqlx::query_scalar::<_, i64>( + "SELECT count(*) FROM pg_stat_activity \ + WHERE datname = current_database() AND wait_event_type = 'Lock' \ + AND query LIKE '%FROM channels%'", + ) + .fetch_one(&admin) + .await + .unwrap() + }; + // Polls `ready` for up to a generous watchdog; timing is not + // part of what the test checks. + async fn eventually(what: &str, mut ready: impl AsyncFnMut() -> bool) { + let start = Instant::now(); + while start.elapsed() < std::time::Duration::from_secs(10) { + if ready().await { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + panic!("timed out waiting for {what}"); + } + let parks = |want: i64| { + eventually("parked open-set reads", async move || { + parked().await >= want + }) + }; + let misses = |n: usize| { + (0..n) + .map(|_| { + let state = Arc::clone(&state); + tokio::spawn(async move { state.open_channels(community).await }) + }) + .collect::>() + }; + let lock = || async { + let mut locker = admin.begin().await.unwrap(); + sqlx::query("LOCK TABLE channels IN ACCESS EXCLUSIVE MODE") + .execute(&mut *locker) + .await + .unwrap(); + locker + }; + + // Starts a leader whose read parks on `locker`, then eight + // misses that have reached its gate. + let cohort = || async { + let locker = lock().await; + let leader = misses(1).remove(0); + parks(1).await; + let gate = state + .open_channels_reads + .get(&community) + .expect("the leader's gate"); + let base = Arc::strong_count(&gate); + let waiters = misses(8); + eventually("eight waiters at the gate", async || { + Arc::strong_count(&gate) >= base + 8 + }) + .await; + (locker, leader, gate, waiters) + }; + + let (locker, leader, _gate, waiters) = cohort().await; + assert_eq!(parked().await, 1, "waiters share the leader's read"); + locker.commit().await.unwrap(); + let mut sets = vec![leader.await.expect("join").expect("open set")]; + for waiter in waiters { + sets.push(waiter.await.expect("join").expect("open set")); + } + assert!( + sets.iter().all(|set| Arc::ptr_eq(set, &sets[0])), + "every miss gets the one read's set" + ); + + state.open_channels_cache.invalidate(&community); + let (locker, leader, gate, orphans) = cohort().await; + leader.abort(); + // Each orphan passes the gate and reads while the lock still holds + // every read, so none may keep the gate across its own read. + eventually("the gate to free", async || gate.try_lock().is_ok()).await; + // The cancelled leader's backend stays parked, plus one per orphan. + parks(9).await; + locker.commit().await.unwrap(); + for orphan in orphans { + orphan + .await + .expect("join") + .expect("an orphan reads on its own"); + } + + state.open_channels_cache.invalidate(&community); + let (locker, hung, _gate, waiters) = cohort().await; + // The leader's read never returns while the lock is held; each + // waiter gives up on it and its own read parks too. + parks(9).await; + assert!(!hung.is_finished(), "the leader's read is still in flight"); + locker.commit().await.unwrap(); + hung.await.expect("join").expect("the leader reads"); + for waiter in waiters { + waiter + .await + .expect("join") + .expect("a waiter reads on its own"); + } + } + /// Cross-pod invalidation is best-effort. With it withheld, a pod may /// grant a channel that left the open set on another pod only until /// its cached open set is [`OPEN_CHANNELS_MAX_AGE`] old; after that, @@ -3770,6 +4051,46 @@ pub(crate) mod tests { assert_eq!(ids, vec![both, open]); } + /// Misses that arrive while a read of the community's open set is in + /// flight wait for it and share its set. The database is unreachable, so + /// any read of a waiter's own fails. (The leaderless case is + /// `open_set_misses_share_one_read_and_orphans_read_concurrently`.) + /// Mutation: drop the waiter path → RED. + #[tokio::test] + async fn concurrent_open_channel_misses_share_the_in_flight_read() { + let state = + test_state_with_unreachable_database_url("postgres://buzz@127.0.0.1:1/buzz").await; + let community = CommunityId::from_uuid(Uuid::from_u128(0xAAAA)); + let open = Uuid::from_u128(9); + let waiters = |state: &Arc| { + (0..8) + .map(|_| { + let state = Arc::clone(state); + tokio::spawn(async move { state.open_channels(community).await }) + }) + .collect::>() + }; + + let gate = state + .open_channels_reads + .get_with(community, || Arc::new(tokio::sync::Mutex::new(()))); + let leader = gate.lock().await; + let shared = waiters(&state); + tokio::task::yield_now().await; + state.open_channels_cache.insert( + community, + Arc::new(OpenChannels { + ids: vec![open], + read_at: Instant::now(), + }), + ); + drop(leader); + for waiter in shared { + let set = waiter.await.expect("join").expect("the leader's set"); + assert_eq!(set.ids, vec![open]); + } + } + #[tokio::test] async fn channel_deleted_invalidation_is_scoped_to_community() { let state = test_state().await; @@ -4424,6 +4745,90 @@ pub(crate) mod tests { ); } + /// A ban's disconnect, local or from another pod's conn-control + /// publish, drops that pubkey's cached restriction row in that + /// community, so the pod's next HTTP request or write rereads the ban. + /// Mutation: drop the invalidation from `disconnect_pubkey_local` → RED. + #[tokio::test] + async fn disconnect_drops_the_cached_restriction_row() { + let state = + test_state_with_unreachable_database_url("postgres://buzz@127.0.0.1:1/buzz").await; + let (community, elsewhere) = ( + CommunityId::from_uuid(Uuid::from_u128(0xb1)), + CommunityId::from_uuid(Uuid::from_u128(0xb2)), + ); + let (target, bystander) = (vec![1u8; 32], vec![2u8; 32]); + let unrestricted = || buzz_db::moderation::RestrictionState { + banned: false, + muted_until: None, + }; + for key in [ + (community, target.clone()), + (community, bystander.clone()), + (elsewhere, target.clone()), + ] { + state.seed_restriction(key, unrestricted()); + } + + state.disconnect_pubkey_local( + community, + &target, + &"0".repeat(64), + "blocked: banned", + false, + ); + + let cached = |key| state.restriction_cache.get(&key).is_some(); + assert!( + !cached((community, target.clone())), + "the ban's row is dropped" + ); + assert!(cached((community, bystander)), "another pubkey's row stays"); + assert!(cached((elsewhere, target)), "another community's row stays"); + } + + /// `has_admitted_connection` follows every path that changes an admitted + /// socket's key: admission, a second admitted socket, re-authentication + /// to another key, and deregistration, per community. + /// Mutation: skip any one index update → RED. + #[tokio::test] + async fn admitted_index_follows_every_key_change() { + let mgr = ConnectionManager::new(); + let community = CommunityId::from_uuid(Uuid::from_u128(0xe)); + let other_community = CommunityId::from_uuid(Uuid::from_u128(0xf)); + let (first_key, second_key) = ([4u8; 32], [5u8; 32]); + + let first = bound_conn(&mgr, community, &first_key); + assert!(!mgr.has_admitted_connection(community, &first_key)); + mgr.mark_admitted(first); + mgr.mark_admitted(first); + let second = bound_conn(&mgr, community, &first_key); + mgr.mark_admitted(second); + let elsewhere = bound_conn(&mgr, other_community, &first_key); + mgr.mark_admitted(elsewhere); + + mgr.deregister(first); + assert!( + mgr.has_admitted_connection(community, &first_key), + "a second admitted socket keeps the key online" + ); + + mgr.set_authenticated_pubkey(second, second_key.to_vec()); + assert!(!mgr.has_admitted_connection(community, &first_key)); + assert!(mgr.has_admitted_connection(community, &second_key)); + assert!(mgr.has_admitted_connection(other_community, &first_key)); + + mgr.deregister(second); + mgr.deregister(second); + mgr.deregister(elsewhere); + assert!(!mgr.has_admitted_connection(community, &second_key)); + assert!(!mgr.has_admitted_connection(other_community, &first_key)); + assert!( + mgr.admitted_by_pubkey.is_empty(), + "drained keys are removed" + ); + } + #[tokio::test] async fn disconnect_pubkey_closes_matching_conns_with_reason() { let (mgr, id, _rx, mut ctrl_rx, cancel, _bp) = setup_conn(8); diff --git a/crates/buzz-relay/src/tenant.rs b/crates/buzz-relay/src/tenant.rs index 88b75f7d6e..8038ec3583 100644 --- a/crates/buzz-relay/src/tenant.rs +++ b/crates/buzz-relay/src/tenant.rs @@ -145,6 +145,36 @@ impl HostResolver for buzz_db::Db { } } +/// Request-path [`HostResolver`]: [`buzz_db::Db`] behind +/// [`crate::state::AppState::tenant_host_cache`]. A connection's handshake and +/// its follow-up HTTP requests resolve the same host many times a second, so +/// a mapped host is reread at most once per +/// [`crate::state::TENANT_HOST_CACHE_MAX_AGE`]. Unmapped hosts and lookup +/// errors are never cached. +impl HostResolver for crate::state::AppState { + type Error = buzz_db::DbError; + + async fn resolve_host( + &self, + normalized_host: &str, + ) -> Result, Self::Error> { + if let Some((community, read_at)) = self.tenant_host_cache.get(normalized_host) { + if read_at.elapsed() < crate::state::TENANT_HOST_CACHE_MAX_AGE { + metrics::counter!("buzz_tenant_host_cache_hits_total").increment(1); + return Ok(Some(community)); + } + } + metrics::counter!("buzz_tenant_host_cache_misses_total").increment(1); + let read_at = std::time::Instant::now(); + let community = self.db.resolve_host(normalized_host).await?; + if let Some(community) = community { + self.tenant_host_cache + .insert(normalized_host.to_owned(), (community, read_at)); + } + Ok(community) + } +} + #[cfg(test)] mod tests { use super::*; @@ -330,4 +360,37 @@ mod tests { assert!(matches!(err, BindError::UnmappedHost)); } } + + /// The request-path resolver serves a mapped host from the cache only + /// while the entry is younger than `TENANT_HOST_CACHE_MAX_AGE`; past + /// that it rereads, and a failed reread fails closed. The database is + /// unreachable, so any reread errors. + /// Mutation: drop the cache read, or its max-age check → RED. + #[tokio::test] + async fn app_state_resolver_serves_fresh_cached_hosts_only() { + let state = crate::state::tests::test_state_with_unreachable_database_url( + "postgres://buzz@127.0.0.1:1/buzz", + ) + .await; + let community = CommunityId::from_uuid(Uuid::from_u128(3)); + state.tenant_host_cache.insert( + "relay.example".into(), + (community, std::time::Instant::now()), + ); + let ctx = bind_community(&*state, "RELAY.example") + .await + .expect("a fresh cached host binds without the database"); + assert_eq!(ctx.community(), community); + + let aged = std::time::Instant::now() + .checked_sub(crate::state::TENANT_HOST_CACHE_MAX_AGE) + .expect("instant past max age"); + state + .tenant_host_cache + .insert("relay.example".into(), (community, aged)); + assert!(matches!( + bind_community(&*state, "relay.example").await, + Err(BindError::Lookup(_)) + )); + } } diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index 02dc94039d..15a0a0fca2 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -333,6 +333,13 @@ run_unit_tests() { run_test_step "buzz-relay community close suppresses losing denial frame test" \ cargo test -p buzz-relay --lib state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame -- --exact --nocapture + run_test_step "buzz-relay read cache tests" \ + cargo test -p buzz-relay --lib -- --exact --nocapture \ + state::tests::admitted_index_follows_every_key_change \ + state::tests::concurrent_open_channel_misses_share_the_in_flight_read \ + tenant::tests::app_state_resolver_serves_fresh_cached_hosts_only \ + state::tests::disconnect_drops_the_cached_restriction_row + run_test_step "buzz-relay router tests" \ cargo test -p buzz-relay --lib router::tests:: -- --nocapture