fix(buzz): prevent reconnect storms from reaped ephemeral channels (#1071)

Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
Co-authored-by: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@sprout-oss.stage.blox.sqprod.co>

crates/buzz-acp/src/relay.rs

@@ -85,6 +85,81 @@ pub struct ChannelInfo {
     pub channel_type: String,
 }
 
+/// Build the discovered-channel subscribe set from the membership UUIDs and the
+/// kind:39000 metadata events, **skipping any channel flagged `archived=true`**.
+///
+/// Archived channels (e.g. auto-archived by the ephemeral-channel reaper) are
+/// unusable: re-offering one on reconnect draws a `CLOSED restricted` and would
+/// re-form the reconnect loop. Dropping them here is the defense-in-depth
+/// backstop to the relay-side live-subscription eviction — it covers a client
+/// that was offline when the channel was reaped and so missed the CLOSED.
+/// A channel with no metadata event defaults to a `stream` named `unknown`,
+/// preserving prior behavior for non-archived channels.
+fn merge_discovered_channels(
+    channel_uuids: Vec<Uuid>,
+    meta_events: &serde_json::Value,
+) -> HashMap<Uuid, ChannelInfo> {
+    let mut meta_map: HashMap<Uuid, (String, String)> = HashMap::new();
+    let mut archived: std::collections::HashSet<Uuid> = std::collections::HashSet::new();
+    if let Some(arr) = meta_events.as_array() {
+        for ev in arr {
+            let tags = match ev.get("tags").and_then(|t| t.as_array()) {
+                Some(t) => t,
+                None => continue,
+            };
+            let mut d_val = None;
+            let mut name = None;
+            let mut is_hidden = false;
+            let mut is_private = false;
+            let mut is_archived = false;
+            for tag in tags {
+                if let Some(arr) = tag.as_array() {
+                    match arr.first().and_then(|v| v.as_str()) {
+                        Some("d") => d_val = arr.get(1).and_then(|v| v.as_str()),
+                        Some("name") => name = arr.get(1).and_then(|v| v.as_str()),
+                        Some("hidden") => is_hidden = true,
+                        Some("private") => is_private = true,
+                        Some("archived") => {
+                            is_archived = arr.get(1).and_then(|v| v.as_str()) == Some("true")
+                        }
+                        _ => {}
+                    }
+                }
+            }
+            if let Some(d) = d_val {
+                if let Ok(uuid) = d.parse::<Uuid>() {
+                    if is_archived {
+                        archived.insert(uuid);
+                        continue;
+                    }
+                    let ch_name = name.unwrap_or("unknown").to_string();
+                    // DMs have the "hidden" tag; private channels have "private".
+                    let ch_type = if is_hidden {
+                        "dm".to_string()
+                    } else if is_private {
+                        "private".to_string()
+                    } else {
+                        "stream".to_string()
+                    };
+                    meta_map.insert(uuid, (ch_name, ch_type));
+                }
+            }
+        }
+    }
+
+    let mut map = HashMap::with_capacity(channel_uuids.len());
+    for uuid in channel_uuids {
+        if archived.contains(&uuid) {
+            continue;
+        }
+        let (name, channel_type) = meta_map
+            .remove(&uuid)
+            .unwrap_or_else(|| ("unknown".to_string(), "stream".to_string()));
+        map.insert(uuid, ChannelInfo { name, channel_type });
+    }
+    map
+}
+
 /// Lightweight HTTP client for pre-prompt context fetches via the Nostr HTTP bridge.
 ///
 /// Extracted from `HarnessRelay` fields so it can be shared (via `Arc`) with
@@ -565,54 +640,8 @@ impl HarnessRelay {
             .custom_tags(d_tag, d_values);
         let meta_events = rest.query(&[meta_filter]).await?;
 
-        // Build UUID → (name, channel_type) from metadata events.
-        let mut meta_map: HashMap<Uuid, (String, String)> = HashMap::new();
-        if let Some(arr) = meta_events.as_array() {
-            for ev in arr {
-                let tags = match ev.get("tags").and_then(|t| t.as_array()) {
-                    Some(t) => t,
-                    None => continue,
-                };
-                let mut d_val = None;
-                let mut name = None;
-                let mut is_hidden = false;
-                let mut is_private = false;
-                for tag in tags {
-                    if let Some(arr) = tag.as_array() {
-                        match arr.first().and_then(|v| v.as_str()) {
-                            Some("d") => d_val = arr.get(1).and_then(|v| v.as_str()),
-                            Some("name") => name = arr.get(1).and_then(|v| v.as_str()),
-                            Some("hidden") => is_hidden = true,
-                            Some("private") => is_private = true,
-                            _ => {}
-                        }
-                    }
-                }
-                if let Some(d) = d_val {
-                    if let Ok(uuid) = d.parse::<Uuid>() {
-                        let ch_name = name.unwrap_or("unknown").to_string();
-                        // DMs have the "hidden" tag; private channels have "private".
-                        let ch_type = if is_hidden {
-                            "dm".to_string()
-                        } else if is_private {
-                            "private".to_string()
-                        } else {
-                            "stream".to_string()
-                        };
-                        meta_map.insert(uuid, (ch_name, ch_type));
-                    }
-                }
-            }
-        }
-
-        // Step 3: Merge into final map.
-        let mut map = HashMap::with_capacity(channel_uuids.len());
-        for uuid in channel_uuids {
-            let (name, channel_type) = meta_map
-                .remove(&uuid)
-                .unwrap_or_else(|| ("unknown".to_string(), "stream".to_string()));
-            map.insert(uuid, ChannelInfo { name, channel_type });
-        }
+        // Step 3: Build the final subscribe set, skipping archived channels.
+        let map = merge_discovered_channels(channel_uuids, &meta_events);
 
         debug!("discovered {} channel(s)", map.len());
         Ok(map)
@@ -1748,6 +1777,15 @@ async fn handle_ws_message(
                     subscription_id,
                     message,
                 } => {
+                    // A per-channel membership denial means THIS channel is
+                    // forbidden, not the whole connection. Drop just this
+                    // channel's subscription and keep the socket — otherwise the
+                    // socket is torn down, the forbidden channel is resubscribed,
+                    // and the same CLOSED arrives again: a tight reconnect loop.
+                    if drop_channel_on_access_denied(state, &subscription_id, &message) {
+                        return true;
+                    }
+
                     // Finding #15: CLOSED needs cleanup and resubscribe, not just logging.
                     // Classify the error to decide how to respond.
                     let is_auth_error = message.starts_with("auth-required")
@@ -2565,6 +2603,46 @@ fn channel_id_from_sub_id(sub_id: &str) -> Option<Uuid> {
         .and_then(|s| s.parse::<Uuid>().ok())
 }
 
+/// Per-channel CLOSED denials: the channel is forbidden but the connection is
+/// fine. Match these EXACT strings, never a `starts_with("restricted")` prefix —
+/// a prefix would also swallow connection-level `restricted: insufficient scope`,
+/// dropping a channel instead of reconnecting. The only CLOSED senders of these
+/// strings are `req.rs:153` (not a channel member) and `side_effects.rs:71`
+/// (channel access revoked, via member eviction / open→private flip).
+/// `ingest.rs` returns these as EVENT-publish `OK(false)`, never as a
+/// subscription CLOSED, so it is not a source here.
+const CHANNEL_ACCESS_DENIED_REASONS: &[&str] = &[
+    "restricted: not a channel member",
+    "restricted: channel access revoked",
+];
+
+/// Handle a CLOSED that denies access to a single channel: drop just that
+/// channel's subscription (the proven Unsubscribe cleanup) and keep the socket.
+///
+/// Returns `true` when the CLOSED was an exact per-channel denial on a `ch-`
+/// subscription and the channel was dropped — the caller keeps the connection
+/// with no reconnect. Returns `false` for everything else (connection-level
+/// `restricted: insufficient scope`, `auth-required`, non-channel subs), which
+/// falls through to the existing reconnect path.
+///
+/// An already-removed channel is a harmless no-op: the remove/clear simply
+/// affect nothing, and the dropped channel is never re-subscribed, so the loop
+/// cannot re-form.
+fn drop_channel_on_access_denied(state: &mut BgState, sub_id: &str, message: &str) -> bool {
+    if !CHANNEL_ACCESS_DENIED_REASONS.contains(&message) {
+        return false;
+    }
+    let Some(channel_id) = channel_id_from_sub_id(sub_id) else {
+        return false;
+    };
+    warn!(
+        "channel {channel_id} access denied by relay: {message} — dropping subscription, keeping connection"
+    );
+    state.active_subscriptions.remove(&channel_id);
+    state.clear_channel_state(&channel_id);
+    true
+}
+
 /// Apply the appropriate auth header to a reqwest request builder.
 /// Parse a raw relay text frame into a typed [`RelayMessage`].
 #[allow(private_interfaces)]
@@ -2918,6 +2996,73 @@ mod tests {
         assert!(channel_id_from_sub_id("").is_none());
     }
 
+    // ── merge_discovered_channels (archived skip) ─────────────────────────────
+
+    fn meta_event(uuid: Uuid, name: &str, extra: &[&str]) -> serde_json::Value {
+        let mut tags = vec![
+            serde_json::json!(["d", uuid.to_string()]),
+            serde_json::json!(["name", name]),
+        ];
+        // `extra` is a flat list of single-value tag names (e.g. archived=true).
+        for pair in extra.chunks(2) {
+            match pair {
+                [k, v] => tags.push(serde_json::json!([k, v])),
+                [k] => tags.push(serde_json::json!([k])),
+                _ => {}
+            }
+        }
+        serde_json::json!({ "tags": tags })
+    }
+
+    #[test]
+    fn merge_discovered_channels_skips_archived_metadata() {
+        let live = Uuid::new_v4();
+        let archived = Uuid::new_v4();
+        let meta = serde_json::json!([
+            meta_event(live, "live", &[]),
+            meta_event(archived, "dead", &["archived", "true"]),
+        ]);
+
+        let map = merge_discovered_channels(vec![live, archived], &meta);
+
+        assert!(map.contains_key(&live), "non-archived channel is kept");
+        assert!(
+            !map.contains_key(&archived),
+            "archived=true channel is skipped from the subscribe set"
+        );
+        assert_eq!(map.len(), 1);
+    }
+
+    #[test]
+    fn merge_discovered_channels_skips_archived_even_when_still_a_member() {
+        // The offline feeder: the agent is still listed as a member
+        // (uuid present in channel_uuids, the kind:39002 membership set), but the
+        // channel was reaped while the agent was offline. Even though the agent
+        // missed the eviction CLOSED, the archived=true kind:39000 makes the
+        // client skip re-subscribing on reconnect — proving (b) closes the loop
+        // independently of the relay-side eviction.
+        let reaped = Uuid::new_v4();
+        let meta = serde_json::json!([meta_event(reaped, "reaped", &["archived", "true"])]);
+
+        let map = merge_discovered_channels(vec![reaped], &meta);
+
+        assert!(
+            map.is_empty(),
+            "a still-member but archived channel is not re-subscribed"
+        );
+    }
+
+    #[test]
+    fn merge_discovered_channels_archived_false_is_kept() {
+        // An explicit archived=false (e.g. after unarchive) must NOT be skipped.
+        let ch = Uuid::new_v4();
+        let meta = serde_json::json!([meta_event(ch, "back", &["archived", "false"])]);
+
+        let map = merge_discovered_channels(vec![ch], &meta);
+
+        assert!(map.contains_key(&ch), "archived=false is treated as live");
+    }
+
     // ── parse_relay_message ───────────────────────────────────────────────────
 
     #[test]
@@ -3478,4 +3623,147 @@ mod tests {
             "after backpressure removal, replay must be accepted"
         );
     }
+
+    // ── drop_channel_on_access_denied (Phenomenon A: reconnect loop) ──────────
+
+    /// Subscribe a channel via the production command path so the test exercises
+    /// real subscription state (active_subscriptions + active_filters + since).
+    fn subscribe_channel(state: &mut BgState, channel_id: Uuid) {
+        apply_command_to_state(
+            state,
+            RelayCommand::Subscribe {
+                channel_id,
+                filter: ChannelFilter {
+                    kinds: Some(vec![9]),
+                    require_mention: false,
+                },
+                replay_since: Some(1_000),
+            },
+        );
+    }
+
+    #[test]
+    fn not_a_channel_member_drops_channel_without_reconnect() {
+        let mut state = BgState::new();
+        let channel_id = Uuid::new_v4();
+        subscribe_channel(&mut state, channel_id);
+
+        let handled = drop_channel_on_access_denied(
+            &mut state,
+            &channel_sub_id(channel_id),
+            "restricted: not a channel member",
+        );
+
+        assert!(handled, "per-channel denial must be handled (no reconnect)");
+        assert!(
+            !state.active_subscriptions.contains_key(&channel_id),
+            "the forbidden channel's subscription must be dropped"
+        );
+        assert!(
+            !state.active_filters.contains_key(&channel_id),
+            "channel state must be cleared (Unsubscribe cleanup)"
+        );
+    }
+
+    #[test]
+    fn channel_access_revoked_drops_channel_without_reconnect() {
+        let mut state = BgState::new();
+        let channel_id = Uuid::new_v4();
+        subscribe_channel(&mut state, channel_id);
+
+        let handled = drop_channel_on_access_denied(
+            &mut state,
+            &channel_sub_id(channel_id),
+            "restricted: channel access revoked",
+        );
+
+        assert!(handled, "per-channel denial must be handled (no reconnect)");
+        assert!(!state.active_subscriptions.contains_key(&channel_id));
+        assert!(!state.active_filters.contains_key(&channel_id));
+    }
+
+    #[test]
+    fn insufficient_scope_is_not_dropped_and_reconnects() {
+        let mut state = BgState::new();
+        let channel_id = Uuid::new_v4();
+        subscribe_channel(&mut state, channel_id);
+
+        let handled = drop_channel_on_access_denied(
+            &mut state,
+            &channel_sub_id(channel_id),
+            "restricted: insufficient scope",
+        );
+
+        assert!(
+            !handled,
+            "connection-level insufficient-scope must fall through to reconnect, not drop the channel"
+        );
+        assert!(
+            state.active_subscriptions.contains_key(&channel_id),
+            "the channel must survive so reconnect can restore it"
+        );
+    }
+
+    #[test]
+    fn auth_required_is_not_dropped_and_reconnects() {
+        let mut state = BgState::new();
+        let channel_id = Uuid::new_v4();
+        subscribe_channel(&mut state, channel_id);
+
+        let handled = drop_channel_on_access_denied(
+            &mut state,
+            &channel_sub_id(channel_id),
+            "auth-required: not authenticated",
+        );
+
+        assert!(
+            !handled,
+            "auth-required must fall through to reconnect, not drop the channel"
+        );
+        assert!(state.active_subscriptions.contains_key(&channel_id));
+    }
+
+    #[test]
+    fn already_removed_channel_is_a_no_op() {
+        let mut state = BgState::new();
+        let channel_id = Uuid::new_v4();
+        // Channel was never subscribed (or already dropped) — a delayed CLOSED.
+
+        let handled = drop_channel_on_access_denied(
+            &mut state,
+            &channel_sub_id(channel_id),
+            "restricted: not a channel member",
+        );
+
+        assert!(
+            handled,
+            "an exact per-channel denial is still handled (keep socket) even if the channel is gone"
+        );
+        assert!(
+            !state.active_subscriptions.contains_key(&channel_id),
+            "no-op: nothing to remove and nothing resurrected"
+        );
+    }
+
+    #[test]
+    fn dropped_channel_is_not_resubscribed_so_loop_cannot_re_form() {
+        let mut state = BgState::new();
+        let channel_id = Uuid::new_v4();
+        subscribe_channel(&mut state, channel_id);
+
+        drop_channel_on_access_denied(
+            &mut state,
+            &channel_sub_id(channel_id),
+            "restricted: not a channel member",
+        );
+
+        // Simulate a reconnect: only channels still in active_subscriptions are
+        // restored. The dropped channel must not be among them — otherwise the
+        // forbidden channel would be resubscribed and earn the same CLOSED again.
+        let resubscribed: Vec<Uuid> = state.active_subscriptions.keys().copied().collect();
+        assert!(
+            !resubscribed.contains(&channel_id),
+            "the dropped channel must not be resubscribed — the loop cannot re-form"
+        );
+    }
 }

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

@@ -97,6 +97,21 @@ async fn evict_non_member_channel_subscriptions(
     Ok(())
 }
 
+/// Close every live subscription on a channel, for all subscribers, sending
+/// `CLOSED restricted: channel access revoked` to each.
+///
+/// Used when a channel is archived (e.g. the ephemeral-channel reaper): the
+/// channel becomes unusable for everyone, so all live subscriptions must close.
+/// The `channel access revoked` reason is in the client's drop-set, so a
+/// 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) {
+    for conn_id in state.sub_registry.channel_subscriber_conns(channel_id) {
+        evict_conn_channel_subscriptions(state, channel_id, conn_id).await;
+    }
+}
+
 /// Dispatch side effects for a stored event.
 pub async fn handle_side_effects(
     kind: u32,

crates/buzz-relay/src/main.rs

@@ -389,6 +389,16 @@ async fn main() -> anyhow::Result<()> {
                     {
                         error!(channel = %channel_id, "reaper discovery update failed: {e}");
                     }
+
+                    // Close live subscriptions so connected clients drop the
+                    // archived channel immediately (CLOSED is in the client's
+                    // drop-set → no reconnect storm). Offline clients are caught
+                    // by the archived=true skip in discover_channels on reconnect.
+                    buzz_relay::handlers::side_effects::evict_all_channel_subscriptions(
+                        &reaper_state,
+                        *channel_id,
+                    )
+                    .await;
                 }
             }
         });

