diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index e4c357a611..5a9986644a 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/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, + meta_events: &serde_json::Value, +) -> HashMap { + let mut meta_map: HashMap = HashMap::new(); + let mut archived: std::collections::HashSet = 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::() { + 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 = 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::() { - 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 { .and_then(|s| s.parse::().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 = state.active_subscriptions.keys().copied().collect(); + assert!( + !resubscribed.contains(&channel_id), + "the dropped channel must not be resubscribed — the loop cannot re-form" + ); + } } diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 8c72142ede..e0825399b3 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/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, 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, diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 1a6b5fbc4a..20437a92bb 100644 --- a/crates/buzz-relay/src/main.rs +++ b/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; } } }); diff --git a/crates/buzz-relay/src/subscription.rs b/crates/buzz-relay/src/subscription.rs index 3b2169340c..1b455bd696 100644 --- a/crates/buzz-relay/src/subscription.rs +++ b/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