perf(relay): drop per-pulse ban reads and owner-link writes on HTTP (#8233)

## Why

Agents send an HTTP typing pulse (`POST /events`, kind 20002) every 3–5
seconds while they work. #8207 already routes the pulse's
write-restriction and serving checks through #8208's short-TTL caches.
Two per-pulse costs were still left at the membership step:

- **A fresh ban read** for the agent, plus its owner when the agent
comes in through NIP-OA (`deny_banned` → `community_ban_outcome`).
- **Four writer round-trips for owner-admitted agents**, on every
request: `materialize_nip_oa_owner` runs 2× `ensure_user` + the
`agent_owner_pubkey` upsert + an `is_agent_owner` confirm, even though
the link already exists.

## What changes

- **Typing's ban check reads the restriction cache.** `submit_event`
already parses the event before the membership step, so typing pulses
call a new `enforce_relay_membership_ephemeral`. Its ban check reads
through `restriction_state_cached`, the same 30s bound the pulse's
write-restriction check already has. Every other HTTP route, persistent
`POST /events`, and socket auth still read fresh (`cached: false`).
- **`materialize_nip_oa_owner` skips a link it already knows.** If
`observer_owner_cache` already holds `(community, agent, owner) = true`,
it returns `true` right away. `agent_owner_pubkey` has one production
writer, which sets it only `WHERE agent_owner_pubkey IS NULL`, and
nothing clears it, so the skipped writes were no-ops. This helps every
caller (HTTP, socket auth, audio). The owner-link disconnect only runs
on the first write, so it still runs.
- **`observer_owner_cache` capacity goes from 1k to 100k**, so
concurrently active agents actually hit it. That's tens of MB per pod at
most, the same scale as `restriction_cache`.

Per-pulse Postgres work at the membership step:

| Agent | Before | After |
|---|---|---|
| Direct relay member | host lookup (writer) + membership + ban | host
lookup (writer) + membership |
| Owner-admitted (NIP-OA) | host lookup + 2 memberships + 2 bans + 4
writer trips | host lookup + 2 memberships |

## Not in this PR

- **Caching host → community.** It's the tenant-binding step every HTTP
request goes through, and it reads from the writer. Community deletion
depends on new requests not binding once a community leaves `active`. A
cache would keep binding a quiescing community for its TTL. The
serving-write leases may make that harmless, but that needs a closer
look than a "safe fixes" PR.
- **Caching relay membership.** Removing a member has no HTTP equivalent
of the socket disconnect, so a cache would delay removal by its TTL.
- **The desktop client reusing one connection.** That's going in a
separate buzz-app PR.

## Unrelated one-liner

The `sadscan` pre-commit hook scans whole staged files. It blocked this
commit on a pre-existing test-only URI in `handlers/auth.rs`
(`postgres://buzz:buzz_dev@127.0.0.1:1/buzz`, a deliberately unreachable
stub). I added the hook's own `// sadscan:disable np.postgres.1` marker
to that line.

## Testing

- New Postgres tests. I confirmed each one fails when its guard is
removed:
- `typing_ban_gate_uses_cached_restriction_row`: a member banned in
Postgres with a cached clear row can still type. Fails if typing reads
fresh.
- `persistent_post_ban_gate_reads_fresh`: a stale cached ban doesn't
refuse a kind-9 post that Postgres allows. Fails if every kind reads the
cache.
- `known_owner_link_skips_materialize_writes`: after the link is cached,
deleting the agent row behind the cache's back stays deleted. Fails
without the short-circuit.
- `scripts/postgres-test-run.sh -p buzz-relay --lib --tests`: 397/397
passed. This was local PG14.
- `cargo test -p buzz-relay`: 1430 passed and 7 failed. The same 7 fail
without this commit in this environment: 6 `api::media` tests can't find
a `buzz` role on :5432, and `mesh_demo` returns a 504.
- `cargo clippy -p buzz-relay --all-targets -D warnings` is clean.
- An independent agent review found no blockers. Its optional
suggestions are test gaps: nothing pins the owner-cascade cached read,
or the `false` literals at socket auth.
- **Not yet done:** a run against a live local relay, and human testing.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Signed-off-by: Honey <8e307ae0076a4dab6b94b036ea3edc7e08f823a625269c1e6919e881a048b4d2@buzz.block.builderlab.xyz>
Co-authored-by: Honey <8e307ae0076a4dab6b94b036ea3edc7e08f823a625269c1e6919e881a048b4d2@buzz.block.builderlab.xyz>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

crates/buzz-relay/src/api/bridge.rs

@@ -1118,17 +1118,33 @@ async fn submit_event_authed(
         }
     };
 
