nostr: start discovery without blocking node startup

CHANGELOG.md

@@ -9,6 +9,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- Nostr discovery startup is now non-blocking. `Node::start` no
+  longer waits for relay connect, subscribe, or initial advert
+  publish before returning. A slow or unreachable relay no longer
+  holds node startup hostage; local transports come up immediately
+  and the relay path catches up asynchronously in background tasks.
+  Subscribe retries with exponential backoff (2 s base, 60 s cap),
+  publish attempts time out at 10 s, and the new tasks are aborted
+  cleanly on `Node::stop`.
 - Sidecar example (`examples/sidecar-nostr-relay`): `udp.mtu` is now
   overridable via the `FIPS_UDP_MTU` environment variable, defaulting to
   1472 (preserving prior behavior). Plumbed through `docker-compose.yml`

src/discovery/nostr/runtime.rs

@@ -11,7 +11,7 @@ use nostr::prelude::{
 };
 use nostr_sdk::{Client, ClientOptions, prelude::RelayPoolNotification};
 use serde::Serialize;
-use tokio::sync::{Mutex, RwLock, Semaphore, broadcast, mpsc, oneshot};
+use tokio::sync::{Mutex, Notify, RwLock, Semaphore, broadcast, mpsc, oneshot};
 use tokio::task::JoinHandle;
 use tracing::{debug, info, trace, warn};
 
