nostr: fix subscription startup race losing relay REQ replays

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

@@ -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<Self>) -> JoinHandle<()> {
+    fn spawn_notify_loop(
+        self: Arc<Self>,
+        mut notifications: broadcast::Receiver<RelayPoolNotification>,
+    ) -> 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");