Merge branch 'master' into next

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

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.*`).

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<u64> {
@@ -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

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<NostrPeerFailureView> {
         self.failure_state

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;
         }
 

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"
+                );
+            }
         }
     }
 

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;
             }

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<TransportId>,
+    /// 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<TransportId, String>,
 
     // === 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);
     }

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<u8> {
+    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"
+    );
+}