From 22731d7a212b441991e0804e81d76fd2557b68a1 Mon Sep 17 00:00:00 2001 From: Warm Beer Date: Mon, 6 Apr 2026 21:41:36 +0200 Subject: [PATCH] fix: advertise dialable addresses to DHT and harden NAT traversal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Splits the NAT-traversal and dialable-address work from the original PR #70 bundle. Root cause and fix below; other correctness fixes bundled. ## Root cause `DhtNetworkManager::local_dht_node()` built its self-entry from the static `NodeConfig::listen_addrs()` derivation, which returns wildcard IPs (`0.0.0.0` / `::`) and the configured port (`0` for ephemeral binds). Receivers' `dialable_addresses()` filter then rejected the entry, so DHT-based peer discovery returned **zero contactable addresses** for any node. Connections only succeeded via side channels (static bootstrap, in-memory cache, relay fallback) — exactly the sporadic-failure pattern reported by `saorsa-node-lmdb`. ## Fix `local_dht_node()` is now an `async fn` that sources its self-entry addresses from `TransportHandle::observed_external_addresses()` — the post-NAT addresses observed via QUIC `OBSERVED_ADDRESS` frames, one per local interface that has an observation on a multi-homed host. An intermediate iteration added a `UdpSocket::connect`-then- `getsockname` helper (`primary_local_ip`) to substitute wildcard IPs, but it was dropped in a follow-up because it ran blocking syscalls inside an async context and returned RFC1918 addresses behind home NAT; the OBSERVED_ADDRESS-only flow supersedes it and needs no interface probing. If no source produces an address, the published self-entry has an empty `addresses` vec. This is the right answer at the publish layer: better to admit "I don't know how to be reached yet" than to lie with a bind-side wildcard. The empty window closes naturally once the first peer connects and OBSERVED_ADDRESS flows. ## Other NAT fixes bundled - **Event-driven relay/address updates** — relay-established and peer-address-update events drain via `tokio::select!` over `TransportHandle::recv_relay_established()` / `recv_peer_address_update()` instead of the previous 1-second polling loop, eliminating the race window where outbound DHT queries advertised no relay address for up to a second after relay setup. - **Identity exchange timeout bumped to 15 s** — `send_dht_request` computes its identity-exchange timeout as `min(request_timeout, IDENTITY_EXCHANGE_TIMEOUT)`; with the constant at 5 s the effective budget was always 5 s, too tight for cross-region or congested cellular links where the two-RTT exchange plus ML-DSA-65 verification can blow past with retransmits. Bumped to 15 s to match the bootstrap budget. - **Observed-address cache fallback** — saorsa-transport exposes the externally-observed address only via active connections, so a node whose every connection has just dropped has no answer for `observed_external_address()`. This introduces `ObservedAddressCache`: an in-memory, frequency- and recency-aware cache populated by `P2pEvent::ExternalAddressDiscovered` events. When the live source is empty, the cache returns the most-frequently-observed address among recent entries, breaking ties by recency. NAT-rebind handling via a two-pass selection (recent window wins over stale counts). Keyed by `(local_bind, observed)` so multi-homed hosts don't cross-pollute observations from different interfaces. The cache is reset on process restart — a fresh node re-discovers its current address from live connections rather than trusting stale state. - **Bridge task split + bounded forwarder channels** — the DHT bridge previously ran `find_closest_nodes_network` inline inside the select loop on the relay branch, blocking the peer-address-update branch for the duration of the iterative lookup; the underlying forwarder mpsc channels were `unbounded_channel` so the queue could grow without limit during the stall. Detached the lookup + publish into its own spawned task per relay event so the select loop keeps polling, and replaced both forwarder channels with bounded `channel(ADDRESS_EVENT_CHANNEL_CAPACITY)` plus a per-event drop counter. - **`dial_addresses()` helper** — loops through every dialable address until one succeeds. Replaces single-address dial sites in `bootstrap_from_peers`, `spawn_bucket_refresh_task`, `trigger_self_lookup`, `find_closest_nodes_network` parallel-query block, and `send_dht_request` fallback so a stale NAT binding on entry #1 no longer kills the dial when entry #2 would have worked. ## Regression tests `tests/dht_self_advertisement.rs` adds six `#[tokio::test]` cases that pin the published self-entry behaviour: - **Loopback bind path** (single-node, `local: true`): - `local_mode_publishes_dialable_loopback_address` - `local_mode_published_self_entry_matches_runtime_listen_addrs` - `local_mode_never_publishes_port_zero` - **Wildcard / OBSERVED_ADDRESS path** (`local: false`): - `wildcard_bind_with_no_peers_publishes_empty_self_entry` — pins the "don't lie when you don't know" contract. - `wildcard_bind_publishes_observed_address_after_peer_connection` — primary regression test for the OBSERVED_ADDRESS flow. - `observed_address_cache_serves_fallback_after_connection_drop` — pins the cache fallback behaviour after all live connections drop. All six FAIL on `rc-2026.4.1` with concrete evidence (e.g. `[Quic(0.0.0.0:0)]` published as the only address) and PASS here. `ObservedAddressCache` ships with 9 unit tests covering empty, single, repeated, frequency, recency tiebreak, NAT-rebinding fallback, long-quiet fallback, eviction, and re-observation-on-full. ## chore: drop needless struct update in IPDiversityConfig test Both fields of `IPDiversityConfig` are explicitly set, so the trailing `..IPDiversityConfig::default()` triggers `clippy::needless_update` under `cargo clippy --all-targets`. Test code only; no functional change. ## Verification - `cargo build --lib`: clean - `cargo fmt --check`: clean - `cargo clippy --lib -- -D warnings -D clippy::unwrap_used -D clippy::expect_used`: clean - `cargo clippy --all-targets -- -D warnings`: clean - `cargo test --lib`: 282 passed, 0 failed - `cargo test --test dht_self_advertisement`: 6/6 passed - `cargo test --test two_node_messaging`: 7/7 passed - `cargo test --test node_lifecycle`: 17/17 passed This is one of two PRs that split the original PR #70, which bundled this NAT-traversal hardening with two 1000-node scale fixes in the DHT touch path and the inbound message dispatcher. The other PR covers the 1000-node perf work. Co-Authored-By: Claude Opus 4.6 (1M context) --- src/dht/security_tests.rs | 1 - src/dht_network_manager.rs | 217 ++++++-- src/network.rs | 160 +++--- src/transport.rs | 5 + src/transport/observed_address_cache.rs | 585 +++++++++++++++++++ src/transport/saorsa_transport_adapter.rs | 207 ++++++- src/transport_handle.rs | 155 +++++- tests/dht_self_advertisement.rs | 649 ++++++++++++++++++++++ 8 files changed, 1823 insertions(+), 156 deletions(-) create mode 100644 src/transport/observed_address_cache.rs create mode 100644 tests/dht_self_advertisement.rs diff --git a/src/dht/security_tests.rs b/src/dht/security_tests.rs index 479beece..52af268c 100644 --- a/src/dht/security_tests.rs +++ b/src/dht/security_tests.rs @@ -186,7 +186,6 @@ async fn test_ipv4_ip_override_raises_limit() -> anyhow::Result<()> { engine.set_ip_diversity_config(IPDiversityConfig { max_per_ip: Some(3), max_per_subnet: Some(usize::MAX), - ..IPDiversityConfig::default() }); for i in 1..=3u8 { diff --git a/src/dht_network_manager.rs b/src/dht_network_manager.rs index 025ed4cf..0397ef88 100644 --- a/src/dht_network_manager.rs +++ b/src/dht_network_manager.rs @@ -31,7 +31,7 @@ use crate::{ use anyhow::Context as _; use serde::{Deserialize, Serialize}; use std::collections::{BTreeMap, HashMap, HashSet}; -use std::net::IpAddr; +use std::net::{IpAddr, SocketAddr}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant, SystemTime}; use tokio::sync::{RwLock, Semaphore, broadcast, oneshot}; @@ -62,7 +62,16 @@ const SELF_RELIABILITY_SCORE: f64 = 1.0; /// Maximum time to wait for the identity-exchange handshake after dialling /// a peer. The actual timeout is `min(request_timeout, this)`. -const IDENTITY_EXCHANGE_TIMEOUT: Duration = Duration::from_secs(5); +/// +/// Identity exchange is two RTTs over a freshly-handshaken QUIC connection +/// plus an ML-DSA-65 signature verification. On a LAN this completes in +/// well under a second; on congested cellular or cross-region links it can +/// blow past 5s with retransmits. Kept in lockstep with +/// `BOOTSTRAP_IDENTITY_TIMEOUT_SECS` in `network.rs` — both budgets exist +/// to absorb the same slow-link failure mode (the bootstrap variant covers +/// the initial join, this one covers every subsequent peer dial via +/// `send_dht_request`). +const IDENTITY_EXCHANGE_TIMEOUT: Duration = Duration::from_secs(15); /// Maximum time to wait for a stale peer's ping response during admission contention. const STALE_REVALIDATION_TIMEOUT: Duration = Duration::from_secs(1); @@ -590,11 +599,8 @@ impl DhtNetworkManager { if dht_node.peer_id == this.config.peer_id { continue; } - if let Some(addr) = - Self::first_dialable_address(&dht_node.addresses) - { - this.dial_candidate(&dht_node.peer_id, &addr, None).await; - } + this.dial_addresses(&dht_node.peer_id, &dht_node.addresses, None) + .await; } } Err(e) => { @@ -626,10 +632,11 @@ impl DhtNetworkManager { if dht_node.peer_id == self_id { continue; } - // Dial if not already connected - if let Some(addr) = Self::first_dialable_address(&dht_node.addresses) { - self.dial_candidate(&dht_node.peer_id, &addr, None).await; - } + // Dial if not already connected — try every advertised + // address, not just the first, so a stale NAT binding on + // one entry doesn't kill the dial. + self.dial_addresses(&dht_node.peer_id, &dht_node.addresses, None) + .await; } Ok(()) } @@ -725,21 +732,24 @@ impl DhtNetworkManager { .first() .and_then(|a| a.dialable_socket_addr()); + // The bootstrap peer is the natural NAT-traversal referrer for + // every node it returns: it has a live connection to us (we just + // queried it) and presumably also to the nodes it tells us about. + // Passing its socket address as the preferred coordinator lets + // hole-punch PUNCH_ME_NOW be relayed through it. let op = DhtNetworkOperation::FindNode { key }; match self.send_dht_request(peer_id, op, None).await { Ok(DhtNetworkResult::NodesFound { nodes, .. }) => { for node in &nodes { - let first = Self::first_dialable_address(&node.addresses); + let dialable = Self::dialable_addresses(&node.addresses); debug!( - "DHT bootstrap: peer={} num_addresses={} first_dialable={:?}", + "DHT bootstrap: peer={} num_addresses={} dialable={}", node.peer_id.to_hex(), node.addresses.len(), - first.as_ref().map(|a| a.to_string()) + dialable.len() ); - if seen.insert(node.peer_id) - && let Some(addr) = first - { - self.dial_candidate(&node.peer_id, &addr, bootstrap_addr) + if seen.insert(node.peer_id) && !dialable.is_empty() { + self.dial_addresses(&node.peer_id, &node.addresses, bootstrap_addr) .await; } } @@ -923,7 +933,7 @@ impl DhtNetworkManager { // back to `count`. Self may displace the farthest peer. let mut nodes = self.find_closest_nodes_local(key, count).await; - nodes.push(self.local_dht_node()); + nodes.push(self.local_dht_node().await); let key_peer = PeerId::from_bytes(*key); nodes.sort_by(|a, b| { @@ -972,7 +982,7 @@ impl DhtNetworkManager { // final K-closest result, but we must never send an RPC to ourselves. // Seed best_nodes with self and mark self as "queried" so the iterative // loop never tries to contact us. - best_nodes.push(self.local_dht_node()); + best_nodes.push(self.local_dht_node().await); self.mark_self_queried(&mut queried_nodes); // Candidates sorted by XOR distance to target (closest first). @@ -1039,16 +1049,21 @@ impl DhtNetworkManager { .iter() .map(|node| { let peer_id = node.peer_id; - let address = Self::first_dialable_address(&node.addresses); + let addresses = node.addresses.clone(); let referrer = referrers.get(&peer_id).copied(); let op = DhtNetworkOperation::FindNode { key: *key }; async move { - if let Some(ref addr) = address { - self.dial_candidate(&peer_id, addr, referrer).await; - } + // Try every dialable address, not just the first. + // If at least one succeeds the peer is connected and + // `send_dht_request` will reuse that channel; if all + // fail, `send_dht_request`'s own fallback will retry + // with the routing-table addresses. + self.dial_addresses(&peer_id, &addresses, referrer).await; + let address_hint = Self::first_dialable_address(&addresses); ( peer_id, - self.send_dht_request(&peer_id, op, address.as_ref()).await, + self.send_dht_request(&peer_id, op, address_hint.as_ref()) + .await, ) } }) @@ -1234,14 +1249,56 @@ impl DhtNetworkManager { /// Build a `DHTNode` representing the local node for inclusion in /// K-closest results. The local node always participates in distance /// ranking but is never queried over the network. - fn local_dht_node(&self) -> DHTNode { - let mut addresses: Vec = self.config.node_config.listen_addrs().to_vec(); - if addresses.is_empty() { - addresses.push(MultiAddr::quic(std::net::SocketAddr::new( - std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED), - 0, - ))); + /// + /// The published address list is sourced from: + /// + /// 1. The transport's externally-observed reflexive address (set by + /// OBSERVED_ADDRESS frames received from peers). This is the only + /// authoritative source for a NAT'd node — it is the actual post-NAT + /// address that remote peers see the connection arrive from. + /// 2. The transport's runtime-bound `listen_addrs`, but **only when the + /// bind address has a specific (non-wildcard) IP**. Wildcard binds + /// (`0.0.0.0` / `[::]`) are bind-side concepts meaning "any interface" + /// and are not dialable, so we skip them entirely and rely on (1). + /// + /// If neither source produces an address, the returned `DHTNode` has an + /// empty `addresses` vec. This is the right answer at the publish layer: + /// it tells consumers "I don't know how to be reached yet" rather than + /// lying with a bind-side wildcard or a guessed LAN IP that won't work + /// from the public internet. The empty window closes naturally once the + /// first peer connects to us and OBSERVED_ADDRESS flows. + async fn local_dht_node(&self) -> DHTNode { + let mut addresses: Vec = Vec::new(); + + // 1. Observed external addresses — the post-NAT addresses peers + // actually see, learned from QUIC OBSERVED_ADDRESS frames. + // Empty until at least one peer has observed us. On a + // multi-homed host this can return multiple addresses (one per + // local interface that has an observation), and we publish all + // of them so peers reaching us via any interface can dial back. + for observed in self.transport.observed_external_addresses() { + let resolved = MultiAddr::quic(observed); + if !addresses.contains(&resolved) { + addresses.push(resolved); + } } + + // 2. Runtime-bound listen addresses with specific IPs only. Wildcards + // and zero ports are pre-bind placeholders or all-interface + // bindings — neither is dialable. + for la in self.transport.listen_addrs().await { + let Some(sa) = la.dialable_socket_addr() else { + continue; + }; + if sa.port() == 0 || sa.ip().is_unspecified() { + continue; + } + let resolved = MultiAddr::quic(sa); + if !addresses.contains(&resolved) { + addresses.push(resolved); + } + } + DHTNode { peer_id: self.config.peer_id, addresses, @@ -1287,6 +1344,43 @@ impl DhtNetworkManager { Self::dialable_addresses(addresses).into_iter().next() } + /// Try dialing each dialable address in `addresses` in order until one + /// succeeds. Returns the channel ID of the first successful dial, or + /// `None` if every address was rejected, failed, or timed out. + /// + /// This is the multi-address counterpart of [`Self::dial_candidate`] + /// and is the right entry point for any code path that has been handed + /// a `DHTNode` (or any peer entry that exposes multiple addresses) — + /// using only the first dialable address means a stale NAT binding, + /// failed relay, or unreachable family kills the connection attempt + /// even when other published addresses would have worked. + async fn dial_addresses( + &self, + peer_id: &PeerId, + addresses: &[MultiAddr], + referrer: Option, + ) -> Option { + let dialable = Self::dialable_addresses(addresses); + if dialable.is_empty() { + debug!( + "dial_addresses: no dialable addresses for {}", + peer_id.to_hex() + ); + return None; + } + for addr in &dialable { + if let Some(channel_id) = self.dial_candidate(peer_id, addr, referrer).await { + return Some(channel_id); + } + } + debug!( + "dial_addresses: all {} address(es) failed for {}", + dialable.len(), + peer_id.to_hex() + ); + None + } + async fn record_peer_failure(&self, peer_id: &PeerId) { if let Some(ref engine) = self.trust_engine { engine.update_node_stats( @@ -1397,23 +1491,39 @@ impl DhtNetworkManager { // `peer_to_channel` mapping is only populated after the asynchronous // identity-exchange handshake completes. Without waiting, the // subsequent `send_message` would fail with `PeerNotFound`. - let resolved_address: Option = if self.transport.is_peer_connected(peer_id).await + // + // Build the candidate address list: caller's hint first (if any), + // then the peer's addresses from the routing table. Trying every + // candidate — instead of stopping at the first — protects against + // stale NAT bindings, single-IP-family failures, and recently-relayed + // peers whose direct address is no longer reachable. + let candidate_addresses: Vec = if self.transport.is_peer_connected(peer_id).await { - None - } else if let Some(hint) = address_hint { - Some(hint.clone()) + Vec::new() } else { - self.peer_addresses_for_dial(peer_id) - .await - .into_iter() - .next() + let mut addrs = Vec::new(); + if let Some(hint) = address_hint { + addrs.push(hint.clone()); + } + for addr in self.peer_addresses_for_dial(peer_id).await { + if !addrs.contains(&addr) { + addrs.push(addr); + } + } + addrs }; - if let Some(ref address) = resolved_address { + + if !candidate_addresses.is_empty() { info!( - "[STEP 1b] {} -> {}: No open channel, dialling {}", - local_hex, peer_hex, address + "[STEP 1b] {} -> {}: No open channel, trying {} dialable address(es)", + local_hex, + peer_hex, + candidate_addresses.len() ); - if let Some(channel_id) = self.dial_candidate(peer_id, address, None).await { + if let Some(channel_id) = self + .dial_addresses(peer_id, &candidate_addresses, None) + .await + { let identity_timeout = self.config.request_timeout.min(IDENTITY_EXCHANGE_TIMEOUT); match self .transport @@ -1423,11 +1533,10 @@ impl DhtNetworkManager { Ok(authenticated) => { if &authenticated != peer_id { warn!( - "[STEP 1b] {} -> {}: identity MISMATCH — dialled {} but authenticated as {}. \ + "[STEP 1b] {} -> {}: identity MISMATCH — authenticated as {}. \ Routing table entry may be stale.", local_hex, peer_hex, - address, authenticated.to_hex() ); if let Ok(mut ops) = self.active_operations.lock() { @@ -1462,15 +1571,22 @@ impl DhtNetworkManager { } } else { warn!( - "[STEP 1b] {} -> {}: dial failed to {}", - local_hex, peer_hex, address + "[STEP 1b] {} -> {}: dial failed for all {} candidate address(es)", + local_hex, + peer_hex, + candidate_addresses.len() ); if let Ok(mut ops) = self.active_operations.lock() { ops.remove(&message_id); } self.record_peer_failure(peer_id).await; return Err(P2PError::Network(NetworkError::PeerNotFound( - format!("failed to dial {} at {}", peer_hex, address).into(), + format!( + "failed to dial {} at any of {} candidate address(es)", + peer_hex, + candidate_addresses.len() + ) + .into(), ))); } } @@ -1590,7 +1706,8 @@ impl DhtNetworkManager { let pid_bytes = *peer_id.to_bytes(); info!( "dial_candidate: setting hole_punch_target_peer_id for {} = {}", - socket_addr, hex::encode(&pid_bytes[..8]) + socket_addr, + hex::encode(&pid_bytes[..8]) ); self.transport .set_hole_punch_target_peer_id(socket_addr, pid_bytes) diff --git a/src/network.rs b/src/network.rs index 434e336e..33b1d154 100644 --- a/src/network.rs +++ b/src/network.rs @@ -133,7 +133,16 @@ const DEFAULT_CONNECTION_TIMEOUT_SECS: u64 = 25; const BOOTSTRAP_PEER_BATCH_SIZE: usize = 20; /// Timeout in seconds for waiting on a bootstrap peer's identity exchange. -const BOOTSTRAP_IDENTITY_TIMEOUT_SECS: u64 = 5; +/// +/// Identity exchange is two RTTs over a freshly-handshaken QUIC connection +/// plus an ML-DSA-65 signature verification. On a LAN this completes in +/// well under a second; on congested cellular or cross-region links it can +/// blow past 5s with retransmits. The previous 5s default fired +/// spuriously on slow networks during testnet validation, forcing +/// reconnect loops that masqueraded as NAT traversal failures, so we +/// budget enough headroom for two QUIC handshake retries on a high-latency +/// link. +const BOOTSTRAP_IDENTITY_TIMEOUT_SECS: u64 = 15; /// Serde helper — returns `true`. const fn default_true() -> bool { @@ -1120,51 +1129,71 @@ impl P2PNode { self.connect_bootstrap_peers(close_group_cache.as_ref()) .await?; - // Spawn background task to forward peer address updates to DHT. - // When a connected peer advertises a new address (e.g., relay), update - // the DHT routing table so future lookups return the new address. + // Spawn background task to forward peer address updates to the DHT. + // + // Two event streams are bridged from the transport layer onto DHT + // routing-table mutations: + // + // - **Relay established**: when THIS node sets up a MASQUE relay, + // perform a DHT self-lookup so the transport's re-advertisement + // loop can ADD_ADDRESS the new relay address to the K closest + // peers — propagating it beyond peers we already happen to be + // connected to. + // - **Peer address update**: when a connected peer advertises a new + // reachable address via ADD_ADDRESS (typically its relay), update + // the DHT routing table so future lookups return that address. // - // Also handles RelayEstablished events: when THIS node sets up a relay, - // it does a DHT self-lookup to connect to K closest peers. The transport - // layer's re-advertisement loop then sends ADD_ADDRESS to those peers, - // propagating the relay address beyond the initial direct connections. + // Both are handled in a `tokio::select!` against the receiver + // futures so updates propagate immediately. The previous + // implementation polled both queues on a 1-second interval, which + // opened a race window in which a freshly-established relay was + // invisible to outbound DHT queries until the next tick — causing + // the first peers to dial direct (and fail) before learning about + // the relay. + // + // **Slow work isolation**: the relay-propagation path runs an + // iterative DHT lookup (`find_closest_nodes_network`) which can + // take many seconds. Doing it inline in the select loop would + // starve the peer-address-update branch and back up the bounded + // forwarder mpsc into drop territory. Instead, the lookup + + // publish is detached into its own task per relay event, so the + // select loop keeps polling both branches. { let transport = Arc::clone(&self.transport); let dht = self.adaptive_dht.dht_manager().clone(); let shutdown = self.shutdown.clone(); tokio::spawn(async move { - let mut interval = tokio::time::interval(std::time::Duration::from_secs(1)); loop { tokio::select! { + biased; _ = shutdown.cancelled() => break, - _ = interval.tick() => { - // Check for relay established events. - // When this node sets up a relay, do a self-lookup to - // connect to K closest peers for relay address propagation. - if let Some(relay_addr) = transport.drain_relay_established().await { - // Normalize IPv6-mapped addresses to IPv4 so the - // published address is dialable by IPv4-only clients. - let normalized = saorsa_transport::shared::normalize_socket_addr(relay_addr); - let relay_multi = - crate::MultiAddr::quic(normalized); - info!( - "DHT_BRIDGE: relay established at {} — self-lookup then PublishAddress to K closest", - relay_addr - ); - let own_key = *dht.peer_id().to_bytes(); - // Self-lookup to discover K closest peers, then - // send PublishAddress to each. This is O(K) messages - // sent once, not a field on every DHT message. - match dht.find_closest_nodes_network(&own_key, dht.k_value()).await { + relay = transport.recv_relay_established() => { + let Some(relay_addr) = relay else { break }; + // Normalize IPv6-mapped addresses to IPv4 so the + // published address is dialable by IPv4-only clients. + let normalized = saorsa_transport::shared::normalize_socket_addr(relay_addr); + let relay_multi = crate::MultiAddr::quic(normalized); + info!( + "DHT_BRIDGE: relay established at {} — spawning self-lookup + PublishAddress", + relay_addr + ); + // Detach the slow work so the select loop is + // free to keep polling peer-address updates. + let dht_for_propagation = dht.clone(); + tokio::spawn(async move { + let own_key = *dht_for_propagation.peer_id().to_bytes(); + match dht_for_propagation + .find_closest_nodes_network(&own_key, dht_for_propagation.k_value()) + .await + { Ok(nodes) => { info!( "DHT_BRIDGE: self-lookup found {} nodes — sending PublishAddress", nodes.len() ); - dht.publish_address_to_peers( - vec![relay_multi], - &nodes, - ).await; + dht_for_propagation + .publish_address_to_peers(vec![relay_multi], &nodes) + .await; } Err(e) => { warn!( @@ -1173,44 +1202,41 @@ impl P2PNode { ); } } + }); + } + update = transport.recv_peer_address_update() => { + let Some((peer_addr, advertised_addr)) = update else { break }; + info!( + "DHT_BRIDGE: processing update peer={} addr={} same_ip={}", + peer_addr, + advertised_addr, + peer_addr.ip() == advertised_addr.ip() + ); + // Only update DHT when the advertised IP differs + // from the peer's connection IP. Same-IP updates + // are just different NATted ports (useless for + // symmetric NAT); different-IP means a relay. + if peer_addr.ip() == advertised_addr.ip() { + continue; } - - let updates = transport.drain_peer_address_updates().await; - if !updates.is_empty() { - info!("DHT_BRIDGE: drained {} address updates", updates.len()); - } - for (peer_addr, advertised_addr) in updates { + // Look up peer ID by address (tries both IPv4 and + // IPv4-mapped IPv6 forms via dual_stack_alternate). + // For symmetric NAT, this may fail because the + // connection's channel key uses a different NATted port. + if let Some(peer_id) = transport.peer_id_for_addr(&peer_addr).await { + let normalized_adv = + saorsa_transport::shared::normalize_socket_addr(advertised_addr); + let multi_addr = crate::MultiAddr::quic(normalized_adv); info!( - "DHT_BRIDGE: processing update peer={} addr={} same_ip={}", - peer_addr, advertised_addr, peer_addr.ip() == advertised_addr.ip() + "Updating DHT: peer {} relay address {} (connection was {})", + peer_id, advertised_addr, peer_addr ); - // Only update DHT when the advertised IP differs - // from the peer's connection IP. Same-IP updates - // are just different NATted ports (useless for - // symmetric NAT); different-IP means a relay. - if peer_addr.ip() == advertised_addr.ip() { - continue; - } - - // Look up peer ID by address (tries both IPv4 and - // IPv4-mapped IPv6 forms via dual_stack_alternate). - // For symmetric NAT, this may fail because the - // connection's channel key uses a different NATted port. - let peer_id = transport.peer_id_for_addr(&peer_addr).await; - if let Some(peer_id) = peer_id { - let normalized_adv = saorsa_transport::shared::normalize_socket_addr(advertised_addr); - let multi_addr = crate::MultiAddr::quic(normalized_adv); - info!( - "Updating DHT: peer {} relay address {} (connection was {})", - peer_id, advertised_addr, peer_addr - ); - dht.touch_node_typed( - &peer_id, - Some(&multi_addr), - crate::dht::AddressType::Relay, - ) - .await; - } + dht.touch_node_typed( + &peer_id, + Some(&multi_addr), + crate::dht::AddressType::Relay, + ) + .await; } } } diff --git a/src/transport.rs b/src/transport.rs index 00514653..4e95ac0a 100644 --- a/src/transport.rs +++ b/src/transport.rs @@ -22,3 +22,8 @@ pub mod saorsa_transport_adapter; // DHT protocol handler for SharedTransport integration pub mod dht_handler; + +// Observed-address cache: records `ExternalAddressDiscovered` events from the +// transport layer and serves as a frequency- and recency-aware fallback when +// no live connection has an observation. +pub(crate) mod observed_address_cache; diff --git a/src/transport/observed_address_cache.rs b/src/transport/observed_address_cache.rs new file mode 100644 index 00000000..69fd4435 --- /dev/null +++ b/src/transport/observed_address_cache.rs @@ -0,0 +1,585 @@ +// Copyright 2024 Saorsa Labs Limited +// +// This software is dual-licensed under: +// - GNU Affero General Public License v3.0 or later (AGPL-3.0-or-later) +// - Commercial License +// +// For AGPL-3.0 license, see LICENSE-AGPL-3.0 +// For commercial licensing, contact: david@saorsalabs.com +// +// Unless required by applicable law or agreed to in writing, software +// distributed under these licenses is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + +//! Observed-address cache for transport-level reflexive address fallback. +//! +//! ## Why this exists +//! +//! `saorsa-transport` exposes the node's externally-observed address (from +//! QUIC `OBSERVED_ADDRESS` frames) only via *active* connections — when every +//! connection drops, the live read returns `None` and the node has no way to +//! tell the DHT how to be reached. The result is a temporary "invisible +//! window" between the connection drop and the next reconnection. +//! +//! This cache fills that gap. It records every `ExternalAddressDiscovered` +//! event the transport emits and serves as a fallback when no live connection +//! has an observation. +//! +//! ## Per-local-bind partitioning (multi-homed safety) +//! +//! On a multi-homed host, different local interfaces (LAN, cellular, WAN +//! uplink, dual-stack v4/v6 binds) can receive different observations from +//! different sets of peers. An observation seen on the cellular interface is +//! not necessarily reachable from peers that connect via the LAN, and vice +//! versa. Mixing them in one keyspace would let a stale observation from +//! one interface be served as the self-entry advertisement when only a +//! different interface is currently usable. +//! +//! The cache therefore keys observations by **`(local_bind, observed)`**. +//! Selection within a local bind is independent of every other local bind: +//! [`Self::most_frequent_recent_per_local_bind`] returns one best address +//! per bind that has any data, so the caller can publish all of them. The +//! single-address [`Self::most_frequent_recent`] accessor remains for +//! callers that only want one (it picks the global best across binds with +//! the same recency-and-frequency rule). +//! +//! ## Frequency-based selection +//! +//! Different peers can legitimately observe a node at different addresses +//! (symmetric NAT, multi-homed hosts, dual-stack divergence). The cache +//! tracks how many distinct events have been received for each address and +//! returns the one with the highest count, breaking ties by recency. The +//! intuition: "the address most peers agree on" is the most likely to be +//! reachable from any new peer. +//! +//! ## Recency window for NAT-rebinding handling +//! +//! Pure frequency would let a long-lived stale address (count: 10000, last +//! seen 24h ago) win over a fresh new address (count: 5, last seen now). +//! That is the wrong answer when a NAT mapping has rebinded. +//! +//! Selection is therefore split into two passes: +//! +//! 1. Among entries observed within [`OBSERVATION_RECENCY_WINDOW`], return +//! the highest-count one (with `last_seen` as the tiebreaker). +//! 2. If nothing is recent, fall back to the global highest-count entry — +//! handles the case where the node has been quiet for longer than the +//! recency window. +//! +//! Eviction is also recency-based: when the cache is full, the entry with +//! the *oldest* `last_seen` is removed. This ensures stale high-count +//! entries get pushed out as fresh observations arrive. +//! +//! ## Bounded +//! +//! The cache is bounded at [`MAX_CACHED_OBSERVATIONS`] entries to keep +//! memory predictable. The bound is chosen to comfortably handle: +//! +//! - Dual-stack (IPv4 + IPv6) observations of the same node +//! - Symmetric-NAT divergence (different external port per peer) +//! - A handful of recent NAT rebindings during the recency window +//! +//! ## Persistence +//! +//! The cache is in-memory only. A node restart resets it. This is +//! intentional: a freshly-started node should re-discover its current +//! address from live connections rather than trusting potentially-stale +//! state from a previous run. + +use std::collections::HashMap; +use std::net::SocketAddr; +use std::time::{Duration, Instant}; + +/// Maximum number of distinct observed addresses retained in the cache. +/// +/// Bounds memory and protects against pathological cases (a buggy peer +/// reporting random addresses). Sized to fit normal operating conditions: +/// dual-stack + symmetric-NAT divergence + a couple of recent rebindings. +pub(crate) const MAX_CACHED_OBSERVATIONS: usize = 16; + +/// Time window during which an observation counts as "recent". +/// +/// Within this window, selection prefers the highest-count entry. Beyond +/// it, the cache treats observations as stale candidates that only matter +/// if nothing recent exists. +/// +/// 10 minutes is long enough to absorb a normal disconnect+reconnect cycle +/// (typically seconds to a minute) and short enough that a NAT rebinding +/// is reflected in the selection within ~10 min, even if the stale address +/// still wins on raw count. +pub(crate) const OBSERVATION_RECENCY_WINDOW: Duration = Duration::from_secs(600); + +/// Per-address bookkeeping inside [`ObservedAddressCache`]. +#[derive(Debug, Clone, Copy)] +struct ObservedEntry { + /// Cumulative count of `ExternalAddressDiscovered` events received for + /// this address. Each (peer, address) pair contributes at most once, + /// per saorsa-transport's own dedup, so this is effectively a count of + /// distinct peers that have agreed on this address. + count: u64, + /// The most recent instant we received an event for this address. + /// Used both for recency-based selection and for LRU eviction. + last_seen: Instant, +} + +/// Composite cache key: an observed external address is always associated +/// with the **local bind** that received it. Two different local interfaces +/// (e.g. v4 and v6 stacks, or LAN and WAN) recording the same observed +/// address get separate entries so their counts and recencies do not +/// cross-contaminate. +type CacheKey = (SocketAddr, SocketAddr); + +/// Bounded cache of observed external addresses with frequency- and +/// recency-aware selection. See module-level docs for the rationale. +#[derive(Debug, Default)] +pub(crate) struct ObservedAddressCache { + entries: HashMap, +} + +impl ObservedAddressCache { + /// Create an empty cache. + pub(crate) fn new() -> Self { + Self { + entries: HashMap::new(), + } + } + + /// Record an observation of `observed` received via `local_bind`. + /// Increments the count for an existing entry or inserts a new one, + /// evicting the oldest entry by `last_seen` if the cache is full. + pub(crate) fn record(&mut self, local_bind: SocketAddr, observed: SocketAddr) { + self.record_at(local_bind, observed, Instant::now()); + } + + /// Record an observation at a caller-provided instant. Exposed for + /// deterministic unit tests; production callers should use [`record`]. + pub(crate) fn record_at(&mut self, local_bind: SocketAddr, observed: SocketAddr, now: Instant) { + let key = (local_bind, observed); + if let Some(entry) = self.entries.get_mut(&key) { + entry.count = entry.count.saturating_add(1); + entry.last_seen = now; + return; + } + + if self.entries.len() >= MAX_CACHED_OBSERVATIONS { + self.evict_oldest(); + } + + self.entries.insert( + key, + ObservedEntry { + count: 1, + last_seen: now, + }, + ); + } + + /// Return one observed address per **local bind** that has at least + /// one cached entry, picking the highest-count recent observation for + /// each bind. Multi-homed callers should publish all addresses + /// returned here so peers reaching the node via *any* interface can + /// dial it. + /// + /// Within a local bind, selection follows the same recency-and- + /// frequency algorithm as [`Self::most_frequent_recent`]: prefer + /// entries inside [`OBSERVATION_RECENCY_WINDOW`], fall back to the + /// highest-count overall if nothing is recent. + pub(crate) fn most_frequent_recent_per_local_bind(&self) -> Vec { + self.most_frequent_recent_per_local_bind_at(Instant::now()) + } + + /// Selection at a caller-provided "now". Exposed for deterministic + /// unit tests; production callers should use the non-`_at` variant. + pub(crate) fn most_frequent_recent_per_local_bind_at(&self, now: Instant) -> Vec { + // Collect distinct local binds, preserving deterministic order + // for callers that may iterate the result. We sort by the local + // bind so the output is reproducible across runs. + let mut binds: Vec = self.entries.keys().map(|(bind, _)| *bind).collect(); + binds.sort(); + binds.dedup(); + + let mut result = Vec::with_capacity(binds.len()); + for bind in binds { + if let Some(addr) = self.best_observed_for_bind_at(bind, now) { + result.push(addr); + } + } + result + } + + /// Best observed address for a single local bind, applying the + /// recent-then-fallback selection rule. + fn best_observed_for_bind_at( + &self, + local_bind: SocketAddr, + now: Instant, + ) -> Option { + let recent = self + .entries + .iter() + .filter(|((bind, _), _)| *bind == local_bind) + .filter(|(_, e)| now.duration_since(e.last_seen) <= OBSERVATION_RECENCY_WINDOW) + .max_by_key(|(_, e)| (e.count, e.last_seen)) + .map(|((_, observed), _)| *observed); + + if recent.is_some() { + return recent; + } + + self.entries + .iter() + .filter(|((bind, _), _)| *bind == local_bind) + .max_by_key(|(_, e)| (e.count, e.last_seen)) + .map(|((_, observed), _)| *observed) + } + + /// Return the **single** address with the highest observation count + /// among entries seen within [`OBSERVATION_RECENCY_WINDOW`], breaking + /// ties by most recent `last_seen`. If no entry is recent, fall back + /// to the highest count overall. + /// + /// This crosses local-bind boundaries — it is the right answer for + /// callers that only want a single address (single-interface hosts, + /// legacy callers). Multi-homed callers should prefer + /// [`Self::most_frequent_recent_per_local_bind`] instead. + pub(crate) fn most_frequent_recent(&self) -> Option { + self.most_frequent_recent_at(Instant::now()) + } + + /// Selection at a caller-provided "now". Exposed for deterministic + /// unit tests; production callers should use [`most_frequent_recent`]. + pub(crate) fn most_frequent_recent_at(&self, now: Instant) -> Option { + let recent = self + .entries + .iter() + .filter(|(_, e)| now.duration_since(e.last_seen) <= OBSERVATION_RECENCY_WINDOW) + .max_by_key(|(_, e)| (e.count, e.last_seen)) + .map(|((_, observed), _)| *observed); + + if recent.is_some() { + return recent; + } + + self.entries + .iter() + .max_by_key(|(_, e)| (e.count, e.last_seen)) + .map(|((_, observed), _)| *observed) + } + + /// Evict the entry with the oldest `last_seen`. No-op on an empty cache. + fn evict_oldest(&mut self) { + let oldest = self + .entries + .iter() + .min_by_key(|(_, e)| e.last_seen) + .map(|(key, _)| *key); + if let Some(key) = oldest { + self.entries.remove(&key); + } + } + + /// Number of distinct addresses currently cached. Exposed for tests + /// and diagnostics. + #[cfg(test)] + pub(crate) fn len(&self) -> usize { + self.entries.len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::{IpAddr, Ipv4Addr}; + + /// Default local bind used by tests that only care about a single + /// interface (the most common case). + const DEFAULT_LOCAL_BIND_PORT: u16 = 7000; + /// Alternate local bind for multi-homed partitioning tests. + const ALT_LOCAL_BIND_PORT: u16 = 7001; + + /// Construct a unique IPv4 socket address for tests. + fn addr(last_octet: u8, port: u16) -> SocketAddr { + SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 0, 2, last_octet)), port) + } + + /// Default local-bind socket used by single-interface tests. + fn default_bind() -> SocketAddr { + SocketAddr::new( + IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), + DEFAULT_LOCAL_BIND_PORT, + ) + } + + /// Alternate local-bind socket used by multi-homed partitioning tests. + fn alt_bind() -> SocketAddr { + SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), ALT_LOCAL_BIND_PORT) + } + + #[test] + fn empty_cache_returns_none() { + let cache = ObservedAddressCache::new(); + assert_eq!(cache.most_frequent_recent(), None); + assert!(cache.most_frequent_recent_per_local_bind().is_empty()); + } + + #[test] + fn single_observation_returns_that_address() { + let mut cache = ObservedAddressCache::new(); + let a = addr(1, 9000); + cache.record(default_bind(), a); + assert_eq!(cache.most_frequent_recent(), Some(a)); + assert_eq!(cache.most_frequent_recent_per_local_bind(), vec![a]); + assert_eq!(cache.len(), 1); + } + + #[test] + fn repeated_observation_increments_count_without_growing() { + let mut cache = ObservedAddressCache::new(); + let a = addr(1, 9000); + cache.record(default_bind(), a); + cache.record(default_bind(), a); + cache.record(default_bind(), a); + assert_eq!(cache.len(), 1); + assert_eq!(cache.most_frequent_recent(), Some(a)); + } + + #[test] + fn higher_count_wins_among_recent_entries() { + let mut cache = ObservedAddressCache::new(); + let popular = addr(1, 9000); + let unpopular = addr(2, 9000); + + // popular: 5 observations, unpopular: 1 observation, all recent. + for _ in 0..5 { + cache.record(default_bind(), popular); + } + cache.record(default_bind(), unpopular); + + assert_eq!(cache.most_frequent_recent(), Some(popular)); + } + + #[test] + fn equal_counts_break_tie_by_recency() { + let mut cache = ObservedAddressCache::new(); + let older = addr(1, 9000); + let newer = addr(2, 9000); + + let base = Instant::now(); + cache.record_at(default_bind(), older, base); + cache.record_at(default_bind(), newer, base + Duration::from_secs(1)); + + assert_eq!( + cache.most_frequent_recent_at(base + Duration::from_secs(2)), + Some(newer) + ); + } + + #[test] + fn stale_high_count_loses_to_recent_low_count() { + // The NAT-rebinding scenario: an old address has a huge count from + // a long session, but a new address has just started accumulating + // observations after the rebind. The cache should prefer the new one + // because the old one is outside the recency window. + let mut cache = ObservedAddressCache::new(); + let stale = addr(1, 9000); + let fresh = addr(2, 9000); + + let base = Instant::now(); + + // 1000 observations of `stale`, all well outside the recency window. + let stale_time = base; + for _ in 0..1000 { + cache.record_at(default_bind(), stale, stale_time); + } + + // 3 observations of `fresh`, all just now. + let fresh_time = base + OBSERVATION_RECENCY_WINDOW + Duration::from_secs(60); + for _ in 0..3 { + cache.record_at(default_bind(), fresh, fresh_time); + } + + let now = fresh_time + Duration::from_secs(1); + assert_eq!(cache.most_frequent_recent_at(now), Some(fresh)); + } + + #[test] + fn falls_back_to_global_highest_count_when_nothing_is_recent() { + // Long-quiet network case: the node has been silent for longer than + // the recency window, so the recent-pass returns nothing. The + // fallback returns the highest-count address overall so the node + // can still publish *something*. + let mut cache = ObservedAddressCache::new(); + let popular = addr(1, 9000); + let unpopular = addr(2, 9000); + + let base = Instant::now(); + for _ in 0..5 { + cache.record_at(default_bind(), popular, base); + } + cache.record_at(default_bind(), unpopular, base); + + // Far in the future — every entry is stale, fallback path engages. + let far_future = base + OBSERVATION_RECENCY_WINDOW * 10; + assert_eq!(cache.most_frequent_recent_at(far_future), Some(popular)); + } + + #[test] + fn eviction_removes_oldest_by_last_seen_when_full() { + let mut cache = ObservedAddressCache::new(); + let base = Instant::now(); + + // Fill the cache with MAX entries, each at a distinct time. + for i in 0..(MAX_CACHED_OBSERVATIONS as u8) { + cache.record_at( + default_bind(), + addr(i + 1, 9000), + base + Duration::from_secs(u64::from(i)), + ); + } + assert_eq!(cache.len(), MAX_CACHED_OBSERVATIONS); + + // The oldest entry is the one inserted at `base` (i = 0, addr=1). + let oldest_key = (default_bind(), addr(1, 9000)); + assert!(cache.entries.contains_key(&oldest_key)); + + // Insert one more — should evict the oldest. + let newcomer_key = (default_bind(), addr(99, 9000)); + cache.record_at( + newcomer_key.0, + newcomer_key.1, + base + Duration::from_secs(MAX_CACHED_OBSERVATIONS as u64), + ); + + assert_eq!(cache.len(), MAX_CACHED_OBSERVATIONS); + assert!( + !cache.entries.contains_key(&oldest_key), + "oldest entry should have been evicted" + ); + assert!( + cache.entries.contains_key(&newcomer_key), + "newcomer should be present" + ); + } + + #[test] + fn re_observing_an_existing_entry_does_not_trigger_eviction() { + // If we record an address that's already in the cache, we just + // bump its count and last_seen — no eviction needed even when the + // cache is full. + let mut cache = ObservedAddressCache::new(); + let base = Instant::now(); + + for i in 0..(MAX_CACHED_OBSERVATIONS as u8) { + cache.record_at( + default_bind(), + addr(i + 1, 9000), + base + Duration::from_secs(u64::from(i)), + ); + } + assert_eq!(cache.len(), MAX_CACHED_OBSERVATIONS); + + // Re-observe the oldest entry, refreshing its last_seen. + let oldest_key = (default_bind(), addr(1, 9000)); + let refresh_time = base + Duration::from_secs(1000); + cache.record_at(oldest_key.0, oldest_key.1, refresh_time); + + // Cache size unchanged; the entry is now the youngest. + assert_eq!(cache.len(), MAX_CACHED_OBSERVATIONS); + let entry = cache.entries.get(&oldest_key).copied().unwrap(); + assert_eq!(entry.count, 2); + assert_eq!(entry.last_seen, refresh_time); + } + + #[test] + fn observations_for_different_local_binds_do_not_collide() { + // Two different local interfaces independently observe the SAME + // external address. They must remain as separate entries so the + // counts and recencies of one cannot leak into the other. + let mut cache = ObservedAddressCache::new(); + let observed = addr(1, 9000); + + cache.record(default_bind(), observed); + cache.record(alt_bind(), observed); + cache.record(alt_bind(), observed); + + assert_eq!(cache.len(), 2); + + // Each bind tracks its own count. + let default_entry = cache.entries.get(&(default_bind(), observed)).unwrap(); + let alt_entry = cache.entries.get(&(alt_bind(), observed)).unwrap(); + assert_eq!(default_entry.count, 1); + assert_eq!(alt_entry.count, 2); + } + + #[test] + fn per_local_bind_returns_one_address_per_distinct_bind() { + // A multi-homed host with two interfaces observing two distinct + // external addresses (one per interface). The plural API must + // return both so the caller can publish all of them. + let mut cache = ObservedAddressCache::new(); + let observed_default = addr(1, 9000); + let observed_alt = addr(2, 9000); + + cache.record(default_bind(), observed_default); + cache.record(alt_bind(), observed_alt); + + let mut result = cache.most_frequent_recent_per_local_bind(); + result.sort(); + let mut expected = vec![observed_default, observed_alt]; + expected.sort(); + assert_eq!(result, expected); + } + + #[test] + fn per_local_bind_picks_best_within_each_bind_independently() { + // For each local bind, the picked address must be the best + // observation for THAT bind, not the global best. + let mut cache = ObservedAddressCache::new(); + let default_winner = addr(1, 9000); + let default_loser = addr(2, 9000); + let alt_winner = addr(3, 9000); + let alt_loser = addr(4, 9000); + + // default_bind: default_winner has 5 observations, default_loser has 1. + for _ in 0..5 { + cache.record(default_bind(), default_winner); + } + cache.record(default_bind(), default_loser); + + // alt_bind: alt_winner has 3 observations, alt_loser has 1. + for _ in 0..3 { + cache.record(alt_bind(), alt_winner); + } + cache.record(alt_bind(), alt_loser); + + let mut result = cache.most_frequent_recent_per_local_bind(); + result.sort(); + let mut expected = vec![default_winner, alt_winner]; + expected.sort(); + assert_eq!(result, expected); + } + + #[test] + fn stale_observation_on_one_bind_does_not_affect_recency_on_another() { + // The multi-homed correctness scenario: bind A has only stale data + // (outside the recency window) while bind B has fresh data. The + // partitioning means each bind's selection runs independently — + // bind A correctly falls back to its global pick, bind B uses + // its recent pick. + let mut cache = ObservedAddressCache::new(); + let stale_for_default = addr(1, 9000); + let fresh_for_alt = addr(2, 9000); + + let base = Instant::now(); + cache.record_at(default_bind(), stale_for_default, base); + let fresh_time = base + OBSERVATION_RECENCY_WINDOW + Duration::from_secs(60); + cache.record_at(alt_bind(), fresh_for_alt, fresh_time); + + let now = fresh_time + Duration::from_secs(1); + let mut result = cache.most_frequent_recent_per_local_bind_at(now); + result.sort(); + let mut expected = vec![stale_for_default, fresh_for_alt]; + expected.sort(); + assert_eq!(result, expected); + } +} diff --git a/src/transport/saorsa_transport_adapter.rs b/src/transport/saorsa_transport_adapter.rs index b9751f27..2040a554 100644 --- a/src/transport/saorsa_transport_adapter.rs +++ b/src/transport/saorsa_transport_adapter.rs @@ -39,15 +39,17 @@ //! automatically enables saorsa-transport's prometheus metrics collection. use crate::error::{GeoRejectionError, GeographicConfig}; +use crate::transport::observed_address_cache::ObservedAddressCache; use anyhow::{Context, Result}; use std::collections::HashMap; use std::net::{IpAddr, SocketAddr, SocketAddrV6}; use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; use tokio::sync::{RwLock, broadcast}; use tokio::time::sleep; use tokio_util::sync::CancellationToken; -use tracing::{debug, info, trace}; +use tracing::{debug, info, trace, warn}; // Import saorsa-transport types using the new LinkTransport API (0.14+) use saorsa_transport::{ @@ -125,6 +127,51 @@ pub struct P2PNetworkNode { /// explicitly configured. pub const DEFAULT_MAX_CONNECTIONS: usize = 100; +/// Bounded capacity for the relay/peer-address forwarder mpsc channels. +/// +/// Replaces the previous `unbounded_channel` so a slow consumer (e.g. the +/// DHT bridge while running an iterative lookup) cannot grow the queue +/// without limit. When the channel is full the forwarder logs and drops +/// the event rather than blocking the receive loop, so we still keep +/// processing newer events. +pub const ADDRESS_EVENT_CHANNEL_CAPACITY: usize = 256; + +/// Log a warning every Nth dropped address-event in the forwarder. +/// +/// `try_send` failures (channel full) increment a counter; logging at +/// every drop would flood the log under sustained pressure, so we +/// coalesce to one warning per `ADDRESS_EVENT_DROP_LOG_INTERVAL` drops. +const ADDRESS_EVENT_DROP_LOG_INTERVAL: u64 = 32; + +/// Increment the drop counter and log periodically when the address-event +/// forwarder fails to push into a bounded channel. +/// +/// Used by the forwarder loop in +/// [`DualStackNetworkNode::spawn_peer_address_update_forwarder`] when the +/// downstream consumer is too slow to drain. Drops are coalesced to one +/// warning per [`ADDRESS_EVENT_DROP_LOG_INTERVAL`] events to avoid log +/// floods under sustained backpressure; the very first drop in any burst +/// is always logged so operators see the onset. +fn handle_address_event_drop( + counter: &AtomicU64, + event_kind: &'static str, + err: &tokio::sync::mpsc::error::TrySendError, +) { + let prev = counter.fetch_add(1, Ordering::Relaxed); + let kind = match err { + tokio::sync::mpsc::error::TrySendError::Full(_) => "channel full", + tokio::sync::mpsc::error::TrySendError::Closed(_) => "consumer closed", + }; + if prev.is_multiple_of(ADDRESS_EVENT_DROP_LOG_INTERVAL) { + tracing::warn!( + event = event_kind, + reason = kind, + total_drops = prev + 1, + "ADDR_FWD: dropped address event" + ); + } +} + #[allow(dead_code)] impl P2PNetworkNode { /// Create a new P2P network node with default P2pLinkTransport @@ -203,17 +250,21 @@ impl P2PNetworkNode { /// This method is specialized for P2pLinkTransport and uses the underlying /// P2pEndpoint's send() method which corresponds with recv() for proper /// bidirectional communication. + /// + /// On failure the underlying transport error is preserved via + /// `anyhow::Context` so callers can inspect the cause (e.g. QUIC + /// `peer did not acknowledge`, `open_uni failed`, `PeerNotFound`). pub async fn send_to_peer_optimized(&self, addr: &SocketAddr, data: &[u8]) -> Result<()> { trace!( "[QUIC SEND] endpoint().send() to {} ({} bytes)", addr, data.len() ); - let result = self.transport.endpoint().send(addr, data).await; - if let Err(ref e) = result { - debug!("[QUIC SEND] send failed to {}: {}", addr, e); - } - result.map_err(|e| anyhow::anyhow!("Send failed: {e}")) + self.transport + .endpoint() + .send(addr, data) + .await + .with_context(|| format!("QUIC send to {} ({} bytes) failed", addr, data.len())) } /// Disconnect a specific peer, closing the underlying QUIC connection. @@ -882,26 +933,51 @@ impl DualStackNetworkNode { } } - /// Spawn background tasks that forward `P2pEvent::PeerAddressUpdated` - /// from each stack's `P2pEndpoint` into a channel. + /// Spawn background tasks that forward address-related `P2pEvent`s from + /// each stack's `P2pEndpoint` to the upper layers. + /// + /// Three event flavours are bridged: /// - /// Call after construction. The returned receiver yields - /// `(peer_addr, advertised_addr)` pairs when any connected peer - /// advertises a new address via ADD_ADDRESS frames. + /// - **`PeerAddressUpdated`**: a connected peer advertised a new + /// reachable address via an ADD_ADDRESS frame (typically a relay). + /// Returned via the first mpsc receiver as + /// `(peer_connection_addr, advertised_addr)`. + /// - **`RelayEstablished`**: this node set up a MASQUE relay and now + /// needs to publish the relay address to the K closest peers. + /// Returned via the second mpsc receiver. + /// - **`ExternalAddressDiscovered`**: a peer reported the address it + /// sees this node at, via a QUIC `OBSERVED_ADDRESS` frame. Recorded + /// directly into the supplied [`ObservedAddressCache`] so the + /// transport layer can fall back to it when no live connection has an + /// observation. See the cache module for the frequency- and + /// recency-aware selection algorithm. + /// + /// Other `P2pEvent` variants are not consumed by saorsa-core and are + /// silently ignored. pub fn spawn_peer_address_update_forwarder( &self, + observed_cache: Arc>, ) -> ( - tokio::sync::mpsc::UnboundedReceiver<(SocketAddr, SocketAddr)>, - tokio::sync::mpsc::UnboundedReceiver, + tokio::sync::mpsc::Receiver<(SocketAddr, SocketAddr)>, + tokio::sync::mpsc::Receiver, ) { - let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); - let (relay_tx, relay_rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, rx) = tokio::sync::mpsc::channel(ADDRESS_EVENT_CHANNEL_CAPACITY); + let (relay_tx, relay_rx) = tokio::sync::mpsc::channel(ADDRESS_EVENT_CHANNEL_CAPACITY); + let drop_counter = Arc::new(AtomicU64::new(0)); for node in [&self.v6, &self.v4].into_iter().flatten() { let mut p2p_rx = node.transport.endpoint().subscribe(); let tx_clone = tx.clone(); let relay_tx_clone = relay_tx.clone(); + let cache_clone = Arc::clone(&observed_cache); + let drops = Arc::clone(&drop_counter); + // Capture which local bind owns this forwarder so the cache can + // partition observations by interface (multi-homed correctness). + let local_bind = node.local_address(); tokio::spawn(async move { - tracing::debug!("ADDR_FWD: peer address update forwarder started"); + tracing::debug!( + local_bind = %local_bind, + "ADDR_FWD: peer address update forwarder started" + ); loop { match p2p_rx.recv().await { Ok(saorsa_transport::P2pEvent::PeerAddressUpdated { @@ -913,17 +989,38 @@ impl DualStackNetworkNode { peer_addr, advertised_addr ); - let _ = tx_clone.send(( + let payload = ( saorsa_transport::shared::normalize_socket_addr(peer_addr), saorsa_transport::shared::normalize_socket_addr(advertised_addr), - )); + ); + if let Err(err) = tx_clone.try_send(payload) { + handle_address_event_drop(&drops, "PeerAddressUpdated", &err); + } } Ok(saorsa_transport::P2pEvent::RelayEstablished { relay_addr }) => { tracing::info!( "ADDR_FWD: received RelayEstablished relay_addr={}", relay_addr ); - let _ = relay_tx_clone.send(relay_addr); + if let Err(err) = relay_tx_clone.try_send(relay_addr) { + handle_address_event_drop(&drops, "RelayEstablished", &err); + } + } + Ok(saorsa_transport::P2pEvent::ExternalAddressDiscovered { addr }) => { + // Convert TransportAddr → SocketAddr for QUIC. + // Non-UDP transports (BLE, LoRa) yield None and + // are skipped — the cache only models routable + // IP addresses. + if let Some(socket_addr) = addr.as_socket_addr() { + let normalized = + saorsa_transport::shared::normalize_socket_addr(socket_addr); + tracing::debug!( + local_bind = %local_bind, + "ADDR_FWD: caching observed external address {}", + normalized + ); + cache_clone.lock().record(local_bind, normalized); + } } Err(broadcast::error::RecvError::Closed) => { tracing::info!("ADDR_FWD: channel closed, exiting"); @@ -934,7 +1031,13 @@ impl DualStackNetworkNode { continue; } Ok(_other) => { - tracing::trace!("ADDR_FWD: ignoring non-address P2pEvent"); + // Other P2pEvent variants (PeerConnected, + // PeerDisconnected, NatTraversalProgress, + // BootstrapStatus, PeerAuthenticated, + // DataReceived, …) are not consumed here. + // They are observed via other channels or are + // simply not relevant to saorsa-core. + continue; } } } @@ -1053,15 +1156,21 @@ impl DualStackNetworkNode { /// In dual-stack mode, converts plain IPv4 addresses to the mapped form /// expected by the v6 transport before sending. pub async fn send_to_peer_optimized(&self, addr: &SocketAddr, data: &[u8]) -> Result<()> { + // Preserve the underlying error(s) from the v6 and v4 stacks so the + // caller can surface them at WARN level. The previous implementation + // dropped both errors and returned a hardcoded + // "send_to_peer_optimized failed on both stacks" which made every + // transport failure look identical in the logs. + let mut v6_err: Option = None; + let mut v4_err: Option = None; + if let Some(v6) = &self.v6 { let wire_addr = self.to_mapped_if_needed(addr); match v6.send_to_peer_optimized(&wire_addr, data).await { Ok(()) => return Ok(()), Err(e) => { - trace!( - "[DUAL SEND] IPv6 failed to {}, falling back to IPv4: {}", - addr, e - ); + warn!("[DUAL SEND] IPv6 send to {} failed: {:#}", addr, e); + v6_err = Some(e); } } } @@ -1069,13 +1178,34 @@ impl DualStackNetworkNode { match v4.send_to_peer_optimized(addr, data).await { Ok(()) => return Ok(()), Err(e) => { - debug!("[DUAL SEND] IPv4 send failed to {}: {}", addr, e); + warn!("[DUAL SEND] IPv4 send to {} failed: {:#}", addr, e); + v4_err = Some(e); } } } - Err(anyhow::anyhow!( - "send_to_peer_optimized failed on both stacks" - )) + + // Produce a single error that preserves the full cause chain from + // whichever stack(s) were actually tried. In dual-stack-over-v6 mode + // (v4 is None) we don't lie about having tried v4. + let err = match (v6_err, v4_err) { + (Some(v6), Some(v4)) => v6.context(format!( + "send_to_peer_optimized to {} failed on both stacks (v4 cause: {:#})", + addr, v4 + )), + (Some(v6), None) => v6.context(format!( + "send_to_peer_optimized to {} failed (v6-only: no v4 stack bound)", + addr + )), + (None, Some(v4)) => v4.context(format!( + "send_to_peer_optimized to {} failed (v4-only: no v6 stack bound)", + addr + )), + (None, None) => anyhow::anyhow!( + "send_to_peer_optimized to {}: neither v6 nor v4 stack available", + addr + ), + }; + Err(err) } /// Disconnect a peer, closing the underlying QUIC connection. @@ -1418,6 +1548,27 @@ impl DualStackNetworkNode { }); raw.map(|a| self.normalize(a)) } + + /// Return observed external addresses for **every** stack that has one. + /// + /// Multi-homed publishing path: each stack (v4 / v6) is queried + /// independently and any address it reports is included in the + /// returned list (deduped, normalised). A multi-homed host that has + /// observations on both v4 and v6 will return both — `local_dht_node` + /// then publishes both so peers reaching the host on either family + /// can dial it. + pub fn get_observed_external_addresses(&self) -> Vec { + let mut out: Vec = Vec::new(); + for stack in [self.v4.as_ref(), self.v6.as_ref()].into_iter().flatten() { + if let Some(raw) = stack.get_observed_external_address() { + let normalized = self.normalize(raw); + if !out.contains(&normalized) { + out.push(normalized); + } + } + } + out + } } /// Normalise addresses in a `ConnectionEvent` (IPv4-mapped → plain IPv4). diff --git a/src/transport_handle.rs b/src/transport_handle.rs index 5ee99a7c..d683051d 100644 --- a/src/transport_handle.rs +++ b/src/transport_handle.rs @@ -28,6 +28,7 @@ use crate::network::{ RequestResponseEnvelope, WireMessage, broadcast_event, normalize_wildcard_to_loopback, parse_protocol_message, register_new_channel, }; +use crate::transport::observed_address_cache::ObservedAddressCache; use crate::transport::saorsa_transport_adapter::{ConnectionEvent, DualStackNetworkNode}; use crate::validation::{RateLimitConfig, RateLimiter}; @@ -117,10 +118,24 @@ pub struct TransportHandle { geo_provider: Arc, shutdown: CancellationToken, /// Peer address updates from ADD_ADDRESS frames (relay address advertisement). + /// + /// Bounded mpsc — see + /// [`crate::transport::saorsa_transport_adapter::ADDRESS_EVENT_CHANNEL_CAPACITY`]. + /// The producer (`spawn_peer_address_update_forwarder`) drops events + /// rather than blocking when the consumer is slow. peer_address_update_rx: - tokio::sync::Mutex>, + tokio::sync::Mutex>, /// Relay established events — received when this node sets up a MASQUE relay. - relay_established_rx: tokio::sync::Mutex>, + /// + /// Bounded mpsc with the same drop semantics as + /// `peer_address_update_rx`. + relay_established_rx: tokio::sync::Mutex>, + /// Frequency- and recency-aware cache of externally-observed addresses. + /// Populated by the address-update forwarder from + /// `P2pEvent::ExternalAddressDiscovered` frames; consulted as a fallback + /// by [`Self::observed_external_address`] when no live connection has + /// an observation. Survives connection drops; reset on process restart. + observed_address_cache: Arc>, connection_timeout: Duration, connection_monitor_handle: Arc>>>, recv_handles: Arc>>>, @@ -193,12 +208,20 @@ impl TransportHandle { let shutdown = CancellationToken::new(); - // Subscribe to P2pEvent::PeerAddressUpdated from the transport layer. - // These are forwarded from ADD_ADDRESS frames when a peer advertises - // a new reachable address (e.g., relay). The P2PNode reads these and - // updates the DHT routing table. + // Cache for externally-observed addresses. The forwarder spawned + // below feeds this cache from `P2pEvent::ExternalAddressDiscovered` + // events; the cache becomes the fallback for + // `observed_external_address()` when no live connection has an + // observation (see TransportHandle::observed_external_address). + let observed_address_cache = Arc::new(parking_lot::Mutex::new(ObservedAddressCache::new())); + + // Subscribe to address-related P2pEvents from the transport layer: + // - PeerAddressUpdated → mpsc, drained by the DHT bridge + // - RelayEstablished → mpsc, drained by the DHT bridge + // - ExternalAddressDiscovered → recorded directly into the + // observed-address cache above let (peer_addr_update_rx, relay_established_rx) = - dual_node.spawn_peer_address_update_forwarder(); + dual_node.spawn_peer_address_update_forwarder(Arc::clone(&observed_address_cache)); // Subscribe to connection events BEFORE spawning the monitor task let connection_event_rx = dual_node.subscribe_connection_events(); @@ -254,6 +277,7 @@ impl TransportHandle { shutdown, peer_address_update_rx: tokio::sync::Mutex::new(peer_addr_update_rx), relay_established_rx: tokio::sync::Mutex::new(relay_established_rx), + observed_address_cache, connection_timeout: config.connection_timeout, connection_monitor_handle, recv_handles: Arc::new(RwLock::new(Vec::new())), @@ -316,13 +340,18 @@ impl TransportHandle { geo_provider: Arc::new(BgpGeoProvider::new()), shutdown: CancellationToken::new(), peer_address_update_rx: { - let (_tx, rx) = tokio::sync::mpsc::unbounded_channel(); + let (_tx, rx) = tokio::sync::mpsc::channel( + crate::transport::saorsa_transport_adapter::ADDRESS_EVENT_CHANNEL_CAPACITY, + ); tokio::sync::Mutex::new(rx) }, relay_established_rx: { - let (_tx, rx) = tokio::sync::mpsc::unbounded_channel(); + let (_tx, rx) = tokio::sync::mpsc::channel( + crate::transport::saorsa_transport_adapter::ADDRESS_EVENT_CHANNEL_CAPACITY, + ); tokio::sync::Mutex::new(rx) }, + observed_address_cache: Arc::new(parking_lot::Mutex::new(ObservedAddressCache::new())), connection_timeout: Duration::from_secs(TEST_CONNECTION_TIMEOUT_SECS), connection_monitor_handle: Arc::new(RwLock::new(None)), recv_handles: Arc::new(RwLock::new(Vec::new())), @@ -364,6 +393,85 @@ impl TransportHandle { self.listen_addrs.read().await.clone() } + /// Returns the node's externally-observed address as reported by peers + /// (via QUIC `OBSERVED_ADDRESS` frames), or `None` if no peer has ever + /// observed this node since process start. + /// + /// This is the most authoritative source of the node's reflexive + /// (post-NAT) address — it is the address remote peers actually saw the + /// connection arrive from. Prefer it over `listen_addrs()` (which only + /// reflects locally-bound socket addresses) when advertising the node to + /// the rest of the network. + /// + /// ## Resolution order + /// + /// 1. **Live**: ask `dual_node.get_observed_external_address()` first. + /// This iterates currently-active connections and returns the + /// observation from the first one (preferring known/bootstrap peers + /// inside saorsa-transport). When at least one connection is up, + /// this is always the freshest answer. + /// 2. **Cache**: if no live connection has an observation (e.g. every + /// connection has just dropped during a network blip), fall back to + /// the in-memory [`ObservedAddressCache`]. The cache returns the + /// most-frequently-observed address among recent entries, breaking + /// ties by recency. See `observed_address_cache.rs` for the full + /// selection algorithm and rationale. + /// + /// The cache is populated by the `ExternalAddressDiscovered` forwarder + /// spawned in [`Self::new`]; it survives connection drops but is reset + /// on process restart. + pub fn observed_external_address(&self) -> Option { + // Prefer the plural accessor's first entry so the single-address + // path stays consistent with multi-homed publishing. + self.observed_external_addresses().into_iter().next() + } + + /// Return **all** externally-observed addresses for this node, one per + /// local interface that has an observation. + /// + /// Resolution order matches [`Self::observed_external_address`]: + /// + /// 1. **Live**: query each stack on `dual_node` independently (v4 and + /// v6) and collect any address it reports. + /// 2. **Cache fallback**: for each `(local_bind, observed)` partition + /// in the [`ObservedAddressCache`] that has no live observation + /// yet, append the cache's per-bind best. + /// + /// The returned list is deduped — if the live source and the cache + /// both report the same address, it appears only once. Order is not + /// part of the contract; callers that need a specific priority should + /// sort the result themselves. + /// + /// This is the right entry point for publishing the node's self-entry + /// to the DHT on a multi-homed host: peers reaching the node via any + /// interface in the returned list will be able to dial back. + pub fn observed_external_addresses(&self) -> Vec { + let mut out: Vec = self.dual_node.get_observed_external_addresses(); + let cached = self + .observed_address_cache + .lock() + .most_frequent_recent_per_local_bind(); + for addr in cached { + if !out.contains(&addr) { + out.push(addr); + } + } + out + } + + /// Returns the cache-only fallback for the observed external address, + /// bypassing the live `dual_node` read entirely. + /// + /// Production code should call [`Self::observed_external_address`] + /// instead — it prefers the live source and only consults the cache + /// when no live observation is available. This accessor exists so that + /// integration tests can poll for cache population without having to + /// race the periodic poll task in saorsa-transport that drives the + /// `ExternalAddressDiscovered` event stream. + pub fn cached_observed_external_address(&self) -> Option { + self.observed_address_cache.lock().most_frequent_recent() + } + /// Get the connection timeout duration. pub fn connection_timeout(&self) -> Duration { self.connection_timeout @@ -519,6 +627,31 @@ impl TransportHandle { rx.try_recv().ok() } + /// Wait for the next peer-address update from an ADD_ADDRESS frame. + /// + /// Returns `(peer_connection_addr, advertised_addr)` when one arrives, + /// or `None` if the underlying channel has closed (transport shut down). + /// + /// Use this in a `tokio::select!` against a shutdown token to react to + /// address updates immediately instead of polling. + pub async fn recv_peer_address_update(&self) -> Option<(SocketAddr, SocketAddr)> { + let mut rx = self.peer_address_update_rx.lock().await; + rx.recv().await + } + + /// Wait for the next relay-established event. + /// + /// Resolves when this node has just set up a MASQUE relay (yielding + /// the relay socket address), or `None` if the underlying channel has + /// closed (transport shut down). + /// + /// Use this in a `tokio::select!` against a shutdown token to react to + /// relay establishment immediately instead of polling. + pub async fn recv_relay_established(&self) -> Option { + let mut rx = self.relay_established_rx.lock().await; + rx.recv().await + } + /// Check if an authenticated peer is connected (has at least one active /// channel). pub async fn is_peer_connected(&self, peer_id: &PeerId) -> bool { @@ -580,7 +713,9 @@ impl TransportHandle { /// Set the target peer ID for a hole-punch attempt to a specific address. /// See [`P2pEndpoint::set_hole_punch_target_peer_id`]. pub async fn set_hole_punch_target_peer_id(&self, target: SocketAddr, peer_id: [u8; 32]) { - self.dual_node.set_hole_punch_target_peer_id(target, peer_id).await; + self.dual_node + .set_hole_punch_target_peer_id(target, peer_id) + .await; } /// Set a preferred coordinator for hole-punching to a specific target. diff --git a/tests/dht_self_advertisement.rs b/tests/dht_self_advertisement.rs new file mode 100644 index 00000000..31b7cb3a --- /dev/null +++ b/tests/dht_self_advertisement.rs @@ -0,0 +1,649 @@ +// Copyright 2024 Saorsa Labs Limited +// +// This software is dual-licensed under: +// - GNU Affero General Public License v3.0 or later (AGPL-3.0-or-later) +// - Commercial License +// +// For AGPL-3.0 license, see LICENSE-AGPL-3.0 +// For commercial licensing, contact: david@saorsalabs.com +// +// Unless required by applicable law or agreed to in writing, software +// distributed under these licenses is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + +//! Regression tests for the addresses a node publishes about itself in the DHT. +//! +//! These tests pin the contract that `DhtNetworkManager::local_dht_node()` is +//! the only place where a node decides what addresses to advertise about +//! itself, and that those addresses must always survive a receiving peer's +//! `dialable_addresses()` filter. +//! +//! The production failure mode this guards against is a node telling other +//! peers "you can reach me at ", which is silently +//! filtered out by every consumer's `dialable_addresses()` check and makes +//! the node invisible to DHT-based peer discovery. +//! +//! Concretely, the production path is: +//! +//! 1. Peer A queries the DHT for peer B. +//! 2. The DHT response contains B's self-entry, built by +//! `DhtNetworkManager::local_dht_node()`. +//! 3. A's `dialable_addresses()` filter rejects any unspecified IP +//! (`0.0.0.0`, `[::]`) and any non-QUIC entry. Port 0 is also undialable. +//! 4. If every address in B's self-entry is filtered out, A cannot reach B +//! via the DHT — it can only reach B via static bootstrap config or a +//! pre-existing in-memory connection. +//! +//! These tests assert (3) succeeds against (2) using the public +//! `find_closest_nodes_local_with_self()` API, which is the same code path the +//! DHT response handler invokes. +//! +//! ## Address sources +//! +//! `local_dht_node()` has exactly two sources: +//! +//! - **Loopback / specific-IP listen binds**: when the transport is bound to +//! a non-wildcard address (e.g. `127.0.0.1:` from `local: true` +//! mode), the bound address is published directly. This path is exercised +//! by the single-node `local_mode_*` tests below. +//! - **OBSERVED_ADDRESS frames**: when the transport is bound to a wildcard +//! (`0.0.0.0` / `[::]`), the bound address is *not* published. Instead the +//! node waits until at least one peer connects and reports back via QUIC's +//! OBSERVED_ADDRESS extension. This path is exercised by the two-node +//! `wildcard_*` tests below. +//! +//! ## Why we don't substitute wildcards with `primary_local_ip()` +//! +//! An earlier iteration of this fix substituted `0.0.0.0` with the host's +//! primary outbound interface IP (via the standard `UdpSocket::connect` +//! trick). That worked for VPS / public-IP hosts and for LAN deployments, +//! but for home-NAT deployments it published an RFC1918 LAN address that +//! internet peers cannot route to — wasting connection attempts on +//! guaranteed-failed dials. The current design publishes nothing until +//! OBSERVED_ADDRESS arrives, which is honest and self-correcting. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +use saorsa_core::{Key, MultiAddr, NodeConfig, P2PNode}; +use std::net::{Ipv4Addr, SocketAddr}; +use std::time::Duration; +use tokio::time::timeout; + +/// How many results to ask for from the local closest-nodes query. Any value +/// >= 1 is fine; we only care that the local self-entry is included. +const QUERY_COUNT: usize = 8; + +/// Brief delay after `start()` to let the listener bind. The two_node_messaging +/// integration tests use the same value. +const POST_START_DELAY: Duration = Duration::from_millis(50); + +/// Maximum time to wait for an OBSERVED_ADDRESS frame to arrive after a +/// peer connection completes its handshake. In practice the frame arrives +/// within tens of milliseconds; the budget is generous to absorb scheduler +/// jitter on slow CI. +const OBSERVED_ADDRESS_TIMEOUT: Duration = Duration::from_secs(5); + +/// Polling interval for waiting on the observed external address. +const OBSERVED_ADDRESS_POLL_INTERVAL: Duration = Duration::from_millis(20); + +/// Hard timeout on `connect_peer` and identity exchange in two-node tests. +const CONNECT_TIMEOUT: Duration = Duration::from_secs(2); + +/// Loopback-mode config used for the single-node local-bind tests. +fn local_mode_config() -> NodeConfig { + NodeConfig::builder() + .local(true) + .port(0) + .ipv6(false) + .build() + .expect("test config should be valid") +} + +/// Public-mode (wildcard bind) config used for the two-node OBSERVED_ADDRESS +/// tests and the empty-self-entry contract test. +fn wildcard_mode_config() -> NodeConfig { + NodeConfig::builder() + .local(false) + .port(0) + .ipv6(false) + .build() + .expect("test config should be valid") +} + +/// Returns the entry for `peer_id` from a list of `DHTNode`s, or panics with a +/// descriptive message. We assert this entry exists separately from the +/// dialability assertions so a missing self-entry is reported clearly. +fn extract_self_entry( + nodes: &[saorsa_core::DHTNode], + peer_id: &saorsa_core::PeerId, +) -> saorsa_core::DHTNode { + nodes + .iter() + .find(|n| n.peer_id == *peer_id) + .cloned() + .unwrap_or_else(|| { + panic!( + "find_closest_nodes_local_with_self did not return the local node \ + (peer_id={peer_id:?}); this means local_dht_node() was not invoked" + ) + }) +} + +/// Returns true if the given `MultiAddr` would survive `dialable_addresses()`'s +/// filter — i.e. it is a QUIC address with a specified IP and a non-zero port. +/// +/// Mirrors the rejection rules in +/// `saorsa-core/src/dht_network_manager.rs::dialable_addresses`. +fn is_dialable(addr: &MultiAddr) -> bool { + let Some(sa) = addr.dialable_socket_addr() else { + return false; // not QUIC + }; + if sa.ip().is_unspecified() { + return false; // 0.0.0.0 / [::] + } + if sa.port() == 0 { + return false; // OS-assigned placeholder, never dialable + } + true +} + +/// Fetches the local self-entry from the node's DHT manager. +async fn fetch_self_entry(node: &P2PNode) -> saorsa_core::DHTNode { + let key: Key = [0u8; 32]; + let nodes = node + .dht_manager() + .find_closest_nodes_local_with_self(&key, QUERY_COUNT) + .await; + extract_self_entry(&nodes, node.peer_id()) +} + +/// Polls `transport.observed_external_address()` until it returns `Some` or +/// the timeout expires. Returns the observed address, or `None` if it never +/// arrived. +async fn wait_for_observed_external_address( + node: &P2PNode, + deadline: Duration, +) -> Option { + let result = timeout(deadline, async { + loop { + if let Some(addr) = node.transport().observed_external_address() { + return addr; + } + tokio::time::sleep(OBSERVED_ADDRESS_POLL_INTERVAL).await; + } + }) + .await; + result.ok() +} + +/// Polls `transport.cached_observed_external_address()` (cache-only, +/// bypassing the live read) until it returns `Some` or the timeout expires. +/// +/// Tests use this to wait for the broadcast `ExternalAddressDiscovered` +/// event to be processed by the forwarder and recorded in the cache. The +/// event is fired by saorsa-transport's `poll_discovery_task` on a 1-second +/// tick, so a generous timeout is needed even though the live read may +/// already return a value within tens of milliseconds. +async fn wait_for_cached_observed_address( + node: &P2PNode, + deadline: Duration, +) -> Option { + let result = timeout(deadline, async { + loop { + if let Some(addr) = node.transport().cached_observed_external_address() { + return addr; + } + tokio::time::sleep(OBSERVED_ADDRESS_POLL_INTERVAL).await; + } + }) + .await; + result.ok() +} + +// --------------------------------------------------------------------------- +// Single-node tests: loopback bind path +// --------------------------------------------------------------------------- + +/// **PRIMARY REGRESSION TEST FOR THE LOOPBACK BIND PATH.** +/// +/// A node bound to a specific loopback address (`local: true` → +/// `127.0.0.1:`) must publish that address directly in its DHT +/// self-entry. The published port must be the *actually-bound* port — not +/// `0`, not the configured port (which is also `0`). If this test fails, +/// peers performing DHT FIND_NODE for this node will receive zero usable +/// addresses and will be unable to connect. +/// +/// On the broken codebase this test failed because `local_dht_node()` read +/// from `NodeConfig::listen_addrs()` (a static `(port, ipv6, local)` +/// derivation that returns the configured port — `0` for `--port 0`) instead +/// of from the runtime-bound listener addresses. +#[tokio::test] +async fn local_mode_publishes_dialable_loopback_address() { + let node = P2PNode::new(local_mode_config()) + .await + .expect("P2PNode::new should succeed"); + node.start().await.expect("node.start() should succeed"); + tokio::time::sleep(POST_START_DELAY).await; + + let self_entry = fetch_self_entry(&node).await; + + assert!( + !self_entry.addresses.is_empty(), + "local DHT self-entry has no addresses at all — peers will see this node \ + as having zero contact information" + ); + + let dialable: Vec<&MultiAddr> = self_entry + .addresses + .iter() + .filter(|a| is_dialable(a)) + .collect(); + + assert!( + !dialable.is_empty(), + "local DHT self-entry has {} address(es) but NONE are dialable: {:?}\n\ + \n\ + This is the root cause of sporadic NAT traversal failure: every address \ + in the self-entry will be filtered out by dialable_addresses() on the \ + receiving peer, so DHT-based peer discovery for this node always returns \ + no contactable address.\n\ + \n\ + Fix: DhtNetworkManager::local_dht_node() must read from the runtime \ + transport state (transport.listen_addrs() and \ + transport.observed_external_address()) instead of NodeConfig::listen_addrs() \ + (which is a pure derivation that returns wildcards in Public mode and \ + zero ports for --port 0).", + self_entry.addresses.len(), + self_entry.addresses, + ); + + node.stop().await.expect("node.stop() should succeed"); +} + +/// The runtime `listen_addrs` read from the transport's RwLock should match +/// the addresses published in the DHT self-entry for loopback binds. If they +/// diverge, the DHT is advertising stale or incorrect contact information. +/// +/// This test catches the case where someone "fixes" `local_dht_node()` by +/// reading from the static `NodeConfig` instead of from the live transport +/// state, which would produce a result that's still wrong but in a different +/// way (e.g. the configured port instead of the bound port). +#[tokio::test] +async fn local_mode_published_self_entry_matches_runtime_listen_addrs() { + let node = P2PNode::new(local_mode_config()) + .await + .expect("P2PNode::new should succeed"); + node.start().await.expect("node.start() should succeed"); + tokio::time::sleep(POST_START_DELAY).await; + + // What the transport actually bound to (real ports on a specific IP). + let runtime_addrs = node.listen_addrs().await; + + // What the node tells the rest of the network about itself. + let self_entry = fetch_self_entry(&node).await; + + // The published ports must be the actually-bound ports — not 0, and not + // some statically configured value that might differ from what the OS + // chose. + let runtime_ports: Vec = runtime_addrs + .iter() + .filter_map(MultiAddr::port) + .filter(|p| *p != 0) + .collect(); + + assert!( + !runtime_ports.is_empty(), + "transport.listen_addrs() returned no non-zero ports — the listener \ + did not bind successfully" + ); + + let published_ports: Vec = self_entry + .addresses + .iter() + .filter_map(MultiAddr::port) + .collect(); + + for port in &runtime_ports { + assert!( + published_ports.contains(port), + "published self-entry does not include the actually-bound port {port}; \ + runtime ports = {runtime_ports:?}, published ports = {published_ports:?}\n\ + \n\ + local_dht_node() is reading from the static NodeConfig instead of \ + from the runtime transport state — peers will dial the wrong port." + ); + } + + node.stop().await.expect("node.stop() should succeed"); +} + +/// A node configured with `local: true` (loopback mode) must never publish a +/// port-0 address. Port 0 is the kernel's "pick any port" placeholder; it is +/// never a valid destination. +#[tokio::test] +async fn local_mode_never_publishes_port_zero() { + let node = P2PNode::new(local_mode_config()) + .await + .expect("P2PNode::new should succeed"); + node.start().await.expect("node.start() should succeed"); + tokio::time::sleep(POST_START_DELAY).await; + + let self_entry = fetch_self_entry(&node).await; + + let zero_port_addrs: Vec<&MultiAddr> = self_entry + .addresses + .iter() + .filter(|a| a.port() == Some(0)) + .collect(); + + assert!( + zero_port_addrs.is_empty(), + "local DHT self-entry contains {} address(es) with port 0: {:?}\n\ + \n\ + Port 0 is the placeholder the kernel uses for 'pick any port'; it is \ + never a valid destination. Publishing it to the DHT means peers will \ + try to dial port 0 and fail.", + zero_port_addrs.len(), + zero_port_addrs, + ); + + node.stop().await.expect("node.stop() should succeed"); +} + +// --------------------------------------------------------------------------- +// Single-node tests: wildcard bind contract +// --------------------------------------------------------------------------- + +/// **CONTRACT: a wildcard-bound node with no observation publishes nothing.** +/// +/// When the transport is bound to `0.0.0.0:` (the production default +/// for VPS / cloud deployments) and no peer has yet connected to send an +/// OBSERVED_ADDRESS frame, `local_dht_node()` must return an empty +/// `addresses` vec — *not* the wildcard, *not* a guessed LAN IP, *not* the +/// configured port-0 placeholder. +/// +/// This pins the "don't lie when you don't know" contract: it is better to +/// publish no contact information than to publish bind-side wildcards or +/// LAN-only addresses that internet peers cannot route to. Once the bootstrap +/// dial completes and the first OBSERVED_ADDRESS frame arrives, future +/// queries return the real address (see the `wildcard_*_two_nodes` tests +/// below). +#[tokio::test] +async fn wildcard_bind_with_no_peers_publishes_empty_self_entry() { + let node = P2PNode::new(wildcard_mode_config()) + .await + .expect("P2PNode::new should succeed"); + node.start().await.expect("node.start() should succeed"); + tokio::time::sleep(POST_START_DELAY).await; + + let self_entry = fetch_self_entry(&node).await; + + assert!( + self_entry.addresses.is_empty(), + "wildcard-bound node with no peers should publish an empty self-entry, \ + but published {} address(es): {:?}\n\ + \n\ + This is a regression: a freshly-started node must not lie about its \ + contact information. The acceptable sources are (1) the transport's \ + observed_external_address() — None until a peer connects, or (2) a \ + specific-IP bind — N/A for wildcard. Anything else (wildcard \ + substitution, primary outbound interface IP, etc.) risks publishing \ + an address that internet peers cannot route to.", + self_entry.addresses.len(), + self_entry.addresses, + ); + + node.stop().await.expect("node.stop() should succeed"); +} + +// --------------------------------------------------------------------------- +// Two-node tests: OBSERVED_ADDRESS path +// --------------------------------------------------------------------------- + +/// Build a loopback dial target for a wildcard-bound node. The node's +/// `listen_addrs()` returns `0.0.0.0:` (not directly dialable), +/// so we substitute `127.0.0.1` as the destination IP — the kernel routes +/// loopback traffic to the wildcard-bound socket. +async fn loopback_dial_target_for(node: &P2PNode) -> MultiAddr { + let port = node + .listen_addrs() + .await + .into_iter() + .find_map(|a| a.dialable_socket_addr()) + .expect("wildcard-bound node should have an IPv4 listen address") + .port(); + assert_ne!(port, 0, "bound port must be non-zero after start()"); + MultiAddr::quic(SocketAddr::new(Ipv4Addr::LOCALHOST.into(), port)) +} + +/// **PRIMARY REGRESSION TEST FOR THE OBSERVED_ADDRESS PATH.** +/// +/// Two wildcard-bound (`0.0.0.0:0`) nodes connect to each other over +/// loopback. After the QUIC handshake completes and OBSERVED_ADDRESS frames +/// flow, each side learns its post-NAT (here: post-loopback) reflexive +/// address. The published DHT self-entry must then include that observed +/// address as a dialable entry. +/// +/// This is the contract that makes Public-mode deployments work: a fresh +/// VPS node binds to `0.0.0.0:10000`, dials a bootstrap peer, and the +/// bootstrap's OBSERVED_ADDRESS frame fills in the node's public IP. From +/// that point forward, every DHT query for this node returns its public +/// IP:port. +/// +/// If this test fails, sporadic NAT traversal failure will return: peers +/// querying the DHT for a wildcard-bound node will receive empty addresses +/// even after the node has been observed. +#[tokio::test] +async fn wildcard_bind_publishes_observed_address_after_peer_connection() { + let node_a = P2PNode::new(wildcard_mode_config()) + .await + .expect("node_a creation should succeed"); + let node_b = P2PNode::new(wildcard_mode_config()) + .await + .expect("node_b creation should succeed"); + + node_a.start().await.expect("node_a.start() should succeed"); + node_b.start().await.expect("node_b.start() should succeed"); + tokio::time::sleep(POST_START_DELAY).await; + + // node_b's listen_addrs returns 0.0.0.0: — substitute + // 127.0.0.1 to produce a dialable address. + let dial_target = loopback_dial_target_for(&node_b).await; + + let channel_id = timeout(CONNECT_TIMEOUT, node_a.connect_peer(&dial_target)) + .await + .expect("connect should not timeout") + .expect("connect should succeed"); + + let _peer_b = timeout( + CONNECT_TIMEOUT, + node_a.wait_for_peer_identity(&channel_id, CONNECT_TIMEOUT), + ) + .await + .expect("identity exchange should not timeout") + .expect("identity exchange should succeed"); + + // Wait for the OBSERVED_ADDRESS frame to populate node_a's reflexive + // address. This is the moment the wildcard-bind path becomes useful. + let observed = wait_for_observed_external_address(&node_a, OBSERVED_ADDRESS_TIMEOUT).await; + + assert!( + observed.is_some(), + "node_a should have received an OBSERVED_ADDRESS frame from node_b within \ + {OBSERVED_ADDRESS_TIMEOUT:?} of identity exchange, but observed_external_address() \ + is still None.\n\ + \n\ + Either the saorsa-transport address-discovery extension is not emitting \ + OBSERVED_ADDRESS frames after handshake completion, or the frames are \ + not being plumbed through to TransportHandle::observed_external_address(). \ + Check src/transport/saorsa_transport_adapter.rs and the saorsa-transport \ + connection layer." + ); + let observed = observed.unwrap(); + + let self_entry = fetch_self_entry(&node_a).await; + + assert!( + !self_entry.addresses.is_empty(), + "node_a's DHT self-entry should contain at least the observed address \ + {observed} after peer connection, but is empty" + ); + + let dialable: Vec<&MultiAddr> = self_entry + .addresses + .iter() + .filter(|a| is_dialable(a)) + .collect(); + + assert!( + !dialable.is_empty(), + "node_a's self-entry has {} address(es) but NONE are dialable: {:?}\n\ + observed external address = {observed}", + self_entry.addresses.len(), + self_entry.addresses, + ); + + let observed_multi = MultiAddr::quic(observed); + assert!( + self_entry.addresses.contains(&observed_multi), + "node_a's self-entry does not include the observed external address \ + {observed}.\n\ + Published addresses: {:?}\n\ + \n\ + local_dht_node() is failing to read from \ + transport.observed_external_address(). Without this, wildcard-bound \ + nodes have no way to advertise themselves to the DHT.", + self_entry.addresses, + ); + + node_a.stop().await.expect("node_a.stop() should succeed"); + node_b.stop().await.expect("node_b.stop() should succeed"); +} + +/// **REGRESSION TEST FOR THE OBSERVED-ADDRESS CACHE FALLBACK.** +/// +/// `saorsa-transport` exposes the live observed external address only via +/// active connections — when every connection drops, the live read returns +/// `None`. Without a fallback, a node that briefly loses connectivity +/// disappears from the DHT until reconnection. +/// +/// `TransportHandle::observed_external_address()` therefore consults a +/// cache populated by `P2pEvent::ExternalAddressDiscovered` events. After +/// a connection drop the cache should still serve the most-recently-observed +/// address, keeping the node visible to DHT queries. +/// +/// This test: +/// +/// 1. Connects two wildcard-bound nodes over loopback. +/// 2. Waits for OBSERVED_ADDRESS to populate `node_a`'s reflexive address. +/// 3. Records the observed value. +/// 4. Stops `node_b`, which drops the only connection on `node_a`. +/// 5. Waits for `node_a` to see zero connected peers (the live source is +/// now empty). +/// 6. Asserts `node_a.transport().observed_external_address()` still +/// returns the same address — proving the cache fallback engaged. +/// 7. Asserts `node_a`'s DHT self-entry still publishes the observed +/// address, completing the end-to-end contract. +#[tokio::test] +async fn observed_address_cache_serves_fallback_after_connection_drop() { + let node_a = P2PNode::new(wildcard_mode_config()) + .await + .expect("node_a creation should succeed"); + let node_b = P2PNode::new(wildcard_mode_config()) + .await + .expect("node_b creation should succeed"); + + node_a.start().await.expect("node_a.start() should succeed"); + node_b.start().await.expect("node_b.start() should succeed"); + tokio::time::sleep(POST_START_DELAY).await; + + let dial_target = loopback_dial_target_for(&node_b).await; + let channel_id = timeout(CONNECT_TIMEOUT, node_a.connect_peer(&dial_target)) + .await + .expect("connect should not timeout") + .expect("connect should succeed"); + let _peer_b = timeout( + CONNECT_TIMEOUT, + node_a.wait_for_peer_identity(&channel_id, CONNECT_TIMEOUT), + ) + .await + .expect("identity exchange should not timeout") + .expect("identity exchange should succeed"); + + // Step 2-3: wait for the OBSERVED_ADDRESS frame to flow all the way + // through to the cache. The live read can return a value as soon as + // the QUIC connection has stored an observed address, but the cache is + // populated by the broadcast `ExternalAddressDiscovered` event which + // is fired by saorsa-transport's `poll_discovery_task` on a 1-second + // tick. We poll the cache-only accessor so we know the broadcast event + // has been received and recorded *before* we disconnect. + let observed = wait_for_cached_observed_address(&node_a, OBSERVED_ADDRESS_TIMEOUT) + .await + .expect( + "ExternalAddressDiscovered event should reach the observed-address cache \ + within the timeout. If this fails, either saorsa-transport's \ + poll_discovery_task is not firing the broadcast event, or the \ + ExternalAddressDiscovered branch in spawn_peer_address_update_forwarder \ + is not feeding the cache.", + ); + + // Sanity check: while connected, the live read agrees with the cache. + assert_eq!( + node_a.transport().observed_external_address(), + Some(observed), + "live + cache should agree on the observed address while connected" + ); + + // Step 4: stop node_b. This drops the QUIC connection node_a was using + // as its only live source of observed-address data. + node_b.stop().await.expect("node_b.stop() should succeed"); + + // Step 5: wait for node_a to notice it has no live peers. Without this, + // the live `dual_node.get_observed_external_address()` may still return + // Some(...) because saorsa-transport's connection cleanup is async. + let drained = timeout(CONNECT_TIMEOUT, async { + loop { + if node_a.connected_peers().await.is_empty() { + return; + } + tokio::time::sleep(OBSERVED_ADDRESS_POLL_INTERVAL).await; + } + }) + .await; + assert!( + drained.is_ok(), + "node_a should observe zero connected peers within {CONNECT_TIMEOUT:?} \ + after stopping node_b" + ); + + // Step 6: the live source is now empty, so any value returned by + // `observed_external_address()` must be coming from the cache. It must + // match the address we recorded while the connection was live. + let after_drop = node_a.transport().observed_external_address(); + assert_eq!( + after_drop, + Some(observed), + "observed_external_address() should still return the cached value \ + {observed} after every live connection has dropped, but returned {after_drop:?}.\n\ + \n\ + Either the ExternalAddressDiscovered forwarder is not feeding the \ + cache (check spawn_peer_address_update_forwarder in \ + saorsa_transport_adapter.rs), or the fallback path in \ + TransportHandle::observed_external_address() is not consulting it." + ); + + // Step 7: end-to-end — the DHT self-entry must still include the + // observed address, so peers querying us via the DHT can still find us + // even though we have no live connections. + let self_entry = fetch_self_entry(&node_a).await; + let observed_multi = MultiAddr::quic(observed); + assert!( + self_entry.addresses.contains(&observed_multi), + "node_a's DHT self-entry should still include the cached observed \ + address {observed} after the live connection dropped.\n\ + Published addresses: {:?}", + self_entry.addresses, + ); + + node_a.stop().await.expect("node_a.stop() should succeed"); +}