fix(relay): send REQ frames only while the request owns its ID

REQ handlers run in separate tasks, so a same-ID REQ can replace one
that is still reading. The replaced history, search or huddle-liveness
request then sent its EVENT and EOSE frames anyway. When the new holder
is an artifact sync, that stale EOSE tells the client a set the sync
never sent is complete. These frames now go through
close::send_if_owner, which queues a frame only while the request still
owns the ID, under the lifecycle lock.

A subscription-limit refusal (ordinary REQ and sync) now cancels the
connection when its CLOSED cannot be queued, as close::refuse does. The
ID is new there, so there is no old holder to retire.

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

crates/buzz-relay/src/connection.rs

@@ -2313,6 +2313,39 @@ pub(crate) mod tests {
         }
     }
 
+    /// A REQ for a new ID refused at the subscription limit cancels the
+    /// connection when its `CLOSED` cannot be queued: the client would
+    /// otherwise wait on a request the relay refused.
+    /// Mutation: ignore the `send` result on that refusal → RED.
+    #[tokio::test]
+    async fn subscription_limit_refusal_with_a_full_queue_cancels_the_connection() {
+        let state = crate::state::tests::test_state().await;
+        let (conn, _rx) = test_conn_with_auth(authenticated_state());
+        {
+            let mut subs = conn.subscriptions.lock().await;
+            for index in 0..handlers::req::MAX_SUBSCRIPTIONS {
+                subs.insert(format!("held-{index}"), 0);
+            }
+        }
+        while conn
+            .send_tx
+            .try_send(WsMessage::Text("filler".into()))
+            .is_ok()
+        {}
+
+        handlers::req::handle_req(
+            "s".into(),
+            vec![nostr::Filter::new().kind(Kind::TextNote)],
+            vec![None],
+            Arc::clone(&conn),
+            Arc::clone(&state),
+        )
+        .await;
+
+        assert!(conn.cancel.is_cancelled());
+        assert!(!conn.subscriptions.lock().await.contains_key("s"));
+    }
+
     /// 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.rs

@@ -466,10 +466,14 @@ async fn claim(
         return None;
     }
     if !subs.contains_key(sub_id) && subs.len() >= super::req::MAX_SUBSCRIPTIONS {
-        conn.send(RelayMessage::closed(
+        // A new ID has no old holder to retire, but a lost CLOSED would leave
+        // the client waiting on a refused sync.
+        if !conn.send(RelayMessage::closed(
             sub_id,
             "error: too many subscriptions",
-        ));
+        )) {
+            conn.cancel.cancel();
+        }
         return None;
     }
     super::close::retire_locked(&mut subs, sub_id, conn, state).await;

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

@@ -621,3 +621,73 @@ async fn refused_or_one_shot_replacement_retires_the_old_holder() {
         no_frame(&mut rx).await;
     }
 }
