diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index 4960c96196..709ad873c8 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -1400,10 +1400,14 @@ async fn handle_text_message(text: String, conn: Arc, state: Ar let permit = match state.handler_semaphore.clone().try_acquire_owned() { Ok(p) => p, Err(_) => { - conn.send(request_rejection_message( - RejectionTarget::Subscription(&sub_id), + // Retires whatever held the ID before `CLOSED`. + handlers::artifact_sync::refuse( + &sub_id, "rate-limited: too many concurrent requests", - )); + &conn, + &state, + ) + .await; return; } }; @@ -2134,6 +2138,79 @@ pub(crate) mod tests { assert_eq!(frame[2], "rate-limited: too many concurrent requests"); } + /// A sync REQ refused before its handler runs still replaces the old + /// holder of its ID (NIP-01): the `CLOSED` must not leave a live + /// subscription that keeps fanning out under a closed ID. Covers handler + /// saturation and WS admission, both through production dispatch. + /// Mutation: send either refusal without `retire_locked` → RED. + #[tokio::test] + async fn pre_handler_sync_refusal_retires_the_old_holder() { + use buzz_pubsub::EventTopic; + for (auth, saturate, reason) in [ + ( + AuthState::Failed, + true, + "rate-limited: too many concurrent requests", + ), + ( + authenticated_state(), + false, + "rate-limited: shared admission unavailable", + ), + ] { + // Redis is unreachable here, so authenticated admission is + // `Unavailable`; an unauthenticated connection skips admission. + let state = crate::state::tests::test_state().await; + let (conn, mut rx) = test_conn_with_auth(auth); + let channel = uuid::Uuid::new_v4(); + let topic = EventTopic::Channel(channel); + handlers::req::claim_live_subscription( + "s", + &[nostr::Filter::new().kind(Kind::TextNote)], + Some(&[channel]), + &conn, + &state, + ) + .await + .expect("claim the old holder"); + let _held = if saturate { + let permits = state.handler_semaphore.available_permits(); + Some( + Arc::clone(&state.handler_semaphore) + .acquire_many_owned(permits as u32) + .await + .expect("hold every handler permit"), + ) + } else { + None + }; + + let raw = serde_json::json!( + ["REQ", "s", {"artifact": "current", "#h": [channel.to_string()]}] + ) + .to_string(); + handle_text_message(raw, Arc::clone(&conn), Arc::clone(&state)).await; + + assert!(conn.subscriptions.lock().await.is_empty(), "{reason}"); + assert!( + state.sub_registry.get_filters(conn.conn_id, "s").is_none(), + "{reason}" + ); + assert!( + !state.artifact_syncs.contains(conn.conn_id, "s"), + "{reason}" + ); + assert_eq!( + state.pubsub.topic_refcount(&conn.tenant, topic).await, + 0, + "{reason}" + ); + let frame = read_frame(&mut rx); + assert_eq!(frame, serde_json::json!(["CLOSED", "s", reason])); + assert!(rx.try_recv().is_err(), "{reason}: only CLOSED is sent"); + } + } + #[derive(Debug, Default)] struct MockSinkState { messages: Vec, diff --git a/crates/buzz-relay/src/handlers/artifact_sync.rs b/crates/buzz-relay/src/handlers/artifact_sync.rs index 22f9ef1ab3..0eefecca77 100644 --- a/crates/buzz-relay/src/handlers/artifact_sync.rs +++ b/crates/buzz-relay/src/handlers/artifact_sync.rs @@ -437,8 +437,9 @@ const OVER_MAX_EVENTS: &str = "error: sync set exceeds max_events"; /// Refuse a sync `REQ` with `CLOSED reason`. A `REQ` replaces whatever held /// its ID (NIP-01), and `CLOSED` tells the client the ID is closed, so the -/// old holder is retired before the frame, under the lifecycle lock. -async fn refuse(sub_id: &str, reason: &str, conn: &ConnectionState, state: &AppState) { +/// old holder is retired before the frame, under the lifecycle lock. Every +/// refusal of a sync `REQ` goes through here, before or inside the handler. +pub(crate) async fn refuse(sub_id: &str, reason: &str, conn: &ConnectionState, state: &AppState) { let mut subs = conn.subscriptions.lock().await; super::close::retire_locked(&mut subs, sub_id, conn, state).await; conn.send(RelayMessage::closed(sub_id, reason)); diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index 94e84d279e..9b2484324c 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -1511,7 +1511,7 @@ async fn handle_huddle_liveness_req( /// owner token, register for fan-out (replacing any same-ID entry), retain its /// topics, and release the replaced scope's. Returns the owner token, or /// `None` once the connection is closing — the caller then exits silently. -async fn claim_live_subscription( +pub(crate) async fn claim_live_subscription( sub_id: &str, filters: &[Filter], channel_ids: Option<&[uuid::Uuid]>, diff --git a/crates/buzz-relay/src/rejection.rs b/crates/buzz-relay/src/rejection.rs index 7821ea3991..d2d3ec9ee0 100644 --- a/crates/buzz-relay/src/rejection.rs +++ b/crates/buzz-relay/src/rejection.rs @@ -84,7 +84,7 @@ pub(crate) async fn enforce_ws_admission( ws_limit, ) .await; - if !send_admission_result(conn, ws_result, msg, "ws_operations") { + if !send_admission_result(conn, state, ws_result, msg, "ws_operations").await { return false; } @@ -106,7 +106,7 @@ pub(crate) async fn enforce_ws_admission( .await; // Only persistable EVENT attempts consume the minute quota. Ephemeral activity // still passes the shared WS flood limit above and normal event auth. - if !send_admission_result(conn, message_result, msg, "messages") { + if !send_admission_result(conn, state, message_result, msg, "messages").await { return false; } } @@ -121,32 +121,38 @@ pub(crate) async fn enforce_ws_admission( /// caller: every quota check in this module must address its rejection to the /// rejected frame's own acknowledgement channel, so there is deliberately no way /// for a call site to name a different one. -fn send_admission_result( +async fn send_admission_result( conn: &ConnectionState, + state: &AppState, result: Result<(), AdmissionError>, msg: &ClientMessage, bucket: &'static str, ) -> bool { - let target = rejection_target_for(msg); - match result { - Ok(()) => true, - Err(AdmissionError::Exceeded { reset_in_secs }) => { - metrics::counter!("buzz_admission_rejections_total", "transport" => "websocket", "reason" => "quota", "bucket" => bucket).increment(1); - conn.send(request_rejection_message( - target, - &format!("rate-limited: quota exceeded; retry in {reset_in_secs}s"), - )); - false + let (reason, label) = match result { + Ok(()) => return true, + Err(AdmissionError::Exceeded { reset_in_secs }) => ( + format!("rate-limited: quota exceeded; retry in {reset_in_secs}s"), + "quota", + ), + Err(AdmissionError::Unavailable) => ( + "rate-limited: shared admission unavailable".to_owned(), + "unavailable", + ), + }; + metrics::counter!("buzz_admission_rejections_total", "transport" => "websocket", "reason" => label, "bucket" => bucket).increment(1); + match msg { + // A refused sync retires whatever held its ID before `CLOSED`. + ClientMessage::ArtifactSync { sub_id, .. } => { + crate::handlers::artifact_sync::refuse(sub_id, &reason, conn, state).await; } - Err(AdmissionError::Unavailable) => { - metrics::counter!("buzz_admission_rejections_total", "transport" => "websocket", "reason" => "unavailable", "bucket" => bucket).increment(1); + _ => { conn.send(request_rejection_message( - target, - "rate-limited: shared admission unavailable", + rejection_target_for(msg), + &reason, )); - false } } + false } #[cfg(test)] @@ -200,17 +206,20 @@ mod tests { /// The regression: an over-quota EVENT must be rejected with /// `OK(event_id, false, reason)` so the client can settle the exact pending /// publish it belongs to. A NOTICE here reintroduces the 25s send stall. - #[test] - fn over_quota_event_is_rejected_with_a_correlated_ok() { + #[tokio::test] + async fn over_quota_event_is_rejected_with_a_correlated_ok() { + let state = crate::state::tests::test_state().await; let (conn, mut rx) = test_conn(); let (msg, event_id) = parsed_event_message(); let admitted = send_admission_result( &conn, + &state, Err(AdmissionError::Exceeded { reset_in_secs: 7 }), &msg, "ws_operations", - ); + ) + .await; assert!(!admitted, "an over-quota frame is not admitted"); let frame = sent_frame(&mut rx); @@ -234,17 +243,20 @@ mod tests { /// The same correlation is required when admission is unavailable rather /// than exceeded — both branches strand a send if they emit a NOTICE. - #[test] - fn event_rejected_for_unavailable_admission_is_also_correlated() { + #[tokio::test] + async fn event_rejected_for_unavailable_admission_is_also_correlated() { + let state = crate::state::tests::test_state().await; let (conn, mut rx) = test_conn(); let (msg, event_id) = parsed_event_message(); send_admission_result( &conn, + &state, Err(AdmissionError::Unavailable), &msg, "ws_operations", - ); + ) + .await; let frame = sent_frame(&mut rx); assert_eq!(frame[0], "OK"); @@ -254,18 +266,21 @@ mod tests { /// A REQ still settles on CLOSED, which carries the subscription id. This /// pins the pre-existing behavior the fix must not disturb. - #[test] - fn over_quota_req_still_closes_the_subscription() { + #[tokio::test] + async fn over_quota_req_still_closes_the_subscription() { + let state = crate::state::tests::test_state().await; let (conn, mut rx) = test_conn(); let raw = serde_json::json!(["REQ", "history-abc", {"kinds": [1]}]).to_string(); let msg = ClientMessage::parse(&raw).expect("parse REQ"); send_admission_result( &conn, + &state, Err(AdmissionError::Exceeded { reset_in_secs: 7 }), &msg, "ws_operations", - ); + ) + .await; let frame = sent_frame(&mut rx); assert_eq!(frame[0], "CLOSED"); @@ -277,18 +292,21 @@ mod tests { } /// NIP-45 uses `CLOSED(query_id, reason)` when a relay refuses a COUNT. - #[test] - fn over_quota_count_closes_the_query() { + #[tokio::test] + async fn over_quota_count_closes_the_query() { + let state = crate::state::tests::test_state().await; let (conn, mut rx) = test_conn(); let raw = serde_json::json!(["COUNT", "count-abc", {"kinds": [1]}]).to_string(); let msg = ClientMessage::parse(&raw).expect("parse COUNT"); send_admission_result( &conn, + &state, Err(AdmissionError::Exceeded { reset_in_secs: 7 }), &msg, "ws_operations", - ); + ) + .await; let frame = sent_frame(&mut rx); assert_eq!(frame[0], "CLOSED");