feat(relay): wire retain_topic/release_topic into subscription lifecycle

The dynamic refcounted Redis subscriber (Max's pubsub seam) only SUBSCRIBEs
a community-scoped topic once a pod declares local interest via retain_topic,
and debounce-UNSUBSCRIBEs on the last release. Until now nothing drove those
calls from real subscription activity, so cross-pod delivery only worked via
a manual retain in the presence test setup.

The SubscriptionRegistry stays synchronous and tenant-agnostic — it does not
grow a PubSubManager or TenantContext. Instead each mutator now reports the
routing topic(s) it added/removed:
  - register     -> Option<EventTopic> displaced by an in-place sub_id replace
  - remove_subscription -> Option<EventTopic> removed
  - remove_connection   -> Vec<EventTopic>, one per removed sub

The call sites — where &conn.tenant and state.pubsub already live — apply the
delta: retain the new topic on REQ, release on CLOSE / connection drop /
channel-access revocation. The pubsub manager's own (community, topic)
refcount collapses N releases to one debounced UNSUBSCRIBE and keeps two
communities sharing a channel UUID isolated, since the community is bound at
the call site via ctx. Every subscription contributes exactly one retain at
creation and one release at teardown — balanced, no registry-side counter.

The presence cross-pod test now drives retain through the real register()
path (register_global_sub is async) instead of a manual workaround, proving
delivery from genuine subscription interest. Five new registry unit tests
pin the topic-reporting contract including the replace-displaces-old case.

cargo test -p buzz-relay: 378 passed, 0 failed. Redis round-trip presence
test exercised against local Redis.

Co-authored-by: Tyler Longwell <tlongwell@block.xyz>
Signed-off-by: Tyler Longwell <tlongwell@block.xyz>

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

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

@@ -14,7 +14,11 @@ pub async fn handle_close(sub_id: String, conn: Arc<ConnectionState>, 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, ""));
 

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<Message>) {
@@ -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.

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");
 

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<AppState>,
+    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<AppState>,
+    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<AppState>,
+    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<AppState>, channel_id: Uuid) {
+pub async fn evict_all_channel_subscriptions(
+    state: &Arc<AppState>,
+    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(

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;

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<Filter>, Option<Uuid>);
 
+/// 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<Uuid>) -> 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<Filter>,
         channel_id: Option<Uuid>,
-    ) {
-        self.remove_subscription(conn_id, &sub_id);
+    ) -> Option<EventTopic> {
+        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<EventTopic> {
         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<EventTopic> {
+        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<SubId> {
         let sub_ids: Vec<SubId> = 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());
+    }
 }