From 326e2301cb4b1edcb8a72d01ac4b19a83545365f Mon Sep 17 00:00:00 2001 From: Bradley Axen Date: Fri, 9 Oct 2026 16:15:16 -0700 Subject: [PATCH] perf(relay): drop per-pulse ban reads and owner-link writes on HTTP (#8233) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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 --- crates/buzz-relay/src/api/bridge.rs | 84 ++++++++++++++++--- crates/buzz-relay/src/api/mod.rs | 66 +++++++++++++-- .../src/api/typing_postgres_tests.rs | 56 +++++++++++++ crates/buzz-relay/src/handlers/auth.rs | 30 +++++-- crates/buzz-relay/src/state.rs | 6 +- 5 files changed, 213 insertions(+), 29 deletions(-) diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 2b8ff56152..d631df7500 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/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 diff --git a/crates/buzz-relay/src/api/mod.rs b/crates/buzz-relay/src/api/mod.rs index 5e068f5fdf..f41712de4f 100644 --- a/crates/buzz-relay/src/api/mod.rs +++ b/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, + ) -> Result, (StatusCode, Json)> { + 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, + ) -> Result, (StatusCode, Json)> { + 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, + cached_ban: bool, ) -> Result, (StatusCode, Json)> { 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, + 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")); @@ -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 } diff --git a/crates/buzz-relay/src/api/typing_postgres_tests.rs b/crates/buzz-relay/src/api/typing_postgres_tests.rs index d6d24be0e3..6c7c7b1982 100644 --- a/crates/buzz-relay/src/api/typing_postgres_tests.rs +++ b/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] diff --git a/crates/buzz-relay/src/handlers/auth.rs b/crates/buzz-relay/src/handlers/auth.rs index c22c64968b..8a94a5827d 100644 --- a/crates/buzz-relay/src/handlers/auth.rs +++ b/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, + 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, 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. diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index f158fb10ab..e762095ca3 100644 --- a/crates/buzz-relay/src/state.rs +++ b/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, Vec), 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(), ),