Production hardening: unwrap safety, 14 new tests, diagnostics

Harden unwrap() calls in handler hot paths (handshake, encrypted,
rekey, session handlers) with proper error propagation.

Add 14 tests: profile rejection (6), MMP forward-compat (4),
discovery min_mtu pruning (3), XX duplicate msg1 dedup (1).

Add NodeProfile Display impl with structured tracing for profile
mismatch logging. Expose bloom compression diagnostics
(total_compressed_bytes, total_raw_bytes) in control socket.

Add module documentation for XX identity timing, profile decision
tree, and bloom codec strategy.

src/node/bloom.rs

@@ -113,6 +113,8 @@ impl Node {
         } else {
             self.stats_mut().bloom.full_sends += 1;
         }
+        self.stats_mut().bloom.total_compressed_bytes += stats.compressed_bytes as u64;
+        self.stats_mut().bloom.total_raw_bytes += (stats.raw_words * 8) as u64;
 
         // Record send and store the filter for change detection
         debug!(

src/node/handlers/encrypted.rs

@@ -52,7 +52,9 @@ impl Node {
         // K-bit flip detection: peer has cut over to the new session.
         // Check and perform cutover in a scoped borrow.
         {
-            let peer = self.peers.get(&node_addr).unwrap();
+            let Some(peer) = self.peers.get(&node_addr) else {
+                return;
+            };
             let k_bit_flipped =
                 received_k_bit != peer.current_k_bit() && peer.pending_new_session().is_some();
 
@@ -63,7 +65,9 @@ impl Node {
                     "Peer K-bit flip detected, promoting new session"
                 );
 
-                let peer = self.peers.get_mut(&node_addr).unwrap();
+                let Some(peer) = self.peers.get_mut(&node_addr) else {
+                    return;
+                };
                 if let Some(_old_our_index) = peer.handle_peer_kbit_flip() {
                     // New index was pre-registered in peers_by_index during
                     // msg1 handling (handshake.rs). Verify, don't duplicate.
@@ -83,7 +87,9 @@ impl Node {
         // Decrypt: try current session first, then previous (drain fallback)
         let ciphertext = &packet.data[header.ciphertext_offset()..];
         let plaintext = {
-            let peer = self.peers.get_mut(&node_addr).unwrap();
+            let Some(peer) = self.peers.get_mut(&node_addr) else {
+                return;
+            };
             let session = match peer.noise_session_mut() {
                 Some(s) => s,
                 None => {

src/node/handlers/handshake.rs

@@ -377,7 +377,11 @@ impl Node {
             return;
         }
 
-        let conn = self.connections.get_mut(&link_id).unwrap();
+        let Some(conn) = self.connections.get_mut(&link_id) else {
+            warn!(link_id = %link_id, "Connection removed during msg2 processing");
+            self.pending_outbound.remove(&key);
+            return;
+        };
 
         // Create FMP negotiation payload for msg3 (includes profile, MMP bits, bloom TLV)
         let neg_payload = NegotiationPayload::fmp(1, 1, self.node_profile).encode();
@@ -403,7 +407,7 @@ impl Node {
             match process_fmp_negotiation(self.node_profile, conn, neg_bytes) {
                 Ok(()) => {}
                 Err(e) => {
-                    warn!(link_id = %link_id, error = %e, "FMP negotiation failed");
+                    warn!(link_id = %link_id, our_profile = %self.node_profile, error = %e, "FMP negotiation failed");
                     conn.mark_failed();
                     return;
                 }
@@ -524,7 +528,11 @@ impl Node {
                     );
 
                     // Update peers_by_index: remove old inbound index, add outbound
-                    let transport_id = peer.transport_id().unwrap();
+                    let Some(transport_id) = peer.transport_id() else {
+                        warn!(peer = %self.peer_display_name(&peer_node_addr), "Active peer missing transport_id during cross-connection");
+                        self.pending_outbound.remove(&key);
+                        return;
+                    };
                     if let Some(old_idx) = old_our_index {
                         self.peers_by_index
                             .remove(&(transport_id, old_idx.as_u32()));
@@ -746,7 +754,7 @@ impl Node {
             match process_fmp_negotiation(self.node_profile, conn, neg_bytes) {
                 Ok(()) => {}
                 Err(e) => {
-                    warn!(link_id = %link_id, error = %e, "FMP negotiation failed");
+                    warn!(link_id = %link_id, our_profile = %self.node_profile, error = %e, "FMP negotiation failed");
                     self.connections.remove(&link_id);
                     self.remove_link(&link_id);
                     return;
@@ -857,7 +865,11 @@ impl Node {
 
                         // Rekey: process as responder, store new session as pending
                         let noise_session = {
-                            let conn = self.connections.get_mut(&link_id).unwrap();
+                            let Some(conn) = self.connections.get_mut(&link_id) else {
+                                warn!(link_id = %link_id, "Connection removed during rekey msg3 processing");
+                                self.links.remove(&link_id);
+                                return;
+                            };
                             conn.take_session()
                         };
                         let our_new_index = our_index;
@@ -1181,7 +1193,9 @@ impl Node {
 
             if this_wins {
                 // This connection wins, replace the existing peer
-                let old_peer = self.peers.remove(&peer_node_addr).unwrap();
+                let Some(old_peer) = self.peers.remove(&peer_node_addr) else {
+                    return Err(NodeError::PeerNotFound(peer_node_addr));
+                };
                 let loser_link_id = old_peer.link_id();
 
                 // Clean up old peer's index from peers_by_index
@@ -1361,8 +1375,8 @@ fn process_fmp_negotiation(
 
     debug!(
         link_id = %conn.link_id(),
-        our_profile = ?our_profile,
-        peer_profile = ?their_profile,
+        our_profile = %our_profile,
+        peer_profile = %their_profile,
         "FMP negotiation complete"
     );
 

src/node/handlers/rekey.rs

@@ -232,8 +232,10 @@ impl Node {
             if !peer.rekey_in_progress() || peer.rekey_msg1().is_none() {
                 continue;
             }
-            if peer.needs_msg1_resend(now_ms) {
-                to_resend.push((*node_addr, peer.rekey_msg1().unwrap().to_vec()));
+            if peer.needs_msg1_resend(now_ms)
+                && let Some(msg1) = peer.rekey_msg1()
+            {
+                to_resend.push((*node_addr, msg1.to_vec()));
             }
         }
 

src/node/handlers/session.rs

@@ -179,7 +179,9 @@ impl Node {
         // K-bit flip detection: peer has cut over to the new session.
         let received_k_bit = header.flags & FSP_FLAG_K != 0;
         {
-            let entry = self.sessions.get(src_addr).unwrap();
+            let Some(entry) = self.sessions.get(src_addr) else {
+                return;
+            };
             let k_bit_flipped =
                 received_k_bit != entry.current_k_bit() && entry.pending_new_session().is_some();
 
@@ -190,7 +192,9 @@ impl Node {
                     "Peer FSP K-bit flip detected, promoting new session"
                 );
                 let now_ms = Self::now_ms();
-                let entry = self.sessions.get_mut(src_addr).unwrap();
+                let Some(entry) = self.sessions.get_mut(src_addr) else {
+                    return;
+                };
                 entry.handle_peer_kbit_flip(now_ms);
             }
         }
@@ -445,8 +449,9 @@ impl Node {
                             src = %self.peer_display_name(src_addr),
                             "Dual FSP rekey initiation: we lose (larger addr), abandoning ours"
                         );
-                        let entry = self.sessions.get_mut(src_addr).unwrap();
-                        entry.abandon_rekey();
+                        if let Some(entry) = self.sessions.get_mut(src_addr) {
+                            entry.abandon_rekey();
+                        }
                     } else if has_pending {
                         // Guard: already have a pending session waiting for K-bit cutover
                         debug!(
@@ -488,9 +493,10 @@ impl Node {
 
                     // Store rekey state on the existing entry
                     let now_ms = Self::now_ms();
-                    let entry = self.sessions.get_mut(src_addr).unwrap();
-                    entry.set_rekey_state(handshake, false);
-                    entry.record_peer_rekey(now_ms);
+                    if let Some(entry) = self.sessions.get_mut(src_addr) {
+                        entry.set_rekey_state(handshake, false);
+                        entry.record_peer_rekey(now_ms);
+                    }
 
                     debug!(
                         src = %self.peer_display_name(src_addr),
@@ -1809,6 +1815,7 @@ impl Node {
         if original_packet.len() < 40 {
             return;
         }
+        // SAFETY: slice is exactly 16 bytes; length validated above (>= 40)
         let src_addr = Ipv6Addr::from(<[u8; 16]>::try_from(&original_packet[8..24]).unwrap());
 
         // Rate limit ICMP PTB messages per source
@@ -1824,6 +1831,7 @@ impl Node {
         // kernel sees the PTB coming from a remote router, not from itself.
         // Linux ignores PTBs whose source matches a local address, which
         // causes a PMTUD blackhole when both src and ICMP-src are local.
+        // SAFETY: slice is exactly 16 bytes; length validated above (>= 40)
         let dest_addr = Ipv6Addr::from(<[u8; 16]>::try_from(&original_packet[24..40]).unwrap());
         if let Some(response) = build_packet_too_big(original_packet, mtu, dest_addr)
             && let Some(tun_tx) = &self.tun_tx

src/node/stats.rs

@@ -222,6 +222,9 @@ pub struct BloomStats {
     pub nacks_received: u64,
     // Adaptive sizing
     pub size_changes: u64,
+    // Compression tracking
+    pub total_compressed_bytes: u64,
+    pub total_raw_bytes: u64,
 }
 
 impl BloomStats {
@@ -241,6 +244,8 @@ impl BloomStats {
             nacks_sent: self.nacks_sent,
             nacks_received: self.nacks_received,
             size_changes: self.size_changes,
+            total_compressed_bytes: self.total_compressed_bytes,
+            total_raw_bytes: self.total_raw_bytes,
         }
     }
 }
@@ -415,6 +420,8 @@ pub struct BloomStatsSnapshot {
     pub nacks_sent: u64,
     pub nacks_received: u64,
     pub size_changes: u64,
+    pub total_compressed_bytes: u64,
+    pub total_raw_bytes: u64,
 }
 
 #[derive(Clone, Debug, Default, Serialize)]

src/node/tests/discovery.rs

@@ -10,7 +10,7 @@ use crate::protocol::{LookupRequest, LookupResponse};
 use crate::tree::TreeCoordinate;
 use spanning_tree::{
     cleanup_nodes, generate_random_edges, process_available_packets, run_tree_test,
-    verify_tree_convergence,
+    run_tree_test_with_mtus, verify_tree_convergence,
 };
 
 // ============================================================================
@@ -882,3 +882,129 @@ async fn test_originator_stores_path_mtu_in_cache() {
         "Originator should store path_mtu from LookupResponse in cache"
     );
 }
+
+// ============================================================================
+// Integration Tests — min_mtu transit pruning
+// ============================================================================
+
+#[tokio::test]
+async fn test_transit_prunes_lookup_by_min_mtu() {
+    // Topology: node0(1280) — node1(800) — node2(1280)
+    // Node0 initiates lookup for node2 with min_mtu=1280 (default TUN MTU).
+    // Node1's transport MTU is 800 < 1280, so node1 should NOT forward
+    // the request to node2. The lookup should fail (no cache entry).
+    let mtus = [1280, 800, 1280];
+    let edges = vec![(0, 1), (1, 2)];
+    let mut nodes = run_tree_test_with_mtus(&mtus, &edges).await;
+
+    let node2_addr = *nodes[2].node.node_addr();
+    let node2_pubkey = nodes[2].node.identity().pubkey_full();
+    nodes[0].node.register_identity(node2_addr, node2_pubkey);
+
+    nodes[0].node.initiate_lookup(&node2_addr, 8).await;
+
+    for _ in 0..10 {
+        tokio::time::sleep(Duration::from_millis(100)).await;
+        process_available_packets(&mut nodes).await;
+    }
+
+    let now_ms = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as u64)
+        .unwrap_or(0);
+
+    assert!(
+        !nodes[0].node.coord_cache().contains(&node2_addr, now_ms),
+        "Node0 should NOT have cached node2 route (transit pruned by min_mtu)"
+    );
+
+    cleanup_nodes(&mut nodes).await;
+}
+
+#[tokio::test]
+async fn test_transit_forwards_when_mtu_sufficient() {
+    // Topology: node0(1280) — node1(1400) — node2(1280)
+    // Node0 initiates lookup for node2 with min_mtu=1280 (default TUN MTU).
+    // Node1's transport MTU is 1400 >= 1280, so the request passes through.
+    // Node1 annotates path_mtu = min(u16::MAX, 1400) = 1400 on response.
+    let mtus = [1280, 1400, 1280];
+    let edges = vec![(0, 1), (1, 2)];
+    let mut nodes = run_tree_test_with_mtus(&mtus, &edges).await;
+
+    let node2_addr = *nodes[2].node.node_addr();
+    let node2_pubkey = nodes[2].node.identity().pubkey_full();
+    nodes[0].node.register_identity(node2_addr, node2_pubkey);
+
+    nodes[0].node.initiate_lookup(&node2_addr, 8).await;
+
+    for _ in 0..10 {
+        tokio::time::sleep(Duration::from_millis(100)).await;
+        process_available_packets(&mut nodes).await;
+    }
+
+    let now_ms = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as u64)
+        .unwrap_or(0);
+
+    assert!(
+        nodes[0].node.coord_cache().contains(&node2_addr, now_ms),
+        "Node0 should have cached node2 route (MTU sufficient)"
+    );
+
+    let entry = nodes[0].node.coord_cache().get_entry(&node2_addr).unwrap();
+    let path_mtu = entry.path_mtu().expect("path_mtu should be set");
+    assert_eq!(
+        path_mtu, 1400,
+        "path_mtu should reflect transit node's transport MTU (1400)"
+    );
+
+    cleanup_nodes(&mut nodes).await;
+}
+
+#[tokio::test]
+async fn test_response_path_mtu_four_node_chain() {
+    // Topology: node0(1280) — node1(1400) — node2(900) — node3(1280)
+    // Node0 initiates lookup for node3. Response travels node3→node2→node1→node0.
+    // Transit nodes apply min(): node2 sees min(u16::MAX, 900) = 900,
+    // node1 sees min(900, 1400) = 900.
+    // Final path_mtu at node0 should be 900 (bottleneck at node2).
+    //
+    // Note: min_mtu=1280 from TUN config. Node2's MTU (900) < 1280 would prune
+    // the forward request at node2, so node3 would never be reached. To test
+    // path_mtu annotation we need all transit links to pass the min_mtu check.
+    // Use MTUs above 1280 to avoid pruning but with different values to verify min().
+    let mtus = [1280, 1500, 1350, 1280];
+    let edges = vec![(0, 1), (1, 2), (2, 3)];
+    let mut nodes = run_tree_test_with_mtus(&mtus, &edges).await;
+
+    let node3_addr = *nodes[3].node.node_addr();
+    let node3_pubkey = nodes[3].node.identity().pubkey_full();
+    nodes[0].node.register_identity(node3_addr, node3_pubkey);
+
+    nodes[0].node.initiate_lookup(&node3_addr, 8).await;
+
+    for _ in 0..15 {
+        tokio::time::sleep(Duration::from_millis(100)).await;
+        process_available_packets(&mut nodes).await;
+    }
+
+    let now_ms = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as u64)
+        .unwrap_or(0);
+
+    assert!(
+        nodes[0].node.coord_cache().contains(&node3_addr, now_ms),
+        "Node0 should have cached node3 route"
+    );
+
+    let entry = nodes[0].node.coord_cache().get_entry(&node3_addr).unwrap();
+    let path_mtu = entry.path_mtu().expect("path_mtu should be set");
+    assert_eq!(
+        path_mtu, 1350,
+        "Four-node chain path_mtu should be min of transit MTUs (1350)"
+    );
+
+    cleanup_nodes(&mut nodes).await;
+}

src/node/tests/handshake.rs

@@ -1,6 +1,7 @@
 //! Integration tests for end-to-end Noise XX handshake scenarios.
 
 use super::*;
+use super::spanning_tree::{cleanup_nodes, drain_all_packets, initiate_handshake, make_test_node};
 
 #[tokio::test]
 async fn test_two_node_handshake_udp() {
@@ -989,3 +990,121 @@ async fn test_duplicate_msg2_dropped() {
     assert_eq!(node.connection_count(), 0);
     assert_eq!(node.peer_count(), 0);
 }
+
+// ===== Profile Rejection Tests =====
+
+/// Helper: create two test nodes, set their profiles, attempt a handshake,
+/// and return whether they successfully peered.
+async fn attempt_profile_handshake(
+    profile_a: crate::protocol::NodeProfile,
+    profile_b: crate::protocol::NodeProfile,
+) -> (usize, usize) {
+    let mut nodes = vec![make_test_node().await, make_test_node().await];
+    nodes[0].node.node_profile = profile_a;
+    nodes[1].node.node_profile = profile_b;
+
+    initiate_handshake(&mut nodes, 0, 1).await;
+    drain_all_packets(&mut nodes, false).await;
+
+    let peers = (nodes[0].node.peer_count(), nodes[1].node.peer_count());
+    cleanup_nodes(&mut nodes).await;
+    peers
+}
+
+#[tokio::test]
+async fn test_nonrouting_nonrouting_rejected() {
+    use crate::protocol::NodeProfile;
+    let (a, b) = attempt_profile_handshake(NodeProfile::NonRouting, NodeProfile::NonRouting).await;
+    assert_eq!(a, 0, "NonRouting↔NonRouting should reject: node A");
+    assert_eq!(b, 0, "NonRouting↔NonRouting should reject: node B");
+}
+
+#[tokio::test]
+async fn test_leaf_leaf_rejected() {
+    use crate::protocol::NodeProfile;
+    let (a, b) = attempt_profile_handshake(NodeProfile::Leaf, NodeProfile::Leaf).await;
+    assert_eq!(a, 0, "Leaf↔Leaf should reject: node A");
+    assert_eq!(b, 0, "Leaf↔Leaf should reject: node B");
+}
+
+#[tokio::test]
+async fn test_nonrouting_leaf_rejected() {
+    use crate::protocol::NodeProfile;
+    let (a, b) = attempt_profile_handshake(NodeProfile::NonRouting, NodeProfile::Leaf).await;
+    assert_eq!(a, 0, "NonRouting↔Leaf should reject: node A");
+    assert_eq!(b, 0, "NonRouting↔Leaf should reject: node B");
+}
+
+#[tokio::test]
+async fn test_leaf_nonrouting_rejected() {
+    use crate::protocol::NodeProfile;
+    let (a, b) = attempt_profile_handshake(NodeProfile::Leaf, NodeProfile::NonRouting).await;
+    assert_eq!(a, 0, "Leaf↔NonRouting should reject: node A");
+    assert_eq!(b, 0, "Leaf↔NonRouting should reject: node B");
+}
+
+#[tokio::test]
+async fn test_full_nonrouting_accepted() {
+    use crate::protocol::NodeProfile;
+    let (a, b) = attempt_profile_handshake(NodeProfile::Full, NodeProfile::NonRouting).await;
+    assert_eq!(a, 1, "Full↔NonRouting should accept: node A");
+    assert_eq!(b, 1, "Full↔NonRouting should accept: node B");
+}
+
+#[tokio::test]
+async fn test_full_leaf_accepted() {
+    use crate::protocol::NodeProfile;
+    let (a, b) = attempt_profile_handshake(NodeProfile::Full, NodeProfile::Leaf).await;
+    assert_eq!(a, 1, "Full↔Leaf should accept: node A");
+    assert_eq!(b, 1, "Full↔Leaf should accept: node B");
+}
+
+// ===== XX Address-Based Dedup Tests =====
+
+#[tokio::test]
+async fn test_xx_duplicate_msg1_resends_msg2() {
+    use crate::node::wire::build_msg1;
+    use crate::transport::ReceivedPacket;
+
+    // Node B with NO transport — msg2 send silently skips (if let Some check),
+    // but the pending connection and link are created.
+    let mut node_b = make_node();
+    let transport_id = TransportId::new(1);
+
+    // Build a valid XX msg1 from an external initiator
+    let initiator = Identity::generate();
+    let mut hs = crate::noise::HandshakeState::new_initiator(initiator.keypair());
+    let noise_msg1 = hs.write_message_1().unwrap();
+    let sender_idx = SessionIndex::new(42);
+    let wire_msg1 = build_msg1(sender_idx, &noise_msg1);
+
+    let remote_addr = TransportAddr::from_string("10.0.0.1:2121");
+
+    // First msg1 → B creates pending inbound connection
+    let first_packet = ReceivedPacket {
+        transport_id,
+        remote_addr: remote_addr.clone(),
+        data: wire_msg1.clone(),
+        timestamp_ms: 1000,
+    };
+    node_b.handle_msg1(first_packet).await;
+
+    assert_eq!(node_b.connection_count(), 1, "B: 1 connection after first msg1");
+    assert_eq!(node_b.peer_count(), 0, "B: 0 peers (XX, no promotion at msg1)");
+
+    // Duplicate msg1 from same address → dedup triggers msg2 resend, not new handshake
+    let dup_packet = ReceivedPacket {
+        transport_id,
+        remote_addr: remote_addr.clone(),
+        data: wire_msg1.clone(),
+        timestamp_ms: 1100,
+    };
+    node_b.handle_msg1(dup_packet).await;
+
+    assert_eq!(
+        node_b.connection_count(),
+        1,
+        "B: still 1 connection after duplicate msg1 (dedup, not new handshake)"
+    );
+    assert_eq!(node_b.peer_count(), 0, "B: still 0 peers");
+}