diff --git a/CHANGELOG.md b/CHANGELOG.md index 09e1c4a..0da4c53 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 direct counters at the bloom / discovery / forwarding sites are retained alongside the new dispatch while the rollout is in progress; a later change collapses the duplicate increment. +- `Node::update_peers` for runtime peer-list refresh, returning an + `UpdatePeersOutcome` summarizing added, removed, and retained peers. + Re-derives active peer connections from a new peer configuration + without dropping links to peers that remain in the set. + `PeerAddress` gains a `seen_at_ms` recency field (with + `with_seen_at_ms`) used to prefer more recently observed addresses. +- Opt-in mDNS / DNS-SD LAN discovery for sub-second pairing of peers on + the same local link, without a relay or NAT-traversal roundtrip. + Disabled by default; operators enable it with + `node.discovery.lan.enabled: true`. Configurable service type and an + optional `node.discovery.lan.scope` that isolates discovery to peers + sharing the same private-network scope. The advertised UDP port is + chosen from a non-bootstrap operational UDP transport using a stable + selector, so it is deterministic across restarts. - `pool_inbound` and `pool_outbound` counters on the TCP and Tor transport stats (`TcpStats`, `TorStats`). Per-direction accounting is updated at every pool-insert and receive-loop-exit site, plus on @@ -41,6 +55,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- `complete_rekey_msg2` now returns the remote peer's startup epoch + alongside the new Noise session, so the rekey path can detect a peer + restart and clear stale session state. +- Active-peer path selection now sorts address candidates by recency + (`seen_at_ms`), preferring the most recently observed address when + racing concurrent path probes. +- Per-tick work budgets bound the connection churn done in a single + node tick: `MAX_DISCOVERY_CONNECTS_PER_TICK`, + `MAX_RETRY_CONNECTIONS_PER_TICK`, and + `MAX_PARALLEL_PATH_CANDIDATES_PER_PEER`. Work beyond a tick's budget + is deferred to the next tick rather than discarded. - 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 @@ -154,6 +179,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- A stale FSP (session-layer) session is now cleared when a peer + restart is detected during FMP rekey or cross-connection promotion. + Previously the old session could linger after the peer came back + with a new startup epoch, leaving the session-layer map out of sync + with the freshly promoted peer. - TCP and Tor `max_inbound_connections` admission cap is now compared against the per-direction inbound count (`pool_inbound`) rather than the combined pool size. Outbound connect-on-send connections share diff --git a/Cargo.lock b/Cargo.lock index 6473808..2524ecc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1083,6 +1083,7 @@ dependencies = [ "hex", "hkdf", "libc", + "mdns-sd", "nostr", "nostr-sdk", "portable-atomic", @@ -1116,6 +1117,17 @@ version = "0.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ce7134b9999ecaf8bcd65542e436736ef32ddca1b3e06094cb6ec5755203b80" +[[package]] +name = "flume" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" +dependencies = [ + "futures-core", + "futures-sink", + "spin", +] + [[package]] name = "fnv" version = "1.0.7" @@ -1515,6 +1527,16 @@ dependencies = [ "icu_properties", ] +[[package]] +name = "if-addrs" +version = "0.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0a05c691e1fae256cf7013d99dad472dc52d5543322761f83ec8d47eab40d2b" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "indexmap" version = "2.14.0" @@ -1774,6 +1796,21 @@ dependencies = [ "regex-automata", ] +[[package]] +name = "mdns-sd" +version = "0.19.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2bb8ce26633738d98ffcef71ec58bff967c6675be50229823c2835f6316e67e" +dependencies = [ + "fastrand", + "flume", + "if-addrs", + "log", + "mio", + "socket-pktinfo", + "socket2", +] + [[package]] name = "memchr" version = "2.8.0" @@ -3018,6 +3055,17 @@ version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" +[[package]] +name = "socket-pktinfo" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "927136cc2ae6a1b0e66ac6b1210902b75c3f726db004a73bc18686dcd0dcd22f" +dependencies = [ + "libc", + "socket2", + "windows-sys 0.60.2", +] + [[package]] name = "socket2" version = "0.6.3" @@ -3028,6 +3076,15 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "spin" +version = "0.9.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" +dependencies = [ + "lock_api", +] + [[package]] name = "stable_deref_trait" version = "1.2.1" @@ -3924,7 +3981,7 @@ version = "0.52.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" dependencies = [ - "windows-targets", + "windows-targets 0.52.6", ] [[package]] @@ -3933,7 +3990,16 @@ version = "0.59.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" dependencies = [ - "windows-targets", + "windows-targets 0.52.6", +] + +[[package]] +name = "windows-sys" +version = "0.60.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb" +dependencies = [ + "windows-targets 0.53.5", ] [[package]] @@ -3951,14 +4017,31 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" dependencies = [ - "windows_aarch64_gnullvm", - "windows_aarch64_msvc", - "windows_i686_gnu", - "windows_i686_gnullvm", - "windows_i686_msvc", - "windows_x86_64_gnu", - "windows_x86_64_gnullvm", - "windows_x86_64_msvc", + "windows_aarch64_gnullvm 0.52.6", + "windows_aarch64_msvc 0.52.6", + "windows_i686_gnu 0.52.6", + "windows_i686_gnullvm 0.52.6", + "windows_i686_msvc 0.52.6", + "windows_x86_64_gnu 0.52.6", + "windows_x86_64_gnullvm 0.52.6", + "windows_x86_64_msvc 0.52.6", +] + +[[package]] +name = "windows-targets" +version = "0.53.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3" +dependencies = [ + "windows-link", + "windows_aarch64_gnullvm 0.53.1", + "windows_aarch64_msvc 0.53.1", + "windows_i686_gnu 0.53.1", + "windows_i686_gnullvm 0.53.1", + "windows_i686_msvc 0.53.1", + "windows_x86_64_gnu 0.53.1", + "windows_x86_64_gnullvm 0.53.1", + "windows_x86_64_msvc 0.53.1", ] [[package]] @@ -3967,48 +4050,96 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53" + [[package]] name = "windows_aarch64_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" +[[package]] +name = "windows_aarch64_msvc" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006" + [[package]] name = "windows_i686_gnu" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" +[[package]] +name = "windows_i686_gnu" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "960e6da069d81e09becb0ca57a65220ddff016ff2d6af6a223cf372a506593a3" + [[package]] name = "windows_i686_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" +[[package]] +name = "windows_i686_gnullvm" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c" + [[package]] name = "windows_i686_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" +[[package]] +name = "windows_i686_msvc" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2" + [[package]] name = "windows_x86_64_gnu" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" +[[package]] +name = "windows_x86_64_gnu" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499" + [[package]] name = "windows_x86_64_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1" + [[package]] name = "windows_x86_64_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" +[[package]] +name = "windows_x86_64_msvc" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650" + [[package]] name = "winreg" version = "0.55.0" diff --git a/Cargo.toml b/Cargo.toml index 9faad4d..d533079 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,6 +29,7 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] } tokio = { version = "1", features = ["rt", "macros", "signal", "sync", "net", "time", "process", "io-util"] } futures = "0.3" simple-dns = "0.11.2" +mdns-sd = "0.19" socket2 = { version = "0.6.2", features = ["all"] } tokio-socks = "0.5" portable-atomic = { version = "1", features = ["std"] } diff --git a/src/config/node.rs b/src/config/node.rs index 0f6421f..984ac71 100644 --- a/src/config/node.rs +++ b/src/config/node.rs @@ -219,6 +219,12 @@ pub struct DiscoveryConfig { /// Nostr-mediated overlay endpoint discovery. #[serde(default = "DiscoveryConfig::default_nostr")] pub nostr: NostrDiscoveryConfig, + /// mDNS / DNS-SD peer discovery on the local link. Identity surface + /// is a strict subset of what `nostr.advertise` already publishes + /// publicly, so there's no marginal privacy cost; the latency win + /// for same-LAN peers is large (sub-second pairing, no relay). + #[serde(default = "DiscoveryConfig::default_lan")] + pub lan: crate::discovery::lan::LanDiscoveryConfig, } impl Default for DiscoveryConfig { @@ -231,6 +237,7 @@ impl Default for DiscoveryConfig { backoff_max_secs: 0, forward_min_interval_secs: 2, nostr: NostrDiscoveryConfig::default(), + lan: crate::discovery::lan::LanDiscoveryConfig::default(), } } } @@ -257,6 +264,9 @@ impl DiscoveryConfig { fn default_nostr() -> NostrDiscoveryConfig { NostrDiscoveryConfig::default() } + fn default_lan() -> crate::discovery::lan::LanDiscoveryConfig { + crate::discovery::lan::LanDiscoveryConfig::default() + } } /// Nostr advert discovery policy. diff --git a/src/config/peer.rs b/src/config/peer.rs index 51a0903..3247737 100644 --- a/src/config/peer.rs +++ b/src/config/peer.rs @@ -45,8 +45,28 @@ pub struct PeerAddress { /// are tried first. #[serde(default = "default_priority")] pub priority: u8, + + /// Wall-clock observation timestamp (Unix ms) for ranking by recency. + /// + /// `None` means "no freshness signal", typically an operator-edited + /// static config. The dialer sorts candidates by this field descending + /// so freshly observed overlay or runtime hints can be tried before stale + /// static addresses. This field is runtime-only and is ignored when + /// comparing peer-address lists for config changes. + #[serde(default, skip_serializing_if = "Option::is_none", skip_deserializing)] + pub seen_at_ms: Option, } +impl PartialEq for PeerAddress { + fn eq(&self, other: &Self) -> bool { + self.transport == other.transport + && self.addr == other.addr + && self.priority == other.priority + } +} + +impl Eq for PeerAddress {} + fn default_priority() -> u8 { 100 } @@ -62,6 +82,7 @@ impl PeerAddress { transport: transport.into(), addr: addr.into(), priority: default_priority(), + seen_at_ms: None, } } @@ -75,8 +96,16 @@ impl PeerAddress { transport: transport.into(), addr: addr.into(), priority, + seen_at_ms: None, } } + + /// Tag this address with a freshness timestamp for source-agnostic + /// candidate ranking. + pub fn with_seen_at_ms(mut self, seen_at_ms: u64) -> Self { + self.seen_at_ms = Some(seen_at_ms); + self + } } /// Configuration for a known peer. diff --git a/src/discovery.rs b/src/discovery.rs index 2d86d22..3612fb5 100644 --- a/src/discovery.rs +++ b/src/discovery.rs @@ -6,6 +6,7 @@ //! it hands the live socket and selected remote endpoint to FIPS so the //! existing Noise/FMP transport path can take over. +pub mod lan; pub mod nostr; use crate::config::UdpConfig; diff --git a/src/discovery/lan/mod.rs b/src/discovery/lan/mod.rs new file mode 100644 index 0000000..f5e3b2d --- /dev/null +++ b/src/discovery/lan/mod.rs @@ -0,0 +1,368 @@ +//! LAN peer discovery via mDNS / DNS-SD (RFC 6762 / RFC 6763). +//! +//! Publishes a `_fips._udp.local.` service advert carrying our `npub` and +//! optional discovery scope on the local link, and concurrently browses for the +//! same service type to learn peers reachable on the same broadcast +//! domain. The result is sub-second peer pairing without any Nostr-relay +//! roundtrip, STUN observation, or NAT traversal — the observed +//! endpoint is by construction routable from the consumer's LAN. +//! +//! ## Trust model +//! +//! mDNS adverts are unauthenticated: anyone on the LAN can multicast a +//! TXT carrying `npub=...`. Identity is still proven end-to-end by the +//! Noise XX handshake the Node initiates against the observed endpoint +//! — a spoofed advert with another peer's npub fails the handshake and +//! is silently dropped. Treat the mDNS advert as a routing hint, not as +//! identity. LAN discovery is link-local mDNS only. It is not a Nostr advert +//! and does not leave the broadcast domain unless the operator's LAN bridges +//! mDNS. +//! +//! ## Scope filtering +//! +//! When a `discovery_scope` is configured, the advert carries it in a +//! `scope=` TXT entry and the browser only surfaces peers with a +//! matching scope. Nodes on the same physical LAN but configured for +//! different mesh networks don't cross-feed each other. + +use std::collections::HashMap; +use std::net::{SocketAddr, SocketAddrV4, SocketAddrV6}; +use std::sync::Arc; +use std::time::Instant; + +use mdns_sd::{ScopedIp, ServiceDaemon, ServiceEvent, ServiceInfo}; +use thiserror::Error; +use tokio::sync::Mutex; +use tracing::{debug, info, warn}; + +use crate::Identity; + +/// DNS-SD service type for the FIPS LAN advert. RFC 6763 §4.1.2: must +/// end with `.local.`. The `_udp` is the IP transport, not the upper +/// protocol — both UDP and TCP FIPS endpoints announce under the same +/// service type because the link-layer punch/handshake travels over UDP +/// either way. +pub const SERVICE_TYPE: &str = "_fips._udp.local."; + +/// TXT key carrying the bech32-encoded npub of the publishing node. +pub const TXT_KEY_NPUB: &str = "npub"; + +/// TXT key carrying the publishing node's `discovery_scope`, if any. +pub const TXT_KEY_SCOPE: &str = "scope"; + +/// TXT key carrying the FIPS protocol version (matches the Nostr advert +/// `PROTOCOL_VERSION`). +pub const TXT_KEY_VERSION: &str = "v"; + +#[derive(Debug, Error)] +pub enum LanDiscoveryError { + #[error("mDNS daemon init failed: {0}")] + Daemon(String), + #[error("mDNS register failed: {0}")] + Register(String), + #[error("mDNS browse failed: {0}")] + Browse(String), + #[error("no advertised UDP port — start a UDP transport first")] + NoAdvertisedPort, + #[error("LAN discovery disabled in config")] + Disabled, +} + +/// A peer we learned about via mDNS. Identity is unverified at this +/// point; the Node initiates a Noise XX handshake against `addr` to +/// confirm `npub` actually controls the matching private key. +#[derive(Debug, Clone)] +pub struct LanDiscoveredPeer { + pub npub: String, + pub scope: Option, + pub addr: SocketAddr, + pub observed_at: Instant, +} + +/// Browser-side events surfaced by `LanDiscovery::drain_events`. +#[derive(Debug, Clone)] +pub enum LanEvent { + Discovered(LanDiscoveredPeer), +} + +/// Runtime configuration for the mDNS responder + browser. +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct LanDiscoveryConfig { + /// Master switch. Default: `false` — LAN discovery is opt-in. Operators + /// who want sub-second same-LAN pairing enable it via + /// `node.discovery.lan.enabled: true`. Default-off avoids reintroducing + /// a per-LAN identity broadcast on nodes that have deliberately disabled + /// other discovery channels, and avoids any multicast surprise on upgrade. + #[serde(default = "LanDiscoveryConfig::default_enabled")] + pub enabled: bool, + /// Overridable service type, primarily so integration tests can run + /// multiple isolated services on the same loopback interface. + #[serde(default = "LanDiscoveryConfig::default_service_type")] + pub service_type: String, + /// Optional application/network scope carried in the LAN-only TXT + /// record. Browsers that set a scope ignore adverts for other scopes. + /// + /// This is intentionally separate from Nostr discovery's public `app` + /// tag so applications can keep relay-visible adverts generic while + /// still isolating LAN discovery per private network. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub scope: Option, +} + +impl Default for LanDiscoveryConfig { + fn default() -> Self { + Self { + enabled: Self::default_enabled(), + service_type: Self::default_service_type(), + scope: None, + } + } +} + +impl LanDiscoveryConfig { + fn default_enabled() -> bool { + false + } + fn default_service_type() -> String { + SERVICE_TYPE.to_string() + } +} + +/// Running mDNS responder + browser bound to the node's UDP advert port. +pub struct LanDiscovery { + daemon: ServiceDaemon, + own_npub: String, + instance_fullname: String, + events_rx: Mutex>, + event_pump: tokio::task::JoinHandle<()>, +} + +impl LanDiscovery { + /// Start the mDNS responder and browser. + /// + /// `advertised_port` is the UDP port the operational UDP transport + /// is bound to — peers receiving our advert will initiate Noise XX + /// against that port. `scope` mirrors the Nostr discovery scope and + /// is used to filter the browser stream. + pub async fn start( + identity: &Identity, + scope: Option, + advertised_port: u16, + config: LanDiscoveryConfig, + ) -> Result, LanDiscoveryError> { + if !config.enabled { + return Err(LanDiscoveryError::Disabled); + } + if advertised_port == 0 { + return Err(LanDiscoveryError::NoAdvertisedPort); + } + + let daemon = ServiceDaemon::new().map_err(|e| LanDiscoveryError::Daemon(e.to_string()))?; + + let npub = identity.npub(); + // mDNS DNS labels are capped at 63 bytes. 16 bech32 chars of npub + // give 80 bits of effective entropy — collisions on a single LAN + // are vanishingly unlikely. Prefixed for human-readable logs. + let label_npub = &npub[..16.min(npub.len())]; + let instance_name = format!("fips-{label_npub}"); + let host_name = format!("{instance_name}.local."); + + let mut props: HashMap = HashMap::new(); + props.insert(TXT_KEY_NPUB.to_string(), npub.clone()); + if let Some(s) = scope.as_deref() + && !s.is_empty() + { + props.insert(TXT_KEY_SCOPE.to_string(), s.to_string()); + } + props.insert( + TXT_KEY_VERSION.to_string(), + super::nostr::PROTOCOL_VERSION.to_string(), + ); + + // host_ipv4 is set to "127.0.0.1" *and* enable_addr_auto() is + // called: the loopback seed makes the advert resolve for + // same-host peers (and same-host integration tests) while the + // auto-flag still appends every non-loopback interface address + // mdns-sd discovers. Belt-and-braces because addr_auto alone + // skips loopback by default on some platforms. + let service_info = ServiceInfo::new( + &config.service_type, + &instance_name, + &host_name, + "127.0.0.1", + advertised_port, + Some(props), + ) + .map_err(|e| LanDiscoveryError::Register(e.to_string()))? + .enable_addr_auto(); + + let instance_fullname = service_info.get_fullname().to_string(); + + daemon + .register(service_info) + .map_err(|e| LanDiscoveryError::Register(e.to_string()))?; + + let browse_rx = daemon + .browse(&config.service_type) + .map_err(|e| LanDiscoveryError::Browse(e.to_string()))?; + + let (events_tx, events_rx) = tokio::sync::mpsc::unbounded_channel(); + let own_npub = npub.clone(); + let scope_filter = scope.clone().filter(|s| !s.is_empty()); + let event_pump = tokio::spawn(async move { + // mdns-sd browse returns a flume::Receiver; pump until the + // daemon shuts down and the channel closes. + loop { + let event = match browse_rx.recv_async().await { + Ok(e) => e, + Err(_) => break, + }; + match event { + ServiceEvent::ServiceResolved(info) => { + let mut peer_npub: Option = None; + let mut peer_scope: Option = None; + for prop in info.get_properties().iter() { + match prop.key() { + TXT_KEY_NPUB => { + peer_npub = Some(prop.val_str().to_string()); + } + TXT_KEY_SCOPE => { + peer_scope = Some(prop.val_str().to_string()); + } + _ => {} + } + } + let Some(peer_npub) = peer_npub else { + debug!( + instance = info.get_fullname(), + "lan: skip advert without npub TXT" + ); + continue; + }; + if peer_npub == own_npub { + // Our own advert echoed back on a loopback + // or multi-homed interface. + continue; + } + if scope_filter.is_some() && scope_filter != peer_scope { + debug!( + npub = %short(&peer_npub), + their_scope = ?peer_scope, + our_scope = ?scope_filter, + "lan: skip cross-scope advert" + ); + continue; + } + let port = info.get_port(); + if port == 0 { + continue; + } + let observed_at = Instant::now(); + // mdns-sd may report multiple interface IPs for + // a multi-homed responder. Surface all routable + // candidates — the Node side filters/dedups and + // only dials addresses compatible with an active + // UDP socket family. IPv6 link-local addresses + // require an interface scope; preserve it when + // mdns-sd provides one, and skip unusable + // scope-less link-local records. + for scoped in info.get_addresses() { + let Some(addr) = socket_addr_from_scoped_ip(scoped, port) else { + debug!( + npub = %short(&peer_npub), + addr = %scoped.to_ip_addr(), + "lan: skip scope-less IPv6 link-local advert" + ); + continue; + }; + if events_tx + .send(LanEvent::Discovered(LanDiscoveredPeer { + npub: peer_npub.clone(), + scope: peer_scope.clone(), + addr, + observed_at, + })) + .is_err() + { + return; + } + } + } + ServiceEvent::ServiceRemoved(_, fullname) => { + debug!(fullname = %fullname, "lan: service removed"); + } + other => { + debug!(?other, "lan: mDNS event"); + } + } + } + }); + + info!( + instance = %instance_fullname, + port = advertised_port, + scope = ?scope, + "lan: mDNS discovery started" + ); + Ok(Arc::new(Self { + daemon, + own_npub: npub, + instance_fullname, + events_rx: Mutex::new(events_rx), + event_pump, + })) + } + + /// Bech32 npub published by this node. + pub fn own_npub(&self) -> &str { + &self.own_npub + } + + /// Drain pending browser events. Called once per Node tick. + pub async fn drain_events(&self) -> Vec { + let mut rx = self.events_rx.lock().await; + let mut events = Vec::new(); + while let Ok(event) = rx.try_recv() { + events.push(event); + } + events + } + + /// Tear down the responder, browser, and event pump. + pub async fn shutdown(self: &Arc) { + if let Err(e) = self.daemon.unregister(&self.instance_fullname) { + warn!(error = %e, "lan: unregister failed"); + } + if let Err(e) = self.daemon.shutdown() { + warn!(error = %e, "lan: daemon shutdown failed"); + } + self.event_pump.abort(); + } +} + +fn short(npub: &str) -> &str { + let end = 16.min(npub.len()); + &npub[..end] +} + +fn socket_addr_from_scoped_ip(scoped: &ScopedIp, port: u16) -> Option { + match scoped { + ScopedIp::V4(v4) => Some(SocketAddr::V4(SocketAddrV4::new(*v4.addr(), port))), + ScopedIp::V6(v6) => { + let ip = *v6.addr(); + let scope_id = v6.scope_id().index; + if ipv6_is_unicast_link_local(ip) && scope_id == 0 { + return None; + } + Some(SocketAddr::V6(SocketAddrV6::new(ip, port, 0, scope_id))) + } + _ => None, + } +} + +fn ipv6_is_unicast_link_local(ip: std::net::Ipv6Addr) -> bool { + (ip.segments()[0] & 0xffc0) == 0xfe80 +} + +#[cfg(test)] +mod tests; diff --git a/src/discovery/lan/tests.rs b/src/discovery/lan/tests.rs new file mode 100644 index 0000000..1e4867c --- /dev/null +++ b/src/discovery/lan/tests.rs @@ -0,0 +1,213 @@ +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; +use std::time::Duration; + +use crate::Identity; +use mdns_sd::ScopedIp; + +use super::{LanDiscovery, LanDiscoveryConfig, LanEvent}; + +/// Distinct service type per test run so concurrent cargo-test workers +/// on the same machine don't cross-feed each other's adverts via the +/// shared 224.0.0.251 multicast group. The trailing `.local.` is +/// required by RFC 6763 — mdns-sd will reject anything else. +fn isolated_service_type(tag: &str) -> String { + let rand: u32 = rand::random(); + format!("_fipstest-{tag}-{rand:08x}._udp.local.") +} + +fn config_for(service_type: String) -> LanDiscoveryConfig { + LanDiscoveryConfig { + enabled: true, + service_type, + scope: None, + } +} + +#[test] +fn scoped_ipv4_advert_becomes_socket_addr() { + let scoped = ScopedIp::from(IpAddr::V4(Ipv4Addr::new(192, 168, 178, 91))); + let addr = super::socket_addr_from_scoped_ip(&scoped, 51820); + + assert_eq!(addr, Some(SocketAddr::from(([192, 168, 178, 91], 51820)))); +} + +#[test] +fn scope_less_ipv6_link_local_advert_is_skipped() { + let scoped = ScopedIp::from(IpAddr::V6("fe80::32c5:99ff:fea7:5fe9".parse().unwrap())); + + assert!(super::socket_addr_from_scoped_ip(&scoped, 51820).is_none()); +} + +#[test] +fn non_link_local_ipv6_advert_is_preserved() { + let scoped = ScopedIp::from(IpAddr::V6(Ipv6Addr::LOCALHOST)); + let addr = super::socket_addr_from_scoped_ip(&scoped, 51820); + + assert_eq!(addr, Some("[::1]:51820".parse().unwrap())); +} + +async fn wait_for_peer( + discovery: &LanDiscovery, + expected_npub: &str, + timeout: Duration, +) -> Option { + let deadline = tokio::time::Instant::now() + timeout; + while tokio::time::Instant::now() < deadline { + for event in discovery.drain_events().await { + let LanEvent::Discovered(peer) = event; + if peer.npub == expected_npub { + return Some(peer); + } + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + None +} + +/// Two LanDiscovery instances on isolated service types — `a` browses +/// only its own type and never sees `b`, and vice versa. Sanity check +/// that the scope-isolation defense works (we'd lose isolation if mdns- +/// sd ever leaked across service types). +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn isolated_service_types_do_not_cross_feed() { + let identity_a = Identity::generate(); + let identity_b = Identity::generate(); + + let service_a = isolated_service_type("isolated-a"); + let service_b = isolated_service_type("isolated-b"); + + let lan_a = LanDiscovery::start( + &identity_a, + Some("scope-x".to_string()), + 61001, + config_for(service_a.clone()), + ) + .await + .expect("start a"); + let lan_b = LanDiscovery::start( + &identity_b, + Some("scope-x".to_string()), + 61002, + config_for(service_b.clone()), + ) + .await + .expect("start b"); + + // Give mDNS multicast time to settle, then confirm neither side saw + // the other (different service type isolates them). + tokio::time::sleep(Duration::from_secs(2)).await; + let saw_b_from_a = wait_for_peer( + &lan_a, + identity_b.npub().as_str(), + Duration::from_millis(500), + ) + .await + .is_some(); + let saw_a_from_b = wait_for_peer( + &lan_b, + identity_a.npub().as_str(), + Duration::from_millis(500), + ) + .await + .is_some(); + + lan_a.shutdown().await; + lan_b.shutdown().await; + + assert!(!saw_b_from_a, "isolated service types must not cross-feed"); + assert!(!saw_a_from_b, "isolated service types must not cross-feed"); +} + +/// Two LanDiscovery instances on the same service type and the same +/// scope: each should observe the other's advert within a few seconds. +/// Exercises the responder + browser + TXT plumbing end-to-end. +/// +/// Ignored by default: relies on multicast-loopback semantics that +/// vary across macOS/Linux/Windows when two `ServiceDaemon` instances +/// run in the same process. Real cross-host LAN deployment exercises +/// the same code path correctly — verify with `cargo test -- --ignored +/// matched_scope_peers_observe_each_other` on a setup where this +/// matters, or via end-to-end integration with two daemons. +#[ignore] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn matched_scope_peers_observe_each_other() { + let identity_a = Identity::generate(); + let identity_b = Identity::generate(); + + let service = isolated_service_type("matched"); + + let lan_a = LanDiscovery::start( + &identity_a, + Some("scope-shared".to_string()), + 61101, + config_for(service.clone()), + ) + .await + .expect("start a"); + let lan_b = LanDiscovery::start( + &identity_b, + Some("scope-shared".to_string()), + 61102, + config_for(service.clone()), + ) + .await + .expect("start b"); + + // Loopback mDNS resolution on macOS/Linux takes a moment. + let observed_b = + wait_for_peer(&lan_a, identity_b.npub().as_str(), Duration::from_secs(10)).await; + let observed_a = + wait_for_peer(&lan_b, identity_a.npub().as_str(), Duration::from_secs(10)).await; + + lan_a.shutdown().await; + lan_b.shutdown().await; + + let observed_b = observed_b.expect("a must see b"); + let observed_a = observed_a.expect("b must see a"); + + assert_eq!(observed_b.scope.as_deref(), Some("scope-shared")); + assert_eq!(observed_a.scope.as_deref(), Some("scope-shared")); + assert_eq!(observed_b.addr.port(), 61102); + assert_eq!(observed_a.addr.port(), 61101); +} + +/// Different scopes on the same service type must be filtered out by +/// the browser — peer in scope X does not surface to a browser in +/// scope Y, even if both adverts arrive on the same multicast group. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn cross_scope_advert_is_filtered() { + let identity_a = Identity::generate(); + let identity_b = Identity::generate(); + + let service = isolated_service_type("cross-scope"); + + let lan_a = LanDiscovery::start( + &identity_a, + Some("scope-a".to_string()), + 61201, + config_for(service.clone()), + ) + .await + .expect("start a"); + let lan_b = LanDiscovery::start( + &identity_b, + Some("scope-b".to_string()), + 61202, + config_for(service.clone()), + ) + .await + .expect("start b"); + + tokio::time::sleep(Duration::from_secs(3)).await; + let saw_b = wait_for_peer( + &lan_a, + identity_b.npub().as_str(), + Duration::from_millis(500), + ) + .await; + + lan_a.shutdown().await; + lan_b.shutdown().await; + + assert!(saw_b.is_none(), "cross-scope advert must be filtered"); +} diff --git a/src/lib.rs b/src/lib.rs index d3906e5..54252f3 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -67,4 +67,4 @@ pub use peer::{ }; // Re-export node types -pub use node::{Node, NodeError, NodeState}; +pub use node::{Node, NodeError, NodeState, UpdatePeersOutcome}; diff --git a/src/node/handlers/discovery.rs b/src/node/handlers/discovery.rs index 3dd227c..d53d3e0 100644 --- a/src/node/handlers/discovery.rs +++ b/src/node/handlers/discovery.rs @@ -12,6 +12,8 @@ use crate::transport::{TransportAddr, TransportId}; use crate::{NodeAddr, PeerIdentity}; use tracing::{debug, info, trace, warn}; +const MAX_RECENT_DISCOVERY_REQUESTS: usize = 4096; + impl Node { /// Handle an incoming LookupRequest from a peer. /// @@ -37,6 +39,7 @@ impl Node { }; let now_ms = Self::now_ms(); + self.purge_expired_requests(now_ms); // Dedup: drop if we've already seen this request_id. // Also serves as loop protection — tree routing is loop-free, @@ -53,13 +56,21 @@ impl Node { return; } + if self.recent_requests.len() >= MAX_RECENT_DISCOVERY_REQUESTS { + debug!( + request_id = request.request_id, + from = %self.peer_display_name(from), + recent_requests = self.recent_requests.len(), + max_recent_requests = MAX_RECENT_DISCOVERY_REQUESTS, + "Discovery request dedup cache full, dropping LookupRequest" + ); + return; + } + // Record for reverse-path forwarding and dedup self.recent_requests .insert(request.request_id, RecentRequest::new(*from, now_ms)); - // Lazy purge expired entries - self.purge_expired_requests(now_ms); - // Are we the target? if request.target == *self.node_addr() { self.stats_mut().discovery.req_target_is_us += 1; diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 91e8f4a..91f0603 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -698,8 +698,15 @@ impl Node { // Complete the rekey handshake on the ActivePeer if let Some(peer) = self.peers.get_mut(&peer_node_addr) { match peer.complete_rekey_msg2(noise_msg2) { - Ok(session) => { + Ok((session, remote_epoch)) => { let our_index = peer.rekey_our_index().unwrap_or(header.receiver_idx); + let remote_epoch_changed = matches!( + (peer.remote_epoch(), remote_epoch), + (Some(old), Some(new)) if old != new + ); + if remote_epoch.is_some() { + peer.set_remote_epoch(remote_epoch); + } peer.set_pending_session(session, our_index, header.sender_idx); if let Some(transport_id) = peer.transport_id() { @@ -707,6 +714,19 @@ impl Node { .insert((transport_id, our_index.as_u32()), peer_node_addr); } + if remote_epoch_changed { + if self.sessions.remove(&peer_node_addr).is_some() { + debug!( + peer = %display_name, + "Cleared stale FSP session after peer restart during FMP rekey" + ); + } + info!( + peer = %display_name, + "Peer restart detected during FMP rekey, replacing stale endpoint session" + ); + } + debug!( peer = %display_name, new_our_index = %our_index, @@ -1083,9 +1103,15 @@ impl Node { if let Some(existing_peer) = self.peers.get(&peer_node_addr) { let existing_link_id = existing_peer.link_id(); - // Determine which connection wins - let this_wins = - cross_connection_winner(self.identity.node_addr(), &peer_node_addr, is_outbound); + let remote_epoch_changed = matches!((existing_peer.remote_epoch(), remote_epoch), (Some(old), Some(new)) if old != new); + + // Determine which connection wins. A peer restart (different + // startup epoch) is not a normal cross-connection: the old link + // and FSP sessions are cryptographically stale, so the freshly + // authenticated connection must replace them regardless of the + // tie-breaker direction. + let this_wins = remote_epoch_changed + || cross_connection_winner(self.identity.node_addr(), &peer_node_addr, is_outbound); if this_wins { // This connection wins, replace the existing peer @@ -1111,6 +1137,21 @@ impl Node { let _ = self.index_allocator.free(old_idx); } + if remote_epoch_changed { + if self.sessions.remove(&peer_node_addr).is_some() { + debug!( + peer = %self.peer_display_name(&peer_node_addr), + "Cleared stale FSP session after peer restart during promotion" + ); + } + info!( + peer = %self.peer_display_name(&peer_node_addr), + winner_link = %link_id, + loser_link = %loser_link_id, + "Peer restart detected during promotion, replacing stale active peer" + ); + } + self.seed_path_mtu_for_link_peer(&peer_node_addr, transport_id, ¤t_addr); let mut new_peer = ActivePeer::with_session( diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index 113c189..941c687 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -260,6 +260,7 @@ impl Node { self.reload_host_map().await; self.poll_pending_connects().await; self.poll_nostr_discovery().await; + self.poll_lan_discovery().await; self.resend_pending_handshakes(now_ms).await; self.resend_pending_rekeys(now_ms).await; self.resend_pending_session_handshakes(now_ms).await; diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index 103b87e..771cb32 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -14,14 +14,185 @@ use crate::protocol::{Disconnect, DisconnectReason}; use crate::transport::{Link, LinkDirection, LinkId, TransportAddr, TransportId, packet_channel}; use crate::upper::tun::{TunDevice, TunState, run_tun_reader, shutdown_tun_interface}; use crate::{NodeAddr, PeerIdentity}; -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; +use std::net::SocketAddr; use std::thread; use std::time::Duration; use tracing::{debug, info, warn}; const OPEN_DISCOVERY_RETRY_LIFETIME_MULTIPLIER: u64 = 2; +const MAX_PARALLEL_PATH_CANDIDATES_PER_PEER: usize = 4; +const MAX_DISCOVERY_CONNECTS_PER_TICK: usize = 16; + +fn socket_addr_families_compatible(local: SocketAddr, remote: SocketAddr) -> bool { + matches!( + (local, remote), + (SocketAddr::V4(_), SocketAddr::V4(_)) | (SocketAddr::V6(_), SocketAddr::V6(_)) + ) +} impl Node { + /// Replace the runtime peer list. + /// + /// Newly added auto-connect peers are dialed immediately, removed peers + /// are dropped from retry bookkeeping, and existing peers get fresh + /// address hints without tearing down an active link. If an existing peer + /// is already connected and a new concrete candidate appears, FIPS starts + /// an alternate handshake in parallel; promotion switches only after that + /// handshake authenticates. + pub async fn update_peers( + &mut self, + new_peers: Vec, + ) -> Result { + let mut new_by_addr: HashMap = + HashMap::with_capacity(new_peers.len()); + for peer in new_peers { + let identity = + PeerIdentity::from_npub(&peer.npub).map_err(|e| NodeError::InvalidPeerNpub { + npub: peer.npub.clone(), + reason: e.to_string(), + })?; + new_by_addr.insert(*identity.node_addr(), peer); + } + + let current_by_addr: HashMap = self + .config + .peers() + .iter() + .filter_map(|peer| { + PeerIdentity::from_npub(&peer.npub) + .ok() + .map(|identity| (*identity.node_addr(), peer.clone())) + }) + .collect(); + + let new_addrs: HashSet<_> = new_by_addr.keys().copied().collect(); + let current_addrs: HashSet<_> = current_by_addr.keys().copied().collect(); + + let removed: Vec<_> = current_addrs.difference(&new_addrs).copied().collect(); + let added: Vec<_> = new_addrs.difference(¤t_addrs).copied().collect(); + let kept: Vec<_> = new_addrs.intersection(¤t_addrs).copied().collect(); + + let mut outcome = crate::node::UpdatePeersOutcome::default(); + let mut refresh_configs = Vec::new(); + + for node_addr in &removed { + if self.retry_pending.remove(node_addr).is_some() { + debug!( + peer = %self.peer_display_name(node_addr), + "Dropping retry entry for peer removed from runtime peer list" + ); + } + self.peer_aliases.remove(node_addr); + outcome.removed += 1; + } + + for node_addr in &kept { + let new_peer = &new_by_addr[node_addr]; + let current_peer = ¤t_by_addr[node_addr]; + let changed = new_peer.addresses != current_peer.addresses + || new_peer.alias != current_peer.alias + || new_peer.connect_policy != current_peer.connect_policy + || new_peer.auto_reconnect != current_peer.auto_reconnect + || new_peer.via_nostr != current_peer.via_nostr; + + if changed { + outcome.updated += 1; + if let Some(state) = self.retry_pending.get_mut(node_addr) { + state.peer_config = new_peer.clone(); + state.retry_after_ms = Self::now_ms(); + } + if let Some(alias) = new_peer.alias.clone() { + self.peer_aliases.insert(*node_addr, alias); + } + } else { + outcome.unchanged += 1; + } + + if new_peer.is_auto_connect() && (!new_peer.addresses.is_empty() || new_peer.via_nostr) + { + refresh_configs.push(new_peer.clone()); + } + } + + let added_configs: Vec<_> = added + .iter() + .map(|node_addr| new_by_addr[node_addr].clone()) + .collect(); + + self.config.peers = new_by_addr.into_values().collect(); + + for peer_config in added_configs { + outcome.added += 1; + let Ok(identity) = PeerIdentity::from_npub(&peer_config.npub) else { + continue; + }; + let name = peer_config + .alias + .clone() + .unwrap_or_else(|| identity.short_npub()); + self.peer_aliases.insert(*identity.node_addr(), name); + self.register_identity(*identity.node_addr(), identity.pubkey_full()); + + if peer_config.is_auto_connect() + && let Err(err) = self.initiate_peer_connection(&peer_config).await + { + debug!( + npub = %peer_config.npub, + error = %err, + "Failed to initiate connection for newly added runtime peer" + ); + self.schedule_retry(*identity.node_addr(), Self::now_ms()); + } + } + + for peer_config in refresh_configs { + let Ok(identity) = PeerIdentity::from_npub(&peer_config.npub) else { + continue; + }; + let node_addr = *identity.node_addr(); + + if self.peers.contains_key(&node_addr) { + match self + .try_active_peer_alternative_addresses(&peer_config, identity) + .await + { + Ok(true) => debug!( + peer = %self.peer_display_name(&node_addr), + "Started alternate-path handshake for active peer" + ), + Ok(false) => {} + Err(err) => debug!( + npub = %peer_config.npub, + error = %err, + "Active peer alternate-path refresh did not start" + ), + } + } else { + match self.initiate_peer_connection(&peer_config).await { + Ok(()) => { + if let Some(state) = self.retry_pending.get_mut(&node_addr) { + state.peer_config = peer_config; + state.retry_after_ms = Self::now_ms().saturating_add( + self.config.node.rate_limit.handshake_timeout_secs * 1000, + ); + } + } + Err(err) => { + debug!( + npub = %peer_config.npub, + error = %err, + "Refreshed peer addresses did not initiate a direct connection" + ); + self.schedule_retry(node_addr, Self::now_ms()); + } + } + } + } + + Ok(outcome) + } + /// Initiate connections to configured static peers. /// /// For each peer configured with AutoConnect policy, creates a link and @@ -142,6 +313,50 @@ impl Node { }) } + fn is_connecting_to_peer_on_path( + &self, + peer_node_addr: &NodeAddr, + transport_id: TransportId, + remote_addr: &TransportAddr, + ) -> bool { + self.connections.values().any(|conn| { + conn.expected_identity() + .map(|id| id.node_addr() == peer_node_addr) + .unwrap_or(false) + && conn.transport_id() == Some(transport_id) + && conn.source_addr() == Some(remote_addr) + }) || self.pending_connects.iter().any(|pending| { + pending.peer_identity.node_addr() == peer_node_addr + && pending.transport_id == transport_id + && pending.remote_addr == *remote_addr + }) + } + + /// Find a UDP transport whose bound socket can send to `remote_addr`. + /// + /// LAN discovery can surface both IPv4 and IPv6 addresses for the same + /// service. A wildcard IPv4 socket cannot send to an IPv6 link-local + /// target, and vice versa, so callers must choose by socket family rather + /// than by transport type alone. + fn find_udp_transport_for_remote_addr( + &self, + remote_addr: SocketAddr, + ) -> Option<(TransportId, SocketAddr)> { + self.transports + .iter() + .filter(|(id, handle)| { + handle.transport_type().name == "udp" + && handle.is_operational() + && !self.bootstrap_transports.contains(id) + }) + .filter_map(|(id, handle)| { + let local_addr = handle.local_addr()?; + socket_addr_families_compatible(local_addr, remote_addr) + .then_some((*id, local_addr)) + }) + .min_by_key(|(id, _)| id.as_u32()) + } + /// Initiate a connection to a peer on a specific transport and address. /// /// For connectionless transports (UDP, Ethernet): allocates a link, starts @@ -340,6 +555,9 @@ impl Node { pub(super) async fn poll_transport_discovery(&mut self) { // Collect discoveries first to avoid borrow conflict with self let mut to_connect = Vec::new(); + let mut queued_per_peer: HashMap = HashMap::new(); + let mut connect_budget = self.discovery_connect_budget(); + let mut skipped_budget = 0usize; for (transport_id, transport) in &self.transports { if !transport.is_operational() { @@ -366,29 +584,80 @@ impl Node { if node_addr == *self.identity.node_addr() { continue; } - // Skip if already connected + + let candidate_transport_id = *transport_id; + let remote_addr = peer.addr; + if self.peers.contains_key(&node_addr) { - continue; - } - // Skip if connection already in progress - let connecting = self.connections.values().any(|c| { - c.expected_identity() - .map(|id| id.node_addr() == &node_addr) - .unwrap_or(false) - }); - if connecting { + let transport_name = transport.transport_type().name; + let candidate = PeerAddress::new(transport_name, remote_addr.to_string()); + if self.active_peer_candidate_is_fresh_enough_to_skip( + &node_addr, + std::slice::from_ref(&candidate), + ) { + continue; + } + if self.is_connecting_to_peer_on_path( + &node_addr, + candidate_transport_id, + &remote_addr, + ) { + continue; + } + let queued_for_peer = queued_per_peer.get(&node_addr).copied().unwrap_or(0); + if connect_budget == 0 + || self + .path_candidate_attempt_budget(&node_addr) + .saturating_sub(queued_for_peer) + == 0 + { + skipped_budget = skipped_budget.saturating_add(1); + continue; + } + to_connect.push((candidate_transport_id, remote_addr, identity, true)); + *queued_per_peer.entry(node_addr).or_default() += 1; + connect_budget = connect_budget.saturating_sub(1); continue; } - to_connect.push((*transport_id, peer.addr, identity)); + if self.is_connecting_to_peer_on_path( + &node_addr, + candidate_transport_id, + &remote_addr, + ) { + continue; + } + let queued_for_peer = queued_per_peer.get(&node_addr).copied().unwrap_or(0); + if connect_budget == 0 + || self + .path_candidate_attempt_budget(&node_addr) + .saturating_sub(queued_for_peer) + == 0 + { + skipped_budget = skipped_budget.saturating_add(1); + continue; + } + + to_connect.push((candidate_transport_id, remote_addr, identity, false)); + *queued_per_peer.entry(node_addr).or_default() += 1; + connect_budget = connect_budget.saturating_sub(1); } } - for (transport_id, remote_addr, identity) in to_connect { + if skipped_budget > 0 { + debug!( + skipped = skipped_budget, + queued = to_connect.len(), + "Transport discovery connect budget exhausted" + ); + } + + for (transport_id, remote_addr, identity, active_refresh) in to_connect { info!( peer = %self.peer_display_name(identity.node_addr()), transport_id = %transport_id, remote_addr = %remote_addr, + active_refresh, "Auto-connecting to discovered peer" ); if let Err(e) = self @@ -593,6 +862,90 @@ impl Node { self.queue_open_discovery_retries(&bootstrap).await; } + /// Resolve the LAN-only discovery scope. Applications with explicit + /// connectivity config can set `node.discovery.lan.scope` without + /// changing the public Nostr discovery `app` tag. The older fallback + /// extracts a scope from the Nostr app tag used by default scoped + /// discovery. + pub(super) fn lan_discovery_scope(&self) -> Option { + if let Some(scope) = self.config.node.discovery.lan.scope.as_deref() { + let scope = scope.trim(); + if !scope.is_empty() { + return Some(scope.to_string()); + } + } + + let app = self.config.node.discovery.nostr.app.trim(); + if app.is_empty() { + return None; + } + if let Some(rest) = app.strip_prefix("fips-overlay-v1:") { + let scope = rest.trim(); + if scope.is_empty() { + None + } else { + Some(scope.to_string()) + } + } else { + Some(app.to_string()) + } + } + + /// Drain mDNS-discovered peers and initiate Noise XX handshakes. + /// The handshake itself is the authentication — a spoofed mDNS advert + /// with someone else's npub fails the XX exchange and is dropped. + pub(super) async fn poll_lan_discovery(&mut self) { + let Some(runtime) = self.lan_discovery.clone() else { + return; + }; + let events = runtime.drain_events().await; + if events.is_empty() { + return; + } + for event in events { + let crate::discovery::lan::LanEvent::Discovered(peer) = event; + let Some((transport_id, local_addr)) = + self.find_udp_transport_for_remote_addr(peer.addr) + else { + debug!( + addr = %peer.addr, + "lan: skip discovered peer with no compatible UDP transport" + ); + continue; + }; + let identity = match crate::PeerIdentity::from_npub(&peer.npub) { + Ok(id) => id, + Err(err) => { + debug!(npub = %peer.npub, error = %err, "lan: skip bad npub"); + continue; + } + }; + let peer_node_addr = *identity.node_addr(); + let remote_addr = crate::transport::TransportAddr::from_string(&peer.addr.to_string()); + if self.peers.contains_key(&peer_node_addr) + || self.is_connecting_to_peer(&peer_node_addr) + { + continue; + } + info!( + npub = %identity.short_npub(), + addr = %peer.addr, + local_addr = %local_addr, + "lan: initiating handshake to discovered peer" + ); + if let Err(err) = self + .initiate_connection(transport_id, remote_addr, identity) + .await + { + debug!( + npub = %peer.npub, + error = %err, + "lan: failed to initiate connection to discovered peer" + ); + } + } + } + /// Poll pending transport connects and initiate handshakes for ready ones. /// /// Called from the tick handler. For each pending connect, queries the @@ -793,6 +1146,47 @@ impl Node { } } + // mDNS / DNS-SD LAN discovery. Independent of Nostr — runs even + // when Nostr is disabled, since it gives us sub-second pairing + // on the same link without any relay or NAT-traversal roundtrip. + if self.config.node.discovery.lan.enabled { + // Advertise the port of a non-bootstrap operational UDP transport. + // Bootstrap transports must be excluded (they are not the node's + // listening data-plane socket), and a stable selector (lowest + // TransportId) is used so the advertised port is deterministic + // across restarts rather than dependent on HashMap iteration + // order. This mirrors find_udp_transport_for_remote_addr. + let advertised_udp_port = self + .transports + .iter() + .filter(|(id, h)| { + h.transport_type().name == "udp" + && h.is_operational() + && !self.bootstrap_transports.contains(id) + }) + .filter_map(|(id, h)| h.local_addr().map(|addr| (*id, addr.port()))) + .min_by_key(|(id, _)| id.as_u32()) + .map(|(_, port)| port) + .unwrap_or(0); + let scope = self.lan_discovery_scope(); + match crate::discovery::lan::LanDiscovery::start( + &self.identity, + scope, + advertised_udp_port, + self.config.node.discovery.lan.clone(), + ) + .await + { + Ok(runtime) => { + self.lan_discovery = Some(runtime); + info!("LAN mDNS discovery enabled"); + } + Err(err) => { + debug!(error = %err, "LAN mDNS discovery not started"); + } + } + } + // Connect to static peers before TUN is active // This allows handshake messages to be sent before we start accepting packets self.initiate_peer_connections().await; @@ -1078,6 +1472,13 @@ impl Node { warn!(error = %e, "Failed to shutdown Nostr overlay discovery"); } + // Tear down LAN mDNS responder + browser. Best-effort: the + // OS will eventually time the advert out via its TTL even if + // we don't get a clean unregister out before the daemon exits. + if let Some(lan) = self.lan_discovery.take() { + lan.shutdown().await; + } + // Shutdown transports (they're packet producers) let transport_ids: Vec<_> = self.transports.keys().cloned().collect(); for transport_id in transport_ids { @@ -1228,8 +1629,10 @@ impl Node { .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) + let Some(candidate) = + Self::overlay_endpoint_to_peer_address(&endpoint, next_priority, seen_at_ms) else { continue; }; @@ -1251,17 +1654,27 @@ impl Node { 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, - )) + 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_discovery.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( @@ -1271,18 +1684,28 @@ impl Node { allow_bootstrap_nat: bool, addresses: &[PeerAddress], ) -> Result<(), NodeError> { + let peer_node_addr = *peer_identity.node_addr(); + let mut attempted = 0usize; + let max_attempts = self.path_candidate_attempt_budget(&peer_node_addr); + if max_attempts == 0 { + return Err(NodeError::NoTransportForType(format!( + "no outbound slots available for {}", + peer_config.npub + ))); + } + for addr in addresses { + if attempted >= max_attempts { + break; + } if addr.transport == "udp" && addr.addr.eq_ignore_ascii_case("nat") { if !allow_bootstrap_nat { continue; } - let Some(bootstrap) = self.nostr_discovery.clone() else { - debug!(npub = %peer_config.npub, "No Nostr overlay runtime for udp:nat address"); - continue; - }; - bootstrap.request_connect(peer_config.clone()).await; - info!(npub = %peer_config.npub, "Started Nostr UDP NAT traversal attempt"); - return Ok(()); + if self.request_nostr_bootstrap(peer_config).await { + attempted = attempted.saturating_add(1); + } + continue; } let (transport_id, remote_addr) = if addr.transport == "ethernet" { @@ -1320,25 +1743,45 @@ impl Node { continue; } } else { - let tid = match self.find_transport_for_type(&addr.transport) { - Some(id) => id, - None => { - debug!( - transport = %addr.transport, - addr = %addr.addr, - "No operational transport for address type" - ); - continue; + let tid = if addr.transport == "udp" + && let Ok(remote_socket_addr) = addr.addr.parse::() + { + match self.find_udp_transport_for_remote_addr(remote_socket_addr) { + Some((id, _)) => id, + None => { + debug!( + transport = %addr.transport, + addr = %addr.addr, + "No compatible operational UDP transport for address" + ); + continue; + } + } + } else { + match self.find_transport_for_type(&addr.transport) { + Some(id) => id, + None => { + debug!( + transport = %addr.transport, + addr = %addr.addr, + "No operational transport for address type" + ); + continue; + } } }; (tid, TransportAddr::from_string(&addr.addr)) }; + if self.is_connecting_to_peer_on_path(&peer_node_addr, transport_id, &remote_addr) { + continue; + } + match self .initiate_connection(transport_id, remote_addr, peer_identity) .await { - Ok(()) => return Ok(()), + Ok(()) => attempted = attempted.saturating_add(1), Err(e @ NodeError::AccessDenied(_)) => return Err(e), Err(e) => { debug!( @@ -1351,6 +1794,10 @@ impl Node { } } + if attempted > 0 { + return Ok(()); + } + Err(NodeError::NoTransportForType(format!( "no operational transport for any of {}'s addresses", peer_config.npub @@ -1427,6 +1874,21 @@ impl Node { } if configured_npubs.contains(&npub) { + if let Ok(peer_identity) = PeerIdentity::from_npub(&npub) { + let node_addr = *peer_identity.node_addr(); + if !self.peers.contains_key(&node_addr) + && !self.is_connecting_to_peer(&node_addr) + && let Some(state) = self.retry_pending.get_mut(&node_addr) + && state.retry_after_ms > now_ms + { + state.retry_after_ms = now_ms; + debug!( + peer = %peer_identity.short_npub(), + caller = %caller, + "open-discovery sweep: fresh configured-peer advert expedited retry" + ); + } + } skipped_configured = skipped_configured.saturating_add(1); continue; } @@ -1467,8 +1929,10 @@ impl Node { let mut addresses = Vec::new(); let mut priority = 120u8; + let seen_at_ms = Self::now_ms(); for endpoint in endpoints { - let Some(candidate) = Self::overlay_endpoint_to_peer_address(&endpoint, priority) + let Some(candidate) = + Self::overlay_endpoint_to_peer_address(&endpoint, priority, seen_at_ms) else { continue; }; @@ -1607,6 +2071,61 @@ impl Node { connection_slots.min(peer_slots) } + fn outbound_handshake_slots(&self) -> usize { + let used = self + .connections + .len() + .saturating_add(self.pending_connects.len()); + if self.max_connections == 0 { + usize::MAX + } else { + self.max_connections.saturating_sub(used) + } + } + + fn outbound_link_slots(&self) -> usize { + if self.max_links == 0 { + usize::MAX + } else { + self.max_links.saturating_sub(self.links.len()) + } + } + + fn path_candidate_attempt_budget(&self, peer_node_addr: &NodeAddr) -> usize { + if !self.peers.contains_key(peer_node_addr) + && self.max_peers > 0 + && self.peers.len() >= self.max_peers + { + return 0; + } + + let in_flight_for_peer = self + .connections + .values() + .filter(|conn| { + conn.expected_identity() + .map(|identity| identity.node_addr() == peer_node_addr) + .unwrap_or(false) + }) + .count() + .saturating_add( + self.pending_connects + .iter() + .filter(|pending| pending.peer_identity.node_addr() == peer_node_addr) + .count(), + ); + + self.outbound_handshake_slots() + .min(self.outbound_link_slots()) + .min(MAX_PARALLEL_PATH_CANDIDATES_PER_PEER.saturating_sub(in_flight_for_peer)) + } + + fn discovery_connect_budget(&self) -> usize { + self.outbound_handshake_slots() + .min(self.outbound_link_slots()) + .min(MAX_DISCOVERY_CONNECTS_PER_TICK) + } + fn open_discovery_enqueue_budget(&self, configured_npubs: &HashSet) -> usize { let current_open_discovery_pending = self .retry_pending @@ -1841,47 +2360,159 @@ impl Node { return Ok(()); } - // Static-first dialing: avoid delaying configured address attempts on - // advert fetch/network latency. - let static_addresses = self.static_peer_addresses(peer_config); + let candidates = self.peer_address_candidates(peer_config).await; + + if candidates.is_empty() { + return Err(NodeError::NoTransportForType(format!( + "no addresses known for {}", + peer_config.npub + ))); + } + if self - .attempt_peer_address_list( - peer_config, - peer_identity, - allow_bootstrap_nat, - &static_addresses, - ) + .attempt_peer_address_list(peer_config, peer_identity, allow_bootstrap_nat, &candidates) .await .is_ok() { return Ok(()); } - { - let fallback = self - .nostr_peer_fallback_addresses(peer_config, &static_addresses) - .await; - if !fallback.is_empty() - && self - .attempt_peer_address_list( - peer_config, - peer_identity, - allow_bootstrap_nat, - &fallback, - ) - .await - .is_ok() - { - return Ok(()); - } - } - Err(NodeError::NoTransportForType(format!( "no operational transport for any of {}'s addresses", peer_config.npub ))) } + async fn try_active_peer_alternative_addresses( + &mut self, + peer_config: &PeerConfig, + peer_identity: PeerIdentity, + ) -> Result { + let peer_node_addr = *peer_identity.node_addr(); + let candidates = self.peer_address_candidates(peer_config).await; + + if candidates.is_empty() { + return Err(NodeError::NoTransportForType(format!( + "no addresses known for {}", + peer_config.npub + ))); + } + + let concrete: Vec<_> = candidates + .into_iter() + .filter(|addr| !(addr.transport == "udp" && addr.addr.eq_ignore_ascii_case("nat"))) + .collect(); + let has_alternative = concrete + .iter() + .any(|addr| !self.active_peer_matches_candidate(&peer_node_addr, addr)); + let attempt_candidates: Vec<_> = if has_alternative { + concrete + .into_iter() + .filter(|addr| !self.active_peer_matches_candidate(&peer_node_addr, addr)) + .collect() + } else if self.active_peer_needs_same_path_refresh(&peer_node_addr) { + concrete + } else { + Vec::new() + }; + + if attempt_candidates.is_empty() { + return Ok(false); + } + + self.attempt_peer_address_list(peer_config, peer_identity, false, &attempt_candidates) + .await?; + Ok(true) + } + + 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) + .await; + + let mut candidates = Vec::with_capacity(overlay_addresses.len() + static_addresses.len()); + for addr in overlay_addresses.into_iter().chain(static_addresses) { + if !candidates.iter().any(|existing: &PeerAddress| { + existing.transport == addr.transport && existing.addr == addr.addr + }) { + candidates.push(addr); + } + } + + candidates.sort_by(|a, b| match (a.seen_at_ms, b.seen_at_ms) { + (Some(a_ts), Some(b_ts)) => b_ts.cmp(&a_ts), + (Some(_), None) => std::cmp::Ordering::Less, + (None, Some(_)) => std::cmp::Ordering::Greater, + (None, None) => std::cmp::Ordering::Equal, + }); + candidates + } + + pub(in crate::node) fn active_peer_candidate_is_fresh_enough_to_skip( + &self, + peer_node_addr: &NodeAddr, + candidates: &[PeerAddress], + ) -> bool { + if !self.active_peer_matches_any_candidate(peer_node_addr, candidates) { + return false; + } + !self.active_peer_needs_same_path_refresh(peer_node_addr) + } + + fn active_peer_needs_same_path_refresh(&self, peer_node_addr: &NodeAddr) -> bool { + let Some(peer) = self.peers.get(peer_node_addr) else { + return false; + }; + let stale_after_ms = self + .config + .node + .heartbeat_interval_secs + .saturating_mul(1000) + .max(1000); + peer.idle_time(Self::now_ms()) > stale_after_ms + } + + fn active_peer_matches_any_candidate( + &self, + peer_node_addr: &NodeAddr, + candidates: &[PeerAddress], + ) -> bool { + candidates + .iter() + .any(|candidate| self.active_peer_matches_candidate(peer_node_addr, candidate)) + } + + fn active_peer_matches_candidate( + &self, + peer_node_addr: &NodeAddr, + candidate: &PeerAddress, + ) -> bool { + let Some(peer) = self.peers.get(peer_node_addr) else { + return false; + }; + let Some(current_addr) = peer.current_addr() else { + return false; + }; + if peer + .transport_id() + .map(|id| self.bootstrap_transports.contains(&id)) + .unwrap_or(false) + { + return false; + } + let current_addr = current_addr.to_string(); + let current_transport = peer + .transport_id() + .and_then(|id| self.transports.get(&id)) + .map(|transport| transport.transport_type().name); + + candidate.addr == current_addr + && current_transport + .map(|transport| transport == candidate.transport) + .unwrap_or(true) + } + // === Control API methods === /// Connect to a peer via the control API. diff --git a/src/node/mod.rs b/src/node/mod.rs index 9fcd614..5776b14 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -202,6 +202,19 @@ impl fmt::Display for NodeState { } } +/// Reports what changed when replacing the runtime peer list. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct UpdatePeersOutcome { + /// Peers present in the new list but not the previous list. + pub added: usize, + /// Peers removed from the previous list. + pub removed: usize, + /// Existing peers whose configured behavior changed. + pub updated: usize, + /// Existing peers whose comparable config did not change. + pub unchanged: usize, +} + /// Recent request tracking for dedup and reverse-path forwarding. /// /// When a LookupRequest is forwarded through a node, the node stores the @@ -453,6 +466,11 @@ pub struct Node { /// Optional Nostr/STUN overlay discovery coordinator for `udp:nat` peers. nostr_discovery: Option>, + /// 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_discovery: 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 @@ -669,6 +687,7 @@ impl Node { retry_pending: HashMap::new(), nostr_discovery: None, nostr_discovery_started_at_ms: None, + lan_discovery: None, startup_open_discovery_sweep_done: false, bootstrap_transports: HashSet::new(), bootstrap_transport_npubs: HashMap::new(), @@ -812,6 +831,7 @@ impl Node { retry_pending: HashMap::new(), nostr_discovery: None, nostr_discovery_started_at_ms: None, + lan_discovery: None, startup_open_discovery_sweep_done: false, bootstrap_transports: HashSet::new(), bootstrap_transport_npubs: HashMap::new(), diff --git a/src/node/retry.rs b/src/node/retry.rs index 34f7e0e..91f2f08 100644 --- a/src/node/retry.rs +++ b/src/node/retry.rs @@ -11,6 +11,7 @@ use crate::identity::NodeAddr; use tracing::{debug, info, warn}; // MAX_BACKOFF_MS is now derived from config: node.retry.max_backoff_secs * 1000 +const MAX_RETRY_CONNECTIONS_PER_TICK: usize = 16; /// Tracks retry state for a peer across connection attempts. pub struct RetryState { @@ -248,8 +249,17 @@ impl Node { .filter(|(_, state)| now_ms >= state.retry_after_ms) .map(|(addr, _)| *addr) .collect(); + let deferred = due.len().saturating_sub(MAX_RETRY_CONNECTIONS_PER_TICK); + if deferred > 0 { + debug!( + due = due.len(), + processing = MAX_RETRY_CONNECTIONS_PER_TICK, + deferred, + "Retry processing budget exhausted; deferring remaining peers" + ); + } - for node_addr in due { + for node_addr in due.into_iter().take(MAX_RETRY_CONNECTIONS_PER_TICK) { // Peer may have connected inbound while we waited if self.peers.contains_key(&node_addr) { self.retry_pending.remove(&node_addr); diff --git a/src/node/tests/unit.rs b/src/node/tests/unit.rs index ed654a4..acbb403 100644 --- a/src/node/tests/unit.rs +++ b/src/node/tests/unit.rs @@ -97,6 +97,58 @@ async fn test_nat_bootstrap_failure_falls_back_to_direct_udp_address() { } } +#[tokio::test] +async fn test_try_peer_addresses_races_all_concrete_udp_candidates() { + let peer_identity = Identity::generate(); + let mut node = make_node(); + let (packet_tx, packet_rx) = packet_channel(64); + node.packet_tx = Some(packet_tx.clone()); + node.packet_rx = Some(packet_rx); + + let transport_id = TransportId::new(1); + let mut udp = UdpTransport::new( + transport_id, + Some("main".to_string()), + crate::config::UdpConfig { + bind_addr: Some("127.0.0.1:0".to_string()), + ..Default::default() + }, + packet_tx, + ); + udp.start_async().await.unwrap(); + node.transports + .insert(transport_id, TransportHandle::Udp(udp)); + + let peer_config = crate::config::PeerConfig { + npub: peer_identity.npub(), + alias: None, + addresses: vec![ + crate::config::PeerAddress::with_priority("udp", "127.0.0.1:9", 1), + crate::config::PeerAddress::with_priority("udp", "127.0.0.1:10", 2), + ], + connect_policy: crate::config::ConnectPolicy::AutoConnect, + auto_reconnect: true, + via_nostr: false, + }; + let peer_identity = PeerIdentity::from_npub(&peer_config.npub).unwrap(); + + node.try_peer_addresses(&peer_config, peer_identity, false) + .await + .unwrap(); + + let mut addrs = node + .connections + .values() + .filter_map(|conn| conn.source_addr().and_then(|addr| addr.as_str())) + .collect::>(); + addrs.sort(); + assert_eq!(addrs, vec!["127.0.0.1:10", "127.0.0.1:9"]); + + for transport in node.transports.values_mut() { + transport.stop().await.ok(); + } +} + #[tokio::test] async fn test_node_state_transitions() { let mut node = make_node(); @@ -712,6 +764,48 @@ fn test_schedule_retry_increments() { assert_eq!(state.retry_after_ms, 11_000 + 20_000); } +/// Retry processing is paced so a large due set cannot start every +/// handshake candidate in one maintenance tick. +#[tokio::test] +async fn test_process_pending_retries_is_budgeted_per_tick() { + let mut node = make_node(); + let mut addrs = Vec::new(); + + for _ in 0..20 { + let identity = Identity::generate(); + let npub = identity.npub(); + let peer_identity = PeerIdentity::from_npub(&npub).unwrap(); + let node_addr = *peer_identity.node_addr(); + node.retry_pending.insert( + node_addr, + crate::node::retry::RetryState { + peer_config: crate::config::PeerConfig::new(npub, "udp", "10.0.0.2:2121"), + retry_count: 0, + retry_after_ms: 0, + reconnect: true, + expires_at_ms: None, + }, + ); + addrs.push(node_addr); + } + + node.process_pending_retries(1).await; + + let processed = addrs + .iter() + .filter(|addr| { + node.retry_pending + .get(addr) + .is_some_and(|state| state.retry_count > 0) + }) + .count(); + let deferred = addrs.len().saturating_sub(processed); + + assert_eq!(processed, 16); + assert_eq!(deferred, 4); + assert_eq!(node.retry_pending.len(), 20); +} + /// Test that auto-connect peers retry indefinitely (never exhaust). #[test] fn test_schedule_retry_auto_connect_never_exhausts() { @@ -864,6 +958,130 @@ async fn test_try_peer_addresses_skips_connecting_peer() { ); } +#[test] +fn active_peer_same_path_discovery_skips_fresh_peer() { + let mut node = make_node(); + let peer_full = Identity::generate(); + let peer_identity = PeerIdentity::from_pubkey_full(peer_full.pubkey_full()); + let peer_node_addr = *peer_identity.node_addr(); + let transport_id = TransportId::new(1); + let current_addr = TransportAddr::from_string("127.0.0.1:9"); + let mut active_peer = ActivePeer::new(peer_identity, LinkId::new(7), Node::now_ms()); + active_peer.set_current_addr(transport_id, current_addr.clone()); + node.peers.insert(peer_node_addr, active_peer); + let candidate = crate::config::PeerAddress::new("udp", "127.0.0.1:9"); + + assert!(node.active_peer_candidate_is_fresh_enough_to_skip( + &peer_node_addr, + std::slice::from_ref(&candidate), + )); +} + +#[test] +fn active_peer_same_path_discovery_refreshes_stale_peer() { + let mut node = make_node(); + let peer_full = Identity::generate(); + let peer_identity = PeerIdentity::from_pubkey_full(peer_full.pubkey_full()); + let peer_node_addr = *peer_identity.node_addr(); + let transport_id = TransportId::new(1); + let current_addr = TransportAddr::from_string("127.0.0.1:9"); + let stale_at = Node::now_ms().saturating_sub( + node.config + .node + .heartbeat_interval_secs + .saturating_add(1) + .saturating_mul(1000), + ); + let mut active_peer = ActivePeer::new(peer_identity, LinkId::new(7), stale_at); + active_peer.set_current_addr(transport_id, current_addr.clone()); + node.peers.insert(peer_node_addr, active_peer); + let candidate = crate::config::PeerAddress::new("udp", "127.0.0.1:9"); + + assert!(!node.active_peer_candidate_is_fresh_enough_to_skip( + &peer_node_addr, + std::slice::from_ref(&candidate), + )); +} + +#[tokio::test] +async fn update_peers_races_new_alternative_without_dropping_active_peer() { + let mut node = make_node(); + let (packet_tx, packet_rx) = packet_channel(64); + node.packet_tx = Some(packet_tx.clone()); + node.packet_rx = Some(packet_rx); + + let transport_id = TransportId::new(1); + let mut udp = UdpTransport::new( + transport_id, + Some("main".to_string()), + crate::config::UdpConfig { + bind_addr: Some("127.0.0.1:0".to_string()), + ..Default::default() + }, + packet_tx, + ); + udp.start_async().await.unwrap(); + node.transports + .insert(transport_id, TransportHandle::Udp(udp)); + + let peer_full = Identity::generate(); + let peer_identity = PeerIdentity::from_pubkey_full(peer_full.pubkey_full()); + let peer_node_addr = *peer_identity.node_addr(); + let current_addr = TransportAddr::from_string("127.0.0.1:9"); + let new_addr = TransportAddr::from_string("127.0.0.1:10"); + let old_link_id = LinkId::new(7); + let mut active_peer = ActivePeer::new(peer_identity, old_link_id, Node::now_ms()); + active_peer.set_current_addr(transport_id, current_addr.clone()); + node.peers.insert(peer_node_addr, active_peer); + node.links.insert( + old_link_id, + Link::connectionless( + old_link_id, + transport_id, + current_addr.clone(), + LinkDirection::Outbound, + Duration::from_millis(100), + ), + ); + + let old_peer = crate::config::PeerConfig { + npub: peer_full.npub(), + alias: None, + addresses: vec![crate::config::PeerAddress::new("udp", "127.0.0.1:9")], + connect_policy: crate::config::ConnectPolicy::AutoConnect, + auto_reconnect: true, + via_nostr: false, + }; + let new_peer = crate::config::PeerConfig { + addresses: vec![ + crate::config::PeerAddress::new("udp", "127.0.0.1:9"), + crate::config::PeerAddress::new("udp", "127.0.0.1:10"), + ], + ..old_peer.clone() + }; + node.config.peers = vec![old_peer]; + + let outcome = node.update_peers(vec![new_peer]).await.unwrap(); + + assert_eq!(outcome.updated, 1); + assert_eq!(node.peer_count(), 1, "existing link must stay live"); + assert_eq!(node.connection_count(), 1); + assert_eq!( + node.connections + .values() + .next() + .and_then(|conn| conn.source_addr()), + Some(&new_addr) + ); + let active = node.get_peer(&peer_node_addr).unwrap(); + assert_eq!(active.link_id(), old_link_id); + assert_eq!(active.current_addr(), Some(¤t_addr)); + + for transport in node.transports.values_mut() { + transport.stop().await.ok(); + } +} + #[tokio::test] async fn test_nostr_traversal_failure_skips_connected_peer() { let mut node = make_node(); diff --git a/src/peer/active.rs b/src/peer/active.rs index 2af00f0..3fa1535 100644 --- a/src/peer/active.rs +++ b/src/peer/active.rs @@ -593,6 +593,13 @@ impl ActivePeer { self.remote_epoch } + /// Update the remote peer's startup epoch after a successful in-place + /// rekey. Initial handshakes set this through `with_session`, but recovery + /// rekeys also exchange epochs and must keep restart detection current. + pub(crate) fn set_remote_epoch(&mut self, remote_epoch: Option<[u8; 8]>) { + self.remote_epoch = remote_epoch; + } + // === Tree Accessors === /// Get the peer's tree coordinates, if known. @@ -1099,7 +1106,10 @@ impl ActivePeer { /// Takes the stored handshake state, reads msg2, and returns the /// completed NoiseSession. Clears the handshake-related fields but /// leaves rekey_our_index for set_pending_session to use. - pub fn complete_rekey_msg2(&mut self, msg2_bytes: &[u8]) -> Result { + pub fn complete_rekey_msg2( + &mut self, + msg2_bytes: &[u8], + ) -> Result<(NoiseSession, Option<[u8; 8]>), NoiseError> { let mut hs = self .rekey_handshake .take() @@ -1109,13 +1119,14 @@ impl ActivePeer { })?; hs.read_message_2(msg2_bytes)?; + let remote_epoch = hs.remote_epoch(); let session = hs.into_session()?; // Clear msg1 resend state self.rekey_msg1 = None; self.rekey_msg1_next_resend = 0; - Ok(session) + Ok((session, remote_epoch)) } /// Check if msg1 needs resending.