fix(relay): retire the old holder on every in-handler REQ refusal

handle_req refused an ordinary REQ (auth, scope, channel count, channel
access, NIP-FI, p-gated, engram, author-only, mixed search, database
errors) with a bare CLOSED before it claimed the ID. A live subscription
or artifact sync that held the same ID kept delivering under an ID the
client was told is closed. These refusals now go through close::refuse,
which retires the old holder under the lifecycle lock before CLOSED and
cancels the connection if CLOSED cannot be queued.

A huddle-liveness REQ answered under the ID without claiming it, so the
old holder also survived it. It now claims the ID as a one-shot, like
search, and its own refusals close only while it still owns the ID.

Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>
Signed-off-by: Logan Johnson <loganj@squareup.com>

crates/buzz-relay/src/connection.rs

@@ -2224,6 +2224,95 @@ pub(crate) mod tests {
         }
     }
 
+    /// An ordinary REQ that `handle_req` itself refuses, before it claims its
+    /// ID, still replaces the old holder (NIP-01): a live subscription or an
+    /// artifact sync must not keep using an ID the client was told is closed.
+    /// Covers the auth, scope, channel-count and database-error refusals.
+    /// Mutation: send any of them as a bare `CLOSED` → RED.
+    #[tokio::test]
+    async fn in_handler_req_refusal_retires_the_old_holder() {
+        use buzz_pubsub::EventTopic;
+        let mut scoped = auth_context();
+        scoped.scopes = vec![buzz_auth::Scope::ChannelsRead];
+        let channel = uuid::Uuid::new_v4();
+        let too_many: Vec<String> = (0..=handlers::req::MAX_EXPLICIT_CHANNEL_VALUES)
+            .map(|_| uuid::Uuid::new_v4().to_string())
+            .collect();
+        let cases = [
+            (
+                AuthState::Failed,
+                serde_json::json!({"kinds": [1]}),
+                "auth-required: not authenticated",
+            ),
+            (
+                AuthState::Authenticated(scoped),
+                serde_json::json!({"kinds": [1]}),
+                "restricted: insufficient scope",
+            ),
+            (
+                authenticated_state(),
+                serde_json::json!({"kinds": [1], "#h": too_many}),
+                "restricted: too many explicit channels",
+            ),
+            (
+                authenticated_state(),
+                serde_json::json!({"search": "x"}),
+                "error: database error",
+            ),
+        ];
+        for ((auth, filter, reason), sync_holder) in cases
+            .into_iter()
+            .flat_map(|case| [(case.clone(), false), (case, true)])
+        {
+            let case = format!("{reason}, sync holder: {sync_holder}");
+            let state =
+                crate::state::tests::test_state_with_database_url("postgres://127.0.0.1:1/none")
+                    .await;
+            let (conn, mut rx) = test_conn_with_auth(auth);
+            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");
+            if sync_holder {
+                state.artifact_syncs.insert_idle(conn.conn_id, "s");
+            }
+
+            let filter = serde_json::from_value(filter).expect("filter");
+            handlers::req::handle_req(
+                "s".into(),
+                vec![filter],
+                vec![None],
+                Arc::clone(&conn),
+                Arc::clone(&state),
+            )
+            .await;
+
+            assert!(conn.subscriptions.lock().await.is_empty(), "{case}");
+            assert!(
+                state.sub_registry.get_filters(conn.conn_id, "s").is_none(),
+                "{case}"
+            );
+            assert!(!state.artifact_syncs.contains(conn.conn_id, "s"), "{case}");
+            assert_eq!(
+                state.pubsub.topic_refcount(&conn.tenant, topic).await,
+                0,
+                "{case}"
+            );
+            let mut frame = read_frame(&mut rx);
+            if frame[0] == "NOTICE" {
+                frame = read_frame(&mut rx);
+            }
+            assert_eq!(frame, serde_json::json!(["CLOSED", "s", reason]), "{case}");
+            assert!(rx.try_recv().is_err(), "{case}: nothing follows CLOSED");
+        }
+    }
+
     /// A pre-handler refusal whose `CLOSED` cannot be queued has already
     /// retired the old holder, so it cancels the connection: the client
     /// learns the ID is gone from the disconnect. The queue is full but

crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs

@@ -521,3 +521,93 @@ async fn close_during_a_live_head_read_sends_nothing_more() {
         .unwrap();
     no_frame(&mut rx).await;
 }
