[codex] Scope derived runtime state by community

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

@@ -42,15 +42,15 @@ const MEDIA_UPLOAD_RATE_WINDOW: Duration = Duration::from_secs(60);
 
 struct UploadPermit {
     _global: tokio::sync::OwnedSemaphorePermit,
-    in_flight: Arc<dashmap::DashMap<[u8; 32], u32>>,
-    pubkey: [u8; 32],
+    in_flight: Arc<dashmap::DashMap<crate::state::ScopedPubkeyKey, u32>>,
+    key: crate::state::ScopedPubkeyKey,
 }
 
 impl Drop for UploadPermit {
     fn drop(&mut self) {
         use dashmap::mapref::entry::Entry;
 
-        if let Entry::Occupied(mut entry) = self.in_flight.entry(self.pubkey) {
+        if let Entry::Occupied(mut entry) = self.in_flight.entry(self.key) {
             if *entry.get() <= 1 {
                 entry.remove();
             } else {
@@ -60,8 +60,12 @@ impl Drop for UploadPermit {
     }
 }
 
-fn upload_rate_limited(state: &AppState, pubkey: &nostr::PublicKey) -> bool {
-    let key: [u8; 32] = pubkey.to_bytes();
+fn upload_rate_limited(
+    state: &AppState,
+    community_id: buzz_core::CommunityId,
+    pubkey: &nostr::PublicKey,
+) -> bool {
+    let key = (community_id, pubkey.to_bytes());
     let now = Instant::now();
     let limit = state.config.media_uploads_per_minute;
     let mut entry = state
@@ -83,6 +87,7 @@ fn upload_rate_limited(state: &AppState, pubkey: &nostr::PublicKey) -> bool {
 
 fn acquire_upload_permit(
     state: &AppState,
+    community_id: buzz_core::CommunityId,
     pubkey: &nostr::PublicKey,
 ) -> Result<UploadPermit, MediaError> {
     let global = state
@@ -91,7 +96,7 @@ fn acquire_upload_permit(
         .try_acquire_owned()
         .map_err(|_| MediaError::UploadConcurrencyLimitReached)?;
 
-    let key: [u8; 32] = pubkey.to_bytes();
+    let key = (community_id, pubkey.to_bytes());
     let mut in_flight = state.media_uploads_in_flight.entry(key).or_insert(0);
     if *in_flight >= state.config.media_max_concurrent_uploads_per_pubkey {
         return Err(MediaError::UploadConcurrencyLimitReached);
@@ -102,7 +107,7 @@ fn acquire_upload_permit(
     Ok(UploadPermit {
         _global: global,
         in_flight: Arc::clone(&state.media_uploads_in_flight),
-        pubkey: key,
+        key,
     })
 }
 
@@ -185,15 +190,16 @@ impl FromRequestParts<Arc<AppState>> for AuthenticatedUpload {
         .await
         .map_err(|_| MediaError::RelayMembershipRequired)?;
 
-        if upload_rate_limited(state, &auth_event.pubkey) {
+        if upload_rate_limited(state, tenant.community(), &auth_event.pubkey) {
             metrics::counter!("buzz_media_upload_rejections_total", "reason" => "rate_limit")
                 .increment(1);
             return Err(MediaError::UploadRateLimitExceeded);
         }
-        let upload_permit = acquire_upload_permit(state, &auth_event.pubkey).inspect_err(|_| {
-            metrics::counter!("buzz_media_upload_rejections_total", "reason" => "concurrency")
-                .increment(1);
-        })?;
+        let upload_permit = acquire_upload_permit(state, tenant.community(), &auth_event.pubkey)
+            .inspect_err(|_| {
+                metrics::counter!("buzz_media_upload_rejections_total", "reason" => "concurrency")
+                    .increment(1);
+            })?;
 
         Ok(AuthenticatedUpload {
             auth_event,
@@ -743,9 +749,88 @@ fn extract_blossom_auth(headers: &HeaderMap) -> Result<nostr::Event, MediaError>
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::sync::Arc;
+
+    use uuid::Uuid;
 
     const VALID_HASH: &str = "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789";
 
+    async fn test_state() -> Arc<AppState> {
+        let mut config = crate::config::Config::from_env().expect("default config loads");
+        config.require_relay_membership = false;
+        config.redis_url = "redis://127.0.0.1:1".to_string();
+        config.media_uploads_per_minute = 1;
+        config.media_max_concurrent_uploads = 2;
+        config.media_max_concurrent_uploads_per_pubkey = 1;
+
+        let pool = sqlx::PgPool::connect_lazy(&config.database_url).expect("lazy pg pool");
+        let db = buzz_db::Db::from_pool(pool.clone());
+        let redis_pool = deadpool_redis::Config::from_url(&config.redis_url)
+            .create_pool(Some(deadpool_redis::Runtime::Tokio1))
+            .expect("redis pool");
+        let pubsub = Arc::new(
+            buzz_pubsub::PubSubManager::new(&config.redis_url, redis_pool.clone())
+                .await
+                .expect("pubsub manager"),
+        );
+        let audit = buzz_audit::AuditService::new(pool.clone());
+        let auth = buzz_auth::AuthService::new(config.auth.clone());
+        let search = buzz_search::SearchService::new(pool.clone());
+        let workflow_engine = Arc::new(buzz_workflow::WorkflowEngine::new(
+            db.clone(),
+            buzz_workflow::WorkflowConfig::default(),
+        ));
+        let media_storage = buzz_media::MediaStorage::new(&config.media).expect("media storage");
+        let (state, _audit_shutdown) = AppState::new(
+            config,
+            db,
+            redis_pool,
+            audit,
+            pubsub,
+            auth,
+            search,
+            workflow_engine,
+            nostr::Keys::generate(),
+            media_storage,
+        );
+        Arc::new(state)
+    }
+
+    #[tokio::test]
+    async fn upload_rate_limiter_is_scoped_by_community() {
+        let state = test_state().await;
+        let pubkey = nostr::Keys::generate().public_key();
+        let community_a = buzz_core::CommunityId::from_uuid(Uuid::from_u128(0xAAAA));
+        let community_b = buzz_core::CommunityId::from_uuid(Uuid::from_u128(0xBBBB));
+
+        assert!(!upload_rate_limited(&state, community_a, &pubkey));
+        assert!(upload_rate_limited(&state, community_a, &pubkey));
+        assert!(
+            !upload_rate_limited(&state, community_b, &pubkey),
+            "A's exhausted upload budget must not rate-limit the same key in B"
+        );
+    }
+
+    #[tokio::test]
+    async fn upload_concurrency_limit_is_scoped_by_community() {
+        let state = test_state().await;
+        let pubkey = nostr::Keys::generate().public_key();
+        let community_a = buzz_core::CommunityId::from_uuid(Uuid::from_u128(0xAAAA));
+        let community_b = buzz_core::CommunityId::from_uuid(Uuid::from_u128(0xBBBB));
+
+        let permit_a =
+            acquire_upload_permit(&state, community_a, &pubkey).expect("first A upload allowed");
+        assert!(matches!(
+            acquire_upload_permit(&state, community_a, &pubkey),
+            Err(MediaError::UploadConcurrencyLimitReached)
+        ));
+        let permit_b = acquire_upload_permit(&state, community_b, &pubkey)
+            .expect("A's in-flight upload must not block B");
+
+        drop(permit_b);
+        drop(permit_a);
+    }
+
     #[test]
     fn test_validate_media_path_bare_hash() {
         assert!(validate_media_path(VALID_HASH).is_ok());

crates/buzz-relay/src/connection.rs

@@ -243,9 +243,10 @@ pub async fn handle_connection(
     }
     state.conn_manager.deregister(conn.conn_id);
     if let AuthState::Authenticated(ref auth_ctx) = *conn.auth_state.read().await {
-        let remaining = state
-            .conn_manager
-            .connection_ids_for_pubkey(auth_ctx.pubkey.to_bytes().as_slice());
+        let remaining = state.conn_manager.connection_ids_for_pubkey_in_community(
+            conn.tenant.community(),
+            auth_ctx.pubkey.to_bytes().as_slice(),
+        );
         if remaining.is_empty() {
             let _ = state
                 .pubsub

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

@@ -815,7 +815,11 @@ async fn handle_ephemeral_event(
     if event_kind_u32(&event) == KIND_MESH_CONNECT_REQUEST {
         // Per-requester rate limit shared with the HTTP door — see
         // `mesh_signaling::connect_request_rate_limited` for rationale.
-        if super::mesh_signaling::connect_request_rate_limited(&state, &auth_pubkey) {
+        if super::mesh_signaling::connect_request_rate_limited(
+            &state,
+            conn.tenant.community(),
+            &auth_pubkey,
+        ) {
             conn.send(RelayMessage::ok(
                 event_id_hex,
                 false,
@@ -924,6 +928,32 @@ struct AgentObserverRoute {
     direction: AgentObserverDirection,
 }
 
+/// Check + bump the per-agent observer telemetry limit (100/sec window).
+///
+/// Observer frames are ephemeral, but the rejection is visible to the sender.
+/// Scope the counter by community so an agent key active in one tenant does not
+/// consume another tenant's logical rate budget.
+fn observer_frame_rate_limited(
+    state: &AppState,
+    community_id: CommunityId,
+    agent_key: [u8; 32],
+) -> bool {
+    let now = std::time::Instant::now();
+    let mut entry = state
+        .observer_rate_limiter
+        .entry((community_id, agent_key))
+        .or_insert((0, now));
+    let (count, window_start) = entry.value_mut();
+    if now.duration_since(*window_start).as_secs() >= 1 {
+        *count = 1;
+        *window_start = now;
+        false
+    } else {
+        *count += 1;
+        *count > 100
+    }
+}
+
 /// Handle encrypted agent observer frames (kind 24200).
 ///
 /// These frames bypass storage and are routed as global ephemeral events. The
@@ -1041,25 +1071,13 @@ async fn handle_agent_observer_event(
     // be starved by bursty telemetry from the agent.
     if matches!(route.direction, AgentObserverDirection::Telemetry) {
         let agent_key: [u8; 32] = agent_bytes.as_slice().try_into().unwrap_or([0u8; 32]);
-        let now = std::time::Instant::now();
-        let mut entry = state
-            .observer_rate_limiter
-            .entry(agent_key)
-            .or_insert((0, now));
-        let (count, window_start) = entry.value_mut();
-        if now.duration_since(*window_start).as_secs() >= 1 {
-            *count = 1;
-            *window_start = now;
-        } else {
-            *count += 1;
-            if *count > 100 {
-                conn.send(RelayMessage::ok(
-                    event_id_hex,
-                    false,
-                    "rate-limited: observer frame rate exceeded (100/sec per agent)",
-                ));
-                return;
-            }
+        if observer_frame_rate_limited(&state, conn.tenant.community(), agent_key) {
+            conn.send(RelayMessage::ok(
+                event_id_hex,
+                false,
+                "rate-limited: observer frame rate exceeded (100/sec per agent)",
+            ));
+            return;
         }
     }
 
@@ -1304,6 +1322,31 @@ mod tests {
         assert!(err.contains("NIP-44"));
     }
 
+    #[tokio::test]
+    async fn observer_frame_rate_limiter_is_scoped_by_community() {
+        let state = fanout_access::test_state().await;
+        let agent_key = Keys::generate().public_key().to_bytes();
+        let community_a = buzz_core::tenant::CommunityId::from_uuid(uuid::Uuid::from_u128(0xAAAA));
+        let community_b = buzz_core::tenant::CommunityId::from_uuid(uuid::Uuid::from_u128(0xBBBB));
+
+        for _ in 0..100 {
+            assert!(!super::observer_frame_rate_limited(
+                &state,
+                community_a,
+                agent_key
+            ));
+        }
+        assert!(super::observer_frame_rate_limited(
+            &state,
+            community_a,
+            agent_key
+        ));
+        assert!(
+            !super::observer_frame_rate_limited(&state, community_b, agent_key),
+            "A's exhausted budget must not rate-limit the same agent key in B"
+        );
+    }
+
     mod pubsub_fanout {
         use std::collections::HashMap;
         use std::sync::atomic::AtomicU8;

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

@@ -38,9 +38,14 @@ use crate::state::AppState;
 /// bound the amplification. 20/sec is far above any real interactive use; a
 /// buggy desktop loop can't storm the relay. Shared by the WS door
 /// (`handlers::event`) and the HTTP door (`handle_mesh_event_http`) — one
-/// limiter, two transports.
-pub(crate) fn connect_request_rate_limited(state: &AppState, pubkey: &nostr::PublicKey) -> bool {
-    let key: [u8; 32] = pubkey.to_bytes();
+/// limiter, two transports. The key includes community because the same Nostr
+/// key can legitimately participate in more than one community.
+pub(crate) fn connect_request_rate_limited(
+    state: &AppState,
+    community_id: CommunityId,
+    pubkey: &nostr::PublicKey,
+) -> bool {
+    let key = (community_id, pubkey.to_bytes());
     let now = std::time::Instant::now();
     let mut entry = state
         .mesh_connect_rate_limiter
@@ -93,7 +98,7 @@ pub async fn handle_mesh_event_http(
             handle_status_report(state, tenant, &pubkey_hex, event).await
         }
         k if k == KIND_MESH_CONNECT_REQUEST => {
-            if connect_request_rate_limited(state, auth_pubkey) {
+            if connect_request_rate_limited(state, tenant.community(), auth_pubkey) {
                 return Err("rate-limited: mesh connect request rate exceeded (20/sec)".to_string());
             }
             handle_connect_request(state, tenant, &pubkey_hex, event).await
@@ -804,14 +809,31 @@ mod tests {
         let pubkey = nostr::Keys::generate().public_key();
         for i in 0..20 {
             assert!(
-                !connect_request_rate_limited(&state, &pubkey),
+                !connect_request_rate_limited(&state, test_tenant().community(), &pubkey),
                 "request {} should pass",
                 i + 1
             );
         }
         assert!(
-            connect_request_rate_limited(&state, &pubkey),
+            connect_request_rate_limited(&state, test_tenant().community(), &pubkey),
             "21st request in the same window must be limited"
         );
     }
+
+    #[tokio::test]
+    async fn connect_request_rate_limiter_is_scoped_by_community() {
+        let state = test_state().await;
+        let pubkey = nostr::Keys::generate().public_key();
+        let community_a = CommunityId::from_uuid(uuid::Uuid::from_u128(0xAAAA));
+        let community_b = CommunityId::from_uuid(uuid::Uuid::from_u128(0xBBBB));
+
+        for _ in 0..20 {
+            assert!(!connect_request_rate_limited(&state, community_a, &pubkey));
+        }
+        assert!(connect_request_rate_limited(&state, community_a, &pubkey));
+        assert!(
+            !connect_request_rate_limited(&state, community_b, &pubkey),
+            "A's exhausted budget must not rate-limit the same key in B"
+        );
+    }
 }

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

@@ -42,7 +42,9 @@ async fn evict_live_channel_subscriptions(
     channel_id: Uuid,
     target_pubkey: &[u8],
 ) {
-    let conn_ids = state.conn_manager.connection_ids_for_pubkey(target_pubkey);
+    let conn_ids = state
+        .conn_manager
+        .connection_ids_for_pubkey_in_community(tenant.community(), target_pubkey);
 
     for conn_id in conn_ids {
         evict_conn_channel_subscriptions(tenant, state, channel_id, conn_id).await;

crates/buzz-relay/src/main.rs

@@ -716,10 +716,10 @@ async fn main() -> anyhow::Result<()> {
             loop {
                 match rx.recv().await {
                     Ok(scoped) => {
-                        // The local moka caches key on globally-unique UUIDs /
-                        // pubkeys, so applying the tenant-local op by key is
-                        // correct regardless of community; the `community_id`
-                        // scope rides the Redis topic, not the moka key.
+                        // The Redis topic carries the originating community,
+                        // and the local moka keys carry that same label. Apply
+                        // only the matching tenant-local drop; a mutation in A
+                        // must not flush B's derived state.
                         state_for_cache
                             .apply_cache_invalidation(scoped.community_id, scoped.invalidation);
                     }