Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 19 additions & 0 deletions crates/buzz-pubsub/src/cache_invalidation.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<u8>,
},
}

/// A cache invalidation received from a community-scoped Redis channel.
Expand Down Expand Up @@ -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::<CacheInvalidation>(&json).unwrap(),
msg
);
}

#[test]
fn unit_variants_roundtrip_through_json() {
for msg in [
Expand Down
23 changes: 23 additions & 0 deletions crates/buzz-relay/src/api/invites.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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!(
Expand Down Expand Up @@ -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}");
}
Expand Down Expand Up @@ -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(),
Expand All @@ -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
Expand Down
8 changes: 3 additions & 5 deletions crates/buzz-relay/src/api/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -80,15 +78,15 @@ 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}"))?;

match buzz_sdk::nip_oa::verify_auth_tag(tag_json, &agent_pubkey) {
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 {
Expand Down
4 changes: 4 additions & 0 deletions crates/buzz-relay/src/api/operator.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
{
Expand Down
3 changes: 3 additions & 0 deletions crates/buzz-relay/src/handlers/ingest.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
7 changes: 7 additions & 0 deletions crates/buzz-relay/src/handlers/relay_admin.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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");
}
Expand Down Expand Up @@ -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");
}
Expand Down
160 changes: 160 additions & 0 deletions crates/buzz-relay/src/state.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<moka::sync::Cache<(CommunityId, Uuid, Vec<u8>), 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<moka::sync::Cache<(CommunityId, Vec<u8>), 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)]
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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<bool, buzz_db::DbError> {
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
Expand Down Expand Up @@ -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);
}
}
}

Expand Down Expand Up @@ -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;
Expand Down
Loading