@@ -75,6 +75,8 @@ struct CachedPublicUdpAddr {
 /// full `advert_refresh_secs` (30 min) for the success-path TTL to
 /// expire. Successful results use the longer per-config TTL.
 const PUBLIC_UDP_ADDR_FAILURE_TTL: Duration = Duration::from_secs(60);
+const RELAY_STARTUP_OP_TIMEOUT: Duration = Duration::from_secs(5);
+const ADVERT_PUBLISH_TIMEOUT: Duration = Duration::from_secs(10);
 
 pub struct NostrDiscovery {
     client: Client,
@@ -91,6 +93,10 @@ pub struct NostrDiscovery {
     offer_slots: Arc<Semaphore>,
     event_tx: mpsc::UnboundedSender<BootstrapEvent>,
     event_rx: Mutex<mpsc::UnboundedReceiver<BootstrapEvent>>,
+    connect_task: Mutex<Option<JoinHandle<()>>>,
+    relay_startup_task: Mutex<Option<JoinHandle<()>>>,
+    publish_task: Mutex<Option<JoinHandle<()>>>,
+    publish_notify: Notify,
     notify_task: Mutex<Option<JoinHandle<()>>>,
     advertise_task: Mutex<Option<JoinHandle<()>>>,
     failure_state: FailureState,
@@ -125,8 +131,6 @@ impl NostrDiscovery {
                 .await
                 .map_err(|e| BootstrapError::Nostr(e.to_string()))?;
         }
-        client.connect().await;
-
         let pubkey = keys.public_key();
         let npub = crate::encode_npub(&identity.pubkey());
         let (event_tx, event_rx) = mpsc::unbounded_channel();
@@ -154,6 +158,10 @@ impl NostrDiscovery {
             offer_slots,
             event_tx,
             event_rx: Mutex::new(event_rx),
+            connect_task: Mutex::new(None),
+            relay_startup_task: Mutex::new(None),
+            publish_task: Mutex::new(None),
+            publish_notify: Notify::new(),
             notify_task: Mutex::new(None),
             advertise_task: Mutex::new(None),
             failure_state,
@@ -170,8 +178,9 @@ impl NostrDiscovery {
         // `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.publish_task.lock().await = Some(runtime.clone().spawn_publish_loop());
+        *runtime.connect_task.lock().await = Some(runtime.clone().spawn_connect_loop());
+        *runtime.relay_startup_task.lock().await = Some(runtime.clone().spawn_relay_startup_loop());
         *runtime.advertise_task.lock().await = Some(runtime.clone().spawn_advertise_loop());
         *runtime.notify_task.lock().await = Some(runtime.clone().spawn_notify_loop(notifications));
 
@@ -452,7 +461,7 @@ impl NostrDiscovery {
     }
 
     pub async fn update_local_advert(
-        &self,
+        self: &Arc<Self>,
         advert: Option<OverlayAdvert>,
     ) -> Result<(), BootstrapError> {
         let changed = {
@@ -467,7 +476,8 @@ impl NostrDiscovery {
         if !changed {
             return Ok(());
         }
-        self.publish_advert().await
+        self.request_publish_advert();
+        Ok(())
     }
 
     pub async fn advert_endpoints_for_peer(
@@ -509,6 +519,15 @@ impl NostrDiscovery {
         if let Some(handle) = self.advertise_task.lock().await.take() {
             handle.abort();
         }
+        if let Some(handle) = self.connect_task.lock().await.take() {
+            handle.abort();
+        }
+        if let Some(handle) = self.relay_startup_task.lock().await.take() {
+            handle.abort();
+        }
+        if let Some(handle) = self.publish_task.lock().await.take() {
+            handle.abort();
+        }
 
         // Don't proactively retract the advert via NIP-09 on shutdown.
         // Parameterized-replaceable semantics handle restart supersedence,
@@ -663,18 +682,92 @@ impl NostrDiscovery {
         tokio::spawn(async move {
             let mut interval =
                 tokio::time::interval(Duration::from_secs(self.config.advert_refresh_secs.max(1)));
-            // Swallow the immediate first tick: Node::start() already publishes
-            // the initial advert via refresh_overlay_advert().
+            // Swallow the immediate first tick: Node::start() requests the
+            // initial advert publish via update_local_advert().
             interval.tick().await;
             loop {
                 interval.tick().await;
-                if let Err(err) = self.publish_advert().await {
-                    warn!(error = %err, "failed to refresh traversal advert");
+                self.request_publish_advert();
+            }
+        })
+    }
+
+    fn spawn_relay_startup_loop(self: Arc<Self>) -> JoinHandle<()> {
+        tokio::spawn(async move {
+            let mut retry_delay = Duration::from_secs(2);
+            loop {
+                let subscribed =
+                    match tokio::time::timeout(RELAY_STARTUP_OP_TIMEOUT, self.subscribe()).await {
+                        Ok(Ok(())) => true,
+                        Ok(Err(err)) => {
+                            warn!(error = %err, "failed to subscribe to Nostr discovery relays");
+                            false
+                        }
+                        Err(_) => {
+                            warn!(
+                                timeout_ms = RELAY_STARTUP_OP_TIMEOUT.as_millis() as u64,
+                                "Nostr discovery relay subscribe timed out"
+                            );
+                            false
+                        }
+                    };
+                match tokio::time::timeout(RELAY_STARTUP_OP_TIMEOUT, self.publish_inbox_relays())
+                    .await
+                {
+                    Ok(Ok(())) => {}
+                    Ok(Err(err)) => {
+                        warn!(error = %err, "failed to publish Nostr inbox relay list");
+                    }
+                    Err(_) => {
+                        warn!(
+                            timeout_ms = RELAY_STARTUP_OP_TIMEOUT.as_millis() as u64,
+                            "Nostr inbox relay publish timed out"
+                        );
+                    }
+                }
+
+                self.request_publish_advert();
+
+                if subscribed {
+                    break;
+                }
+
+                tokio::time::sleep(retry_delay).await;
+                retry_delay = retry_delay.saturating_mul(2).min(Duration::from_secs(60));
+            }
+        })
+    }
+
+    fn spawn_connect_loop(self: Arc<Self>) -> JoinHandle<()> {
+        tokio::spawn(async move {
+            self.client.connect().await;
+        })
+    }
+
+    fn spawn_publish_loop(self: Arc<Self>) -> JoinHandle<()> {
+        tokio::spawn(async move {
+            loop {
+                self.publish_notify.notified().await;
+                match tokio::time::timeout(ADVERT_PUBLISH_TIMEOUT, self.publish_advert()).await {
+                    Ok(Ok(())) => {}
+                    Ok(Err(err)) => {
+                        warn!(error = %err, "failed to publish traversal advert");
+                    }
+                    Err(_) => {
+                        warn!(
+                            timeout_ms = ADVERT_PUBLISH_TIMEOUT.as_millis() as u64,
+                            "Nostr traversal advert publish timed out"
+                        );
+                    }
                 }
             }
         })
     }
 
+    fn request_publish_advert(&self) {
+        self.publish_notify.notify_one();
+    }
+
     fn punch_hint(&self) -> PunchHint {
         PunchHint {
             start_at_ms: now_ms() + self.config.punch_start_delay_ms,
@@ -1503,6 +1596,10 @@ impl NostrDiscovery {
             offer_slots,
             event_tx,
             event_rx: Mutex::new(event_rx),
+            connect_task: Mutex::new(None),
+            relay_startup_task: Mutex::new(None),
+            publish_task: Mutex::new(None),
+            publish_notify: Notify::new(),
             notify_task: Mutex::new(None),
             advertise_task: Mutex::new(None),
             failure_state,

src/node/tests/unit.rs

@@ -113,6 +113,35 @@ async fn test_node_state_transitions() {
     assert_eq!(node.state(), NodeState::Stopped);
 }
 
+#[tokio::test]
+async fn test_node_start_does_not_wait_for_nostr_relay_startup() {
+    let mut config = Config::new();
+    config.node.control.enabled = false;
+    config.node.discovery.nostr.enabled = true;
+    config.node.discovery.nostr.advertise = true;
+    config.node.discovery.nostr.policy = crate::config::NostrDiscoveryPolicy::Open;
+    config.node.discovery.nostr.advert_relays = vec!["wss://127.0.0.1:9".to_string()];
+    config.node.discovery.nostr.dm_relays = vec!["wss://127.0.0.1:9".to_string()];
+    config.transports.udp = crate::config::TransportInstances::Single(crate::config::UdpConfig {
+        bind_addr: Some("127.0.0.1:0".to_string()),
+        advertise_on_nostr: Some(true),
+        public: Some(false),
+        accept_connections: Some(true),
+        ..Default::default()
+    });
+
+    let mut node = Node::new(config).unwrap();
+    tokio::time::timeout(std::time::Duration::from_millis(500), node.start())
+        .await
+        .expect("node start should not wait for relay I/O")
+        .unwrap();
+
+    assert!(node.is_running());
+    assert!(node.nostr_discovery_handle().is_some());
+
+    node.stop().await.unwrap();
+}
+
 #[tokio::test]
 async fn test_node_double_start() {
     let mut node = make_node();