crates/buzz-relay/src/subscription.rs

@@ -987,6 +987,51 @@ mod tests {
         assert_eq!(conns, expected);
     }
 
+    #[test]
+    fn test_evict_all_subscribers_leaves_no_channel_subscriptions() {
+        // Models the reaper's evict_all_channel_subscriptions: remove every
+        // subscriber conn of a channel, leaving the channel with zero live subs
+        // while an unrelated channel is untouched. This is the registry behavior
+        // that lets an auto-archived channel stop fanning out without a storm.
+        let registry = SubscriptionRegistry::new();
+        let conn_a = Uuid::new_v4();
+        let conn_b = Uuid::new_v4();
+        let reaped = Uuid::new_v4();
+        let survivor = Uuid::new_v4();
+
+        registry.register(
+            conn_a,
+            "a".to_string(),
+            vec![Filter::new().kind(Kind::TextNote)],
+            Some(reaped),
+        );
+        registry.register(
+            conn_b,
+            "b".to_string(),
+            vec![Filter::new().kind(Kind::TextNote)],
+            Some(reaped),
+        );
+        registry.register(
+            conn_a,
+            "keep".to_string(),
+            vec![Filter::new().kind(Kind::TextNote)],
+            Some(survivor),
+        );
+
+        for conn_id in registry.channel_subscriber_conns(reaped) {
+            registry.remove_channel_subscriptions(conn_id, reaped);
+        }
+
+        assert!(registry.channel_subscriber_conns(reaped).is_empty());
+        assert!(registry
+            .fan_out(&make_stored_event(Kind::TextNote, Some(reaped)))
+            .is_empty());
+        // Unrelated channel still fans out.
+        let survivor_matches = registry.fan_out(&make_stored_event(Kind::TextNote, Some(survivor)));
+        assert_eq!(survivor_matches.len(), 1);
+        assert_eq!(survivor_matches[0].1, "keep");
+    }
+
     #[test]
     fn test_global_kind_index_fan_out() {
         // Global subscriptions with explicit kinds should use the global_kind_index