fix(relay): fence stale historical subscription output

Signed-off-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz>
Signed-off-by: loganj <loganj@squareup.com>

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<PendingSubscription>,
+        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));

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<PendingSubscription>,
 ) {
     // 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.