From 8094a51a82d6669e429c5a759a0c83ee4b39d385 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sun, 10 May 2026 01:03:54 +0000 Subject: [PATCH] nostr: fix subscription startup race losing relay REQ replays MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Freshly-restarted nodes with policy: open silently lost the historical event replay that relays send in response to subscribe(). The broadcast::Receiver was created INSIDE spawn_notify_loop, which the tokio runtime starts at some indeterminate point after subscribe() returns. tokio's broadcast channel only delivers messages sent after the receiver is created; messages dispatched in the gap between subscribe() issuing the REQ and the spawned task calling client.notifications() were dropped by external_notification_sender.send returning Err(SendError) with no subscribers attached. Symptom on a node with policy: open: non-configured peers were not discovered until they next re-published their advert (default advert_refresh_secs = 1800s = 30 min). Configured peers were unaffected because fetch_advert (relay-fetch path) caches them at startup-sweep time. The bug has been latent since 34e00b9 added Nostr discovery — relay-fetch covered the common case for configured-peer setups. Fix: create the broadcast::Receiver in start() before subscribe() and pass it into spawn_notify_loop. The receiver now exists when the REQ replay arrives, so historical events flow through the cache path. Also handle broadcast::error::RecvError::Lagged separately from ::Closed. The previous `while let Ok(...) = recv().await` exited the loop on any Err, so a single lag event would silently kill the entire subscription consumer with no recovery. Lagged now logs a warn (with the skipped count) and continues; only Closed exits the loop. Add two info-level log lines for in-field observability of the loop's liveness. "nostr notify loop entered" fires once at task start; "nostr notify loop received first event" fires once after the first successful recv() with elapsed_ms since loop entry. Together these turn the previous silent-failure shape (zero advert: peer cached log lines indistinguishable between dead loop and idle channel) into an immediately greppable startup signal — operators can confirm the loop is alive and see how long it took to receive its first event, catching any future regression in the subscription codepath in seconds rather than waiting one advert_refresh_secs interval. No public API change; the test fixture (NostrDiscovery::new_for_test) does not call spawn_notify_loop and is unaffected. --- src/discovery/nostr/runtime.rs | 48 +++++++++++++++++++++++++++++----- 1 file changed, 42 insertions(+), 6 deletions(-) diff --git a/src/discovery/nostr/runtime.rs b/src/discovery/nostr/runtime.rs index 3ad700c..eea46f5 100644 --- a/src/discovery/nostr/runtime.rs +++ b/src/discovery/nostr/runtime.rs @@ -11,9 +11,9 @@ use nostr::prelude::{ }; use nostr_sdk::{Client, ClientOptions, prelude::RelayPoolNotification}; use serde::Serialize; -use tokio::sync::{Mutex, RwLock, Semaphore, mpsc, oneshot}; +use tokio::sync::{Mutex, RwLock, Semaphore, broadcast, mpsc, oneshot}; use tokio::task::JoinHandle; -use tracing::{debug, trace, warn}; +use tracing::{debug, info, trace, warn}; use super::failure_state::FailureState; use super::signal::{ @@ -160,10 +160,20 @@ impl NostrDiscovery { public_udp_addr_cache: RwLock::new(HashMap::new()), }); + // Subscribe to the relay-pool broadcast channel BEFORE issuing the + // Nostr REQs. tokio's broadcast channel only delivers messages sent + // after the receiver is created — historical events that arrive in + // response to subscribe() (REQ replays) would otherwise be dropped + // by the pool's `external_notification_sender.send(...)` returning + // `Err(SendError)` when no subscriber exists yet. Without this, + // freshly-restarted nodes with `policy: open` waited up to one + // `advert_refresh_secs` interval (default 30 min) for non-configured + // peers to re-publish before discovering them. + let notifications = runtime.client.notifications(); runtime.subscribe().await?; runtime.publish_inbox_relays().await?; *runtime.advertise_task.lock().await = Some(runtime.clone().spawn_advertise_loop()); - *runtime.notify_task.lock().await = Some(runtime.clone().spawn_notify_loop()); + *runtime.notify_task.lock().await = Some(runtime.clone().spawn_notify_loop(notifications)); Ok(runtime) } @@ -515,10 +525,36 @@ impl NostrDiscovery { Ok(()) } - fn spawn_notify_loop(self: Arc) -> JoinHandle<()> { + fn spawn_notify_loop( + self: Arc, + mut notifications: broadcast::Receiver, + ) -> JoinHandle<()> { tokio::spawn(async move { - let mut notifications = self.client.notifications(); - while let Ok(notification) = notifications.recv().await { + let started_at = Instant::now(); + let mut first_event_seen = false; + info!("nostr notify loop entered"); + loop { + let notification = match notifications.recv().await { + Ok(notification) => notification, + Err(broadcast::error::RecvError::Lagged(skipped)) => { + warn!( + skipped, + "nostr notification channel lagged; advert/signal events dropped" + ); + continue; + } + Err(broadcast::error::RecvError::Closed) => { + warn!("nostr notification channel closed; notify loop exiting"); + break; + } + }; + if !first_event_seen { + first_event_seen = true; + info!( + elapsed_ms = started_at.elapsed().as_millis() as u64, + "nostr notify loop received first event" + ); + } if let RelayPoolNotification::Event { event, .. } = notification { if event.kind == Kind::Custom(ADVERT_KIND) { let author_npub = event.pubkey.to_bech32().expect("infallible");