+
+/// A same-ID sync that replaces a history, search or huddle-liveness `REQ`
+/// while that request waits to send gets only its own frames: the old
+/// request's late frames, `EOSE` included, must not reach the sync's
+/// client. The old request pauses just before its first frame.
+/// Mutation: drop the owner check in `close::send_if_owner` → stale frames
+/// follow the sync's `EOSE` → RED.
+#[tokio::test]
+#[ignore = "requires Postgres and Redis"]
+async fn a_replaced_req_sends_nothing_under_a_sync_that_took_its_id() {
+    use super::close::test_seam::{Pause, SEND_PAUSE};
+    let f = fixture().await;
+    let owner = Keys::generate();
+    let history: fn(Uuid) -> Vec<Value> = |home| vec![json!({"kinds": [45010], "#h": [home]})];
+    let search: fn(Uuid) -> Vec<Value> = |home| vec![json!({"search": "x", "#h": [home]})];
+    let liveness: fn(Uuid) -> Vec<Value> =
+        |home| vec![json!({"kinds": [buzz_core::kind::KIND_HUDDLE_LIVENESS], "#h": [home]})];
+    for (name, old_req) in [
+        ("history", history),
+        ("search", search),
+        ("liveness", liveness),
+    ] {
+        let home = create_channel(&f.state, &f.tenant, &owner, &[]).await;
+        let created = revision(&f, &owner, home, Uuid::new_v4(), "create", None).await;
+        let (conn, mut rx) = connect(&f, &owner);
+        let pause = Arc::new(Pause::default());
+        let stale = tokio::spawn({
+            let (f_conn, state) = (Arc::clone(&conn), Arc::clone(&f.state));
+            let filters: Vec<nostr::Filter> = old_req(home)
+                .into_iter()
+                .map(|filter| serde_json::from_value(filter).expect("filter"))
+                .collect();
+            let before_ids = vec![None; filters.len()];
+            SEND_PAUSE.scope(
+                Arc::clone(&pause),
+                super::req::handle_req("s".into(), filters, before_ids, f_conn, state),
+            )
+        });
+        tokio::time::timeout(Duration::from_secs(10), pause.reached.notified())
+            .await
+            .unwrap_or_else(|_| panic!("{name}: the old REQ reaches its first send"));
+
+        let sync = start_sync(&f, &conn, sync_filter(&[home]));
+        assert_eq!(
+            initial_set(&mut rx).await,
+            vec![created.id.to_hex()],
+            "{name}"
+        );
+        // The seam pauses before every send; release each one.
+        tokio::time::timeout(Duration::from_secs(10), async {
+            while !stale.is_finished() {
+                pause.resume.notify_one();
+                tokio::time::sleep(Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .unwrap_or_else(|_| panic!("{name}: the old REQ ends"));
+        stale.await.unwrap();
+        no_frame(&mut rx).await;
+        assert!(
+            f.state.artifact_syncs.contains(conn.conn_id, "s"),
+            "{name}: the sync still holds the ID"
+        );
+        super::close::handle_close("s".into(), Arc::clone(&conn), Arc::clone(&f.state)).await;
+        tokio::time::timeout(Duration::from_secs(5), sync)
+            .await
+            .expect("sync task ends on CLOSE")
+            .unwrap();
+    }
+}

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

@@ -39,6 +39,22 @@ pub(crate) async fn refuse(sub_id: &str, reason: &str, conn: &ConnectionState, s
     }
 }
 
+/// Queue `frame` for `sub_id` only while the request holding `owner` still
+/// owns the ID. A request that a newer same-ID REQ replaced sends nothing:
+/// its late `EOSE` would otherwise tell a replacing sync that a set it never
+/// sent is complete. Returns whether the frame was queued.
+pub(crate) async fn send_if_owner(
+    sub_id: &str,
+    owner: u64,
+    frame: String,
+    conn: &ConnectionState,
+) -> bool {
+    #[cfg(test)]
+    test_seam::pause_at(&test_seam::SEND_PAUSE).await;
+    let subs = conn.subscriptions.lock().await;
+    subs.get(sub_id) == Some(&owner) && conn.send(frame)
+}
+
 /// Monotonic source of subscription owner tokens. Each REQ claims its sub ID
 /// with a fresh token, so a superseded request can tell it no longer owns it.
 static NEXT_OWNER: AtomicU64 = AtomicU64::new(1);
@@ -151,6 +167,7 @@ pub(crate) mod test_seam {
         pub(crate) static PAUSE: Arc<Pause>;
         pub(crate) static RELEASE_PAUSE: Arc<Pause>;
         pub(crate) static CLEANUP_PAUSE: Arc<Pause>;
+        pub(crate) static SEND_PAUSE: Arc<Pause>;
     }
 
     pub(crate) async fn pause_at(point: &'static tokio::task::LocalKey<Arc<Pause>>) {

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

@@ -74,10 +74,14 @@ pub async fn handle_req(
 
                 let subs = conn.subscriptions.lock().await;
                 if !subs.contains_key(&sub_id) && subs.len() >= MAX_SUBSCRIPTIONS {
-                    conn.send(RelayMessage::closed(
+                    // A new ID has no old holder to retire, but a lost CLOSED
+                    // would leave the client waiting on a refused request.
+                    if !conn.send(RelayMessage::closed(
                         &sub_id,
                         "error: too many subscriptions",
-                    ));
+                    )) {
+                        conn.cancel.cancel();
+                    }
                     return;
                 }
 
@@ -499,7 +503,8 @@ pub async fn handle_req(
                     if e.is_statement_cancelled() {
                         return true;
                     }
-                    conn.send(RelayMessage::eose(&sub_id));
+                    super::close::send_if_owner(&sub_id, owner, RelayMessage::eose(&sub_id), &conn)
+                        .await;
                     return false;
                 }
             };
@@ -576,7 +581,7 @@ pub async fn handle_req(
                 }
 
                 let msg = RelayMessage::event(&sub_id, &stored.event);
-                if !conn.send(msg) {
+                if !super::close::send_if_owner(&sub_id, owner, msg, &conn).await {
                     return false;
                 }
                 total_sent += 1;
@@ -586,7 +591,7 @@ pub async fn handle_req(
             }
         }
 
-        conn.send(RelayMessage::eose(&sub_id));
+        super::close::send_if_owner(&sub_id, owner, RelayMessage::eose(&sub_id), &conn).await;
 
         debug!(
             conn_id = %conn_id,
@@ -784,7 +789,7 @@ async fn handle_search_req(
     if build_search_channel_scope_filter(implicit_channels, include_global).is_none()
         && named_only.is_empty()
     {
-        conn.send(RelayMessage::eose(sub_id));
+        super::close::send_if_owner(sub_id, owner, RelayMessage::eose(sub_id), conn).await;
         return;
     }
 
@@ -988,7 +993,8 @@ async fn handle_search_req(
                     if !seen_ids.insert(stored.event.id) {
                         continue;
                     }
-                    if !conn.send(RelayMessage::event(sub_id, &stored.event)) {
+                    let frame = RelayMessage::event(sub_id, &stored.event);
+                    if !super::close::send_if_owner(sub_id, owner, frame, conn).await {
                         return;
                     }
                     emitted += 1;
@@ -1001,7 +1007,7 @@ async fn handle_search_req(
         }
     }
 
-    conn.send(RelayMessage::eose(sub_id));
+    super::close::send_if_owner(sub_id, owner, RelayMessage::eose(sub_id), conn).await;
 }
 
 /// Convert a single NIP-01 filter into an [`EventQuery`] for the database.
@@ -1526,12 +1532,13 @@ async fn handle_huddle_liveness_req(
                 return;
             }
         };
-        if !conn.send(RelayMessage::event(sub_id, &event)) {
+        let frame = RelayMessage::event(sub_id, &event);
+        if !super::close::send_if_owner(sub_id, owner, frame, conn).await {
             return;
         }
     }
 
-    conn.send(RelayMessage::eose(sub_id));
+    super::close::send_if_owner(sub_id, owner, RelayMessage::eose(sub_id), conn).await;
 }
 
 /// Claim `sub_id` for a live REQ under the lifecycle lock: record a fresh