Merge master into next (coord cache surgical invalidation)

# Conflicts:
#	CHANGELOG.md

src/node/handlers/forwarding.rs

@@ -12,7 +12,8 @@ use crate::node::session_wire::{
 };
 use crate::node::{Node, NodeError};
 use crate::protocol::{
-    CoordsRequired, MtuExceeded, PathBroken, SessionAck, SessionDatagram, SessionSetup,
+    CoordsRequired, MtuExceeded, PathBroken, SessionAck, SessionDatagram, SessionDatagramRef,
+    SessionSetup,
 };
 use std::time::{Duration, Instant};
 use tracing::{debug, warn};
@@ -30,7 +31,7 @@ impl Node {
     ) {
         self.stats_mut().forwarding.record_received(payload.len());
 
-        let mut datagram = match SessionDatagram::decode(payload) {
+        let datagram_ref = match SessionDatagramRef::decode(payload) {
             Ok(dg) => dg,
             Err(e) => {
                 self.stats_mut()
@@ -41,35 +42,41 @@ impl Node {
             }
         };
 
-        // TTL enforcement: decrement and drop if exhausted
-        if !datagram.decrement_ttl() {
+        // TTL enforcement: decrement for forwarding and drop only if the
+        // received datagram was already exhausted.
+        if datagram_ref.ttl == 0 {
             self.stats_mut()
                 .forwarding
                 .record_ttl_exhausted(payload.len());
             debug!(
-                src = %datagram.src_addr,
-                dest = %datagram.dest_addr,
+                src = %datagram_ref.src_addr,
+                dest = %datagram_ref.dest_addr,
                 "SessionDatagram TTL exhausted, dropping"
             );
             return;
         }
+        let forwarded_ttl = datagram_ref.ttl - 1;
 
         // Coordinate cache warming from plaintext session-layer headers
-        self.try_warm_coord_cache(&datagram);
+        self.try_warm_coord_cache_ref(&datagram_ref);
 
-        // Local delivery: dispatch to session layer handlers
-        if datagram.dest_addr == *self.node_addr() {
+        // Local delivery: dispatch to session layer handlers without
+        // materializing an owned SessionDatagram payload Vec.
+        if datagram_ref.dest_addr == *self.node_addr() {
             self.stats_mut().forwarding.record_delivered(payload.len());
             self.handle_session_payload(
-                &datagram.src_addr,
-                &datagram.payload,
-                datagram.path_mtu,
+                &datagram_ref.src_addr,
+                datagram_ref.payload,
+                datagram_ref.path_mtu,
                 incoming_ce,
             )
             .await;
             return;
         }
 
+        let mut datagram = datagram_ref.into_owned();
+        datagram.ttl = forwarded_ttl;
+
         // Find next hop toward destination
         let next_hop_addr = match self.find_next_hop(&datagram.dest_addr) {
             Some(peer) => *peer.node_addr(),
@@ -153,8 +160,8 @@ impl Node {
     ///
     /// Decode failures are logged and silently ignored — they don't block
     /// forwarding.
-    fn try_warm_coord_cache(&mut self, datagram: &SessionDatagram) {
-        let prefix = match FspCommonPrefix::parse(&datagram.payload) {
+    fn try_warm_coord_cache_ref(&mut self, datagram: &SessionDatagramRef<'_>) {
+        let prefix = match FspCommonPrefix::parse(datagram.payload) {
             Some(p) => p,
             None => return,
         };

src/node/handlers/mmp.rs

@@ -155,7 +155,9 @@ impl Node {
                     warn!(error = %e, "Failed to sign declaration after first-RTT parent eval");
                     return;
                 }
-                self.coord_cache.clear();
+                // Surgical invalidation — see CoordCache::invalidate_via_node doc.
+                self.coord_cache
+                    .invalidate_via_node(self.identity.node_addr());
                 self.reset_discovery_backoff();
                 self.stats_mut().tree.parent_switched += 1;
                 self.stats_mut().tree.parent_switches += 1;
@@ -180,7 +182,9 @@ impl Node {
                     warn!(error = %e, "Failed to sign self-root declaration after first-RTT");
                     return;
                 }
-                self.coord_cache.clear();
+                // Surgical invalidation — see CoordCache::invalidate_other_roots doc.
+                self.coord_cache
+                    .invalidate_other_roots(self.identity.node_addr());
                 self.reset_discovery_backoff();
                 self.stats_mut().tree.parent_switched += 1;
                 self.stats_mut().tree.parent_switches += 1;

src/node/lifecycle.rs

@@ -122,12 +122,7 @@ impl Node {
         }
 
         // Check if connection already in progress to this peer
-        let already_connecting = self.connections.values().any(|conn| {
-            conn.expected_identity()
-                .map(|id| id.node_addr() == &peer_node_addr)
-                .unwrap_or(false)
-        });
-        if already_connecting {
+        if self.is_connecting_to_peer(&peer_node_addr) {
             debug!(
                 npub = %peer_config.npub,
                 "Connection already in progress, skipping"
@@ -139,6 +134,14 @@ impl Node {
             .await
     }
 
+    fn is_connecting_to_peer(&self, peer_node_addr: &NodeAddr) -> bool {
+        self.connections.values().any(|conn| {
+            conn.expected_identity()
+                .map(|id| id.node_addr() == peer_node_addr)
+                .unwrap_or(false)
+        })
+    }
+
     /// Initiate a connection to a peer on a specific transport and address.
     ///
     /// For connectionless transports (UDP, Ethernet): allocates a link, starts
@@ -451,6 +454,23 @@ impl Node {
             match event {
                 BootstrapEvent::Established { traversal } => {
                     let peer_npub = traversal.peer_npub.clone();
+                    if let Ok(peer_identity) = PeerIdentity::from_npub(&peer_npub) {
+                        let peer_addr = *peer_identity.node_addr();
+                        if self.peers.contains_key(&peer_addr) {
+                            debug!(
+                                peer_npub = %peer_npub,
+                                "Ignoring established NAT traversal for already-connected peer"
+                            );
+                            continue;
+                        }
+                        if self.is_connecting_to_peer(&peer_addr) {
+                            debug!(
+                                peer_npub = %peer_npub,
+                                "Ignoring established NAT traversal while peer handshake is already in progress"
+                            );
+                            continue;
+                        }
+                    }
                     match self.adopt_established_traversal(traversal).await {
                         Ok(_) => {
                             info!(peer_npub = %peer_npub, "Adopted NAT traversal socket");
@@ -467,6 +487,28 @@ impl Node {
                     peer_config,
                     reason,
                 } => {
+                    let peer_identity = match PeerIdentity::from_npub(&peer_config.npub) {
+                        Ok(identity) => identity,
+                        Err(_) => continue,
+                    };
+                    let node_addr = *peer_identity.node_addr();
+                    if self.peers.contains_key(&node_addr) {
+                        debug!(
+                            npub = %peer_config.npub,
+                            error = %reason,
+                            "Ignoring failed NAT traversal for already-connected peer"
+                        );
+                        continue;
+                    }
+                    if self.is_connecting_to_peer(&node_addr) {
+                        debug!(
+                            npub = %peer_config.npub,
+                            error = %reason,
+                            "Ignoring failed NAT traversal while peer handshake is already in progress"
+                        );
+                        continue;
+                    }
+
                     let now_ms = Self::now_ms();
                     let decision = bootstrap.record_traversal_failure(&peer_config.npub, now_ms);
                     if decision.should_warn {
@@ -517,11 +559,6 @@ impl Node {
                         });
                     }
 
-                    let peer_identity = match PeerIdentity::from_npub(&peer_config.npub) {
-                        Ok(identity) => identity,
-                        Err(_) => continue,
-                    };
-
                     if self
                         .try_peer_addresses(&peer_config, peer_identity, false)
                         .await
@@ -530,7 +567,6 @@ impl Node {
                         continue;
                     }
 
-                    let node_addr = *peer_identity.node_addr();
                     self.schedule_retry(node_addr, now_ms);
                     if let Some(cooldown_until_ms) = decision.cooldown_until_ms
                         && let Some(state) = self.retry_pending.get_mut(&node_addr)
@@ -1737,6 +1773,22 @@ impl Node {
         peer_identity: PeerIdentity,
         allow_bootstrap_nat: bool,
     ) -> Result<(), NodeError> {
+        let peer_node_addr = *peer_identity.node_addr();
+        if self.peers.contains_key(&peer_node_addr) {
+            debug!(
+                npub = %peer_config.npub,
+                "Peer already exists, skipping address attempts"
+            );
+            return Ok(());
+        }
+        if self.is_connecting_to_peer(&peer_node_addr) {
+            debug!(
+                npub = %peer_config.npub,
+                "Connection already in progress, skipping address attempts"
+            );
+            return Ok(());
+        }
+
         // Static-first dialing: avoid delaying configured address attempts on
         // advert fetch/network latency.
         let static_addresses = self.static_peer_addresses(peer_config);
@@ -1880,6 +1932,20 @@ impl Node {
             }
         })?;
         let peer_node_addr = *peer_identity.node_addr();
+        if self.peers.contains_key(&peer_node_addr) {
+            debug!(
+                peer_npub = %traversal.peer_npub,
+                "Ignoring NAT traversal handoff for already-connected peer"
+            );
+            return Err(NodeError::PeerAlreadyExists(peer_node_addr));
+        }
+        if self.is_connecting_to_peer(&peer_node_addr) {
+            debug!(
+                peer_npub = %traversal.peer_npub,
+                "Ignoring NAT traversal handoff while peer handshake is already in progress"
+            );
+            return Err(NodeError::PeerAlreadyExists(peer_node_addr));
+        }
 
         self.peer_aliases
             .insert(peer_node_addr, peer_identity.short_npub());

src/node/tests/bootstrap.rs

@@ -133,6 +133,51 @@ async fn test_failed_adopted_traversal_cleans_up_transport() {
     );
 }
 
+#[tokio::test]
+async fn test_adopted_traversal_skips_already_connected_peer() {
+    let mut node = make_node();
+    let (packet_tx, packet_rx) = packet_channel(64);
+    node.packet_tx = Some(packet_tx);
+    node.packet_rx = Some(packet_rx);
+    node.state = NodeState::Running;
+
+    let transport_id = TransportId::new(1);
+    let link_id = LinkId::new(1);
+    let (conn, peer_identity) = make_completed_connection(&mut node, link_id, transport_id, 1_000);
+    let peer_node_addr = *peer_identity.node_addr();
+    node.add_connection(conn).unwrap();
+    node.promote_connection(link_id, peer_identity, 2_000)
+        .unwrap();
+
+    let link_count = node.link_count();
+    let transport_count = node.transport_count();
+
+    let adopted_socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
+    let handoff = EstablishedTraversal::new(
+        "sess-stale",
+        peer_identity.npub(),
+        "127.0.0.1:9".parse().unwrap(),
+        adopted_socket,
+    )
+    .with_transport_name("nostr-stale");
+
+    let result = node.adopt_established_traversal(handoff).await;
+    assert!(
+        matches!(result, Err(NodeError::PeerAlreadyExists(addr)) if addr == peer_node_addr),
+        "stale traversal handoff should be ignored once the peer is already active"
+    );
+    assert_eq!(
+        node.link_count(),
+        link_count,
+        "ignored traversal must not create a duplicate link"
+    );
+    assert_eq!(
+        node.transport_count(),
+        transport_count,
+        "ignored traversal must not leak an adopted transport"
+    );
+}
+
 #[tokio::test]
 async fn test_third_peer_can_handshake_via_adopted_transport_socket() {
     let mut node_a = make_node(); // Existing traversal peer (Alice)

src/node/tests/unit.rs

@@ -1,7 +1,9 @@
 use super::*;
+use crate::discovery::nostr::{BootstrapEvent, NostrDiscovery};
 use crate::peer::PromotionResult;
 use crate::transport::udp::UdpTransport;
 use crate::transport::{TransportHandle, packet_channel};
+use std::sync::Arc;
 
 #[test]
 fn test_node_creation() {
@@ -782,6 +784,135 @@ fn test_schedule_retry_skips_connected_peer() {
     );
 }
 
+#[tokio::test]
+async fn test_try_peer_addresses_skips_connected_peer() {
+    let mut node = make_node();
+    let transport_id = TransportId::new(1);
+    let link_id = LinkId::new(1);
+    let (conn, peer_identity) = make_completed_connection(&mut node, link_id, transport_id, 1000);
+    let peer_config = crate::config::PeerConfig::new(peer_identity.npub(), "udp", "127.0.0.1:9");
+
+    node.add_connection(conn).unwrap();
+    node.promote_connection(link_id, peer_identity, 2000)
+        .unwrap();
+    let link_count = node.link_count();
+    let connection_count = node.connection_count();
+
+    node.try_peer_addresses(&peer_config, peer_identity, true)
+        .await
+        .unwrap();
+
+    assert_eq!(
+        node.link_count(),
+        link_count,
+        "stale retry/traversal fallback must not create a duplicate link"
+    );
+    assert_eq!(
+        node.connection_count(),
+        connection_count,
+        "stale retry/traversal fallback must not create a duplicate handshake"
+    );
+}
+
+#[tokio::test]
+async fn test_try_peer_addresses_skips_connecting_peer() {
+    let mut node = make_node();
+    let peer_identity = make_peer_identity();
+    let peer_config = crate::config::PeerConfig::new(peer_identity.npub(), "udp", "127.0.0.1:9");
+    let pending = PeerConnection::outbound(LinkId::new(1), peer_identity, 1000);
+    node.add_connection(pending).unwrap();
+
+    node.try_peer_addresses(&peer_config, peer_identity, true)
+        .await
+        .unwrap();
+
+    assert_eq!(
+        node.connection_count(),
+        1,
+        "stale retry/traversal fallback must not start a second handshake"
+    );
+    assert_eq!(
+        node.link_count(),
+        0,
+        "stale retry/traversal fallback must not allocate a link while a handshake is pending"
+    );
+}
+
+#[tokio::test]
+async fn test_nostr_traversal_failure_skips_connected_peer() {
+    let mut node = make_node();
+    let transport_id = TransportId::new(1);
+    let link_id = LinkId::new(1);
+    let (conn, peer_identity) = make_completed_connection(&mut node, link_id, transport_id, 1000);
+    node.add_connection(conn).unwrap();
+    node.promote_connection(link_id, peer_identity, 2000)
+        .unwrap();
+
+    let bootstrap = Arc::new(NostrDiscovery::new_for_test());
+    bootstrap.push_event_for_test(BootstrapEvent::Failed {
+        peer_config: crate::config::PeerConfig::new(peer_identity.npub(), "udp", "127.0.0.1:9"),
+        reason: "stale traversal failure".to_string(),
+    });
+    node.nostr_discovery = Some(bootstrap.clone());
+
+    node.poll_nostr_discovery().await;
+
+    assert!(
+        bootstrap.failure_state_snapshot().is_empty(),
+        "stale failures for connected peers must not affect traversal cooldown"
+    );
+    assert!(
+        node.retry_pending.is_empty(),
+        "stale failures for connected peers must not enqueue reconnect attempts"
+    );
+}
+
+#[tokio::test]
+async fn test_nostr_traversal_established_skips_connected_peer() {
+    use crate::discovery::EstablishedTraversal;
+    use std::net::UdpSocket;
+
+    let mut node = make_node();
+    let transport_id = TransportId::new(1);
+    let link_id = LinkId::new(1);
+    let (conn, peer_identity) = make_completed_connection(&mut node, link_id, transport_id, 1000);
+    node.add_connection(conn).unwrap();
+    node.promote_connection(link_id, peer_identity, 2000)
+        .unwrap();
+    let link_count = node.link_count();
+    let connection_count = node.connection_count();
+
+    let bootstrap = Arc::new(NostrDiscovery::new_for_test());
+    let socket = UdpSocket::bind("127.0.0.1:0").expect("bind local UDP socket");
+    let remote_addr = "127.0.0.1:9999".parse().expect("parse remote addr");
+    bootstrap.push_event_for_test(BootstrapEvent::Established {
+        traversal: EstablishedTraversal::new(
+            "test-session",
+            peer_identity.npub(),
+            remote_addr,
+            socket,
+        ),
+    });
+    node.nostr_discovery = Some(bootstrap.clone());
+
+    node.poll_nostr_discovery().await;
+
+    assert_eq!(
+        node.link_count(),
+        link_count,
+        "stale established handoff must not allocate a new link"
+    );
+    assert_eq!(
+        node.connection_count(),
+        connection_count,
+        "stale established handoff must not start a new handshake"
+    );
+    assert!(
+        node.retry_pending.is_empty(),
+        "stale established handoff must not enqueue a reconnect"
+    );
+}
+
 #[tokio::test]
 async fn test_process_pending_retries_drops_expired_entries() {
     let mut node = make_node();

src/node/tree.rs

@@ -255,7 +255,9 @@ impl Node {
                 warn!(error = %e, "Failed to sign declaration after parent switch");
                 return;
             }
-            self.coord_cache.clear();
+            // Surgical invalidation — see CoordCache::invalidate_via_node doc.
+            self.coord_cache
+                .invalidate_via_node(self.identity.node_addr());
             self.reset_discovery_backoff();
 
             self.stats_mut().tree.parent_switched += 1;
@@ -266,7 +268,7 @@ impl Node {
                 new_seq = new_seq,
                 new_root = %self.tree_state.root(),
                 depth = self.tree_state.my_coords().depth(),
-                "Parent switched, flushed coord cache, announcing to all peers"
+                "Parent switched, invalidated downstream coord cache entries, announcing to all peers"
             );
             if flap_dampened {
                 self.stats_mut().tree.flap_dampened += 1;
@@ -286,7 +288,9 @@ impl Node {
                 warn!(error = %e, "Failed to sign self-root declaration");
                 return;
             }
-            self.coord_cache.clear();
+            // Surgical invalidation — see CoordCache::invalidate_other_roots doc.
+            self.coord_cache
+                .invalidate_other_roots(self.identity.node_addr());
             self.reset_discovery_backoff();
             self.stats_mut().tree.parent_switched += 1;
             self.stats_mut().tree.parent_switches += 1;
@@ -320,7 +324,12 @@ impl Node {
                         warn!(error = %e, "Failed to sign declaration after loop detection");
                         return;
                     }
-                    self.coord_cache.clear();
+                    // handle_parent_lost may promote to root OR find new parent;
+                    // cover both invalidation classes.
+                    self.coord_cache
+                        .invalidate_via_node(self.identity.node_addr());
+                    self.coord_cache
+                        .invalidate_other_roots(self.tree_state.root());
                     self.reset_discovery_backoff();
                     self.send_tree_announce_to_all().await;
                 }
@@ -355,7 +364,9 @@ impl Node {
                 warn!(error = %e, "Failed to sign declaration after parent update");
                 return;
             }
-            self.coord_cache.clear();
+            // Surgical invalidation — see CoordCache::invalidate_via_node doc.
+            self.coord_cache
+                .invalidate_via_node(self.identity.node_addr());
             self.reset_discovery_backoff();
 
             let new_addrs: Vec<NodeAddr> =
@@ -445,7 +456,9 @@ impl Node {
                 warn!(error = %e, "Failed to sign declaration after periodic parent re-eval");
                 return;
             }
-            self.coord_cache.clear();
+            // Surgical invalidation — see CoordCache::invalidate_via_node doc.
+            self.coord_cache
+                .invalidate_via_node(self.identity.node_addr());
             self.reset_discovery_backoff();
 
             self.stats_mut().tree.parent_switched += 1;
@@ -474,7 +487,9 @@ impl Node {
                 warn!(error = %e, "Failed to sign self-root declaration in periodic reeval");
                 return;
             }
-            self.coord_cache.clear();
+            // Surgical invalidation — see CoordCache::invalidate_other_roots doc.
+            self.coord_cache
+                .invalidate_other_roots(self.identity.node_addr());
             self.reset_discovery_backoff();
             self.stats_mut().tree.parent_switched += 1;
             self.stats_mut().tree.parent_switches += 1;