diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index b7773e04ef..a03708e594 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -2402,6 +2402,13 @@ async fn handle_ws_message( subscription_id, message, } => { + // Every CLOSED for an exact channel subscription revokes its + // accepted-route authority before any branch can keep the + // socket, park a retry, drop membership, or reconnect. + if let Some(channel_id) = channel_id_from_sub_id(&subscription_id) { + state.admitted_routes.write().await.remove(&channel_id); + } + // 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 @@ -2437,10 +2444,6 @@ async fn handle_ws_message( return true; // keep the socket } - // CLOSED revokes accepted route before any recovery decision. - if let Some(channel_id) = channel_id_from_sub_id(&subscription_id) { - state.admitted_routes.write().await.remove(&channel_id); - } // CLOSED needs cleanup and resubscribe, not just logging. let is_auth_error = message.starts_with("auth-required") || message.starts_with("restricted") diff --git a/crates/buzz-acp/src/relay/recovery_tests.rs b/crates/buzz-acp/src/relay/recovery_tests.rs index 0d2e63eedd..32db33a563 100644 --- a/crates/buzz-acp/src/relay/recovery_tests.rs +++ b/crates/buzz-acp/src/relay/recovery_tests.rs @@ -35,6 +35,35 @@ async fn dispatch( ); } +#[tokio::test] +async fn exact_channel_closed_revokes_route_before_keep_socket_recovery_branches() { + let (mut client, _server) = test_ws_pair().await; + let mut state = BgState::new(); + let (tx, _rx) = mpsc::channel(1); + + for reason in [ + "restricted: channel access revoked", + "rate-limited: retry in 5s", + ] { + let channel = Uuid::new_v4(); + seed_test_subscription(&mut state, channel); + let sub = channel_sub_id(channel); + + dispatch(&mut client, &mut state, &tx, json!(["EOSE", sub])).await; + assert!(state.admitted_routes.read().await.contains(&channel)); + + dispatch(&mut client, &mut state, &tx, json!(["CLOSED", sub, reason])).await; + assert!( + !state.admitted_routes.read().await.contains(&channel), + "{reason} must revoke before its keep-socket recovery branch returns" + ); + assert!( + reason.starts_with("restricted") || state.rate_limited_pending.contains_key(&channel), + "rate-limited replacement must remain parked after authority is revoked" + ); + } +} + #[tokio::test] async fn repeated_overflow_recovers_only_affected_channel_after_capacity() { let (mut client, mut server) = test_ws_pair().await;