perf(relay): cache ephemeral-path serving and restriction reads (#8208)

## 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 <baxen@squareup.com>
Signed-off-by: Honey <8e307ae0076a4dab6b94b036ea3edc7e08f823a625269c1e6919e881a048b4d2@buzz.block.builderlab.xyz>
Signed-off-by: ART <ea39c1c8acf57180161ef4ba4ed68abf8589f5291956b909a5aa991fc66664ca@buzz.block.builderlab.xyz>
Co-authored-by: Honey <8e307ae0076a4dab6b94b036ea3edc7e08f823a625269c1e6919e881a048b4d2@buzz.block.builderlab.xyz>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: ART <ea39c1c8acf57180161ef4ba4ed68abf8589f5291956b909a5aa991fc66664ca@buzz.block.builderlab.xyz>

crates/buzz-relay/src/handlers/event.rs

@@ -672,11 +672,19 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
     }
 
     // Observer frames and ephemeral kinds return before ingest, so they take
-    // the ban / timeout write-block here. Persistent kinds get it in ingest.
-    if kind_u32 == KIND_AGENT_OBSERVER_FRAME || is_ephemeral(kind_u32) {
-        if let Err(e) =
-            super::ingest::enforce_write_restriction(&state, &conn.tenant, kind_u32, &auth_pubkey)
-                .await
+    // the ban / timeout write-block here, over a short-TTL cache because it
+    // runs per frame. Persistent kinds get the uncached check in ingest, and
+    // so does NIP-43 leave (28936), which is ephemeral-range but ingested.
+    if kind_u32 == KIND_AGENT_OBSERVER_FRAME
+        || (is_ephemeral(kind_u32) && kind_u32 != buzz_core::kind::KIND_NIP43_LEAVE_REQUEST)
+    {
+        if let Err(e) = super::ingest::enforce_cached_write_restriction(
+            &state,
+            &conn.tenant,
+            kind_u32,
+            &auth_pubkey,
+        )
+        .await
         {
             let (message, reason) = match e {
                 IngestError::Internal(_) => (
@@ -750,8 +758,10 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
             ));
             return;
         }
-        match buzz_deletion::store(&state.db)
-            .is_serving_active(conn.tenant.community())
+        // Cached (10s): this runs per ephemeral event. Persistent ingest keeps
+        // the uncached read as its durable write fence.
+        match state
+            .is_serving_active_cached(conn.tenant.community())
             .await
         {
             Ok(true) => {}
@@ -3192,6 +3202,7 @@ mod tests {
         fn authed_conn(
             keys: &nostr::Keys,
             tenant: buzz_core::tenant::TenantContext,
+            scopes: Vec<buzz_auth::Scope>,
         ) -> (
             std::sync::Arc<crate::connection::ConnectionState>,
             tokio::sync::mpsc::Receiver<axum::extract::ws::Message>,
@@ -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<crate::state::AppState>,
+            tenant: &buzz_core::tenant::TenantContext,
+            keys: &nostr::Keys,
+            event: nostr::Event,
+            scopes: Vec<buzz_auth::Scope>,
+        ) -> 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<crate::state::AppState>,
+            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:

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());

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?;
 

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<Tag>| {
+            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()
+            );
+        }
+    }
 }

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<moka::sync::Cache<(CommunityId, Vec<u8>), 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<moka::sync::Cache<CommunityId, bool>>,
+    /// 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<moka::sync::Cache<(CommunityId, Vec<u8>), 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<bool, buzz_db::DbError> {
+        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<buzz_db::moderation::RestrictionState, buzz_db::DbError> {
+        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