jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-28 21:11:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(acp): revoke route authority on every channel close

Browse Source 
Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.local>
Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
Logan Johnson
2026-09-18 12:45:54 -04:00
parent 33ea81fc51
commit f76aad97ce
2 changed files with 36 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
crates/buzz-acp/src/relay.rs
View File

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

29
crates/buzz-acp/src/relay/recovery_tests.rs
View File

@@ -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;