From 28bad17e3e2ab5cc98bc87a8e676f6eacfb7b019 Mon Sep 17 00:00:00 2001 From: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Date: Mon, 17 Aug 2026 18:26:11 -0400 Subject: [PATCH] fix(relay): fence stale historical subscription output Signed-off-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Signed-off-by: loganj --- crates/buzz-relay/src/connection.rs | 118 ++++++++++++++++++++++++++ crates/buzz-relay/src/handlers/req.rs | 32 +++++-- 2 files changed, 144 insertions(+), 6 deletions(-) diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index ac794ba895..5b4ae9d275 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -220,6 +220,26 @@ impl ConnectionState { } } + pub(crate) async fn send_historical_if_permitted( + &self, + sub_id: &str, + pending: &Arc, + msg: String, + ) -> bool { + let pending_subscriptions = self.pending_subscriptions.lock().await; + let permitted = pending_subscriptions + .get(sub_id) + .is_some_and(|current| PendingSubscription::permits_commit(current, pending)); + if self.cancel.is_cancelled() || !permitted { + return false; + } + + // Keep the lifecycle lock through queue insertion. Replacement commit + // and CLOSE take the same lock, so neither can publish its cutover and + // then be followed by a stale historical frame from this lease. + self.send(msg) + } + /// Sends a data message to this connection's outbound channel. /// /// On a full buffer, increments the backpressure counter. The first @@ -992,6 +1012,104 @@ mod tests { assert!(conn.pending_subscriptions.lock().await.is_empty()); } + #[tokio::test] + async fn replacement_commit_fences_predecessor_historical_output() { + let (send_tx, mut send_rx) = mpsc::channel(8); + let conn = Arc::new(ConnectionState { + conn_id: Uuid::new_v4(), + tenant: TenantContext::resolved( + buzz_core::CommunityId::from_uuid(Uuid::new_v4()), + "lease.test", + ), + remote_addr: "127.0.0.1:1234".parse().expect("socket address"), + auth_state: RwLock::new(AuthState::Failed), + subscriptions: Arc::new(tokio::sync::Mutex::new(HashMap::new())), + pending_subscriptions: Arc::new(tokio::sync::Mutex::new(HashMap::new())), + send_tx, + ctrl_tx: mpsc::channel(1).0, + cancel: CancellationToken::new(), + backpressure_count: Arc::new(AtomicU8::new(0)), + grace_limit: 3, + }); + let predecessor = conn.begin_pending_subscription("same-id").await; + let replacement = conn.begin_pending_subscription("same-id").await; + let lifecycle = conn.pending_subscriptions.lock().await; + + let old_event = tokio::spawn({ + let conn = Arc::clone(&conn); + let predecessor = Arc::clone(&predecessor); + async move { + conn.send_historical_if_permitted("same-id", &predecessor, "old-event".into()) + .await + } + }); + let old_eose = tokio::spawn({ + let conn = Arc::clone(&conn); + let predecessor = Arc::clone(&predecessor); + async move { + conn.send_historical_if_permitted("same-id", &predecessor, "old-eose".into()) + .await + } + }); + tokio::task::yield_now().await; + + replacement.commit(); + assert!(conn.send("new-owner".into())); + drop(lifecycle); + + assert!(!old_event.await.expect("old event task")); + assert!(!old_eose.await.expect("old EOSE task")); + assert_eq!( + send_rx.recv().await, + Some(WsMessage::Text("new-owner".into())) + ); + assert!(send_rx.try_recv().is_err()); + } + + #[tokio::test] + async fn close_fences_predecessor_historical_output_before_closed() { + let (send_tx, mut send_rx) = mpsc::channel(8); + let conn = Arc::new(ConnectionState { + conn_id: Uuid::new_v4(), + tenant: TenantContext::resolved( + buzz_core::CommunityId::from_uuid(Uuid::new_v4()), + "lease.test", + ), + remote_addr: "127.0.0.1:1234".parse().expect("socket address"), + auth_state: RwLock::new(AuthState::Failed), + subscriptions: Arc::new(tokio::sync::Mutex::new(HashMap::new())), + pending_subscriptions: Arc::new(tokio::sync::Mutex::new(HashMap::new())), + send_tx, + ctrl_tx: mpsc::channel(1).0, + cancel: CancellationToken::new(), + backpressure_count: Arc::new(AtomicU8::new(0)), + grace_limit: 3, + }); + let pending = conn.begin_pending_subscription("same-id").await; + let mut lifecycle = conn.pending_subscriptions.lock().await; + + let old_event = tokio::spawn({ + let conn = Arc::clone(&conn); + let pending = Arc::clone(&pending); + async move { + conn.send_historical_if_permitted("same-id", &pending, "old-event".into()) + .await + } + }); + tokio::task::yield_now().await; + + lifecycle + .remove("same-id") + .expect("pending owner") + .cancel_lineage(); + assert!(conn.send("closed".into())); + drop(lifecycle); + + assert!(!old_event.await.expect("old event task")); + assert_eq!(send_rx.recv().await, Some(WsMessage::Text("closed".into()))); + assert!(send_rx.try_recv().is_err()); + } + #[tokio::test] async fn committed_same_id_replacement_cancels_the_previous_lease() { let original = Arc::new(PendingSubscription::new(None)); diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index 5dc56637a7..ee06765ff2 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -286,6 +286,7 @@ pub(crate) async fn handle_req( &conn, &state, trace_state.as_ref(), + &pending, ) .await; return; @@ -380,7 +381,9 @@ pub(crate) async fn handle_req( Ok(evs) => evs, Err(e) => { warn!(conn_id = %conn_id, sub_id = %sub_id, "Historical query failed: {e}"); - conn.send(RelayMessage::eose(&sub_id)); + let _ = conn + .send_historical_if_permitted(&sub_id, &pending, RelayMessage::eose(&sub_id)) + .await; return; } }; @@ -457,7 +460,10 @@ pub(crate) async fn handle_req( } let msg = RelayMessage::event(&sub_id, &stored.event); - if !conn.send(msg) { + if !conn + .send_historical_if_permitted(&sub_id, &pending, msg) + .await + { return; } total_sent += 1; @@ -467,7 +473,9 @@ pub(crate) async fn handle_req( } } - conn.send(RelayMessage::eose(&sub_id)); + let _ = conn + .send_historical_if_permitted(&sub_id, &pending, RelayMessage::eose(&sub_id)) + .await; debug!( conn_id = %conn_id, @@ -641,6 +649,7 @@ async fn handle_search_req( conn: &ConnectionState, state: &AppState, trace_state: Option<&crate::conformance::AbstractState>, + pending: &Arc, ) { // The community-wide channel scope (no #h tag on the filter). `None` means // "no accessible channels and no global access" → EOSE, exactly as the @@ -649,7 +658,9 @@ async fn handle_search_req( match build_search_channel_scope_filter(accessible_channels, include_global) { Some(scope) => scope, None => { - conn.send(RelayMessage::eose(sub_id)); + let _ = conn + .send_historical_if_permitted(sub_id, pending, RelayMessage::eose(sub_id)) + .await; return; } }; @@ -839,7 +850,14 @@ async fn handle_search_req( if !seen_ids.insert(stored.event.id) { continue; } - if !conn.send(RelayMessage::event(sub_id, &stored.event)) { + if !conn + .send_historical_if_permitted( + sub_id, + pending, + RelayMessage::event(sub_id, &stored.event), + ) + .await + { return; } emitted += 1; @@ -852,7 +870,9 @@ async fn handle_search_req( } } - conn.send(RelayMessage::eose(sub_id)); + let _ = conn + .send_historical_if_permitted(sub_id, pending, RelayMessage::eose(sub_id)) + .await; } /// Convert a single NIP-01 filter into an [`EventQuery`] for the database.