From 1208f6a5c2b398f2249a7c685edd05c6e84e68d2 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Thu, 9 Jul 2026 22:13:56 +0000 Subject: [PATCH] Consolidate the nostr rendezvous driver into src/nostr Pull the rendezvous driver state and the movable driver logic out of the Node struct into the src/nostr home. A new RendezvousDriver owns the engine handle and the four bookkeeping fields (traversal start time, startup-sweep latch, adopted bootstrap-transport set and their npubs) that previously sat loose on Node; Node holds a single driver field and reaches the same data through thin accessors, including the rx-loop hot-path protocol-mismatch hook. The advert build/refresh, the via_nostr fallback-address resolve, the overlay-endpoint-to-PeerAddress mapping, and the bootstrap request move onto the driver, taking their Node inputs explicitly (a transport- endpoint snapshot for advert building) rather than reading Node fields directly. Transport/connection-table-bound work (traversal adoption, bootstrap-transport cleanup, the open-discovery sweep, and the outbound budget calculators) stays on Node as thin glue that calls into the driver. Behavior-neutral relocation: statements moved verbatim, no logic, wire, config-key, or metric changes. lifecycle.rs shrinks ~230 lines. cargo fmt/build/clippy clean; lib suite 1547 passing (baseline unchanged). --- src/node/handlers/rx_loop.rs | 8 +- src/node/lifecycle.rs | 298 ++++++-------------------- src/node/mod.rs | 47 ++--- src/node/retry.rs | 4 +- src/node/tests/unit.rs | 6 +- src/nostr/driver.rs | 399 +++++++++++++++++++++++++++++++++++ src/nostr/mod.rs | 2 + 7 files changed, 486 insertions(+), 278 deletions(-) create mode 100644 src/nostr/driver.rs diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index f7b3288..756aa9a 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -319,10 +319,12 @@ impl Node { // though no msg1/msg2 exchange can ever succeed. Bump the // discovery-layer cooldown to the long protocol-mismatch // window and emit a single WARN per fresh observation. - if self.bootstrap_transports.contains(&packet.transport_id) + if self + .nostr_rendezvous + .is_bootstrap_transport(&packet.transport_id) && let Some(npub) = self - .bootstrap_transport_npubs - .get(&packet.transport_id) + .nostr_rendezvous + .bootstrap_transport_npub(&packet.transport_id) .cloned() && let Some(handle) = self.nostr_rendezvous_handle() { diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index bfd97e1..200ef4d 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -4,10 +4,7 @@ use super::{Node, NodeError, NodeState}; use crate::config::{ConnectPolicy, PeerAddress, PeerConfig}; use crate::node::acl::PeerAclContext; use crate::node::wire::build_msg1; -use crate::nostr::{ - ADVERT_IDENTIFIER, ADVERT_VERSION, BootstrapEvent, NostrRendezvous, OverlayAdvert, - OverlayEndpointAdvert, OverlayTransportKind, -}; +use crate::nostr::{BootstrapEvent, NostrRendezvous}; use crate::nostr::{BootstrapHandoffResult, EstablishedTraversal}; use crate::peer::PeerConnection; use crate::proto::fmp::{Disconnect, DisconnectReason}; @@ -266,7 +263,7 @@ impl Node { // would loop on the same dead address until expiry. Force a // re-fetch so the next retry tick picks up fresh endpoints. if matches!(e, crate::node::NodeError::NoTransportForType(_)) - && let Some(bootstrap) = self.nostr_rendezvous.clone() + && let Some(bootstrap) = self.nostr_rendezvous.engine_arc() { let npub = peer_config.npub.clone(); tokio::spawn(async move { @@ -357,7 +354,7 @@ impl Node { .filter(|(id, handle)| { handle.transport_type().name == "udp" && handle.is_operational() - && !self.bootstrap_transports.contains(id) + && !self.nostr_rendezvous.is_bootstrap_transport(id) }) .filter_map(|(id, handle)| { let local_addr = handle.local_addr()?; @@ -680,7 +677,7 @@ impl Node { } pub(super) async fn poll_nostr_rendezvous(&mut self) { - let Some(bootstrap) = self.nostr_rendezvous.clone() else { + let Some(bootstrap) = self.nostr_rendezvous.engine_arc() else { return; }; @@ -1149,8 +1146,8 @@ impl Node { if let Err(err) = self.refresh_overlay_advert(&runtime).await { warn!(error = %err, "Failed to publish initial Nostr overlay advert"); } - self.nostr_rendezvous = Some(runtime); - self.nostr_rendezvous_started_at_ms = Some(Self::now_ms()); + self.nostr_rendezvous.set_engine(runtime); + self.nostr_rendezvous.set_started_at_ms(Self::now_ms()); info!("Nostr overlay discovery enabled"); } Err(err) => { @@ -1175,7 +1172,7 @@ impl Node { .filter(|(id, h)| { h.transport_type().name == "udp" && h.is_operational() - && !self.bootstrap_transports.contains(id) + && !self.nostr_rendezvous.is_bootstrap_transport(id) }) .filter_map(|(id, h)| h.local_addr().map(|addr| (*id, addr.port()))) .min_by_key(|(id, _)| id.as_u32()) @@ -1479,7 +1476,7 @@ impl Node { .await; // Stop Nostr overlay discovery background work and withdraw any advert. - if let Some(bootstrap) = self.nostr_rendezvous.take() + if let Some(bootstrap) = self.nostr_rendezvous.take_engine() && let Err(e) = bootstrap.shutdown().await { warn!(error = %e, "Failed to shutdown Nostr overlay discovery"); @@ -1621,89 +1618,6 @@ impl Node { .collect() } - async fn nostr_peer_fallback_addresses( - &self, - peer_config: &PeerConfig, - existing: &[PeerAddress], - ) -> Vec { - if !self.config().node.rendezvous.nostr.enabled - || !peer_config.via_nostr - || self.config().node.rendezvous.nostr.policy - == crate::config::NostrRendezvousPolicy::Disabled - { - return Vec::new(); - } - - let Some(bootstrap) = self.nostr_rendezvous.clone() else { - return Vec::new(); - }; - let endpoints = match bootstrap.advert_endpoints_for_peer(&peer_config.npub).await { - Ok(endpoints) => endpoints, - Err(err) => { - debug!( - npub = %peer_config.npub, - error = %err, - "Failed to resolve Nostr advert endpoints for configured peer" - ); - return Vec::new(); - } - }; - - let mut fallback = Vec::new(); - let mut next_priority = existing - .iter() - .map(|addr| addr.priority) - .max() - .unwrap_or(100) - .saturating_add(1); - let seen_at_ms = Self::now_ms(); - for endpoint in endpoints { - let Some(candidate) = - Self::overlay_endpoint_to_peer_address(&endpoint, next_priority, seen_at_ms) - else { - continue; - }; - if existing - .iter() - .any(|addr| addr.transport == candidate.transport && addr.addr == candidate.addr) - || fallback.iter().any(|addr: &PeerAddress| { - addr.transport == candidate.transport && addr.addr == candidate.addr - }) - { - continue; - } - fallback.push(candidate); - next_priority = next_priority.saturating_add(1); - } - fallback - } - - fn overlay_endpoint_to_peer_address( - endpoint: &OverlayEndpointAdvert, - priority: u8, - seen_at_ms: u64, - ) -> Option { - let transport = match endpoint.transport { - OverlayTransportKind::Udp => "udp", - OverlayTransportKind::Tcp => "tcp", - OverlayTransportKind::Tor => "tor", - }; - Some( - PeerAddress::with_priority(transport, endpoint.addr.clone(), priority) - .with_seen_at_ms(seen_at_ms), - ) - } - - async fn request_nostr_bootstrap(&self, peer_config: &PeerConfig) -> bool { - let Some(bootstrap) = self.nostr_rendezvous.clone() else { - debug!(npub = %peer_config.npub, "No Nostr overlay runtime for udp:nat address"); - return false; - }; - bootstrap.request_connect(peer_config.clone()).await; - info!(npub = %peer_config.npub, "Started Nostr UDP NAT traversal attempt"); - true - } - async fn attempt_peer_address_list( &mut self, peer_config: &PeerConfig, @@ -1729,7 +1643,11 @@ impl Node { if !allow_bootstrap_nat { continue; } - if self.request_nostr_bootstrap(peer_config).await { + if self + .nostr_rendezvous + .request_nostr_bootstrap(peer_config) + .await + { attempted = attempted.saturating_add(1); } continue; @@ -1960,7 +1878,9 @@ impl Node { let seen_at_ms = Self::now_ms(); for endpoint in endpoints { let Some(candidate) = - Self::overlay_endpoint_to_peer_address(&endpoint, priority, seen_at_ms) + crate::nostr::RendezvousDriver::overlay_endpoint_to_peer_address( + &endpoint, priority, seen_at_ms, + ) else { continue; }; @@ -2048,7 +1968,7 @@ impl Node { &mut self, bootstrap: &std::sync::Arc, ) { - if self.startup_open_discovery_sweep_done { + if self.nostr_rendezvous.startup_sweep_done() { return; } if !self.config().node.rendezvous.nostr.enabled @@ -2056,10 +1976,10 @@ impl Node { != crate::config::NostrRendezvousPolicy::Open { // Mark done so we don't keep re-checking on every tick. - self.startup_open_discovery_sweep_done = true; + self.nostr_rendezvous.set_startup_sweep_done(); return; } - let Some(started_at_ms) = self.nostr_rendezvous_started_at_ms else { + let Some(started_at_ms) = self.nostr_rendezvous.started_at_ms() else { return; }; let now_ms = Self::now_ms(); @@ -2082,7 +2002,7 @@ impl Node { .startup_sweep_max_age_secs; self.run_open_discovery_sweep(bootstrap, Some(max_age_secs), "startup") .await; - self.startup_open_discovery_sweep_done = true; + self.nostr_rendezvous.set_startup_sweep_done(); } fn available_outbound_slots(&self) -> usize { @@ -2190,163 +2110,66 @@ impl Node { ) } - async fn build_overlay_advert( - &self, - bootstrap: &std::sync::Arc, - ) -> Option { - if !self.config().node.rendezvous.nostr.enabled { - return None; - } - - let mut endpoints = Vec::new(); - let mut has_udp_nat = false; - + /// Capture the advertisable-endpoint inputs of every operational + /// transport into a snapshot the rendezvous driver can turn into an + /// `OverlayAdvert` without borrowing the transport table across the + /// STUN await. Iteration order matches `self.transports.values()`, and + /// only transports whose type matched a configured listener are + /// included, mirroring the original per-transport branch structure. + fn advert_transport_snapshot(&self) -> Vec { + use crate::nostr::AdvertTransportSnapshot; + let mut snapshot = Vec::new(); for handle in self.transports.values() { if !handle.is_operational() { continue; } - match handle.transport_type().name { "udp" => { let Some(cfg) = self.lookup_udp_config(handle.name()) else { continue; }; - if !cfg.advertise_on_nostr() { - continue; - } - if cfg.is_public() { - // Precedence: - // 1. operator-supplied `external_addr` (skips STUN) - // 2. non-wildcard `local_addr` (operator bound to - // a specific public IP directly) - // 3. STUN auto-discovery against ephemeral socket - // 4. loud warn + omit endpoint - if let Some(explicit) = cfg.external_advert_addr() { - endpoints.push(OverlayEndpointAdvert { - transport: OverlayTransportKind::Udp, - addr: explicit.to_string(), - }); - } else { - match handle.local_addr() { - Some(addr) if !addr.ip().is_unspecified() => { - endpoints.push(OverlayEndpointAdvert { - transport: OverlayTransportKind::Udp, - addr: addr.to_string(), - }); - } - Some(addr) => { - let key = handle.transport_id().as_u32(); - let port = addr.port(); - if let Some(public) = - bootstrap.learn_public_udp_addr(key, port).await - { - endpoints.push(OverlayEndpointAdvert { - transport: OverlayTransportKind::Udp, - addr: public.to_string(), - }); - } else { - warn!( - transport_id = key, - bind_addr = %addr, - "advert: udp public=true bound to wildcard but \ - STUN observation failed; advertising no UDP \ - endpoint. Either set transports.udp.external_addr, \ - bind to a specific public IP, or ensure \ - node.rendezvous.nostr.stun_servers is reachable" - ); - } - } - None => {} - } - } - } else { - endpoints.push(OverlayEndpointAdvert { - transport: OverlayTransportKind::Udp, - addr: "nat".to_string(), - }); - has_udp_nat = true; - } + snapshot.push(AdvertTransportSnapshot::Udp { + advertise: cfg.advertise_on_nostr(), + is_public: cfg.is_public(), + external_addr: cfg.external_advert_addr(), + local_addr: handle.local_addr(), + transport_key: handle.transport_id().as_u32(), + }); } "tcp" => { let Some(cfg) = self.lookup_tcp_config(handle.name()) else { continue; }; - if !cfg.advertise_on_nostr() { - continue; - } - // Precedence: - // 1. operator-supplied `external_addr` (only path that - // works on cloud-NAT setups where the public IP is - // not on a host interface). - // 2. non-wildcard `local_addr` (operator bound to a - // specific public IP directly). - // 3. loud warn + omit endpoint (no TCP STUN equivalent). - if let Some(explicit) = cfg.external_advert_addr() { - endpoints.push(OverlayEndpointAdvert { - transport: OverlayTransportKind::Tcp, - addr: explicit.to_string(), - }); - } else { - match handle.local_addr() { - Some(addr) if !addr.ip().is_unspecified() => { - endpoints.push(OverlayEndpointAdvert { - transport: OverlayTransportKind::Tcp, - addr: addr.to_string(), - }); - } - Some(addr) => { - warn!( - bind_addr = %addr, - "advert: tcp advertise_on_nostr=true bound to wildcard \ - and no transports.tcp.external_addr set; advertising no \ - TCP endpoint. Either set external_addr to the public \ - IP (recommended for cloud 1:1-NAT setups) or bind \ - explicitly to the public IP" - ); - } - None => {} - } - } + snapshot.push(AdvertTransportSnapshot::Tcp { + advertise: cfg.advertise_on_nostr(), + external_addr: cfg.external_advert_addr(), + local_addr: handle.local_addr(), + }); } "tor" => { let Some(cfg) = self.lookup_tor_config(handle.name()) else { continue; }; - if !cfg.advertise_on_nostr() { - continue; - } - if let Some(addr) = handle.onion_address() { - endpoints.push(OverlayEndpointAdvert { - transport: OverlayTransportKind::Tor, - addr: format!("{}:{}", addr, cfg.advertised_port()), - }); - } + snapshot.push(AdvertTransportSnapshot::Tor { + advertise: cfg.advertise_on_nostr(), + onion_addr: handle.onion_address().map(|s| s.to_string()), + advertised_port: cfg.advertised_port(), + }); } _ => {} } } - - if endpoints.is_empty() { - return None; - } - - Some(OverlayAdvert { - identifier: ADVERT_IDENTIFIER.to_string(), - version: ADVERT_VERSION, - endpoints, - signal_relays: has_udp_nat - .then(|| self.config().node.rendezvous.nostr.dm_relays.clone()), - stun_servers: has_udp_nat - .then(|| self.config().node.rendezvous.nostr.stun_servers.clone()), - }) + snapshot } async fn refresh_overlay_advert( &self, bootstrap: &std::sync::Arc, ) -> Result<(), crate::nostr::BootstrapError> { - let advert = self.build_overlay_advert(bootstrap).await; - bootstrap.update_local_advert(advert).await + let snapshot = self.advert_transport_snapshot(); + self.nostr_rendezvous + .refresh_overlay_advert(bootstrap, snapshot, &self.config().node.rendezvous.nostr) + .await } fn lookup_udp_config(&self, transport_name: Option<&str>) -> Option<&crate::config::UdpConfig> { @@ -2463,7 +2286,13 @@ impl Node { async fn peer_address_candidates(&self, peer_config: &PeerConfig) -> Vec { let static_addresses = self.static_peer_addresses(peer_config); let overlay_addresses = self - .nostr_peer_fallback_addresses(peer_config, &static_addresses) + .nostr_rendezvous + .nostr_peer_fallback_addresses( + peer_config, + &static_addresses, + &self.config().node.rendezvous.nostr, + Self::now_ms(), + ) .await; let mut candidates = Vec::with_capacity(overlay_addresses.len() + static_addresses.len()); @@ -2531,7 +2360,7 @@ impl Node { }; if peer .transport_id() - .map(|id| self.bootstrap_transports.contains(&id)) + .map(|id| self.nostr_rendezvous.is_bootstrap_transport(&id)) .unwrap_or(false) { return false; @@ -2727,17 +2556,16 @@ impl Node { transport_id, crate::transport::TransportHandle::Udp(transport), ); - self.bootstrap_transports.insert(transport_id); - self.bootstrap_transport_npubs - .insert(transport_id, traversal.peer_npub.clone()); + self.nostr_rendezvous + .insert_bootstrap_transport(transport_id, traversal.peer_npub.clone()); let remote_addr = TransportAddr::from_string(&traversal.remote_addr.to_string()); if let Err(err) = self .initiate_connection(transport_id, remote_addr.clone(), peer_identity) .await { - self.bootstrap_transports.remove(&transport_id); - self.bootstrap_transport_npubs.remove(&transport_id); + self.nostr_rendezvous + .remove_bootstrap_transport(&transport_id); if let Some(mut handle) = self.transports.remove(&transport_id) { let _ = handle.stop().await; } diff --git a/src/node/mod.rs b/src/node/mod.rs index 14c7fc0..2b699a8 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -65,7 +65,7 @@ use crate::upper::tun::{TunError, TunOutboundRx, TunState, TunTx}; use crate::utils::index::IndexAllocator; use crate::{Config, ConfigError, Identity, IdentityError, NodeAddr, PeerIdentity, TreeCoordinate}; use rand::Rng; -use std::collections::{BTreeSet, HashMap, HashSet, VecDeque}; +use std::collections::{BTreeSet, HashMap, VecDeque}; use std::fmt; use std::sync::Arc; use std::thread::JoinHandle; @@ -444,31 +444,16 @@ pub struct Node { /// are exhausted. retry_pending: HashMap, - /// Optional Nostr/STUN overlay discovery coordinator for `udp:nat` peers. - nostr_rendezvous: Option>, + /// Node-side driver state for the Nostr overlay peer-rendezvous + /// subsystem: the engine handle, its startup timestamp, the one-shot + /// startup-sweep latch, and the per-peer bootstrap-transport bookkeeping + /// adopted from NAT-traversal handoffs. + nostr_rendezvous: crate::nostr::RendezvousDriver, /// mDNS / DNS-SD responder + browser for local-link peer discovery. /// Identity is unverified at this layer — the Noise XX handshake /// initiated against an mDNS-observed endpoint is what proves the /// peer holds the matching private key. lan_rendezvous: Option>, - /// Wall-clock ms when Nostr discovery successfully started, used to - /// schedule the one-shot startup advert sweep after a settle delay. - /// `None` until discovery comes up; remains `None` if discovery is - /// disabled or failed to start. - nostr_rendezvous_started_at_ms: Option, - /// Whether the one-shot startup advert sweep has run. Set to true - /// after the first sweep fires (under `policy: open`); thereafter - /// only the per-tick `queue_open_discovery_retries` continues. - startup_open_discovery_sweep_done: bool, - /// Per-peer UDP transports adopted from NAT traversal handoff. - bootstrap_transports: HashSet, - /// Originating peer npub (bech32) for each adopted bootstrap - /// transport, captured at `adopt_established_traversal` time. - /// Populated alongside `bootstrap_transports`; cleared in - /// `cleanup_bootstrap_transport_if_unused`. Used by the rx loop to - /// route fatal-protocol-mismatch observations back to the - /// Nostr-discovery `failure_state` for long cooldown application. - bootstrap_transport_npubs: HashMap, // === Periodic Parent Re-evaluation === /// Timestamp of last periodic parent re-evaluation (for pacing). @@ -684,12 +669,8 @@ impl Node { ), pending_connects: Vec::new(), retry_pending: HashMap::new(), - nostr_rendezvous: None, - nostr_rendezvous_started_at_ms: None, + nostr_rendezvous: crate::nostr::RendezvousDriver::default(), lan_rendezvous: None, - startup_open_discovery_sweep_done: false, - bootstrap_transports: HashSet::new(), - bootstrap_transport_npubs: HashMap::new(), last_parent_reeval: None, last_congestion_log: None, estimated_mesh_size: None, @@ -846,12 +827,8 @@ impl Node { lookup: Lookup::new(LookupBackoff::new(), LookupForwardRateLimiter::new()), pending_connects: Vec::new(), retry_pending: HashMap::new(), - nostr_rendezvous: None, - nostr_rendezvous_started_at_ms: None, + nostr_rendezvous: crate::nostr::RendezvousDriver::default(), lan_rendezvous: None, - startup_open_discovery_sweep_done: false, - bootstrap_transports: HashSet::new(), - bootstrap_transport_npubs: HashMap::new(), last_parent_reeval: None, last_congestion_log: None, estimated_mesh_size: None, @@ -2292,7 +2269,7 @@ impl Node { } pub(crate) fn cleanup_bootstrap_transport_if_unused(&mut self, transport_id: TransportId) { - if !self.bootstrap_transports.contains(&transport_id) { + if !self.nostr_rendezvous.is_bootstrap_transport(&transport_id) { return; } @@ -2322,8 +2299,8 @@ impl Node { "bootstrap transport has no remaining references; dropping" ); - self.bootstrap_transports.remove(&transport_id); - self.bootstrap_transport_npubs.remove(&transport_id); + self.nostr_rendezvous + .remove_bootstrap_transport(&transport_id); self.transport_drops.remove(&transport_id); self.transports.remove(&transport_id); } @@ -2399,7 +2376,7 @@ impl Node { /// Used by control queries (`show_peers` per-peer Nostr-traversal /// state) to read failure-state without taking shared ownership. pub fn nostr_rendezvous_handle(&self) -> Option<&crate::nostr::NostrRendezvous> { - self.nostr_rendezvous.as_deref() + self.nostr_rendezvous.engine() } /// Iterate over all peer node IDs. diff --git a/src/node/retry.rs b/src/node/retry.rs index 9fffdb5..ba422aa 100644 --- a/src/node/retry.rs +++ b/src/node/retry.rs @@ -282,7 +282,7 @@ impl Node { // evicts if the relay has nothing, otherwise leaves it. Cheap // (one Filter fetch with 2s timeout) and bounded by the retry // backoff cadence. - if let Some(bootstrap) = self.nostr_rendezvous.clone() { + if let Some(bootstrap) = self.nostr_rendezvous.engine_arc() { let _ = bootstrap .refetch_advert_for_stale_check(&peer_config.npub) .await; @@ -318,7 +318,7 @@ impl Node { // entry expires. Force a re-fetch so the next retry tick // picks up fresh endpoints. if matches!(e, NodeError::NoTransportForType(_)) - && let Some(bootstrap) = self.nostr_rendezvous.clone() + && let Some(bootstrap) = self.nostr_rendezvous.engine_arc() { let npub = peer_config.npub.clone(); tokio::spawn(async move { diff --git a/src/node/tests/unit.rs b/src/node/tests/unit.rs index 6408dab..c857094 100644 --- a/src/node/tests/unit.rs +++ b/src/node/tests/unit.rs @@ -1130,7 +1130,7 @@ async fn test_nostr_traversal_failure_skips_connected_peer() { peer_config: crate::config::PeerConfig::new(peer_identity.npub(), "udp", "127.0.0.1:9"), reason: "stale traversal failure".to_string(), }); - node.nostr_rendezvous = Some(bootstrap.clone()); + node.nostr_rendezvous.set_engine(bootstrap.clone()); node.poll_nostr_rendezvous().await; @@ -1170,7 +1170,7 @@ async fn test_nostr_traversal_established_skips_connected_peer() { socket, ), }); - node.nostr_rendezvous = Some(bootstrap.clone()); + node.nostr_rendezvous.set_engine(bootstrap.clone()); node.poll_nostr_rendezvous().await; @@ -1727,7 +1727,7 @@ async fn poll_nostr_rendezvous_established_gated_at_capacity() { socket, ), }); - node.nostr_rendezvous = Some(bootstrap.clone()); + node.nostr_rendezvous.set_engine(bootstrap.clone()); let before_peers = node.peer_count(); let before_links = node.link_count(); diff --git a/src/nostr/driver.rs b/src/nostr/driver.rs new file mode 100644 index 0000000..e595181 --- /dev/null +++ b/src/nostr/driver.rs @@ -0,0 +1,399 @@ +//! Node-side driver state for the Nostr overlay peer-rendezvous subsystem. +//! +//! [`RendezvousDriver`] consolidates the rendezvous-subsystem state that +//! previously lived as loose fields on the `Node` struct: the engine handle, +//! its startup timestamp, the one-shot startup-sweep latch, and the +//! per-peer bootstrap-transport bookkeeping adopted from NAT-traversal +//! handoffs. Keeping it in the `nostr` module gives the subsystem a single +//! home while leaving the transport/connection-table mutations that consume +//! this state on `Node`. + +use std::collections::{HashMap, HashSet}; +use std::net::SocketAddr; +use std::sync::Arc; + +use tracing::{debug, info, warn}; + +use crate::config::{NostrRendezvousConfig, NostrRendezvousPolicy, PeerAddress, PeerConfig}; +use crate::transport::TransportId; + +use super::{ + ADVERT_IDENTIFIER, ADVERT_VERSION, BootstrapError, NostrRendezvous, OverlayAdvert, + OverlayEndpointAdvert, OverlayTransportKind, +}; + +/// Snapshot of a single operational transport's advertisable endpoint +/// inputs, captured on `Node` at advert-build time so the driver can +/// assemble the overlay advert without borrowing the transport table. +/// Only transports whose type matched a configured listener are included; +/// the `advertise` gate is carried verbatim so the driver reproduces the +/// original per-transport branch logic exactly. +pub enum AdvertTransportSnapshot { + Udp { + advertise: bool, + is_public: bool, + external_addr: Option, + local_addr: Option, + transport_key: u32, + }, + Tcp { + advertise: bool, + external_addr: Option, + local_addr: Option, + }, + Tor { + advertise: bool, + onion_addr: Option, + advertised_port: u16, + }, +} + +/// Node-side rendezvous-subsystem state and bootstrap-transport bookkeeping. +#[derive(Default)] +pub struct RendezvousDriver { + /// Optional Nostr/STUN overlay discovery coordinator for `udp:nat` peers. + engine: Option>, + /// Wall-clock ms when Nostr discovery successfully started, used to + /// schedule the one-shot startup advert sweep after a settle delay. + /// `None` until discovery comes up; remains `None` if discovery is + /// disabled or failed to start. + started_at_ms: Option, + /// Whether the one-shot startup advert sweep has run. Set to true + /// after the first sweep fires (under `policy: open`); thereafter + /// only the per-tick `queue_open_discovery_retries` continues. + startup_sweep_done: bool, + /// Per-peer UDP transports adopted from NAT traversal handoff. + bootstrap_transports: HashSet, + /// Originating peer npub (bech32) for each adopted bootstrap + /// transport, captured at `adopt_established_traversal` time. + /// Populated alongside `bootstrap_transports`; cleared in + /// `cleanup_bootstrap_transport_if_unused`. Used by the rx loop to + /// route fatal-protocol-mismatch observations back to the + /// Nostr-discovery `failure_state` for long cooldown application. + bootstrap_transport_npubs: HashMap, +} + +impl RendezvousDriver { + /// Borrow the engine handle if discovery is running. + pub fn engine(&self) -> Option<&NostrRendezvous> { + self.engine.as_deref() + } + + /// Clone the engine `Arc` handle if discovery is running. + pub fn engine_arc(&self) -> Option> { + self.engine.clone() + } + + /// Install the engine handle once discovery starts. + pub fn set_engine(&mut self, engine: Arc) { + self.engine = Some(engine); + } + + /// Take the engine handle for shutdown, clearing it. + pub fn take_engine(&mut self) -> Option> { + self.engine.take() + } + + /// Record the wall-clock ms at which discovery successfully started. + pub fn set_started_at_ms(&mut self, now_ms: u64) { + self.started_at_ms = Some(now_ms); + } + + /// Wall-clock ms when discovery started, if it has. + pub fn started_at_ms(&self) -> Option { + self.started_at_ms + } + + /// Whether the one-shot startup sweep has already run. + pub fn startup_sweep_done(&self) -> bool { + self.startup_sweep_done + } + + /// Latch the one-shot startup sweep as done. + pub fn set_startup_sweep_done(&mut self) { + self.startup_sweep_done = true; + } + + /// Whether `transport_id` is an adopted bootstrap transport. + pub fn is_bootstrap_transport(&self, transport_id: &TransportId) -> bool { + self.bootstrap_transports.contains(transport_id) + } + + /// Originating peer npub for an adopted bootstrap transport, if any. + pub fn bootstrap_transport_npub(&self, transport_id: &TransportId) -> Option<&String> { + self.bootstrap_transport_npubs.get(transport_id) + } + + /// Register an adopted bootstrap transport and its originating npub. + pub fn insert_bootstrap_transport(&mut self, transport_id: TransportId, npub: String) { + self.bootstrap_transports.insert(transport_id); + self.bootstrap_transport_npubs.insert(transport_id, npub); + } + + /// Drop an adopted bootstrap transport from both bookkeeping maps. + pub fn remove_bootstrap_transport(&mut self, transport_id: &TransportId) { + self.bootstrap_transports.remove(transport_id); + self.bootstrap_transport_npubs.remove(transport_id); + } + + /// Convert an advertised overlay endpoint into a `PeerAddress` candidate. + /// Pure mapping; `seen_at_ms` is supplied by the caller. + pub fn overlay_endpoint_to_peer_address( + endpoint: &OverlayEndpointAdvert, + priority: u8, + seen_at_ms: u64, + ) -> Option { + let transport = match endpoint.transport { + OverlayTransportKind::Udp => "udp", + OverlayTransportKind::Tcp => "tcp", + OverlayTransportKind::Tor => "tor", + }; + Some( + PeerAddress::with_priority(transport, endpoint.addr.clone(), priority) + .with_seen_at_ms(seen_at_ms), + ) + } + + /// Kick off a Nostr-mediated UDP NAT-traversal attempt for `peer_config`. + /// Returns whether an attempt was started (false if discovery is down). + pub async fn request_nostr_bootstrap(&self, peer_config: &PeerConfig) -> bool { + let Some(bootstrap) = self.engine_arc() else { + debug!(npub = %peer_config.npub, "No Nostr overlay runtime for udp:nat address"); + return false; + }; + bootstrap.request_connect(peer_config.clone()).await; + info!(npub = %peer_config.npub, "Started Nostr UDP NAT traversal attempt"); + true + } + + /// Resolve additional overlay `PeerAddress` candidates for a `via_nostr` + /// configured peer by fetching its published advert endpoints. `existing` + /// is the already-known static address list (used for priority and dedup); + /// `now_ms` stamps the returned candidates' `seen_at`. + pub async fn nostr_peer_fallback_addresses( + &self, + peer_config: &PeerConfig, + existing: &[PeerAddress], + nostr_cfg: &NostrRendezvousConfig, + now_ms: u64, + ) -> Vec { + if !nostr_cfg.enabled + || !peer_config.via_nostr + || nostr_cfg.policy == NostrRendezvousPolicy::Disabled + { + return Vec::new(); + } + + let Some(bootstrap) = self.engine_arc() else { + return Vec::new(); + }; + let endpoints = match bootstrap.advert_endpoints_for_peer(&peer_config.npub).await { + Ok(endpoints) => endpoints, + Err(err) => { + debug!( + npub = %peer_config.npub, + error = %err, + "Failed to resolve Nostr advert endpoints for configured peer" + ); + return Vec::new(); + } + }; + + let mut fallback = Vec::new(); + let mut next_priority = existing + .iter() + .map(|addr| addr.priority) + .max() + .unwrap_or(100) + .saturating_add(1); + let seen_at_ms = now_ms; + for endpoint in endpoints { + let Some(candidate) = + Self::overlay_endpoint_to_peer_address(&endpoint, next_priority, seen_at_ms) + else { + continue; + }; + if existing + .iter() + .any(|addr| addr.transport == candidate.transport && addr.addr == candidate.addr) + || fallback.iter().any(|addr: &PeerAddress| { + addr.transport == candidate.transport && addr.addr == candidate.addr + }) + { + continue; + } + fallback.push(candidate); + next_priority = next_priority.saturating_add(1); + } + fallback + } + + /// Publish (or withdraw) the local overlay advert built from `snapshot`. + /// `bootstrap` is passed explicitly because the startup path refreshes the + /// advert before the engine handle is installed on the driver. + pub async fn refresh_overlay_advert( + &self, + bootstrap: &Arc, + snapshot: Vec, + nostr_cfg: &NostrRendezvousConfig, + ) -> Result<(), BootstrapError> { + let advert = self + .build_overlay_advert(bootstrap, snapshot, nostr_cfg) + .await; + bootstrap.update_local_advert(advert).await + } + + /// Assemble the local `OverlayAdvert` from the per-transport `snapshot`. + /// The STUN `learn_public_udp_addr` await for wildcard-bound public UDP + /// sockets is reached through the `bootstrap` handle. + async fn build_overlay_advert( + &self, + bootstrap: &Arc, + snapshot: Vec, + nostr_cfg: &NostrRendezvousConfig, + ) -> Option { + if !nostr_cfg.enabled { + return None; + } + + let mut endpoints = Vec::new(); + let mut has_udp_nat = false; + + for entry in snapshot { + match entry { + AdvertTransportSnapshot::Udp { + advertise, + is_public, + external_addr, + local_addr, + transport_key, + } => { + if !advertise { + continue; + } + if is_public { + // Precedence: + // 1. operator-supplied `external_addr` (skips STUN) + // 2. non-wildcard `local_addr` (operator bound to + // a specific public IP directly) + // 3. STUN auto-discovery against ephemeral socket + // 4. loud warn + omit endpoint + if let Some(explicit) = external_addr { + endpoints.push(OverlayEndpointAdvert { + transport: OverlayTransportKind::Udp, + addr: explicit.to_string(), + }); + } else { + match local_addr { + Some(addr) if !addr.ip().is_unspecified() => { + endpoints.push(OverlayEndpointAdvert { + transport: OverlayTransportKind::Udp, + addr: addr.to_string(), + }); + } + Some(addr) => { + let key = transport_key; + let port = addr.port(); + if let Some(public) = + bootstrap.learn_public_udp_addr(key, port).await + { + endpoints.push(OverlayEndpointAdvert { + transport: OverlayTransportKind::Udp, + addr: public.to_string(), + }); + } else { + warn!( + transport_id = key, + bind_addr = %addr, + "advert: udp public=true bound to wildcard but \ + STUN observation failed; advertising no UDP \ + endpoint. Either set transports.udp.external_addr, \ + bind to a specific public IP, or ensure \ + node.rendezvous.nostr.stun_servers is reachable" + ); + } + } + None => {} + } + } + } else { + endpoints.push(OverlayEndpointAdvert { + transport: OverlayTransportKind::Udp, + addr: "nat".to_string(), + }); + has_udp_nat = true; + } + } + AdvertTransportSnapshot::Tcp { + advertise, + external_addr, + local_addr, + } => { + if !advertise { + continue; + } + // Precedence: + // 1. operator-supplied `external_addr` (only path that + // works on cloud-NAT setups where the public IP is + // not on a host interface). + // 2. non-wildcard `local_addr` (operator bound to a + // specific public IP directly). + // 3. loud warn + omit endpoint (no TCP STUN equivalent). + if let Some(explicit) = external_addr { + endpoints.push(OverlayEndpointAdvert { + transport: OverlayTransportKind::Tcp, + addr: explicit.to_string(), + }); + } else { + match local_addr { + Some(addr) if !addr.ip().is_unspecified() => { + endpoints.push(OverlayEndpointAdvert { + transport: OverlayTransportKind::Tcp, + addr: addr.to_string(), + }); + } + Some(addr) => { + warn!( + bind_addr = %addr, + "advert: tcp advertise_on_nostr=true bound to wildcard \ + and no transports.tcp.external_addr set; advertising no \ + TCP endpoint. Either set external_addr to the public \ + IP (recommended for cloud 1:1-NAT setups) or bind \ + explicitly to the public IP" + ); + } + None => {} + } + } + } + AdvertTransportSnapshot::Tor { + advertise, + onion_addr, + advertised_port, + } => { + if !advertise { + continue; + } + if let Some(addr) = onion_addr { + endpoints.push(OverlayEndpointAdvert { + transport: OverlayTransportKind::Tor, + addr: format!("{}:{}", addr, advertised_port), + }); + } + } + } + } + + if endpoints.is_empty() { + return None; + } + + Some(OverlayAdvert { + identifier: ADVERT_IDENTIFIER.to_string(), + version: ADVERT_VERSION, + endpoints, + signal_relays: has_udp_nat.then(|| nostr_cfg.dm_relays.clone()), + stun_servers: has_udp_nat.then(|| nostr_cfg.stun_servers.clone()), + }) + } +} diff --git a/src/nostr/mod.rs b/src/nostr/mod.rs index dcece0d..24ec3cf 100644 --- a/src/nostr/mod.rs +++ b/src/nostr/mod.rs @@ -1,3 +1,4 @@ +mod driver; mod failure_state; mod handoff; mod runtime; @@ -9,6 +10,7 @@ mod types; #[cfg(test)] mod tests; +pub use driver::{AdvertTransportSnapshot, RendezvousDriver}; pub use handoff::{BootstrapHandoffResult, EstablishedTraversal, is_punch_packet}; pub use runtime::NostrRendezvous; pub use types::{