feat(pubsub): expose link state and confirm a topic subscription

LinkState carries an epoch that advances whenever the Redis event link connects, drops, or skips a message it cannot decode. confirm_topic waits until Redis has confirmed a retained topic and returns the epoch it was confirmed on. Sync uses both to register live interest before its snapshot and to detect possible loss.

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

crates/buzz-pubsub/src/lib.rs

@@ -51,7 +51,7 @@ use std::time::Duration;
 
 use buzz_core::TenantContext;
 use nostr::PublicKey;
-use tokio::sync::{broadcast, mpsc, Mutex};
+use tokio::sync::{broadcast, mpsc, oneshot, watch, Mutex};
 
 use crate::cache_invalidation::{
     cache_invalidation_channel, CacheInvalidation, ScopedCacheInvalidation,
@@ -72,6 +72,20 @@ pub struct ChannelEvent {
     pub event: nostr::Event,
 }
 
+/// Health of this node's Redis event subscription link.
+///
+/// `epoch` advances on every event that can lose messages without a replay:
+/// the link connects, drops, or skips a message it could not decode. A reader
+/// that needs every message records the epoch when its interest is confirmed
+/// and treats any later change as a possible loss.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct LinkState {
+    /// Advances on every possible loss.
+    pub epoch: u64,
+    /// Whether the link is connected and its desired topics are subscribed.
+    pub up: bool,
+}
+
 /// Configuration for the pub/sub subsystem.
 #[derive(Debug, Clone)]
 pub struct PubSubConfig {
@@ -115,6 +129,7 @@ pub struct PubSubManager {
     cache_invalidation_tx: broadcast::Sender<ScopedCacheInvalidation>,
     conn_control_tx: broadcast::Sender<ScopedConnControl>,
     nip_fi_disconnect_tx: broadcast::Sender<NipFiDisconnect>,
+    link: watch::Sender<LinkState>,
 }
 
 impl PubSubManager {
@@ -145,6 +160,7 @@ impl PubSubManager {
             cache_invalidation_tx,
             conn_control_tx,
             nip_fi_disconnect_tx,
+            link: watch::Sender::new(LinkState::default()),
         })
     }
 
@@ -163,6 +179,7 @@ impl PubSubManager {
             self.broadcast_tx.clone(),
             self.desired_topics.clone(),
             subscription_rx,
+            self.link.clone(),
         )
         .await;
     }
@@ -232,6 +249,28 @@ impl PubSubManager {
         }
     }
 
+    /// Watch this node's Redis event link (see [`LinkState`]).
+    pub fn link_state(&self) -> watch::Receiver<LinkState> {
+        self.link.subscribe()
+    }
+
+    /// Wait until Redis has confirmed a subscription to an already retained
+    /// topic, and return the link epoch it was confirmed on. Fails when the
+    /// link drops before it answers; the caller bounds the wait.
+    pub async fn confirm_topic(
+        &self,
+        ctx: &TenantContext,
+        topic: EventTopic,
+    ) -> Result<u64, PubSubError> {
+        let (reply, answer) = oneshot::channel();
+        let topic_key = EventTopicKey::from_context(ctx, topic);
+        self.subscription_tx
+            .send(subscriber::SubscriptionCommand::Confirm(topic_key, reply))
+            .await
+            .map_err(|_| PubSubError::SubscriberStopped)?;
+        answer.await.map_err(|_| PubSubError::SubscriberStopped)
+    }
+
     /// Release local interest in a scoped Redis event topic.
     ///
     /// When the last retain is released, unsubscribe is delayed by the configured

crates/buzz-pubsub/src/subscriber.rs

@@ -6,10 +6,10 @@ use std::time::Duration;
 
 use futures_util::StreamExt;
 use nostr::JsonUtil;
-use tokio::sync::{broadcast, mpsc, Mutex};
+use tokio::sync::{broadcast, mpsc, oneshot, watch, Mutex};
 
 use crate::topic::EventTopicKey;
-use crate::ChannelEvent;
+use crate::{ChannelEvent, LinkState};
 
 /// Initial reconnect backoff (1 second).
 const BACKOFF_INITIAL_SECS: u64 = 1;
