diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index 58573163ad..bead324a46 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -205,7 +205,11 @@ pub async fn handle_connection( let _ = send_task.await; let _ = heartbeat_task.await; - state.sub_registry.remove_connection(conn.conn_id); + // Release every topic this connection held local interest in so the pubsub + // manager can debounce-UNSUBSCRIBE topics with no remaining pod interest. + for topic in state.sub_registry.remove_connection(conn.conn_id) { + state.pubsub.release_topic(&conn.tenant, topic).await; + } state.conn_manager.deregister(conn.conn_id); if let AuthState::Authenticated(ref auth_ctx) = *conn.auth_state.read().await { let remaining = state diff --git a/crates/buzz-relay/src/handlers/close.rs b/crates/buzz-relay/src/handlers/close.rs index c0c89062c5..ab6a27c330 100644 --- a/crates/buzz-relay/src/handlers/close.rs +++ b/crates/buzz-relay/src/handlers/close.rs @@ -14,7 +14,11 @@ pub async fn handle_close(sub_id: String, conn: Arc, state: Arc // Deregister from the fan-out index before sending CLOSED so no new // messages are routed to this sub after the client's CLOSE is acknowledged. - state.sub_registry.remove_subscription(conn_id, &sub_id); + // Release the subscription's topic so the pubsub manager can drop the Redis + // subscription once this pod has no remaining local interest in it. + if let Some(topic) = state.sub_registry.remove_subscription(conn_id, &sub_id) { + state.pubsub.release_topic(&conn.tenant, topic).await; + } conn.send(RelayMessage::closed(&sub_id, "")); diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index ebede8d75f..3620cf9f33 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -1096,7 +1096,7 @@ mod tests { ) } - fn register_global_sub( + async fn register_global_sub( state: &AppState, sub_id: &str, filter: Filter, @@ -1115,13 +1115,25 @@ mod tests { if let Some(pubkey) = pubkey { state.conn_manager.set_authenticated_pubkey(conn_id, pubkey); } + let displaced = + state + .sub_registry + .register(conn_id, sub_id.to_string(), vec![filter], None); + // Drive the dynamic Redis subscription through the same retain/release + // path the real REQ handler uses, so the cross-pod presence test + // proves delivery from genuine subscription interest — not a manual + // retain workaround. Global sub → `EventTopic::Global`. state - .sub_registry - .register(conn_id, sub_id.to_string(), vec![filter], None); + .pubsub + .retain_topic(&test_ctx(), EventTopic::Global) + .await; + if let Some(old) = displaced { + state.pubsub.release_topic(&test_ctx(), old).await; + } (conn_id, rx) } - fn register_presence_sub( + async fn register_presence_sub( state: &AppState, sub_id: &str, ) -> (Uuid, mpsc::Receiver) { @@ -1131,9 +1143,10 @@ mod tests { Filter::new().kind(Kind::Custom(KIND_PRESENCE_UPDATE as u16)), None, ) + .await } - fn register_membership_sub( + async fn register_membership_sub( state: &AppState, sub_id: &str, target: &Keys, @@ -1146,6 +1159,7 @@ mod tests { .pubkey(target.public_key()), Some(target.public_key().to_bytes().to_vec()), ) + .await } fn presence_event(status: &str) -> nostr::Event { @@ -1176,7 +1190,7 @@ mod tests { #[tokio::test] async fn global_presence_pubsub_event_fans_out_to_local_subscribers() { let state = test_state().await; - let (_conn_id, mut rx) = register_presence_sub(&state, "presence"); + let (_conn_id, mut rx) = register_presence_sub(&state, "presence").await; let event = presence_event("online"); let event_id = event.id; @@ -1198,7 +1212,7 @@ mod tests { #[tokio::test] async fn local_echo_presence_pubsub_event_is_not_delivered_twice() { let state = test_state().await; - let (_conn_id, mut rx) = register_presence_sub(&state, "presence"); + let (_conn_id, mut rx) = register_presence_sub(&state, "presence").await; let event = presence_event("online"); state.mark_local_event(&test_ctx(), &event.id); @@ -1224,9 +1238,9 @@ mod tests { let target = Keys::generate(); let other = Keys::generate(); let (_target_conn, mut target_rx) = - register_membership_sub(&state, "membership-target", &target); + register_membership_sub(&state, "membership-target", &target).await; let (_other_conn, mut other_rx) = - register_membership_sub(&state, "membership-other", &other); + register_membership_sub(&state, "membership-other", &other).await; let event = membership_event(&target, Uuid::new_v4()); let event_id = event.id; @@ -1290,22 +1304,13 @@ mod tests { let origin_fanout = spawn_pubsub_fanout_loop(origin.clone()); let receiver_fanout = spawn_pubsub_fanout_loop(receiver.clone()); - // Both relays declare interest in the community-scoped Global topic — - // the dynamic-subscription seam the relay wiring drives from a live - // presence/global subscription registering. Without retain, the new - // refcounted subscriber never SUBSCRIBEs and nothing is delivered. - origin - .pubsub - .retain_topic(&test_ctx(), EventTopic::Global) - .await; - receiver - .pubsub - .retain_topic(&test_ctx(), EventTopic::Global) - .await; - - let (_origin_conn, mut origin_rx) = register_presence_sub(&origin, "origin-presence"); + // Both relays' presence subscriptions drive `retain_topic` through + // the real `register()` path inside `register_presence_sub`, so the + // refcounted subscriber SUBSCRIBEs from genuine subscription interest. + let (_origin_conn, mut origin_rx) = + register_presence_sub(&origin, "origin-presence").await; let (_receiver_conn, mut receiver_rx) = - register_presence_sub(&receiver, "receiver-presence"); + register_presence_sub(&receiver, "receiver-presence").await; // Match buzz-pubsub's own Redis round-trip test: give PSUBSCRIBE a // bounded moment to attach before publishing the single test event. diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index 76120a5a63..26b423014b 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -196,9 +196,22 @@ pub async fn handle_req( subs.insert(sub_id.clone(), filters.clone()); } - state - .sub_registry - .register(conn_id, sub_id.clone(), filters.clone(), channel_id); + let displaced_topic = + state + .sub_registry + .register(conn_id, sub_id.clone(), filters.clone(), channel_id); + + // Drive the dynamic Redis subscription: retain this subscription's topic + // (first local interest in `(community, topic)` triggers SUBSCRIBE), and + // release any topic displaced by replacing a same-`sub_id` subscription so + // the pubsub refcount stays balanced across an in-place REQ replace. + let new_topic = channel_id + .map(buzz_pubsub::EventTopic::Channel) + .unwrap_or(buzz_pubsub::EventTopic::Global); + state.pubsub.retain_topic(&conn.tenant, new_topic).await; + if let Some(old_topic) = displaced_topic { + state.pubsub.release_topic(&conn.tenant, old_topic).await; + } debug!(conn_id = %conn_id, sub_id = %sub_id, "Subscription registered"); diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 024883d479..8077a57409 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -34,13 +34,14 @@ pub fn is_side_effect_kind(kind: u32) -> bool { async fn evict_live_channel_subscriptions( state: &Arc, + ctx: &buzz_core::TenantContext, channel_id: Uuid, target_pubkey: &[u8], ) { let conn_ids = state.conn_manager.connection_ids_for_pubkey(target_pubkey); for conn_id in conn_ids { - evict_conn_channel_subscriptions(state, channel_id, conn_id).await; + evict_conn_channel_subscriptions(state, ctx, channel_id, conn_id).await; } } @@ -48,6 +49,7 @@ async fn evict_live_channel_subscriptions( /// the connection's local map and sending `CLOSED restricted` for each. async fn evict_conn_channel_subscriptions( state: &Arc, + ctx: &buzz_core::TenantContext, channel_id: Uuid, conn_id: uuid::Uuid, ) { @@ -58,6 +60,14 @@ async fn evict_conn_channel_subscriptions( return; } + // Every evicted sub was channel-scoped, so each held interest in the same + // `Channel(channel_id)` topic. Release once per removed sub to balance the + // pubsub refcount; the manager debounce-UNSUBSCRIBEs when interest hits zero. + let topic = buzz_pubsub::EventTopic::Channel(channel_id); + for _ in &removed { + state.pubsub.release_topic(ctx, topic).await; + } + if let Some(subscriptions) = state.conn_manager.subscriptions_for(conn_id) { let mut conn_subscriptions = subscriptions.lock().await; for sub_id in &removed { @@ -80,6 +90,7 @@ async fn evict_conn_channel_subscriptions( /// subscriptions promptly so clients stop treating the channel as live. async fn evict_non_member_channel_subscriptions( state: &Arc, + ctx: &buzz_core::TenantContext, channel_id: Uuid, ) -> anyhow::Result<()> { let members = state.db.get_members(channel_id).await?; @@ -92,7 +103,7 @@ async fn evict_non_member_channel_subscriptions( None => false, }; if !is_member { - evict_conn_channel_subscriptions(state, channel_id, conn_id).await; + evict_conn_channel_subscriptions(state, ctx, channel_id, conn_id).await; } } Ok(()) @@ -107,9 +118,13 @@ async fn evict_non_member_channel_subscriptions( /// connected agent drops just that channel and keeps its socket — no reconnect /// storm. Offline/reconnecting clients are covered by the discovery-time /// `archived=true` skip in `discover_channels`. -pub async fn evict_all_channel_subscriptions(state: &Arc, channel_id: Uuid) { +pub async fn evict_all_channel_subscriptions( + state: &Arc, + ctx: &buzz_core::TenantContext, + channel_id: Uuid, +) { for conn_id in state.sub_registry.channel_subscriber_conns(channel_id) { - evict_conn_channel_subscriptions(state, channel_id, conn_id).await; + evict_conn_channel_subscriptions(state, ctx, channel_id, conn_id).await; } } @@ -1003,7 +1018,7 @@ async fn handle_remove_user( .remove_member(channel_id, &target_pubkey, &actor_bytes) .await?; state.invalidate_membership(ctx, channel_id, &target_pubkey); - evict_live_channel_subscriptions(state, channel_id, &target_pubkey).await; + evict_live_channel_subscriptions(state, ctx, channel_id, &target_pubkey).await; let actor_hex = hex::encode(&actor_bytes); let target_hex = hex::encode(&target_pubkey); @@ -1132,7 +1147,7 @@ async fn handle_edit_metadata( // for an immediate CLOSED on this node. The fan-out access // filter is the cluster-wide correctness backstop. if was_open && val == "private" { - evict_non_member_channel_subscriptions(state, channel_id).await?; + evict_non_member_channel_subscriptions(state, ctx, channel_id).await?; } emit_system_message( state, @@ -1589,7 +1604,7 @@ async fn handle_leave_request( .remove_member(channel_id, &actor_bytes, &actor_bytes) .await?; state.invalidate_membership(ctx, channel_id, &actor_bytes); - evict_live_channel_subscriptions(state, channel_id, &actor_bytes).await; + evict_live_channel_subscriptions(state, ctx, channel_id, &actor_bytes).await; let actor_hex = hex::encode(&actor_bytes); emit_system_message( diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index a29d0a0967..d8d45dbbbe 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -436,6 +436,7 @@ async fn main() -> anyhow::Result<()> { // by the archived=true skip in discover_channels on reconnect. buzz_relay::handlers::side_effects::evict_all_channel_subscriptions( &reaper_state, + &ctx, *channel_id, ) .await; diff --git a/crates/buzz-relay/src/subscription.rs b/crates/buzz-relay/src/subscription.rs index 1b455bd696..ba909a2117 100644 --- a/crates/buzz-relay/src/subscription.rs +++ b/crates/buzz-relay/src/subscription.rs @@ -7,6 +7,7 @@ use nostr::{Alphabet, Filter, Kind, SingleLetterTag}; use uuid::Uuid; use buzz_core::{filter::filters_match, StoredEvent}; +use buzz_pubsub::EventTopic; /// Connection identifier — a UUID assigned to each WebSocket connection. pub type ConnId = Uuid; @@ -15,6 +16,22 @@ pub type SubId = String; /// Stored subscription entry: filters paired with an optional channel scope. pub type SubEntry = (Vec, Option); +/// The pub/sub routing topic a subscription declares local interest in. +/// +/// A channel-scoped subscription wants its exact channel topic; a global +/// subscription wants the community-global topic. The registry reports these on +/// register/remove so the caller can drive `retain_topic`/`release_topic` +/// against the pubsub manager with that connection's tenant. The topic is +/// community-agnostic here — the community is bound at the call site via +/// `TenantContext`, and the pubsub manager's own refcount is keyed by the full +/// `(community, topic)` pair. +fn sub_topic(channel_id: Option) -> EventTopic { + match channel_id { + Some(id) => EventTopic::Channel(id), + None => EventTopic::Global, + } +} + /// Index key combining a channel and event kind for O(1) fan-out lookups. #[derive(Debug, Clone, PartialEq, Eq, Hash)] pub struct IndexKey { @@ -54,14 +71,20 @@ impl SubscriptionRegistry { } /// Replaces any existing subscription with the same sub_id (NIP-01). + /// + /// Returns the routing topic of the subscription this call *displaced* (the + /// prior sub with the same `sub_id`), if any. The caller retains the new + /// subscription's topic and releases the displaced one against the pubsub + /// manager, keeping the per-`(community, topic)` refcount balanced across an + /// in-place replace. The new topic is [`sub_topic`] of `channel_id`. pub fn register( &self, conn_id: ConnId, sub_id: SubId, filters: Vec, channel_id: Option, - ) { - self.remove_subscription(conn_id, &sub_id); + ) -> Option { + let displaced = self.remove_subscription(conn_id, &sub_id); self.subs .entry(conn_id) @@ -129,30 +152,49 @@ impl SubscriptionRegistry { } } } + + displaced } /// Remove a single subscription and clean up its index entries. - pub fn remove_subscription(&self, conn_id: ConnId, sub_id: &str) { + /// + /// Returns the routing topic of the removed subscription, if one existed, so + /// the caller can release that topic against the pubsub manager. + pub fn remove_subscription(&self, conn_id: ConnId, sub_id: &str) -> Option { if let Some(mut conn_subs) = self.subs.get_mut(&conn_id) { if let Some((filters, channel_id)) = conn_subs.remove(sub_id) { self.remove_from_index(conn_id, sub_id, &filters, channel_id); metrics::gauge!("buzz_subscriptions_active").decrement(1.0); + return Some(sub_topic(channel_id)); } } + None } /// Remove all subscriptions for a connection and clean up index entries. - pub fn remove_connection(&self, conn_id: ConnId) { + /// + /// Returns the routing topic of every removed subscription (one entry per + /// subscription, so a topic with N subs on this connection appears N times) + /// so the caller can release each against the pubsub manager. The pubsub + /// refcount collapses the N releases to a single debounced UNSUBSCRIBE. + pub fn remove_connection(&self, conn_id: ConnId) -> Vec { + let mut released = Vec::new(); if let Some((_, conn_subs)) = self.subs.remove(&conn_id) { let count = conn_subs.len(); for (sub_id, (filters, channel_id)) in &conn_subs { self.remove_from_index(conn_id, sub_id, filters, *channel_id); + released.push(sub_topic(*channel_id)); } metrics::gauge!("buzz_subscriptions_active").decrement(count as f64); } + released } /// Remove all subscriptions on `conn_id` scoped to `channel_id`. + /// + /// Returns the removed sub ids. Every removed subscription shares the same + /// routing topic — `EventTopic::Channel(channel_id)` — so the caller + /// releases that topic once per returned id to balance the pubsub refcount. pub fn remove_channel_subscriptions(&self, conn_id: ConnId, channel_id: Uuid) -> Vec { let sub_ids: Vec = self .subs @@ -1212,4 +1254,141 @@ mod tests { "only channel sub sees channel event" ); } + + // --- Topic retain/release accounting: the registry reports the exact + // topics the caller must retain on register and release on remove, so + // the pubsub manager's `(community, topic)` refcount stays balanced. --- + + #[test] + fn test_register_channel_sub_reports_no_displaced_topic() { + let registry = SubscriptionRegistry::new(); + let conn_id = Uuid::new_v4(); + let channel_id = Uuid::new_v4(); + + // First registration of a sub_id displaces nothing. + let displaced = registry.register( + conn_id, + "s".to_string(), + vec![Filter::new().kind(Kind::TextNote)], + Some(channel_id), + ); + assert_eq!(displaced, None); + } + + #[test] + fn test_register_replace_same_topic_reports_displaced() { + // Replacing a sub_id with another sub on the SAME channel reports the + // displaced topic, so the caller does retain(new)+release(old): a + // net-zero on the shared topic, no SUBSCRIBE churn. + let registry = SubscriptionRegistry::new(); + let conn_id = Uuid::new_v4(); + let channel_id = Uuid::new_v4(); + + registry.register( + conn_id, + "s".to_string(), + vec![Filter::new().kind(Kind::TextNote)], + Some(channel_id), + ); + let displaced = registry.register( + conn_id, + "s".to_string(), + vec![Filter::new().kind(Kind::Metadata)], + Some(channel_id), + ); + assert_eq!(displaced, Some(EventTopic::Channel(channel_id))); + } + + #[test] + fn test_register_replace_channel_to_global_reports_old_channel_topic() { + // Replacing a channel-scoped sub with a global one displaces the channel + // topic; the caller releases Channel(id) and retains Global. + let registry = SubscriptionRegistry::new(); + let conn_id = Uuid::new_v4(); + let channel_id = Uuid::new_v4(); + + registry.register( + conn_id, + "s".to_string(), + vec![Filter::new().kind(Kind::TextNote)], + Some(channel_id), + ); + let displaced = registry.register( + conn_id, + "s".to_string(), + vec![Filter::new().kind(Kind::TextNote)], + None, + ); + assert_eq!(displaced, Some(EventTopic::Channel(channel_id))); + } + + #[test] + fn test_remove_subscription_reports_topic() { + let registry = SubscriptionRegistry::new(); + let conn_id = Uuid::new_v4(); + let channel_id = Uuid::new_v4(); + + registry.register( + conn_id, + "ch".to_string(), + vec![Filter::new().kind(Kind::TextNote)], + Some(channel_id), + ); + registry.register(conn_id, "gl".to_string(), vec![Filter::new()], None); + + assert_eq!( + registry.remove_subscription(conn_id, "ch"), + Some(EventTopic::Channel(channel_id)) + ); + assert_eq!( + registry.remove_subscription(conn_id, "gl"), + Some(EventTopic::Global) + ); + // Removing an absent sub reports nothing — caller releases nothing. + assert_eq!(registry.remove_subscription(conn_id, "missing"), None); + } + + #[test] + fn test_remove_connection_reports_every_subscription_topic() { + // One entry per subscription (not per distinct topic) so the caller's + // release count matches the retain count it issued at register time. + let registry = SubscriptionRegistry::new(); + let conn_id = Uuid::new_v4(); + let channel_a = Uuid::new_v4(); + let channel_b = Uuid::new_v4(); + + registry.register( + conn_id, + "a1".to_string(), + vec![Filter::new().kind(Kind::TextNote)], + Some(channel_a), + ); + registry.register( + conn_id, + "a2".to_string(), + vec![Filter::new().kind(Kind::Metadata)], + Some(channel_a), + ); + registry.register( + conn_id, + "b1".to_string(), + vec![Filter::new().kind(Kind::TextNote)], + Some(channel_b), + ); + registry.register(conn_id, "g1".to_string(), vec![Filter::new()], None); + + let mut released = registry.remove_connection(conn_id); + released.sort_by_key(|t| format!("{t:?}")); + let mut expected = vec![ + EventTopic::Channel(channel_a), + EventTopic::Channel(channel_a), + EventTopic::Channel(channel_b), + EventTopic::Global, + ]; + expected.sort_by_key(|t| format!("{t:?}")); + assert_eq!(released, expected); + + // A second drop reports nothing — no double-release. + assert!(registry.remove_connection(conn_id).is_empty()); + } }