fix(relay): harden pending subscription replacement

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

crates/buzz-pubsub/src/lib.rs

@@ -200,10 +200,19 @@ impl PubSubManager {
         };
 
         if should_subscribe {
-            let _ = self
+            // `desired_topics` is the source of truth. A full command queue is
+            // only a stale live-connection hint: reconnect snapshots desired
+            // topics, while blocking here can deadlock REQ/CLOSE lifecycle
+            // commits during a Redis outage.
+            match self
                 .subscription_tx
-                .send(subscriber::SubscriptionCommand::Subscribe(topic_key))
-                .await;
+                .try_send(subscriber::SubscriptionCommand::Subscribe(topic_key))
+            {
+                Ok(()) | Err(mpsc::error::TrySendError::Closed(_)) => {}
+                Err(mpsc::error::TrySendError::Full(_)) => {
+                    tracing::warn!(?topic_key, "pubsub subscription command queue full");
+                }
+            }
         }
     }
 

crates/buzz-relay/src/connection.rs

@@ -2,7 +2,7 @@
 
 use std::collections::HashMap;
 use std::net::SocketAddr;
-use std::sync::atomic::{AtomicU8, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -43,12 +43,16 @@ pub(crate) struct RestartClose {
 #[derive(Debug)]
 pub(crate) struct PendingSubscription {
     cancel: CancellationToken,
+    committed: AtomicBool,
+    predecessor: Option<Arc<PendingSubscription>>,
 }
 
 impl PendingSubscription {
-    fn new() -> Self {
+    fn new(predecessor: Option<Arc<PendingSubscription>>) -> Self {
         Self {
             cancel: CancellationToken::new(),
+            committed: AtomicBool::new(false),
+            predecessor,
         }
     }
 
@@ -56,6 +60,31 @@ impl PendingSubscription {
         self.cancel.cancel();
     }
 
+    /// Cancel this request and every older same-ID request it superseded.
+    pub(crate) fn cancel_lineage(&self) {
+        let mut current = Some(self);
+        while let Some(request) = current {
+            request.cancel();
+            current = request.predecessor.as_deref();
+        }
+    }
+
+    /// Supersede older same-ID requests after this replacement commits.
+    pub(crate) fn commit(&self) {
+        self.committed.store(true, Ordering::Release);
+        if let Some(predecessor) = self.predecessor.as_ref() {
+            predecessor.cancel_lineage();
+        }
+    }
+
+    fn live_predecessor(&self) -> Option<Arc<PendingSubscription>> {
+        (!self.committed.load(Ordering::Acquire))
+            .then_some(self.predecessor.as_ref())
+            .flatten()
+            .filter(|predecessor| !predecessor.is_cancelled())
+            .cloned()
+    }
+
     pub(crate) fn is_cancelled(&self) -> bool {
         self.cancel.is_cancelled()
     }
@@ -122,20 +151,16 @@ pub struct ConnectionState {
 
 impl ConnectionState {
     /// Install a pending REQ before its handler task starts. A repeated REQ
-    /// with the same subscription ID supersedes and cancels the older task.
+    /// with the same subscription ID becomes the new owner, but the previous
+    /// request stays alive until the replacement commits successfully.
     pub(crate) async fn begin_pending_subscription(
         &self,
         sub_id: &str,
     ) -> Arc<PendingSubscription> {
-        let pending = Arc::new(PendingSubscription::new());
-        if let Some(replaced) = self
-            .pending_subscriptions
-            .lock()
-            .await
-            .insert(sub_id.to_owned(), Arc::clone(&pending))
-        {
-            replaced.cancel();
-        }
+        let mut pending_subscriptions = self.pending_subscriptions.lock().await;
+        let predecessor = pending_subscriptions.get(sub_id).cloned();
+        let pending = Arc::new(PendingSubscription::new(predecessor));
+        pending_subscriptions.insert(sub_id.to_owned(), Arc::clone(&pending));
         pending
     }
 
@@ -143,7 +168,7 @@ impl ConnectionState {
     async fn cancel_all_pending_subscriptions(&self) {
         let mut pending = self.pending_subscriptions.lock().await;
         for (_, request) in pending.drain() {
-            request.cancel();
+            request.cancel_lineage();
         }
     }
 
@@ -158,7 +183,11 @@ impl ConnectionState {
             .get(sub_id)
             .is_some_and(|current| Arc::ptr_eq(current, completed))
         {
-            pending.remove(sub_id);
+            if let Some(predecessor) = completed.live_predecessor() {
+                pending.insert(sub_id.to_owned(), predecessor);
+            } else {
+                pending.remove(sub_id);
+            }
         }
     }
 
@@ -861,6 +890,53 @@ mod tests {
     use super::*;
     use std::sync::{Arc, Mutex};
 
+    #[tokio::test]
+    async fn failed_same_id_replacement_restores_the_previous_lease() {
+        let conn = 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: mpsc::channel(1).0,
+            ctrl_tx: mpsc::channel(1).0,
+            cancel: CancellationToken::new(),
+            backpressure_count: Arc::new(AtomicU8::new(0)),
+            grace_limit: 3,
+        };
+
+        let original = conn.begin_pending_subscription("same-id").await;
+        let replacement = conn.begin_pending_subscription("same-id").await;
+        assert!(!original.is_cancelled());
+
+        conn.finish_pending_subscription("same-id", &replacement)
+            .await;
+        let current = conn
+            .pending_subscriptions
+            .lock()
+            .await
+            .get("same-id")
+            .cloned()
+            .expect("original lease restored");
+        assert!(Arc::ptr_eq(&current, &original));
+        assert!(!original.is_cancelled());
+    }
+
+    #[tokio::test]
+    async fn committed_same_id_replacement_cancels_the_previous_lease() {
+        let original = Arc::new(PendingSubscription::new(None));
+        let replacement = PendingSubscription::new(Some(Arc::clone(&original)));
+
+        replacement.commit();
+
+        assert!(original.is_cancelled());
+        assert!(!replacement.is_cancelled());
+    }
+
     #[derive(Debug, Default)]
     struct MockSinkState {
         messages: Vec<WsMessage>,

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

@@ -29,7 +29,7 @@ pub(crate) async fn remove_subscription(
 ) {
     let mut pending_subscriptions = conn.pending_subscriptions.lock().await;
     if let Some(pending) = pending_subscriptions.remove(sub_id) {
-        pending.cancel();
+        pending.cancel_lineage();
     }
 
     conn.subscriptions.lock().await.remove(sub_id);

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

@@ -525,6 +525,10 @@ async fn register_subscription_if_current(
         .retain_topic(&conn.tenant, topic_for_subscription(channel_id))
         .await;
 
+    // Only a successful replacement commit supersedes historical delivery from
+    // the previous same-ID REQ. Validation/search failures leave it untouched.
+    pending.commit();
+
     drop(pending_subscriptions);
     true
 }