+
+/// Run an ordinary `REQ "s"` through the real handler.
+async fn ordinary_req(f: &Fixture, conn: &Arc<ConnectionState>, filters: Vec<Value>) {
+    let filters: Vec<nostr::Filter> = filters
+        .into_iter()
+        .map(|filter| serde_json::from_value(filter).expect("filter"))
+        .collect();
+    let before_ids = vec![None; filters.len()];
+    super::req::handle_req(
+        "s".into(),
+        filters,
+        before_ids,
+        Arc::clone(conn),
+        Arc::clone(&f.state),
+    )
+    .await;
+}
+
+/// An ordinary `REQ` that replaces a live sync or a live subscription with
+/// the same ID retires it, also when `handle_req` refuses the new request
+/// (mixed search) or answers it once (huddle liveness) without registering
+/// it. A later change in the channel then sends nothing under that ID.
+/// Mutation: refuse mixed search with a bare `CLOSED`, or answer liveness
+/// without claiming the ID → RED.
+#[tokio::test]
+#[ignore = "requires Postgres and Redis"]
+async fn refused_or_one_shot_replacement_retires_the_old_holder() {
+    let f = fixture().await;
+    let owner = Keys::generate();
+    let home = create_channel(&f.state, &f.tenant, &owner, &[]).await;
+    let mixed = vec![
+        json!({"search": "x", "#h": [home]}),
+        json!({"kinds": [45010], "#h": [home]}),
+    ];
+    let liveness = vec![json!({"kinds": [buzz_core::kind::KIND_HUDDLE_LIVENESS], "#h": [home]})];
+    for (sync_holder, replacement, terminal) in [
+        (
+            true,
+            &mixed,
+            json!([
+                "CLOSED",
+                "s",
+                "error: mixed search and non-search filters not supported"
+            ]),
+        ),
+        (
+            false,
+            &mixed,
+            json!([
+                "CLOSED",
+                "s",
+                "error: mixed search and non-search filters not supported"
+            ]),
+        ),
+        (true, &liveness, json!(["EOSE", "s"])),
+        (false, &liveness, json!(["EOSE", "s"])),
+    ] {
+        let case = format!("sync holder: {sync_holder}, replacement: {terminal}");
+        let (conn, mut rx) = connect(&f, &owner);
+        if sync_holder {
+            start_sync(&f, &conn, sync_filter(&[home])).await.unwrap();
+        } else {
+            ordinary_req(
+                &f,
+                &conn,
+                vec![json!({"kinds": [45010], "#h": [home], "limit": 0})],
+            )
+            .await;
+        }
+        assert!(initial_set(&mut rx).await.is_empty(), "{case}");
+
+        ordinary_req(&f, &conn, replacement.clone()).await;
+        assert_eq!(frame(&mut rx).await, terminal, "{case}");
+        assert!(conn.subscriptions.lock().await.is_empty(), "{case}");
+        assert!(
+            !f.state.artifact_syncs.contains(conn.conn_id, "s"),
+            "{case}"
+        );
+        assert!(
+            f.state
+                .sub_registry
+                .get_filters(conn.conn_id, "s")
+                .is_none(),
+            "{case}"
+        );
+
+        revision(&f, &owner, home, Uuid::new_v4(), "create", None).await;
+        no_frame(&mut rx).await;
+    }
+}

crates/buzz-relay/src/handlers/close.rs

