From bcc46fe8359a2bfd2bc26e6a7a1a1f8c29d44149 Mon Sep 17 00:00:00 2001 From: Bradley Axen Date: Fri, 9 Oct 2026 14:53:27 -0700 Subject: [PATCH] perf(relay): cache ephemeral-path serving and restriction reads (#8208) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Typing pulses and agent observer frames each ran two Postgres reads on the WebSocket path in `handlers/event.rs`: whether the community still serves writes (`is_serving_active`), and the sender's ban/timeout row (`moderation_restriction_state`). Both now read through short-TTL moka caches on `AppState`, following the membership cache's pattern. - **Serving state, keyed per community, 10 s TTL.** Every typer in a community shares one entry, so a pod does about one read per community per 10 s. - **Raw restriction row, keyed per (community, pubkey), 30 s TTL.** It caches the row, not the verdict, so the verdict is still taken at "now" and a cached timeout lifts the second it ends. One exception: `banned` is computed when the row is read, so a ban with an expiry can outlast that expiry by up to the TTL. - **The TTL is the staleness bound.** On every pod, and for every way a ban or timeout changes (signed commands, the admin API, report resolution, an owner's moderation reaching their agents), the ephemeral path sees the change within 30 s (10 s for archive/delete). As a best effort, signed ban/unban/timeout/untimeout (9040–9043) also drop the target's own entry on the pod that handles the command. Bans also keep their existing cluster-wide socket disconnect, and archiving a community disconnects all of its sockets. - **Hit/miss counters** `buzz_serving_active_cache_{hits,misses}_total` and `buzz_restriction_cache_{hits,misses}_total`, matching the membership cache, so the saved reads can be confirmed in prod. - **The durable fence stays uncached.** Persistent ingest, moderation commands, auth, and NIP-43 leave (28936, ephemeral-range but ingested) still read Postgres directly. The worst staleness case is a just-timed-out user's typing dots or observer frames getting through on another pod for up to 30 s. Real messages are never affected. ### Related issue N/A. This is the ephemeral-path Postgres read cost raised in the typing-indicator load discussion. Searched for existing PRs and found none. ### Testing - New Postgres tests: - The ephemeral gate honours a cached ban and admits an expired cached timeout. - The ephemeral fence honours a cached "fenced" community. - A cache miss fills both caches. - At the `handle_event` seam, a stored write (kind 1) is accepted despite a cached ban or a cached "fenced" state, and NIP-43 leave skips the cached ban. Mutation-checked: switching either ingest check to its cached variant, or dropping the 28936 carve-out, turns one red. - Timeout and untimeout commands drop the cached row. Mutation-checked: removing the invalidation turns it red. - Live test on a local relay: 100 typing pulses made 0 restriction/serving reads; 10 real messages made 10 of each. A timeout written straight to the DB refused messages at once and typing after ~30 s; an archive written straight to the DB fenced typing after ~10 s. - `scripts/postgres-test-run.sh -p buzz-relay --lib --tests`: 384/384 passed (local PG14). - `cargo nextest run -p buzz-relay --lib --tests`: 1470 passed. The 6 `api::media` failures are environmental: those tests seed through the default `:5432` database, which locally has no `buzz` role. - `cargo clippy -p buzz-relay --all-targets -D warnings` is clean. - Review follow-up commit pushed with the full pre-push hook passing. - Also adds a `sadscan:disable` marker to an existing dev-only test URL in `ingest.rs`. The commit hook flagged it once the file was staged, and the identical URLs elsewhere in that file already carry the marker. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Bradley Axen Signed-off-by: Honey <8e307ae0076a4dab6b94b036ea3edc7e08f823a625269c1e6919e881a048b4d2@buzz.block.builderlab.xyz> Signed-off-by: ART Co-authored-by: Honey <8e307ae0076a4dab6b94b036ea3edc7e08f823a625269c1e6919e881a048b4d2@buzz.block.builderlab.xyz> Co-authored-by: Claude Opus 5.5 Co-authored-by: ART --- crates/buzz-relay/src/handlers/event.rs | 207 +++++++++++++++++- crates/buzz-relay/src/handlers/ingest.rs | 23 +- .../src/handlers/moderation_commands.rs | 4 + crates/buzz-relay/src/handlers/relay_admin.rs | 57 +++++ crates/buzz-relay/src/state.rs | 76 +++++++ 5 files changed, 357 insertions(+), 10 deletions(-) diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 5b7b257c19..a243a872ef 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -672,11 +672,19 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc ( @@ -750,8 +758,10 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc {} @@ -3192,6 +3202,7 @@ mod tests { fn authed_conn( keys: &nostr::Keys, tenant: buzz_core::tenant::TenantContext, + scopes: Vec, ) -> ( std::sync::Arc, tokio::sync::mpsc::Receiver, @@ -3210,7 +3221,7 @@ mod tests { auth_state: std::sync::Mutex::new(crate::connection::AuthState::Authenticated( buzz_auth::AuthContext { pubkey: keys.public_key(), - scopes: vec![], + scopes, channel_ids: None, auth_method: buzz_auth::AuthMethod::Nip42, agent_owner_pubkey: None, @@ -3240,7 +3251,19 @@ mod tests { keys: &nostr::Keys, event: nostr::Event, ) -> String { - let (conn, mut rx) = authed_conn(keys, tenant.clone()); + ok_frame_scoped(state, tenant, keys, event, vec![]).await + } + + /// [`ok_frame`] with explicit token scopes, so a stored kind gets past + /// ingest's scope check to the restriction and fence checks. + async fn ok_frame_scoped( + state: &std::sync::Arc, + tenant: &buzz_core::tenant::TenantContext, + keys: &nostr::Keys, + event: nostr::Event, + scopes: Vec, + ) -> String { + let (conn, mut rx) = authed_conn(keys, tenant.clone(), scopes); super::super::handle_event(event, conn, std::sync::Arc::clone(state)).await; match rx .try_recv() @@ -3344,6 +3367,172 @@ mod tests { ); } + /// Fresh community with one unrestricted user. + async fn unrestricted_fixture() -> ( + std::sync::Arc, + buzz_core::tenant::TenantContext, + nostr::Keys, + ) { + let state = crate::state::tests::test_state().await; + let host = format!("ws-cache-gate-{}.test", uuid::Uuid::new_v4().simple()); + let community = state + .db + .ensure_configured_community(&host) + .await + .expect("ensure community") + .id; + let tenant = buzz_core::tenant::TenantContext::resolved(community, host); + let user = nostr::Keys::generate(); + state + .db + .ensure_user(community, user.public_key().as_bytes()) + .await + .expect("ensure user"); + (state, tenant, user) + } + + fn typing_event(keys: &nostr::Keys) -> nostr::Event { + nostr::EventBuilder::new(nostr::Kind::Custom(20_555), "typing") + .sign_with_keys(keys) + .expect("sign ephemeral") + } + + /// The ephemeral gate reads the cached restriction row: a cached ban + /// refuses a sender Postgres has never restricted, and a cached + /// timeout that has since expired admits (the verdict is taken now). + /// Mutation: call the uncached `enforce_write_restriction` → RED. + #[tokio::test] + #[ignore = "requires Postgres — runs in postgres-ci nextest lane"] + async fn ephemeral_gate_uses_cached_restriction_row() { + let (state, tenant, user) = unrestricted_fixture().await; + let key = (tenant.community(), user.public_key().to_bytes().to_vec()); + + state.restriction_cache.insert( + key.clone(), + buzz_db::moderation::RestrictionState { + banned: true, + muted_until: None, + }, + ); + let frame = ok_frame(&state, &tenant, &user, typing_event(&user)).await; + assert!( + frame.contains("false") && frame.contains("blocked: you are banned"), + "a cached ban must refuse the ephemeral event; got {frame}" + ); + + state.restriction_cache.insert( + key, + buzz_db::moderation::RestrictionState { + banned: false, + muted_until: Some(chrono::Utc::now() - chrono::Duration::seconds(1)), + }, + ); + let frame = ok_frame(&state, &tenant, &user, typing_event(&user)).await; + assert!( + frame.contains(",true,"), + "a cached timeout that has expired must not refuse; got {frame}" + ); + } + + /// The ephemeral community fence reads the cached serving state. + /// Mutation: call the uncached `is_serving_active` → RED. + #[tokio::test] + #[ignore = "requires Postgres — runs in postgres-ci nextest lane"] + async fn ephemeral_fence_uses_cached_serving_state() { + let (state, tenant, user) = unrestricted_fixture().await; + state.serving_active_cache.insert(tenant.community(), false); + let frame = ok_frame(&state, &tenant, &user, typing_event(&user)).await; + assert!( + frame.contains("false") && frame.contains("community writes are fenced"), + "a cached fenced community must refuse the ephemeral event; got {frame}" + ); + } + + /// A miss fills both caches from Postgres, so the next pulse skips it. + #[tokio::test] + #[ignore = "requires Postgres — runs in postgres-ci nextest lane"] + async fn ephemeral_gates_fill_caches_on_miss() { + let (state, tenant, user) = unrestricted_fixture().await; + ok_frame(&state, &tenant, &user, typing_event(&user)).await; + assert_eq!( + state.serving_active_cache.get(&tenant.community()), + Some(true) + ); + assert_eq!( + state + .restriction_cache + .get(&(tenant.community(), user.public_key().to_bytes().to_vec())), + Some(buzz_db::moderation::RestrictionState::default()) + ); + } + + fn text_note(keys: &nostr::Keys) -> nostr::Event { + nostr::EventBuilder::new(nostr::Kind::TextNote, "stored") + .sign_with_keys(keys) + .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. + #[tokio::test] + #[ignore = "requires Postgres — runs in postgres-ci nextest lane"] + async fn stored_writes_ignore_cached_ban() { + 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 scopes = buzz_auth::Scope::all_non_admin(); + + 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}" + ); + + 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; + assert!( + frame.contains("relay membership is not enabled"), + "NIP-43 leave must skip the cached gate and pass ingest's uncached one; got {frame}" + ); + } + + /// Stored writes skip the cached serving state: a cached "fenced" + /// for a community Postgres says is active does not refuse them. + /// Mutation: make ingest's fence call `is_serving_active_cached` → RED. + #[tokio::test] + #[ignore = "requires Postgres — runs in postgres-ci nextest lane"] + async fn stored_writes_ignore_cached_fence() { + let (state, tenant, user) = unrestricted_fixture().await; + state.serving_active_cache.insert(tenant.community(), false); + let frame = ok_frame_scoped( + &state, + &tenant, + &user, + text_note(&user), + buzz_auth::Scope::all_non_admin(), + ) + .await; + assert!( + frame.contains(",true,"), + "a stored write must read the serving state from Postgres; got {frame}" + ); + } + // W2 full witness: event-ingest barrier + durable absence + publication oracle. // // Extends the unit-level W2 barrier test with two Postgres-required assertions: diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index b46911b4cb..d4dcbaaece 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -2261,6 +2261,27 @@ pub(crate) async fn enforce_write_restriction( 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> { + let restriction = state + .restriction_state_cached(tenant.community(), pubkey.as_bytes()) + .await + .map_err(|e| { + IngestError::Internal(format!( + "error: internal error checking restriction state: {e}" + )) + })?; + write_restriction_denial(kind, &restriction, Utc::now()).map_or(Ok(()), Err) +} + /// Ingest a signed Nostr event through the full validation pipeline. /// /// Shared by WebSocket and HTTP transports. The caller constructs [`IngestAuth`] @@ -3559,7 +3580,7 @@ mod postgres_tests { #[ignore = "requires Postgres"] async fn check_channel_write_denies_when_channel_lookup_fails() { let state = crate::state::tests::test_state_with_database_url( - "postgres://buzz:buzz_dev@127.0.0.1:1/buzz", + "postgres://buzz:buzz_dev@127.0.0.1:1/buzz", // sadscan:disable np.postgres.1 ) .await; let community = buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()); diff --git a/crates/buzz-relay/src/handlers/moderation_commands.rs b/crates/buzz-relay/src/handlers/moderation_commands.rs index e1b5a77ec7..e10bbfed0c 100644 --- a/crates/buzz-relay/src/handlers/moderation_commands.rs +++ b/crates/buzz-relay/src/handlers/moderation_commands.rs @@ -176,6 +176,7 @@ 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 @@ -261,6 +262,7 @@ async fn handle_unban( if !lifted { return Err(invalid("member is not banned")); } + state.invalidate_restriction_cache(tenant.community(), &target); insert_audit(state, tenant, actor, "unban", Some(&target), None, None).await?; @@ -303,6 +305,7 @@ async fn handle_timeout( ) .await .map_err(|e| error(format!("database error: {e}")))?; + state.invalidate_restriction_cache(tenant.community(), &target); let action_id = insert_audit( state, @@ -366,6 +369,7 @@ async fn handle_untimeout( if !cleared { return Err(invalid("member is not timed out")); } + state.invalidate_restriction_cache(tenant.community(), &target); insert_audit(state, tenant, actor, "untimeout", Some(&target), None, None).await?; diff --git a/crates/buzz-relay/src/handlers/relay_admin.rs b/crates/buzz-relay/src/handlers/relay_admin.rs index 8daede44d1..ea8fc901b8 100644 --- a/crates/buzz-relay/src/handlers/relay_admin.rs +++ b/crates/buzz-relay/src/handlers/relay_admin.rs @@ -1228,4 +1228,61 @@ mod postgres_tests { assert!(target_socket.is_cancelled(), "the banned socket closes"); assert!(!bystander_socket.is_cancelled(), "the bystander stays"); } + + /// A timeout and an untimeout each drop the target's cached restriction + /// row on the pod that handles them, so typing and observer frames see + /// the change at once instead of after the cache TTL. + /// Mutation: remove `invalidate_restriction_cache` from either → RED. + #[tokio::test] + #[ignore = "requires Postgres — runs in postgres-ci nextest lane"] + async fn timeout_commands_drop_the_cached_restriction_row() { + let host = format!("timeout-cache-{}.example", uuid::Uuid::new_v4().simple()); + let (state, tenant) = workspace_profile_test_state(&host, true).await; + let (owner, target) = (Keys::generate(), Keys::generate()); + for (keys, role) in [(&owner, "owner"), (&target, "member")] { + state + .db + .add_relay_member(tenant.community(), &keys.public_key().to_hex(), role, None) + .await + .expect("seed member"); + } + let key = (tenant.community(), target.public_key().to_bytes().to_vec()); + let until = (chrono::Utc::now() + chrono::Duration::hours(1)).timestamp(); + let command = |kind: u32, tags: Vec| { + EventBuilder::new(Kind::Custom(kind as u16), "") + .tags(tags) + .sign_with_keys(&owner) + .expect("sign moderation command") + }; + + for event in [ + command( + buzz_core::kind::KIND_MODERATION_TIMEOUT, + vec![ + Tag::public_key(target.public_key()), + Tag::parse(["expiration", &until.to_string()]).expect("expiration tag"), + ], + ), + command( + buzz_core::kind::KIND_MODERATION_UNTIMEOUT, + vec![Tag::public_key(target.public_key())], + ), + ] { + state.restriction_cache.insert( + key.clone(), + buzz_db::moderation::RestrictionState::default(), + ); + crate::handlers::moderation_commands::handle_moderation_command( + &tenant, &state, &event, + ) + .await + .unwrap_or_else(|e| panic!("kind {} must apply: {e}", event.kind.as_u16())); + assert_eq!( + state.restriction_cache.get(&key), + None, + "kind {} must drop the cached row", + event.kind.as_u16() + ); + } + } } diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index ff506fb426..f158fb10ab 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -1317,6 +1317,23 @@ pub struct AppState { /// first-write-wins and set during auth before an agent's first event, /// so a short TTL only bounds staleness for the rare backfill race. pub author_type_cache: Arc), bool>>, + /// Ephemeral-path cache of `is_serving_active` (community not archived or + /// deleted). Key: community. TTL only (10s). Persistent ingest keeps the + /// 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. + #[allow(clippy::type_complexity)] + pub restriction_cache: + Arc), buzz_db::moderation::RestrictionState>>, /// Runtime conformance tracer. Production binds [`crate::conformance::NoopTracer`] /// (zero cost). Conformance tests bind [`crate::conformance::JsonlTracer`] to @@ -1556,6 +1573,18 @@ impl AppState { .time_to_live(std::time::Duration::from_secs(300)) .build(), ), + serving_active_cache: Arc::new( + moka::sync::Cache::builder() + .max_capacity(10_000) + .time_to_live(std::time::Duration::from_secs(10)) + .build(), + ), + restriction_cache: Arc::new( + moka::sync::Cache::builder() + .max_capacity(100_000) + .time_to_live(std::time::Duration::from_secs(30)) + .build(), + ), // Default to NoopTracer: production builds pay zero cost. // Conformance tests overwrite this with a JsonlTracer after // construction (see test helpers in @@ -1657,6 +1686,53 @@ impl AppState { Ok(result) } + /// `is_serving_active` with a 10-second cache, for the ephemeral path only. + /// Errors are not cached. + pub async fn is_serving_active_cached( + &self, + community_id: CommunityId, + ) -> Result { + if let Some(cached) = self.serving_active_cache.get(&community_id) { + metrics::counter!("buzz_serving_active_cache_hits_total").increment(1); + return Ok(cached); + } + metrics::counter!("buzz_serving_active_cache_misses_total").increment(1); + let result = buzz_deletion::store(&self.db) + .is_serving_active(community_id) + .await?; + self.serving_active_cache.insert(community_id, result); + Ok(result) + } + + /// `moderation_restriction_state` with a 30-second cache, for the + /// ephemeral path only. Errors are not cached. + pub async fn restriction_state_cached( + &self, + community_id: CommunityId, + pubkey: &[u8], + ) -> 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); + } + metrics::counter!("buzz_restriction_cache_misses_total").increment(1); + let result = self + .db + .moderation_restriction_state(community_id, pubkey) + .await?; + self.restriction_cache.insert(key, result.clone()); + Ok(result) + } + + /// 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. + pub fn invalidate_restriction_cache(&self, community_id: CommunityId, pubkey: &[u8]) { + self.restriction_cache + .invalidate(&(community_id, pubkey.to_vec())); + } + /// Invalidate caches after a membership change (add/remove member). /// /// Drops the local moka entries AND fire-and-forget publishes the same drop