fix(relay): retire the old holder on pre-handler sync refusals

A sync REQ refused before its handler runs (handler saturation or WS
admission) sent CLOSED and left the previous holder of the ID live, so
fan-out could continue under a closed ID. Both paths now go through
artifact_sync::refuse, which retires the old holder under the lifecycle
lock before CLOSED.

Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Logan Johnson <loganj@squareup.com>

crates/buzz-relay/src/connection.rs

@@ -1400,10 +1400,14 @@ async fn handle_text_message(text: String, conn: Arc<ConnectionState>, 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<WsMessage>,

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

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]>,

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