+    let kind_u32 = buzz_core::kind::event_kind_u32(&event);
+    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 nip_oa_owner = match super::relay_members::enforce_relay_membership(
-        state,
-        tenant.community(),
-        &pubkey_bytes,
-        auth_tag,
-        signed_auth_created_at,
-    )
-    .await
-    {
+    let membership = if is_typing {
+        super::relay_members::enforce_relay_membership_ephemeral(
+            state,
+            tenant.community(),
+            &pubkey_bytes,
+            auth_tag,
+            signed_auth_created_at,
+        )
+        .await
+    } else {
+        super::relay_members::enforce_relay_membership(
+            state,
+            tenant.community(),
+            &pubkey_bytes,
+            auth_tag,
+            signed_auth_created_at,
+        )
+        .await
+    };
+    let nip_oa_owner = match membership {
         Ok(owner) => owner.or_else(|| {
             if !state.config.require_relay_membership {
                 super::relay_members::extract_nip_oa_owner(
@@ -1151,7 +1167,6 @@ async fn submit_event_authed(
         super::relay_members::materialize_nip_oa_owner(state, tenant, &pubkey, &owner).await;
     }
 
-    let kind_u32 = buzz_core::kind::event_kind_u32(&event);
     let auth = IngestAuth::Http {
         pubkey,
         scopes: buzz_auth::Scope::all_known(), // Pure Nostr: full scopes, channel access via membership
@@ -1159,7 +1174,7 @@ async fn submit_event_authed(
     };
 
     // Typing indicators are ephemeral: broadcast to the channel, never stored.
-    let result = if kind_u32 == buzz_core::kind::KIND_TYPING_INDICATOR {
+    let result = if is_typing {
         let event_id = event.id.to_hex();
         crate::handlers::event::publish_http_typing(state, tenant, event, pubkey)
             .await
@@ -8308,6 +8323,53 @@ pub(crate) mod postgres_tests {
         fx.drop().await;
     }
 
+    /// HTTP agents reach `materialize_nip_oa_owner` on every request. Once the
+    /// owner link is confirmed in `observer_owner_cache` the call writes
+    /// nothing: the link is set once and never cleared.
+    /// Mutation: drop the cache short-circuit → the agent row is recreated → RED.
+    #[tokio::test]
+    #[ignore = "requires Postgres and Redis"]
+    async fn known_owner_link_skips_materialize_writes() {
+        let state = bridge_handler_test_state()
+            .await
+            .expect("local Postgres and Redis");
+        let host = format!("owner-link-{}.local", uuid::Uuid::new_v4());
+        state.db.ensure_configured_community(&host).await.unwrap();
+        let tenant = crate::tenant::bind_community(&state.db, &host)
+            .await
+            .unwrap();
+        let (owner, agent_key) = (Keys::generate().public_key(), Keys::generate().public_key());
+        let materialize = || {
+            super::super::relay_members::materialize_nip_oa_owner(
+                &state, &tenant, &agent_key, &owner,
+            )
+        };
+        let agent_rows = || async {
+            sqlx::query_scalar::<_, i64>(
+                "SELECT COUNT(*) FROM users WHERE community_id = $1 AND pubkey = $2",
+            )
+            .bind(tenant.community().as_uuid())
+            .bind(agent_key.to_bytes().as_slice())
+            .fetch_one(state.db.pool())
+            .await
+            .unwrap()
+        };
+
+        assert!(materialize().await, "first call records the link");
+        assert_eq!(agent_rows().await, 1);
+
+        // Remove the row behind the cache's back: a call that still wrote
+        // would recreate it.
+        sqlx::query("DELETE FROM users WHERE community_id = $1 AND pubkey = $2")
+            .bind(tenant.community().as_uuid())
+            .bind(agent_key.to_bytes().as_slice())
+            .execute(state.db.pool())
+            .await
+            .unwrap();
+        assert!(materialize().await, "a known link is still confirmed");
+        assert_eq!(agent_rows().await, 0, "a known link writes nothing");
+    }
+
     /// An agent socket admitted with no owner, then linked to its owner by a
     /// later `POST /events` carrying NIP-OA, closes when the owner is banned
     /// or removed even though the owner-to-agent lookup fails: recording the

crates/buzz-relay/src/api/mod.rs

@@ -246,6 +246,47 @@ pub mod relay_members {
         pubkey_bytes: &[u8],
         auth_tag_header: Option<&str>,
         signed_auth_created_at: Option<u64>,
+    ) -> Result<Option<nostr::PublicKey>, (StatusCode, Json<serde_json::Value>)> {
+        enforce_membership(
+            state,
+            community,
+            pubkey_bytes,
+            auth_tag_header,
+            signed_auth_created_at,
+            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,
+        community: CommunityId,
+        pubkey_bytes: &[u8],
+        auth_tag_header: Option<&str>,
+        signed_auth_created_at: Option<u64>,
+    ) -> Result<Option<nostr::PublicKey>, (StatusCode, Json<serde_json::Value>)> {
+        enforce_membership(
+            state,
+            community,
+            pubkey_bytes,
+            auth_tag_header,
+            signed_auth_created_at,
+            true,
+        )
+        .await
+    }
+
+    async fn enforce_membership(
+        state: &AppState,
+        community: CommunityId,
+        pubkey_bytes: &[u8],
+        auth_tag_header: Option<&str>,
+        signed_auth_created_at: Option<u64>,
+        cached_ban: bool,
     ) -> Result<Option<nostr::PublicKey>, (StatusCode, Json<serde_json::Value>)> {
         match check_relay_membership(
             state,
@@ -263,6 +304,7 @@ pub mod relay_members {
                     pubkey_bytes,
                     auth_tag_header,
                     signed_auth_created_at,
+                    cached_ban,
                 )
                 .await?;
                 Ok(None)
@@ -274,6 +316,7 @@ pub mod relay_members {
                     pubkey_bytes,
                     auth_tag_header,
                     signed_auth_created_at,
+                    cached_ban,
                 )
                 .await?;
                 Ok(Some(owner))
@@ -301,6 +344,7 @@ pub mod relay_members {
         pubkey_bytes: &[u8],
         auth_tag_header: Option<&str>,
         signed_auth_created_at: Option<u64>,
+        cached: bool,
     ) -> Result<(), (StatusCode, Json<serde_json::Value>)> {
         let Ok(pubkey) = nostr::PublicKey::from_slice(pubkey_bytes) else {
             return Err(super::internal_error("invalid pubkey for ban check"));
@@ -311,6 +355,7 @@ pub mod relay_members {
             pubkey,
             auth_tag_header,
             signed_auth_created_at,
+            cached,
         )
         .await
         {
@@ -359,12 +404,24 @@ pub mod relay_members {
     /// Both principals are ensured first because `agent_owner_pubkey` has a
     /// community-scoped foreign key. The mapping is first-write-wins; an
     /// existing mapping is accepted only when it names the same owner.
+    ///
+    /// A mapping already confirmed in `observer_owner_cache` returns early:
+    /// `agent_owner_pubkey` is set once and never cleared, so the writes
+    /// below would all be no-ops. HTTP agents call this on every request.
     pub async fn materialize_nip_oa_owner(
         state: &AppState,
         tenant: &TenantContext,
         agent: &nostr::PublicKey,
         owner: &nostr::PublicKey,
     ) -> bool {
+        let owner_key = (
+            tenant.community(),
+            agent.to_bytes().to_vec(),
+            owner.to_bytes().to_vec(),
+        );
+        if state.observer_owner_cache.get(&owner_key) == Some(true) {
+            return true;
+        }
         for (role, pubkey) in [("agent", agent), ("owner", owner)] {
             match state
                 .db
@@ -417,14 +474,7 @@ pub mod relay_members {
             state
                 .author_type_cache
                 .insert((tenant.community(), agent.to_bytes().to_vec()), true);
-            state.observer_owner_cache.insert(
-                (
-                    tenant.community(),
-                    agent.to_bytes().to_vec(),
-                    owner.to_bytes().to_vec(),
-                ),
-                true,
-            );
+            state.observer_owner_cache.insert(owner_key, true);
         }
         materialized
     }

crates/buzz-relay/src/api/typing_postgres_tests.rs

@@ -176,6 +176,62 @@ async fn typing_uses_cached_restriction_row() {
     assert_eq!(status, axum::http::StatusCode::FORBIDDEN, "{body}");
 }
 
+/// The membership-stage ban gate for HTTP typing reads the same cache: a
+/// cached clear row admits a member Postgres banned within the TTL.
+/// Mutation: route typing through `enforce_relay_membership` (fresh ban
+/// read) → RED.
+#[tokio::test]
+#[ignore = "requires Postgres and Redis"]
+async fn typing_ban_gate_uses_cached_restriction_row() {
+    let f = Fixture::new().await;
+    let member = f.member.public_key().to_bytes().to_vec();
+    f.state
+        .db
+        .ban_community_member(
+            f.community,
+            &member,
+            Keys::generate().public_key().as_bytes(),
+            None,
+            None,
+        )
+        .await
+        .unwrap();
+    f.state.restriction_cache.insert(
+        (f.community, member),
+        buzz_db::moderation::RestrictionState {
+            banned: false,
+            muted_until: None,
+        },
+    );
+    let (status, body) = f
+        .post(&f.member, &f.typing(&f.member, Some(f.channel)))
+        .await;
+    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.
+#[tokio::test]
+#[ignore = "requires Postgres and Redis"]
+async fn persistent_post_ban_gate_reads_fresh() {
+    let f = Fixture::new().await;
+    f.state.restriction_cache.insert(
+        (f.community, f.member.public_key().to_bytes().to_vec()),
+        buzz_db::moderation::RestrictionState {
+            banned: true,
+            muted_until: None,
+        },
+    );
+    let message = EventBuilder::new(Kind::Custom(9), "hello")
+        .tags([Tag::parse(["h".to_string(), f.channel.to_string()]).unwrap()])
+        .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}");
+}
+
 /// HTTP typing reads the cached serving state, like the WebSocket path.
 /// Mutation: call the uncached `is_serving_active` → RED.
 #[tokio::test]

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

@@ -62,23 +62,35 @@ fn classify_relay_membership(
 /// Fails closed: a DB error is `DbError`, never `Clear`. NIP-OA cascade: a ban
 /// 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.
 pub(crate) async fn community_ban_outcome(
     state: &AppState,
     community: buzz_core::CommunityId,
     pubkey: nostr::PublicKey,
     auth_tag_json: Option<&str>,
     signed_auth_created_at: Option<u64>,
+    cached: bool,
 ) -> BanOutcome {
     async fn lookup(
         state: &AppState,
         community: buzz_core::CommunityId,
         key: &nostr::PublicKey,
+        cached: bool,
     ) -> BanOutcome {
-        match state
-            .db
-            .moderation_restriction_state(community, key.as_bytes())
-            .await
-        {
+        let restriction = if cached {
+            state
+                .restriction_state_cached(community, key.as_bytes())
+                .await
+        } else {
+            state
+                .db
+                .moderation_restriction_state(community, key.as_bytes())
+                .await
+        };
+        match restriction {
             Ok(restriction) if restriction.banned => BanOutcome::Banned,
             Ok(_) => BanOutcome::Clear,
             Err(e) => {
@@ -87,7 +99,7 @@ pub(crate) async fn community_ban_outcome(
             }
         }
     }
-    let outcome = lookup(state, community, &pubkey).await;
+    let outcome = lookup(state, community, &pubkey, cached).await;
     if outcome != BanOutcome::Clear {
         return outcome;
     }
@@ -96,7 +108,7 @@ pub(crate) async fn community_ban_outcome(
         auth_tag_json,
         signed_auth_created_at,
     ) {
-        Some(owner) => lookup(state, community, &owner).await,
+        Some(owner) => lookup(state, community, &owner, cached).await,
         None => BanOutcome::Clear,
     }
 }
@@ -146,6 +158,7 @@ pub(crate) async fn final_admission_denial(
         pubkey,
         auth_tag_json,
         signed_auth_created_at,
+        false,
     )
     .await;
     if let Some((metric, reason, outcome)) = ban_denial(ban) {
@@ -402,6 +415,7 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc<ConnectionState>, state:
                     pubkey,
                     auth_tag_json.as_deref(),
                     Some(signed_auth_created_at),
+                    false,
                 )
                 .await;
 
@@ -877,7 +891,7 @@ mod tests {
         // Fix 5: use Config::for_test() which holds NIP_FI_ENV_LOCK internally. [FI-TRACE-ENV-RACE]
         let mut config = crate::config::Config::for_test();
         config.require_relay_membership = false;
-        config.database_url = "postgres://buzz:buzz_dev@127.0.0.1:1/buzz".to_string();
+        config.database_url = "postgres://buzz:buzz_dev@127.0.0.1:1/buzz".to_string(); // sadscan:disable np.postgres.1
         config.redis_url = "redis://127.0.0.1:1".to_string();
         // 100ms acquire timeout: a request that falls through to the stub
         // pool still waits, but for 100ms instead of sqlx's 30s default.

crates/buzz-relay/src/state.rs

@@ -1308,7 +1308,9 @@ pub struct AppState {
     /// Key: (community_id, agent_pubkey_bytes, owner_pubkey_bytes). Value: is_owner.
     /// `agent_owner_pubkey` is immutable inside one community, so a long TTL
     /// (5 min) is safe once the community label is part of the key.
-    /// Prevents repeated DB lookups from bursty observer traffic.
+    /// Prevents repeated DB lookups from bursty observer traffic, and lets
+    /// `materialize_nip_oa_owner` skip its writes for a known mapping. Sized
+    /// for every concurrently active agent so per-request HTTP agents hit.
     #[allow(clippy::type_complexity)]
     pub observer_owner_cache: Arc<moka::sync::Cache<(CommunityId, Vec<u8>, Vec<u8>), bool>>,
     /// Cache for the `author_type` metric label on the ingest path.
@@ -1563,7 +1565,7 @@ impl AppState {
             media_uploads_in_flight: Arc::new(DashMap::new()),
             observer_owner_cache: Arc::new(
                 moka::sync::Cache::builder()
-                    .max_capacity(1_000)
+                    .max_capacity(100_000)
                     .time_to_live(std::time::Duration::from_secs(300))
                     .build(),
             ),