From 3a34a3d2073339cd8cb567659f990803754ef9f8 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Wed, 21 Oct 2026 15:56:41 -0400 Subject: [PATCH] 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 --- crates/buzz-pubsub/src/lib.rs | 41 ++++++++++++++++++++++++++- crates/buzz-pubsub/src/subscriber.rs | 42 ++++++++++++++++++++++++---- 2 files changed, 76 insertions(+), 7 deletions(-) diff --git a/crates/buzz-pubsub/src/lib.rs b/crates/buzz-pubsub/src/lib.rs index 487ea9b630..b078699abb 100644 --- a/crates/buzz-pubsub/src/lib.rs +++ b/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, conn_control_tx: broadcast::Sender, nip_fi_disconnect_tx: broadcast::Sender, + link: watch::Sender, } 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 { + 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 { + 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 diff --git a/crates/buzz-pubsub/src/subscriber.rs b/crates/buzz-pubsub/src/subscriber.rs index 88826ed99b..0c34dba8d1 100644 --- a/crates/buzz-pubsub/src/subscriber.rs +++ b/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>>; /// 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), +} + +/// 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, 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, desired_topics: DesiredTopics, mut subscription_rx: mpsc::Receiver, + link: watch::Sender, ) { 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, desired_topics: DesiredTopics, subscription_rx: &mut mpsc::Receiver, + link: &watch::Sender, ) -> 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; } };