From 510116afb9b743ce9caafef1c09aa366cfb20d9b Mon Sep 17 00:00:00 2001 From: Aleksandr Bogomiakov Date: Wed, 30 Sep 2026 22:51:08 +0100 Subject: [PATCH] Add automatic protocol fallback with backoff and silence DHT bootstrap error When multiple data plane protocols (such as tcp-tls and wg-quic) are configured, the agent previously attempted only the first mutually supported protocol in served order. If that protocol failed to connect or dropped (due to NAT, firewall, or handshake failure), ensure_links retried the exact same failed protocol on every tick without ever falling back to alternative supported protocols. Additionally, tcp-tls mistakenly included 127.0.0.1 in peer candidate lists, causing peers to connect to their own local listener and fail certificate verification (NotValidForName). Finally, BitTorrent DHT bootstrap failures in the third-party mainline crate were logged as noisy ERRORs despite overlay connectivity operating normally. Key improvements: - In crates/tsunagi/src/agent/network.rs, implement ProtocolDialState with exponential backoff (3s, 6s, 12s, 24s, 48s, capped at 60s). - In ensure_links, select the highest-priority mutually supported protocol that is not currently on backoff cooldown. If a protocol fails, immediately record the failure and fall back to alternative mutually supported protocols without waiting for the periodic ticker. - In crates/tsunagi-tcp-tls, remove localhost loopback from local_ip_candidates and capability announcements, and prevent open() from connecting to our own listening port. - In crates/tsunagi-cli/src/main.rs, set mainline=off in the default tracing filter so benign BitTorrent DHT routing table bootstrap errors do not alarm users. Validation on Linux: - cargo fmt --all -- --check passed. - cargo clippy --locked --workspace --all-targets -- -D warnings passed. - cargo test --locked --workspace --all-targets passed (349 tests passed). - cargo build --release passed. --- crates/tsunagi-cli/src/main.rs | 2 +- crates/tsunagi-tcp-tls/src/plugin.rs | 8 -- crates/tsunagi-tcp-tls/src/transport.rs | 3 + crates/tsunagi/src/agent/network.rs | 118 ++++++++++++++++++++---- 4 files changed, 105 insertions(+), 26 deletions(-) diff --git a/crates/tsunagi-cli/src/main.rs b/crates/tsunagi-cli/src/main.rs index a6b21a1..e1df07d 100644 --- a/crates/tsunagi-cli/src/main.rs +++ b/crates/tsunagi-cli/src/main.rs @@ -37,7 +37,7 @@ struct Cli { long, global = true, env = "TSUNAGI_LOG", - default_value = "info,iroh=warn,quinn=warn,rustls=warn,boringtun=warn,mainline=warn" + default_value = "info,iroh=warn,quinn=warn,rustls=warn,boringtun=warn,mainline=off" )] log: String, diff --git a/crates/tsunagi-tcp-tls/src/plugin.rs b/crates/tsunagi-tcp-tls/src/plugin.rs index 80deecd..8a0daf0 100644 --- a/crates/tsunagi-tcp-tls/src/plugin.rs +++ b/crates/tsunagi-tcp-tls/src/plugin.rs @@ -80,7 +80,6 @@ pub fn local_ip_candidates() -> Vec { } } } - ips.push(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)); ips } @@ -198,13 +197,6 @@ impl TunnelCodec for TcpTlsCodec { self.transport .set_peer_addr(peer, std::net::SocketAddr::new(*ip, announcement.port)); } - self.transport.set_peer_addr( - peer, - std::net::SocketAddr::new( - std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST), - announcement.port, - ), - ); let peer_noise_pubkey = PublicKey::from(announcement.noise_public); let cipher = Arc::new(PeerCipher::new( diff --git a/crates/tsunagi-tcp-tls/src/transport.rs b/crates/tsunagi-tcp-tls/src/transport.rs index 5fa86df..0fa4d68 100644 --- a/crates/tsunagi-tcp-tls/src/transport.rs +++ b/crates/tsunagi-tcp-tls/src/transport.rs @@ -527,6 +527,9 @@ impl PacketTransport for TcpTlsTransport { let mut connected_socket = None; let mut connected_addr = None; for addr in &target_addrs { + if addr.ip().is_loopback() && addr.port() == self.listen_port { + continue; + } match tokio::time::timeout(Duration::from_secs(3), TcpStream::connect(*addr)).await { Ok(Ok(socket)) => { diff --git a/crates/tsunagi/src/agent/network.rs b/crates/tsunagi/src/agent/network.rs index 5430948..76c7298 100644 --- a/crates/tsunagi/src/agent/network.rs +++ b/crates/tsunagi/src/agent/network.rs @@ -239,6 +239,24 @@ const REACH_EXPIRY: Duration = Duration::from_secs(90); /// can arrive and be adopted instead. const RANGE_PROPOSAL_GRACE: Duration = Duration::from_secs(3); +#[derive(Debug, Clone)] +struct ProtocolDialState { + failures: u32, + cooldown_until: std::time::Instant, +} + +const DATA_LINK_BASE_BACKOFF: Duration = Duration::from_secs(3); +const DATA_LINK_MAX_BACKOFF: Duration = Duration::from_secs(60); + +fn data_link_backoff(failures: u32) -> Duration { + if failures == 0 { + return Duration::ZERO; + } + let shift = (failures - 1).min(5); + let secs = (DATA_LINK_BASE_BACKOFF.as_secs() << shift).min(DATA_LINK_MAX_BACKOFF.as_secs()); + Duration::from_secs(secs) +} + struct Runtime { discovery_worker: Option, discovery_candidates: mpsc::Receiver, @@ -284,6 +302,8 @@ struct Runtime { opening: HashSet<(EndpointId, String)>, link_results_tx: mpsc::Sender, link_results_rx: mpsc::Receiver, + /// Protocol dial failure count and backoff cooldown per peer and protocol. + protocol_dial_states: HashMap<(EndpointId, String), ProtocolDialState>, /// Peers already told about a protocol version that cannot match, so it /// is said once rather than on every announcement. reported_mismatch: HashSet<(EndpointId, String)>, @@ -346,6 +366,7 @@ impl Runtime { opening: HashSet::new(), link_results_tx, link_results_rx, + protocol_dial_states: HashMap::new(), reported_mismatch: HashSet::new(), released: false, reported_missing: None, @@ -422,6 +443,7 @@ impl Runtime { self.session_events_rx.close(); self.hub.close(); self.links.clear(); + self.protocol_dial_states.clear(); let peers: Vec = self.sessions.keys().copied().collect(); for peer in peers { if let Some(session) = self.sessions.remove(&peer) { @@ -1173,6 +1195,25 @@ impl Runtime { .collect() } + fn record_protocol_failure(&mut self, peer: EndpointId, protocol: &str) { + let key = (peer, protocol.to_string()); + let entry = self + .protocol_dial_states + .entry(key) + .or_insert(ProtocolDialState { + failures: 0, + cooldown_until: std::time::Instant::now(), + }); + entry.failures = entry.failures.saturating_add(1); + let backoff = data_link_backoff(entry.failures); + entry.cooldown_until = std::time::Instant::now() + backoff; + } + + fn record_protocol_success(&mut self, peer: EndpointId, protocol: &str) { + self.protocol_dial_states + .remove(&(peer, protocol.to_string())); + } + /// Opens whatever data plane links are missing, and forgets dead ones. /// /// Only one side dials, chosen by a rule both sides compute the same way, @@ -1198,6 +1239,7 @@ impl Runtime { // reachable through somebody is decided below; the protocol's // link stays either way. self.hub.clear_direct(peer, &protocol); + self.record_protocol_failure(peer, &protocol); self.emit(Event::DataLinkDown { network: self.network_id, peer, @@ -1208,6 +1250,9 @@ impl Runtime { // Choose the highest priority mutually supported protocol for each peer. // If a direct data link is already open or opening, do not dial another. + // If a protocol recently failed, fall back to other mutually supported + // protocols according to backoff cooldown. + let now = std::time::Instant::now(); let mut wanted: Vec<(EndpointId, String)> = Vec::new(); for session in self.sessions.values() { let peer = session.peer; @@ -1216,23 +1261,6 @@ impl Runtime { { continue; } - for (name, ours) in &served { - let matches = session - .capabilities - .iter() - .any(|c| c.enabled && c.protocol == *name && c.version == *ours); - if matches { - wanted.push((peer, name.clone())); - break; - } - } - } - - for (peer, protocol) in wanted { - let key = (peer, protocol.clone()); - if self.links.contains_key(&key) || self.opening.contains(&key) { - continue; - } if self.no_direct_data(peer) { continue; } @@ -1241,6 +1269,39 @@ impl Runtime { if self.local_id.as_bytes() >= peer.as_bytes() { continue; } + + let supported: Vec = served + .iter() + .filter(|(name, ours)| { + session + .capabilities + .iter() + .any(|c| c.enabled && c.protocol == *name && c.version == *ours) + }) + .map(|(name, _)| name.clone()) + .collect(); + + if supported.is_empty() { + continue; + } + + let chosen = supported.iter().find(|proto| { + match self.protocol_dial_states.get(&(peer, (*proto).clone())) { + Some(state) => state.cooldown_until <= now, + None => true, + } + }); + + if let Some(proto) = chosen { + wanted.push((peer, proto.clone())); + } + } + + for (peer, protocol) in wanted { + let key = (peer, protocol.clone()); + if self.links.contains_key(&key) || self.opening.contains(&key) { + continue; + } self.opening.insert(key); let results = self.link_results_tx.clone(); @@ -1475,6 +1536,7 @@ impl Runtime { self.opening.remove(&key); match outcome.result { Ok(link) => { + self.record_protocol_success(outcome.peer, &outcome.protocol); self.adopt_link(outcome.peer, outcome.protocol, link); // A new link is both a path for us and one we can offer // others, so it changes what we route and what we say. @@ -1484,12 +1546,15 @@ impl Runtime { Err(reason) => { // A data plane that cannot be set up is reported, never fatal. self.metrics.data_link_failures += 1; + self.record_protocol_failure(outcome.peer, &outcome.protocol); self.emit(Event::DataLinkDown { network: self.network_id, peer: outcome.peer, protocol: outcome.protocol, reason, }); + // Fall back to other mutually supported protocols immediately. + self.ensure_links(); } } } @@ -1517,6 +1582,7 @@ impl Runtime { if self.no_direct_data(inbound.peer) { return; } + self.record_protocol_success(inbound.peer, &inbound.protocol); self.adopt_link(inbound.peer, inbound.protocol, inbound.link); self.update_paths(); self.announce_reach(false); @@ -1564,6 +1630,7 @@ impl Runtime { // protocol's link goes with it. self.hub.remove_peer(peer); self.reachable.remove(&peer); + self.protocol_dial_states.retain(|(p, _), _| *p != peer); let keys: Vec<(EndpointId, String)> = self .links .keys() @@ -2079,3 +2146,20 @@ fn merge_addr(existing: &mut iroh::EndpointAddr, incoming: &iroh::EndpointAddr) existing.addrs.insert(addr.clone()); } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_data_link_backoff_progression() { + assert_eq!(data_link_backoff(0), Duration::ZERO); + assert_eq!(data_link_backoff(1), Duration::from_secs(3)); + assert_eq!(data_link_backoff(2), Duration::from_secs(6)); + assert_eq!(data_link_backoff(3), Duration::from_secs(12)); + assert_eq!(data_link_backoff(4), Duration::from_secs(24)); + assert_eq!(data_link_backoff(5), Duration::from_secs(48)); + assert_eq!(data_link_backoff(6), Duration::from_secs(60)); + assert_eq!(data_link_backoff(10), Duration::from_secs(60)); + } +}