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