diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index db1315ab40..c9e5d06d41 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/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 diff --git a/crates/buzz-relay/src/handlers/artifact_sync.rs b/crates/buzz-relay/src/handlers/artifact_sync.rs index 9c3a1afaa8..182b12ab92 100644 --- a/crates/buzz-relay/src/handlers/artifact_sync.rs +++ b/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; diff --git a/crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs b/crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs index 2314409fed..6696877c4e 100644 --- a/crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs +++ b/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 = |home| vec![json!({"kinds": [45010], "#h": [home]})]; + let search: fn(Uuid) -> Vec = |home| vec![json!({"search": "x", "#h": [home]})]; + let liveness: fn(Uuid) -> Vec = + |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 = 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(); + } +} diff --git a/crates/buzz-relay/src/handlers/close.rs b/crates/buzz-relay/src/handlers/close.rs index 053442e08c..62e60b3953 100644 --- a/crates/buzz-relay/src/handlers/close.rs +++ b/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; pub(crate) static RELEASE_PAUSE: Arc; pub(crate) static CLEANUP_PAUSE: Arc; + pub(crate) static SEND_PAUSE: Arc; } pub(crate) async fn pause_at(point: &'static tokio::task::LocalKey>) { diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index 9b5785f13b..e88cffce9b 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/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