diff --git a/crates/buzz-pubsub/src/cache_invalidation.rs b/crates/buzz-pubsub/src/cache_invalidation.rs index f48158c9c5..c94eabfbe4 100644 --- a/crates/buzz-pubsub/src/cache_invalidation.rs +++ b/crates/buzz-pubsub/src/cache_invalidation.rs @@ -76,6 +76,13 @@ pub enum CacheInvalidation { /// Drop all membership / accessible / visibility caches. Mirrors /// `invalidate_channel_deleted`. ChannelDeleted, + /// Drop one pubkey's relay-membership entry. Mirrors + /// `invalidate_relay_membership` (invite claim, admin add/remove, + /// self-leave, ownership transfer). + RelayMembership { + /// Affected member's pubkey bytes. + pubkey: Vec, + }, } /// A cache invalidation received from a community-scoped Redis channel. @@ -232,6 +239,18 @@ mod tests { ); } + #[test] + fn relay_membership_roundtrips_through_json() { + let msg = CacheInvalidation::RelayMembership { + pubkey: vec![5, 6, 7, 8], + }; + let json = serde_json::to_string(&msg).unwrap(); + assert_eq!( + serde_json::from_str::(&json).unwrap(), + msg + ); + } + #[test] fn unit_variants_roundtrip_through_json() { for msg in [ diff --git a/crates/buzz-relay/src/api/invites.rs b/crates/buzz-relay/src/api/invites.rs index 6104171cca..1de80f99b2 100644 --- a/crates/buzz-relay/src/api/invites.rs +++ b/crates/buzz-relay/src/api/invites.rs @@ -406,6 +406,9 @@ pub async fn claim_invite( member = %claimer_hex, "relay member added via v2 invite" ); + // Drop any cached negative membership so the new member is + // admitted immediately, not after cache TTL. + state.invalidate_relay_membership(&tenant, &claimer_hex); // NIP-43 side effects only on Joined, never on other outcomes. if let Err(e) = publish_nip43_member_added(&tenant, &state, &claimer_hex).await { tracing::warn!( @@ -484,6 +487,9 @@ pub async fn claim_invite( member = %claimer_hex, "relay member added via invite" ); + // Drop any cached negative membership so the new member is admitted + // immediately, not after cache TTL. + state.invalidate_relay_membership(&tenant, &claimer_hex); if let Err(e) = publish_nip43_member_added(&tenant, &state, &claimer_hex).await { tracing::warn!("failed to publish NIP-43 member-added delta after claim: {e}"); } @@ -1115,6 +1121,16 @@ mod tests { assert!(url.contains("/invite/"), "unexpected url: {url}"); // Claim on a closed relay by a pubkey that is not yet a member. + // + // Wiring guard: seed a cached negative membership for the joiner + // first, exactly what a pre-join auth check would have left behind. + // The claim handler must call invalidate_relay_membership, or the + // joiner stays locked out until cache TTL — deleting that callsite + // must fail here. + state.relay_membership_cache.insert( + (community_id, joiner.public_key().to_bytes().to_vec()), + false, + ); let claim_body = serde_json::json!({ "code": code }).to_string(); let response = post_json( state.clone(), @@ -1128,6 +1144,13 @@ mod tests { let json = read_json(response).await; assert_eq!(json.get("status").and_then(Value::as_str), Some("joined")); assert_eq!(json.get("role").and_then(Value::as_str), Some("member")); + assert_eq!( + state + .relay_membership_cache + .get(&(community_id, joiner.public_key().to_bytes().to_vec())), + None, + "claim must invalidate the joiner's cached membership entry" + ); let member = state .db diff --git a/crates/buzz-relay/src/api/mod.rs b/crates/buzz-relay/src/api/mod.rs index d9f829433b..cd64cf20cb 100644 --- a/crates/buzz-relay/src/api/mod.rs +++ b/crates/buzz-relay/src/api/mod.rs @@ -68,10 +68,8 @@ pub mod relay_members { return Ok(MembershipDecision::OpenRelay); } - let pubkey_hex = hex::encode(pubkey_bytes); let is_member = state - .db - .is_relay_member(community, &pubkey_hex) + .is_relay_member_cached(community, pubkey_bytes) .await .map_err(|e| format!("relay membership check failed: {e}"))?; if is_member { @@ -80,6 +78,7 @@ pub mod relay_members { if state.config.allow_nip_oa_auth { if let Some(tag_json) = auth_tag_header { + let pubkey_hex = hex::encode(pubkey_bytes); let agent_pubkey = nostr::PublicKey::from_slice(pubkey_bytes) .map_err(|e| format!("invalid agent pubkey for NIP-OA check: {e}"))?; @@ -87,8 +86,7 @@ pub mod relay_members { Ok(owner_pubkey) => { let owner_hex = owner_pubkey.to_hex(); let owner_is_member = state - .db - .is_relay_member(community, &owner_hex) + .is_relay_member_cached(community, &owner_pubkey.to_bytes()) .await .map_err(|e| format!("relay membership check (owner) failed: {e}"))?; if owner_is_member { diff --git a/crates/buzz-relay/src/api/operator.rs b/crates/buzz-relay/src/api/operator.rs index 5b69a43874..fb68233b13 100644 --- a/crates/buzz-relay/src/api/operator.rs +++ b/crates/buzz-relay/src/api/operator.rs @@ -439,6 +439,10 @@ pub async fn transfer_community( .map_err(|e| internal_error(&format!("lookup community host: {e}")))? { let tenant = TenantContext::resolved(community, host); + // The step-4 upsert can insert a brand-new member row; drop any + // cached negative entry so the incoming owner is admitted + // immediately, not after cache TTL. + state.invalidate_relay_membership(&tenant, &new_owner_pubkey); if let Err(error) = crate::handlers::side_effects::publish_nip43_membership_list(&tenant, &state).await { diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 39ecbe18e4..df92b7170d 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -1981,6 +1981,9 @@ async fn ingest_event_inner( } // Publish NIP-43 announcements — fire-and-forget. + // Drop the cached positive membership so the leave takes effect + // fleet-wide now, not after cache TTL. + state.invalidate_relay_membership(tenant, &sender_hex); if let Err(e) = crate::handlers::side_effects::publish_nip43_member_removed(tenant, state, &sender_hex) .await diff --git a/crates/buzz-relay/src/handlers/relay_admin.rs b/crates/buzz-relay/src/handlers/relay_admin.rs index 3f58a9c2aa..ca53684b8d 100644 --- a/crates/buzz-relay/src/handlers/relay_admin.rs +++ b/crates/buzz-relay/src/handlers/relay_admin.rs @@ -297,6 +297,9 @@ async fn execute_relay_admin_command( // Only publish NIP-43 announcements when the row was actually inserted — // skip on no-op re-adds to avoid spurious kind:8000 events. if was_inserted { + // Drop any cached negative membership so the new member is + // admitted immediately, not after cache TTL. + state.invalidate_relay_membership(tenant, &target_hex); if let Err(e) = publish_nip43_member_added(tenant, state, &target_hex).await { warn!(error = %e, "failed to publish NIP-43 member added event"); } @@ -357,6 +360,10 @@ async fn execute_relay_admin_command( "relay member removed" ); + // Drop the cached positive membership so removal takes effect + // fleet-wide now, not after cache TTL. + state.invalidate_relay_membership(tenant, &target_hex); + if let Err(e) = publish_nip43_member_removed(tenant, state, &target_hex).await { warn!(error = %e, "failed to publish NIP-43 member removed event"); } diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 58a869a995..b8a846b542 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -543,6 +543,11 @@ pub struct AppState { /// Short TTL (10s) — membership changes are rare but must propagate. #[allow(clippy::type_complexity)] pub membership_cache: Arc), bool>>, + /// Relay membership cache: (community_id, pubkey_bytes) → is_relay_member. + /// Short TTL (10s) — checked on every authenticated HTTP request and WS + /// AUTH, so this keeps the hot auth path off the writer pool. Invalidated + /// on invite claim, admin add/remove, self-leave, and ownership transfer. + pub relay_membership_cache: Arc), bool>>, /// Accessible channel IDs cache: (community_id, pubkey_bytes) → channel UUIDs. /// Short TTL (10s) — invalidated on membership or channel visibility changes. #[allow(clippy::type_complexity)] @@ -745,6 +750,15 @@ impl AppState { .support_invalidation_closures() .build(), ), + // No `support_invalidation_closures()`: this cache is only ever + // dropped by exact key, and the flag adds an is-invalidated check + // to every read on the hottest cache in the relay. + relay_membership_cache: Arc::new( + moka::sync::Cache::builder() + .max_capacity(100_000) + .time_to_live(std::time::Duration::from_secs(10)) + .build(), + ), accessible_channels_cache: Arc::new( moka::sync::Cache::builder() .max_capacity(10_000) @@ -842,6 +856,62 @@ impl AppState { Ok(result) } + /// Check relay membership with a 10-second cache. Falls back to DB on miss. + /// + /// Runs on every authenticated HTTP request and WS AUTH (see + /// `check_relay_membership`), so a hit removes a writer-pool SELECT from + /// the hottest auth path in the relay. Negative results are cached too: + /// they are dropped by `invalidate_relay_membership` when the pubkey joins + /// (invite claim, admin add), so a fresh member is admitted immediately + /// rather than after TTL expiry. + pub async fn is_relay_member_cached( + &self, + community_id: CommunityId, + pubkey: &[u8], + ) -> Result { + let key = (community_id, pubkey.to_vec()); + if let Some(cached) = self.relay_membership_cache.get(&key) { + metrics::counter!("buzz_relay_membership_cache_hits_total").increment(1); + return Ok(cached); + } + metrics::counter!("buzz_relay_membership_cache_misses_total").increment(1); + let result = self + .db + .is_relay_member(community_id, &hex::encode(pubkey)) + .await?; + self.relay_membership_cache.insert(key, result); + Ok(result) + } + + /// Invalidate one pubkey's relay-membership entry after a membership + /// change (invite claim, admin add/remove, self-leave, ownership + /// transfer). + /// + /// Local drop plus fire-and-forget cross-pod publish, same contract as + /// [`invalidate_membership`]: a dropped publish degrades to the 10s TTL, + /// never a permanent leak. + pub fn invalidate_relay_membership(&self, tenant: &TenantContext, pubkey_hex: &str) { + let Ok(pubkey) = hex::decode(pubkey_hex) else { + // Every callsite passes validated 64-char hex; a non-hex input + // can't have a cache entry (keys are raw bytes), so nothing to drop. + debug_assert!(false, "invalidate_relay_membership: non-hex pubkey"); + return; + }; + self.invalidate_relay_membership_local(tenant.community(), &pubkey); + self.spawn_cache_invalidation(tenant, CacheInvalidation::RelayMembership { pubkey }); + } + + /// Local-only relay-membership drop. The cross-pod consumer calls this + /// directly so applying a received drop never re-publishes it. + pub(crate) fn invalidate_relay_membership_local( + &self, + community_id: CommunityId, + pubkey: &[u8], + ) { + self.relay_membership_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 @@ -996,6 +1066,9 @@ impl AppState { CacheInvalidation::ChannelDeleted => { self.invalidate_channel_deleted_local(community_id); } + CacheInvalidation::RelayMembership { pubkey } => { + self.invalidate_relay_membership_local(community_id, &pubkey); + } } } @@ -1477,6 +1550,93 @@ mod tests { assert_eq!(mgr.pubkey_for_conn(Uuid::new_v4()), None); } + #[tokio::test] + async fn relay_membership_cache_hit_and_scoped_invalidation() { + let state = test_state().await; + let community_a = CommunityId::from_uuid(Uuid::from_u128(0xAAAA)); + let community_b = CommunityId::from_uuid(Uuid::from_u128(0xBBBB)); + let member = vec![0xAAu8; 32]; + let other = vec![0xBBu8; 32]; + + // Seed as the cached method would after DB reads. + state + .relay_membership_cache + .insert((community_a, member.clone()), true); + state + .relay_membership_cache + .insert((community_a, other.clone()), false); + state + .relay_membership_cache + .insert((community_b, member.clone()), true); + + // A cached entry is served without touching the DB: the lazy test + // pool points at nothing, so a DB fallback would error, not return. + assert!(state + .is_relay_member_cached(community_a, &member) + .await + .expect("cache hit must not touch the DB"),); + // Negative entries are cached and served the same way. + assert!(!state + .is_relay_member_cached(community_a, &other) + .await + .expect("negative cache hit must not touch the DB"),); + + // Invalidation drops exactly the (community, pubkey) entry: same + // pubkey in another community and other pubkeys are untouched. + state.invalidate_relay_membership_local(community_a, &member); + assert_eq!( + state + .relay_membership_cache + .get(&(community_a, member.clone())), + None, + "invalidated entry must be dropped" + ); + assert_eq!( + state + .relay_membership_cache + .get(&(community_b, member.clone())), + Some(true), + "same pubkey in another community must survive" + ); + assert_eq!( + state.relay_membership_cache.get(&(community_a, other)), + Some(false), + "other pubkeys in the same community must survive" + ); + + // The mutation-flow entry point takes the hex string every handler + // holds and must decode it to the raw-byte cache key. This is the + // seam every invalidation callsite goes through. + state + .relay_membership_cache + .insert((community_a, member.clone()), true); + let tenant_a = buzz_core::TenantContext::resolved(community_a, "a.test.example"); + state.invalidate_relay_membership(&tenant_a, &hex::encode(&member)); + assert_eq!( + state + .relay_membership_cache + .get(&(community_a, member.clone())), + None, + "hex-keyed invalidation must drop the raw-byte entry" + ); + + // The cross-pod path applies the same local drop. + state + .relay_membership_cache + .insert((community_a, member.clone()), true); + state.apply_cache_invalidation( + community_a, + buzz_pubsub::cache_invalidation::CacheInvalidation::RelayMembership { + pubkey: member.clone(), + }, + ); + assert_eq!( + state.relay_membership_cache.get(&(community_a, member)), + None, + "cross-pod drop must clear the entry" + ); + } + #[tokio::test] async fn accessible_channel_invalidation_is_scoped_to_community() { let state = test_state().await;