@@ -21,12 +21,24 @@ pub(crate) type DesiredTopics = Arc<Mutex<HashMap<EventTopicKey, usize>>>;
 
 /// Commands sent from relay subscription registration/removal to the Redis
 /// pub/sub task.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug)]
 pub(crate) enum SubscriptionCommand {
     /// Ensure the topic is subscribed on the current Redis pub/sub connection.
     Subscribe(EventTopicKey),
     /// Unsubscribe only if the desired refcount is still zero when processed.
     UnsubscribeIfIdle(EventTopicKey),
+    /// Subscribe if needed, then answer with the link epoch on which the
+    /// server has confirmed the subscription.
+    Confirm(EventTopicKey, oneshot::Sender<u64>),
+}
+
+/// Record a possible loss of messages: the epoch advances, so every reader
+/// that started on the old epoch can tell.
+fn mark_loss(link: &watch::Sender<LinkState>, up: bool) {
+    link.send_modify(|state| {
+        state.epoch += 1;
+        state.up = up;
+    });
 }
 
 /// Runs a dynamically scoped subscriber and forwards events to broadcast.
@@ -39,18 +51,24 @@ pub(crate) async fn run_subscriber(
     broadcast_tx: broadcast::Sender<ChannelEvent>,
     desired_topics: DesiredTopics,
     mut subscription_rx: mpsc::Receiver<SubscriptionCommand>,
+    link: watch::Sender<LinkState>,
 ) {
     let mut backoff_secs = BACKOFF_INITIAL_SECS;
 
     loop {
-        match connect_and_subscribe(
+        let result = connect_and_subscribe(
             &redis_url,
             &broadcast_tx,
             desired_topics.clone(),
             &mut subscription_rx,
+            &link,
         )
-        .await
-        {
+        .await;
+        if link.borrow().up {
+            // Messages published while the link is down are not replayed.
+            mark_loss(&link, false);
+        }
+        match result {
             Ok(()) => {
                 // Stream ended cleanly (Redis returned None). The connection was
                 // established and ran successfully, so reset backoff to the initial
@@ -77,6 +95,7 @@ async fn connect_and_subscribe(
     broadcast_tx: &broadcast::Sender<ChannelEvent>,
     desired_topics: DesiredTopics,
     subscription_rx: &mut mpsc::Receiver<SubscriptionCommand>,
+    link: &watch::Sender<LinkState>,
 ) -> Result<(), redis::RedisError> {
     let client = redis::Client::open(redis_url)?;
     let conn = client.get_async_pubsub().await?;
@@ -101,6 +120,7 @@ async fn connect_and_subscribe(
         topic_count = active_topics.len(),
         "Redis pub/sub subscriber connected with dynamic scoped subscriptions"
     );
+    mark_loss(link, true);
 
     loop {
         tokio::select! {
@@ -112,6 +132,14 @@ async fn connect_and_subscribe(
                             sink.subscribe(&channel).await?;
                         }
                     }
+                    SubscriptionCommand::Confirm(topic, reply) => {
+                        let channel = topic.redis_channel();
+                        if !active_topics.contains(&channel) {
+                            sink.subscribe(&channel).await?;
+                            active_topics.insert(channel);
+                        }
+                        let _ = reply.send(link.borrow().epoch);
+                    }
                     SubscriptionCommand::UnsubscribeIfIdle(topic) => {
                         if desired_refcount(&desired_topics, topic).await == 0 {
                             let channel = topic.redis_channel();
@@ -132,6 +160,7 @@ async fn connect_and_subscribe(
                     Ok(p) => p,
                     Err(e) => {
                         tracing::warn!("Failed to get pub/sub message payload: {e}");
+                        mark_loss(link, true);
                         continue;
                     }
                 };
@@ -149,6 +178,7 @@ async fn connect_and_subscribe(
                     Ok(e) => e,
                     Err(e) => {
                         tracing::warn!("Failed to deserialize event from pub/sub: {e}");
+                        mark_loss(link, true);
                         continue;
                     }
                 };