diff --git a/CHANGELOG.md b/CHANGELOG.md index 51dec2f..05350ed 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -468,6 +468,33 @@ with v0.2.x peers. the TUN-side `path_mtu_lookup` so later flows pick up forward-path bottlenecks without re-discovery. Windows TUN reader receives the same per-destination plumbing. +- Proactive end-to-end `PathMtuNotification` now mirrors into the + TUN-side `path_mtu_lookup` (TCP MSS clamp store), parallel to the + reactive `MtuExceeded` mirror that already existed. Previously the + proactive handler only updated the session-canonical + `MmpSessionState.path_mtu`; on stable long-lived paths where the + destination's echo had tightened the session MTU but no transit + router had emitted a fresh `MtuExceeded` (because all current + traffic was already sized by the tighter session value), new TCP + flows opened in that window kept getting clamped by the staler + discovery-time value. The proactive mirror closes that gap with + the same tighter-only semantics — never loosens the clamp. +- Nostr-discovered peers running an FMP-protocol version we cannot + speak no longer trigger an indefinite retraversal storm. Open- + discovery NAT-traversal succeeds at the UDP layer regardless of + protocol version, so the daemon would adopt the punched socket, + drop every incoming packet at `Unknown FMP version`, idle out + after 31s, and re-fire the full STUN-offer-answer-punch sequence + ~30s later — every minute, forever, against peers the handshake + literally cannot complete with. The rx loop now detects mismatched- + version packets arriving on adopted bootstrap transports, reverse- + maps to the originating npub, and applies a long structural + cooldown to the discovery layer's `failure_state` so the next + open-discovery sweep skips the peer until either side upgrades. + One-shot WARN per fresh observation; subsequent mismatches inside + the cooldown window are silent. New `protocol_mismatch_cooldown_secs` + config field under `node.discovery.nostr` (default 86400 = 24h), + separate from the transient-failure `extended_cooldown_secs`. - Auto-connect peers now reconnect after a graceful `Disconnect` notification from the remote side. `handle_disconnect` previously removed the peer without scheduling a reconnect, orphaning the diff --git a/src/config/node.rs b/src/config/node.rs index dee7d79..5add977 100644 --- a/src/config/node.rs +++ b/src/config/node.rs @@ -388,6 +388,14 @@ pub struct NostrDiscoveryConfig { /// failure time) evicted when the cap is exceeded. Default: 4096. #[serde(default = "NostrDiscoveryConfig::default_failure_state_max_entries")] pub failure_state_max_entries: usize, + /// Cooldown applied after observing a fatal protocol mismatch on a + /// Nostr-adopted bootstrap transport (e.g. `Unknown FMP version` + /// from a peer running a different FMP-protocol version). Independent + /// of `extended_cooldown_secs` and much longer because the mismatch + /// is structural — re-traversing the peer is wasted effort until one + /// side upgrades. Default: 86400 (24 hours). + #[serde(default = "NostrDiscoveryConfig::default_protocol_mismatch_cooldown_secs")] + pub protocol_mismatch_cooldown_secs: u64, } impl Default for NostrDiscoveryConfig { @@ -419,6 +427,7 @@ impl Default for NostrDiscoveryConfig { extended_cooldown_secs: Self::default_extended_cooldown_secs(), warn_log_interval_secs: Self::default_warn_log_interval_secs(), failure_state_max_entries: Self::default_failure_state_max_entries(), + protocol_mismatch_cooldown_secs: Self::default_protocol_mismatch_cooldown_secs(), } } } @@ -527,6 +536,10 @@ impl NostrDiscoveryConfig { fn default_failure_state_max_entries() -> usize { 4_096 } + + fn default_protocol_mismatch_cooldown_secs() -> u64 { + 86_400 + } } /// Spanning tree (`node.tree.*`). diff --git a/src/discovery/nostr/failure_state.rs b/src/discovery/nostr/failure_state.rs index 57d2b0e..331b21f 100644 --- a/src/discovery/nostr/failure_state.rs +++ b/src/discovery/nostr/failure_state.rs @@ -176,6 +176,52 @@ impl FailureState { } } + /// Record a fatal protocol mismatch against `npub` and apply + /// `cooldown_ms` immediately (independent of the streak threshold). + /// + /// Returns `true` when this is a fresh mismatch entry (caller should + /// log a one-shot WARN) or `false` if a comparable mismatch cooldown + /// is already in place (caller should remain silent — repeat + /// observations of the same mismatch are uninteresting). + /// + /// Used when the rx loop sees an unhandshakable packet (e.g., + /// `Unknown FMP version`) on a Nostr-adopted bootstrap transport: + /// re-traversing the peer at the next sweep cycle is wasted effort + /// because the peer cannot accept our handshake until one side + /// upgrades. The cooldown is much longer than the transient-failure + /// `extended_cooldown_ms` because the mismatch is structural. + pub(super) fn record_protocol_mismatch( + &self, + npub: &str, + now_ms: u64, + cooldown_ms: u64, + ) -> bool { + let mut map = self.inner.lock().expect("failure-state mutex poisoned"); + let entry = map + .entry(npub.to_string()) + .or_insert_with(|| NpubFailureRecord::new(now_ms)); + // Treat the mismatch as crossing the streak threshold so other + // visibility paths (e.g. show_peers JSON) reflect the failed state. + entry.consecutive_failures = entry.consecutive_failures.max(self.threshold); + entry.last_failure_at_ms = now_ms; + + let cooldown_until = now_ms.saturating_add(cooldown_ms); + // "Fresh" means we weren't already inside a comparable cooldown + // window. Use the existing-cooldown's remaining time as the test + // so that an entry shifted forward by a few seconds doesn't keep + // re-triggering WARNs. + let already_suppressed = entry + .cooldown_until_ms + .is_some_and(|t| t > now_ms && t.saturating_sub(now_ms) >= cooldown_ms / 2); + entry.cooldown_until_ms = Some(cooldown_until); + + if map.len() > self.max_entries { + evict_oldest(&mut map, self.max_entries); + } + + !already_suppressed + } + /// Return cooldown_until_ms if the peer is currently in extended /// cooldown. pub(super) fn cooldown_until(&self, npub: &str, now_ms: u64) -> Option { @@ -294,6 +340,67 @@ mod tests { assert_eq!(rec.consecutive_failures, 0); } + #[test] + fn record_protocol_mismatch_fresh_entry_returns_true() { + let s = fs(); + // 24h cooldown + let cooldown_ms = 24 * 60 * 60 * 1000; + assert!( + s.record_protocol_mismatch("npub1mismatch", 1000, cooldown_ms), + "first mismatch must signal fresh — caller should WARN" + ); + assert_eq!( + s.cooldown_until("npub1mismatch", 2000), + Some(1000 + cooldown_ms), + "cooldown applied immediately" + ); + } + + #[test] + fn record_protocol_mismatch_repeat_inside_window_returns_false() { + let s = fs(); + let cooldown_ms = 24 * 60 * 60 * 1000; + s.record_protocol_mismatch("npub1mismatch", 1000, cooldown_ms); + // 30s later, same mismatch — caller should NOT re-WARN + assert!( + !s.record_protocol_mismatch("npub1mismatch", 31_000, cooldown_ms), + "second mismatch inside the existing cooldown must NOT signal fresh" + ); + // Cooldown extends forward. + assert_eq!( + s.cooldown_until("npub1mismatch", 32_000), + Some(31_000 + cooldown_ms), + ); + } + + #[test] + fn record_protocol_mismatch_pins_streak_at_threshold() { + let s = fs(); + s.record_protocol_mismatch("npub1mismatch", 1000, 60_000); + // Snapshot reflects the threshold pin so show_peers renders the + // entry as crossed-threshold. + let snap = s.snapshot(); + let (_, rec) = snap + .iter() + .find(|(n, _)| n == "npub1mismatch") + .expect("entry present"); + assert!(rec.consecutive_failures >= 3); + } + + #[test] + fn record_protocol_mismatch_after_old_cooldown_lapsed_signals_fresh() { + let s = fs(); + let cooldown_ms = 24 * 60 * 60 * 1000; + s.record_protocol_mismatch("npub1mismatch", 1000, cooldown_ms); + // Far in the future after cooldown elapsed: a *new* observation + // is fresh again so the operator gets a fresh WARN log. + let later = 1000 + cooldown_ms + 1; + assert!( + s.record_protocol_mismatch("npub1mismatch", later, cooldown_ms), + "after the cooldown window has elapsed, the next mismatch is fresh" + ); + } + #[test] fn size_cap_evicts_oldest_by_last_failure_at() { let s = fs(); // cap = 8 diff --git a/src/discovery/nostr/runtime.rs b/src/discovery/nostr/runtime.rs index 4a64300..3ad700c 100644 --- a/src/discovery/nostr/runtime.rs +++ b/src/discovery/nostr/runtime.rs @@ -215,6 +215,30 @@ impl NostrDiscovery { self.failure_state.cooldown_until(npub, now_ms) } + /// Record a fatal protocol mismatch (e.g. `Unknown FMP version` on a + /// Nostr-adopted bootstrap transport). Returns `true` if this is a + /// fresh observation worth a WARN log; `false` if the peer is already + /// inside a comparable mismatch cooldown. + /// + /// The cooldown is `protocol_mismatch_cooldown_secs` from config — + /// much longer than `extended_cooldown_secs` because mismatches are + /// structural (only resolves when one side upgrades) rather than + /// transient. + pub fn record_protocol_mismatch(&self, npub: &str, now_ms: u64) -> bool { + let cooldown_ms = self + .config + .protocol_mismatch_cooldown_secs + .saturating_mul(1000); + self.failure_state + .record_protocol_mismatch(npub, now_ms, cooldown_ms) + } + + /// Configured protocol-mismatch cooldown in seconds. Exposed so log + /// emitters can include the duration without re-reading config. + pub fn protocol_mismatch_cooldown_secs(&self) -> u64 { + self.config.protocol_mismatch_cooldown_secs + } + /// Snapshot of per-npub failure state for `show_peers` rendering. pub fn failure_state_snapshot(&self) -> Vec { self.failure_state diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index d5254b6..1f53912 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -162,6 +162,34 @@ impl Node { transport_id = %packet.transport_id, "Unknown FMP version, dropping" ); + + // If the packet arrived on an adopted Nostr-NAT bootstrap + // transport, the originating peer is necessarily on a + // different FMP-protocol version than us — the discovery + // sweep would otherwise re-traverse them every cycle even + // though no msg1/msg2 exchange can ever succeed. Bump the + // discovery-layer cooldown to the long protocol-mismatch + // window and emit a single WARN per fresh observation. + if self.bootstrap_transports.contains(&packet.transport_id) + && let Some(npub) = self + .bootstrap_transport_npubs + .get(&packet.transport_id) + .cloned() + && let Some(handle) = self.nostr_discovery_handle() + { + let now_ms = Self::now_ms(); + let cooldown_secs = handle.protocol_mismatch_cooldown_secs(); + if handle.record_protocol_mismatch(&npub, now_ms) { + warn!( + peer_npub = %npub, + transport_id = %packet.transport_id, + peer_version = prefix.version, + our_version = FMP_VERSION, + cooldown_secs, + "Nostr-discovered peer speaks a different FMP version; suppressing retraversal" + ); + } + } return; } diff --git a/src/node/handlers/session.rs b/src/node/handlers/session.rs index 2631fbd..0206ce3 100644 --- a/src/node/handlers/session.rs +++ b/src/node/handlers/session.rs @@ -1029,7 +1029,11 @@ impl Node { /// /// The destination is telling us the path MTU has changed. /// Apply source-side rules (decrease immediate, increase validated). - fn handle_session_path_mtu_notification(&mut self, src_addr: &NodeAddr, body: &[u8]) { + pub(in crate::node) fn handle_session_path_mtu_notification( + &mut self, + src_addr: &NodeAddr, + body: &[u8], + ) { let notif = match PathMtuNotification::decode(body) { Ok(n) => n, Err(e) => { @@ -1053,16 +1057,59 @@ impl Node { let old_mtu = mmp.path_mtu.current_mtu(); let now = std::time::Instant::now(); - mmp.path_mtu.apply_notification(notif.path_mtu, now); + let changed = mmp.path_mtu.apply_notification(notif.path_mtu, now); let new_mtu = mmp.path_mtu.current_mtu(); - if new_mtu != old_mtu { - debug!( - src = %peer_name, - old_mtu, - new_mtu, - "Path MTU changed via notification" - ); + if !changed { + return; + } + + debug!( + src = %peer_name, + old_mtu, + new_mtu, + "Path MTU changed via notification" + ); + + // Mirror the new effective MTU into the FipsAddress-keyed lookup used + // by the TUN reader/writer at TCP MSS clamp time. Without this, new + // TCP flows opened on a path the proactive end-to-end echo has + // already tightened keep getting clamped by the staler discovery- + // time value until a reactive MtuExceeded happens to fire. Keep the + // tighter of existing-or-new — never loosen the clamp. + let fips_addr = crate::FipsAddress::from_node_addr(src_addr); + match self.path_mtu_lookup.write() { + Ok(mut map) => match map.get(&fips_addr).copied() { + Some(existing) if existing <= new_mtu => { + debug!( + dest = %peer_name, + fips_addr = %fips_addr, + new_mtu, + existing, + "PathMtuNotification: keeping tighter existing path_mtu_lookup value" + ); + } + other => { + map.insert(fips_addr, new_mtu); + debug!( + dest = %peer_name, + fips_addr = %fips_addr, + new_mtu, + prior = ?other, + map_len = map.len(), + "PathMtuNotification: tightened path_mtu_lookup" + ); + } + }, + Err(e) => { + warn!( + dest = %peer_name, + fips_addr = %fips_addr, + new_mtu, + error = %e, + "path_mtu_lookup write lock poisoned; PathMtuNotification not reflected" + ); + } } } diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index 61bd9ca..b2a65d8 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -1895,6 +1895,8 @@ impl Node { crate::transport::TransportHandle::Udp(transport), ); self.bootstrap_transports.insert(transport_id); + self.bootstrap_transport_npubs + .insert(transport_id, traversal.peer_npub.clone()); let remote_addr = TransportAddr::from_string(&traversal.remote_addr.to_string()); if let Err(err) = self @@ -1902,6 +1904,7 @@ impl Node { .await { self.bootstrap_transports.remove(&transport_id); + self.bootstrap_transport_npubs.remove(&transport_id); if let Some(mut handle) = self.transports.remove(&transport_id) { let _ = handle.stop().await; } diff --git a/src/node/mod.rs b/src/node/mod.rs index bcf9ec0..c6c7d6d 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -458,6 +458,13 @@ pub struct Node { startup_open_discovery_sweep_done: bool, /// Per-peer UDP transports adopted from NAT traversal handoff. bootstrap_transports: HashSet, + /// Originating peer npub (bech32) for each adopted bootstrap + /// transport, captured at `adopt_established_traversal` time. + /// Populated alongside `bootstrap_transports`; cleared in + /// `cleanup_bootstrap_transport_if_unused`. Used by the rx loop to + /// route fatal-protocol-mismatch observations back to the + /// Nostr-discovery `failure_state` for long cooldown application. + bootstrap_transport_npubs: HashMap, // === Periodic Parent Re-evaluation === /// Timestamp of last periodic parent re-evaluation (for pacing). @@ -626,6 +633,7 @@ impl Node { nostr_discovery_started_at_ms: None, startup_open_discovery_sweep_done: false, bootstrap_transports: HashSet::new(), + bootstrap_transport_npubs: HashMap::new(), last_parent_reeval: None, last_congestion_log: None, estimated_mesh_size: None, @@ -758,6 +766,7 @@ impl Node { nostr_discovery_started_at_ms: None, startup_open_discovery_sweep_done: false, bootstrap_transports: HashSet::new(), + bootstrap_transport_npubs: HashMap::new(), last_parent_reeval: None, last_congestion_log: None, estimated_mesh_size: None, @@ -1496,6 +1505,7 @@ impl Node { ); self.bootstrap_transports.remove(&transport_id); + self.bootstrap_transport_npubs.remove(&transport_id); self.transport_drops.remove(&transport_id); self.transports.remove(&transport_id); } diff --git a/src/node/tests/session.rs b/src/node/tests/session.rs index c6dcb49..6ecb502 100644 --- a/src/node/tests/session.rs +++ b/src/node/tests/session.rs @@ -2206,3 +2206,123 @@ async fn test_handle_mtu_exceeded_keeps_tighter_existing_path_mtu_lookup() { "MtuExceeded with looser bottleneck must not loosen a tighter existing value" ); } + +// ============================================================================ +// Proactive PathMtuNotification → path_mtu_lookup focused unit tests +// +// These exercise the receive-side write path that mirrors the proactive +// end-to-end echo into `path_mtu_lookup`. Without this mirror, new TCP +// flows opened on a path the proactive notification has tightened keep +// getting clamped by the staler discovery-time value until a reactive +// MtuExceeded fires for those flows — long-lived stable paths can sit +// in the gap indefinitely. +// ============================================================================ + +/// Build a PathMtuNotification body (2 bytes: path_mtu LE). +fn build_path_mtu_notification_body(mtu: u16) -> Vec { + mtu.to_le_bytes().to_vec() +} + +/// Insert an Established session with MMP initialized so the proactive +/// PathMtuNotification handler can apply notifications. +fn install_established_session_with_mmp(node: &mut Node, remote: &Identity) { + let session = make_noise_session(node.identity(), remote); + let remote_addr = *remote.node_addr(); + let mut entry = crate::node::session::SessionEntry::new( + remote_addr, + remote.pubkey_full(), + EndToEndState::Established(session), + 1000, + true, + ); + entry.init_mmp(&node.config.node.session_mmp); + node.sessions.insert(remote_addr, entry); +} + +#[test] +fn test_handle_path_mtu_notification_writes_path_mtu_lookup_when_empty() { + let mut node = make_node(); + let remote = Identity::generate(); + let remote_addr = *remote.node_addr(); + let remote_fips = crate::FipsAddress::from_node_addr(&remote_addr); + + install_established_session_with_mmp(&mut node, &remote); + + assert!( + node.path_mtu_lookup_get(&remote_fips).is_none(), + "lookup should start empty for this destination" + ); + + let body = build_path_mtu_notification_body(1280); + node.handle_session_path_mtu_notification(&remote_addr, &body); + + assert_eq!( + node.path_mtu_lookup_get(&remote_fips), + Some(1280), + "PathMtuNotification should populate path_mtu_lookup with the reported MTU" + ); +} + +#[test] +fn test_handle_path_mtu_notification_tightens_existing_path_mtu_lookup() { + let mut node = make_node(); + let remote = Identity::generate(); + let remote_addr = *remote.node_addr(); + let remote_fips = crate::FipsAddress::from_node_addr(&remote_addr); + + install_established_session_with_mmp(&mut node, &remote); + + // Pre-seed with a generous value (e.g., from the discovery seed at link + // promotion time, before the destination's proactive echo arrived). + node.path_mtu_lookup_insert(remote_fips, 1500); + + let body = build_path_mtu_notification_body(1280); + node.handle_session_path_mtu_notification(&remote_addr, &body); + + assert_eq!( + node.path_mtu_lookup_get(&remote_fips), + Some(1280), + "PathMtuNotification with smaller MTU must tighten the lookup" + ); +} + +#[test] +fn test_handle_path_mtu_notification_keeps_tighter_existing_path_mtu_lookup() { + let mut node = make_node(); + let remote = Identity::generate(); + let remote_addr = *remote.node_addr(); + let remote_fips = crate::FipsAddress::from_node_addr(&remote_addr); + + install_established_session_with_mmp(&mut node, &remote); + + // Pre-seed with a tighter value than what the proactive notification + // reports (e.g., from a prior reactive MtuExceeded on a narrower hop). + // The mirror must never loosen the clamp. + node.path_mtu_lookup_insert(remote_fips, 1200); + + let body = build_path_mtu_notification_body(1400); + node.handle_session_path_mtu_notification(&remote_addr, &body); + + assert_eq!( + node.path_mtu_lookup_get(&remote_fips), + Some(1200), + "PathMtuNotification with looser MTU must not loosen a tighter existing value" + ); +} + +#[test] +fn test_handle_path_mtu_notification_no_session_no_op() { + let mut node = make_node(); + let remote = Identity::generate(); + let remote_addr = *remote.node_addr(); + let remote_fips = crate::FipsAddress::from_node_addr(&remote_addr); + + // No session installed. The handler should drop the notification entirely. + let body = build_path_mtu_notification_body(1280); + node.handle_session_path_mtu_notification(&remote_addr, &body); + + assert!( + node.path_mtu_lookup_get(&remote_fips).is_none(), + "PathMtuNotification with no session must not touch path_mtu_lookup" + ); +}