@@ -28,7 +28,7 @@ pub async fn handle_close(sub_id: String, conn: Arc<ConnectionState>, state: Arc
 /// Refuse a `REQ` or 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. Every refusal before a request's handler runs goes through here. If
+/// lock. Every refusal before a request claims its ID goes through here. If
 /// the frame cannot be queued, the connection is cancelled: the client would
 /// otherwise never learn that the old holder is gone.
 pub(crate) async fn refuse(sub_id: &str, reason: &str, conn: &ConnectionState, state: &AppState) {

crates/buzz-relay/src/handlers/req.rs

@@ -65,10 +65,8 @@ pub async fn handle_req(
             AuthState::Authenticated(ctx) => {
                 if !ctx.scopes.is_empty() && !ctx.scopes.contains(&Scope::MessagesRead) {
                     conn.send(RelayMessage::notice("restricted: insufficient scope"));
-                    conn.send(RelayMessage::closed(
-                        &sub_id,
-                        "restricted: insufficient scope",
-                    ));
+                    super::close::refuse(&sub_id, "restricted: insufficient scope", &conn, &state)
+                        .await;
                     return;
                 }
 
@@ -89,10 +87,8 @@ pub async fn handle_req(
                 conn.send(RelayMessage::notice(
                     "auth-required: authenticate before subscribing",
                 ));
-                conn.send(RelayMessage::closed(
-                    &sub_id,
-                    "auth-required: not authenticated",
-                ));
+                super::close::refuse(&sub_id, "auth-required: not authenticated", &conn, &state)
+                    .await;
                 return;
             }
         }
@@ -102,10 +98,13 @@ pub async fn handle_req(
     let requested_channel_ids = match extract_channel_ids_from_filters_limited(&filters) {
         Ok(ids) => ids,
         Err(()) => {
-            conn.send(RelayMessage::closed(
+            super::close::refuse(
                 &sub_id,
                 "restricted: too many explicit channels",
-            ));
+                &conn,
+                &state,
+            )
+            .await;
             return;
         }
     };
@@ -122,7 +121,7 @@ pub async fn handle_req(
             Ok(ids) => ids,
             Err(e) => {
                 warn!(conn_id = %conn_id, "Failed to get accessible channels: {e}");
-                conn.send(RelayMessage::closed(&sub_id, "error: database error"));
+                super::close::refuse(&sub_id, "error: database error", &conn, &state).await;
                 return;
             }
         }
@@ -146,7 +145,7 @@ pub async fn handle_req(
         Ok(ids) => ids,
         Err(e) => {
             warn!(conn_id = %conn_id, "Failed to get readable named-only channels: {e}");
-            conn.send(RelayMessage::closed(&sub_id, "error: database error"));
+            super::close::refuse(&sub_id, "error: database error", &conn, &state).await;
             return;
         }
     };
@@ -185,7 +184,7 @@ pub async fn handle_req(
             Ok(confirmed) => confirmed,
             Err(e) => {
                 warn!(conn_id = %conn_id, "Channel membership confirmation failed: {e}");
-                conn.send(RelayMessage::closed(&sub_id, "error: database error"));
+                super::close::refuse(&sub_id, "error: database error", &conn, &state).await;
                 return;
             }
         };
@@ -204,7 +203,7 @@ pub async fn handle_req(
         .await
         {
             warn!(conn_id = %conn_id, "Failed to classify repaired channels: {e}");
-            conn.send(RelayMessage::closed(&sub_id, "error: database error"));
+            super::close::refuse(&sub_id, "error: database error", &conn, &state).await;
             return;
         }
     }
@@ -229,10 +228,7 @@ pub async fn handle_req(
         .as_ref()
         .is_some_and(|authorized| authorized.is_empty())
     {
-        conn.send(RelayMessage::closed(
-            &sub_id,
-            "restricted: not a channel member",
-        ));
+        super::close::refuse(&sub_id, "restricted: not a channel member", &conn, &state).await;
         return;
     }
 
@@ -251,17 +247,20 @@ pub async fn handle_req(
             Err(crate::nip_fi_gate::SessionExpired) => {
                 // Fix 4: [FI-TRACE-DENIAL-ORACLE] gate is off_mode when no assertion
                 // exists, so SessionExpired here always implies an active FI session.
-                conn.send(RelayMessage::closed(
-                    &sub_id,
-                    "restricted: authorization denied",
-                ));
+                super::close::refuse(&sub_id, "restricted: authorization denied", &conn, &state)
+                    .await;
                 return;
             }
         };
+        // One-shot like search: accepting it replaces the ID's old holder.
+        let Some(owner) = claim_search_subscription(&sub_id, &conn, &state).await else {
+            return;
+        };
         unless_cancelled(
             &conn,
             handle_huddle_liveness_req(
                 &sub_id,
+                owner,
                 &filters,
                 authorized_requested_channels.as_deref().unwrap_or_default(),
                 &conn,
@@ -269,6 +268,7 @@ pub async fn handle_req(
             ),
         )
         .await;
+        super::close::close_if_owner(&sub_id, owner, None, &conn, &state).await;
         return;
     }
 
@@ -283,24 +283,33 @@ pub async fn handle_req(
     if channel_id.is_none() {
         let authed_pubkey_hex = hex::encode(&pubkey_bytes);
         if !p_gated_filters_authorized(&filters, &authed_pubkey_hex) {
-            conn.send(RelayMessage::closed(
+            super::close::refuse(
                 &sub_id,
                 "restricted: p-gated events require #p matching your pubkey",
-            ));
+                &conn,
+                &state,
+            )
+            .await;
             return;
         }
         if !engram_filters_authorized(&filters, &authed_pubkey_hex) {
-            conn.send(RelayMessage::closed(
+            super::close::refuse(
                 &sub_id,
                 "restricted: agent-engram reads require authors=[self] or #p=[self]",
-            ));
+                &conn,
+                &state,
+            )
+            .await;
             return;
         }
         if !author_only_filters_authorized(&filters, &authed_pubkey_hex) {
-            conn.send(RelayMessage::closed(
+            super::close::refuse(
                 &sub_id,
                 "restricted: author-only kinds require authors=[self]",
-            ));
+                &conn,
+                &state,
+            )
+            .await;
             return;
         }
     }
@@ -312,10 +321,13 @@ pub async fn handle_req(
     let has_search = filters.iter().any(|f| f.search.is_some());
     if has_search {
         if filters.iter().any(|f| f.search.is_none()) {
-            conn.send(RelayMessage::closed(
+            super::close::refuse(
                 &sub_id,
                 "error: mixed search and non-search filters not supported",
-            ));
+                &conn,
+                &state,
+            )
+            .await;
             return;
         }
         // IMPORTANT 6: acquire a REQ effect permit before the search query and
@@ -327,10 +339,8 @@ pub async fn handle_req(
             Ok(permit) => permit,
             Err(crate::nip_fi_gate::SessionExpired) => {
                 // Fix 4: [FI-TRACE-DENIAL-ORACLE]
-                conn.send(RelayMessage::closed(
-                    &sub_id,
-                    "restricted: authorization denied",
-                ));
+                super::close::refuse(&sub_id, "restricted: authorization denied", &conn, &state)
+                    .await;
                 return;
             }
         };
@@ -376,10 +386,7 @@ pub async fn handle_req(
         Ok(permit) => permit,
         Err(crate::nip_fi_gate::SessionExpired) => {
             // Fix 4: [FI-TRACE-DENIAL-ORACLE]
-            conn.send(RelayMessage::closed(
-                &sub_id,
-                "restricted: authorization denied",
-            ));
+            super::close::refuse(&sub_id, "restricted: authorization denied", &conn, &state).await;
             return;
         }
     };
@@ -1413,16 +1420,21 @@ fn huddle_liveness_session_ids(filters: &[Filter]) -> Vec<uuid::Uuid> {
 
 async fn handle_huddle_liveness_req(
     sub_id: &str,
+    owner: u64,
     filters: &[Filter],
     parent_channel_ids: &[uuid::Uuid],
     conn: &ConnectionState,
     state: &AppState,
 ) {
     if parent_channel_ids.is_empty() {
-        conn.send(RelayMessage::closed(
+        super::close::close_if_owner(
             sub_id,
-            "restricted: huddle liveness requires an authorized #h channel",
-        ));
+            owner,
+            Some("restricted: huddle liveness requires an authorized #h channel"),
+            conn,
+            state,
+        )
+        .await;
         return;
     }
 
@@ -1439,7 +1451,8 @@ async fn handle_huddle_liveness_req(
         Ok(links) => links,
         Err(error) => {
             warn!("Huddle liveness linkage batch failed: {error}");
-            conn.send(RelayMessage::closed(sub_id, "error: database error"));
+            super::close::close_if_owner(sub_id, owner, Some("error: database error"), conn, state)
+                .await;
             return;
         }
     };
@@ -1457,7 +1470,14 @@ async fn handle_huddle_liveness_req(
                 Ok(_) => continue,
                 Err(error) => {
                     warn!(session_id = %session_id, "Huddle liveness lease lookup failed: {error}");
-                    conn.send(RelayMessage::closed(sub_id, "error: liveness unavailable"));
+                    super::close::close_if_owner(
+                        sub_id,
+                        owner,
+                        Some("error: liveness unavailable"),
+                        conn,
+                        state,
+                    )
+                    .await;
                     return;
                 }
             }
@@ -1495,7 +1515,14 @@ async fn handle_huddle_liveness_req(
             Ok(event) => event,
             Err(error) => {
                 warn!(session_id = %session_id, "Huddle liveness signing failed: {error}");
-                conn.send(RelayMessage::closed(sub_id, "error: signing failed"));
+                super::close::close_if_owner(
+                    sub_id,
+                    owner,
+                    Some("error: signing failed"),
+                    conn,
+                    state,
+                )
+                .await;
                 return;
             }
         };
@@ -1566,10 +1593,10 @@ pub(crate) async fn claim_live_subscription(
     Some(owner)
 }
 
-/// Claim `sub_id` for a one-shot search REQ under the lifecycle lock.
-/// Accepting it retires whatever holds the ID (NIP-01 replacement), even
-/// though search never registers for fan-out. `None` once the connection is
-/// closing.
+/// Claim `sub_id` for a one-shot search or huddle-liveness REQ under the
+/// lifecycle lock. Accepting it retires whatever holds the ID (NIP-01
+/// replacement), even though neither registers for fan-out. `None` once the
+/// connection is closing.
 async fn claim_search_subscription(
     sub_id: &str,
     conn: &ConnectionState,