diff --git a/crates/buzz-relay/src/handlers/close.rs b/crates/buzz-relay/src/handlers/close.rs index f655c328d9..0e61e006e6 100644 --- a/crates/buzz-relay/src/handlers/close.rs +++ b/crates/buzz-relay/src/handlers/close.rs @@ -36,7 +36,9 @@ pub(crate) async fn remove_subscription( if let Some(removed) = registry.remove_subscription(conn.conn_id, sub_id) { if removed.scope.is_global() { - pubsub.release_topic(&conn.tenant, buzz_pubsub::EventTopic::Global).await; + pubsub + .release_topic(&conn.tenant, buzz_pubsub::EventTopic::Global) + .await; } for &channel_id in removed.scope.channel_ids() { pubsub @@ -47,4 +49,3 @@ pub(crate) async fn remove_subscription( drop(pending_subscriptions); } - diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index ee06765ff2..e8e5bd44d6 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -292,10 +292,14 @@ pub(crate) async fn handle_req( return; } + let subscription_scope = authorized_requested_channels + .clone() + .map(crate::subscription::SubscriptionScope::Channels) + .unwrap_or(crate::subscription::SubscriptionScope::Global); if !register_subscription_if_current( &sub_id, &filters, - authorized_requested_channels.as_deref(), + &subscription_scope, &conn, &state.sub_registry, state.pubsub.as_ref(), @@ -305,7 +309,6 @@ pub(crate) async fn handle_req( { return; } - } debug!(conn_id = %conn_id, sub_id = %sub_id, "Subscription registered"); @@ -498,7 +501,7 @@ const SEARCH_PAGE_SIZE: u32 = 100; async fn register_subscription_if_current( sub_id: &str, filters: &[Filter], - channel_id: Option, + scope: &crate::subscription::SubscriptionScope, conn: &ConnectionState, registry: &crate::subscription::SubscriptionRegistry, pubsub: &dyn SubscriptionTopics, @@ -517,21 +520,38 @@ async fn register_subscription_if_current( .await .insert(sub_id.to_owned(), filters.to_vec()); - let replaced = registry.register_scoped( - conn.tenant.community(), - conn.conn_id, - sub_id.to_owned(), - filters.to_vec(), - channel_id, - ); + let replaced = match scope { + crate::subscription::SubscriptionScope::Channels(channel_ids) => registry + .register_channels_scoped( + conn.tenant.community(), + conn.conn_id, + sub_id.to_owned(), + filters.to_vec(), + channel_ids.clone(), + ), + crate::subscription::SubscriptionScope::Global => registry.register_scoped( + conn.tenant.community(), + conn.conn_id, + sub_id.to_owned(), + filters.to_vec(), + None, + ), + }; if let Some(replaced) = replaced { - pubsub - .release_topic(&conn.tenant, topic_for_subscription(replaced.channel_id)) - .await; + release_subscription_topics(pubsub, &conn.tenant, &replaced.scope).await; + } + match scope { + crate::subscription::SubscriptionScope::Channels(channel_ids) => { + for &channel_id in channel_ids { + pubsub + .retain_topic(&conn.tenant, EventTopic::Channel(channel_id)) + .await; + } + } + crate::subscription::SubscriptionScope::Global => { + pubsub.retain_topic(&conn.tenant, EventTopic::Global).await; + } } - pubsub - .retain_topic(&conn.tenant, topic_for_subscription(channel_id)) - .await; // Only a successful replacement commit supersedes historical delivery from // the previous same-ID REQ. Validation/search failures leave it untouched. @@ -1204,16 +1224,15 @@ pub(crate) fn extract_channel_ids_from_filters(filters: &[Filter]) -> Option