From c0ccedb491462dfb0abb3b00c50fa89485be058c Mon Sep 17 00:00:00 2001 From: Martti Malmi Date: Mon, 18 May 2026 16:52:59 +0000 Subject: [PATCH] nostr: start discovery without blocking node startup --- CHANGELOG.md | 8 +++ src/discovery/nostr/runtime.rs | 119 ++++++++++++++++++++++++++++++--- src/node/tests/unit.rs | 29 ++++++++ 3 files changed, 145 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6be1685..982a5bc 100644 --- a/CHANGELOG.md +++ b/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` diff --git a/src/discovery/nostr/runtime.rs b/src/discovery/nostr/runtime.rs index 26f3b4b..b6b15a2 100644 --- a/src/discovery/nostr/runtime.rs +++ b/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, event_tx: mpsc::UnboundedSender, event_rx: Mutex>, + connect_task: Mutex>>, + relay_startup_task: Mutex>>, + publish_task: Mutex>>, + publish_notify: Notify, notify_task: Mutex>>, advertise_task: Mutex>>, 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, advert: Option, ) -> 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) -> 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) -> JoinHandle<()> { + tokio::spawn(async move { + self.client.connect().await; + }) + } + + fn spawn_publish_loop(self: Arc) -> 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, diff --git a/src/node/tests/unit.rs b/src/node/tests/unit.rs index a4a7a2b..db3341a 100644 --- a/src/node/tests/unit.rs +++ b/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();