Fix FSP rekey cutover race and MMP metric discontinuity

The FSP XK rekey handshake has a race condition where the initiator
can cut over (K-bit flip) and send data encrypted with the new session
before msg3 reaches the responder. The responder has no pending session
yet, so K-bit detection fails and packets are dropped.

Defer FSP initiator cutover by 2 seconds after handshake completion
(FSP_CUTOVER_DELAY_MS) to give msg3 time to traverse the mesh. FMP
(IK, 2 messages) is unaffected since the responder completes during
msg1 processing.

Also fix MMP metric corruption after rekey cutover: the new session
starts with counter 0 but MMP state carries highest_counter and
GapTracker.expected_next from the old session, producing false
reorder counts, jitter spikes, and invalid OWD trends. Add
reset_for_rekey() methods that clear counter-dependent state while
preserving RTT estimates.

Additional fixes:
- Remove stale peers_by_index entry on abandon_rekey error path
- Replace redundant peers_by_index inserts with debug assertions
  verifying the pre-registration invariant
- Tighten rekey integration test to zero tolerance (was 4 failures)

src/node/handlers/encrypted.rs

@@ -64,13 +64,16 @@ impl Node {
                 );
 
                 let peer = self.peers.get_mut(&node_addr).unwrap();
-                if let Some(_old_our_index) = peer.handle_peer_kbit_flip()
-                    && let (Some(transport_id), Some(new_our_index)) =
-                        (peer.transport_id(), peer.our_index())
-                {
-                    self.peers_by_index.insert(
-                        (transport_id, new_our_index.as_u32()),
-                        node_addr,
+                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.
+                    debug_assert!(
+                        peer.transport_id().is_some()
+                            && peer.our_index().is_some()
+                            && self.peers_by_index.contains_key(
+                                &(peer.transport_id().unwrap(), peer.our_index().unwrap().as_u32())
+                            ),
+                        "peers_by_index should contain pre-registered new index after K-bit flip"
                     );
                 }
             }

src/node/handlers/handshake.rs

@@ -537,6 +537,9 @@ impl Node {
                                 "Rekey msg2 processing failed"
                             );
                             if let Some(idx) = peer.abandon_rekey() {
+                                if let Some(tid) = peer.transport_id() {
+                                    self.peers_by_index.remove(&(tid, idx.as_u32()));
+                                }
                                 let _ = self.index_allocator.free(idx);
                             }
                         }

src/node/handlers/rekey.rs

@@ -19,6 +19,10 @@ const DRAIN_WINDOW_SECS: u64 = 10;
 /// a peer's rekey msg1.
 const REKEY_DAMPENING_SECS: u64 = 30;
 
+/// Delay FSP initiator cutover after handshake completion to allow
+/// XK msg3 to reach the responder before K-bit-flipped data arrives.
+const FSP_CUTOVER_DELAY_MS: u64 = 2000;
+
 impl Node {
     /// Periodic rekey check. Called from the tick loop.
     ///
@@ -79,14 +83,16 @@ impl Node {
             if let Some(peer) = self.peers.get_mut(&node_addr)
                 && let Some(_old_our_index) = peer.cutover_to_new_session()
             {
-                if let (Some(transport_id), Some(new_our_index)) =
-                    (peer.transport_id(), peer.our_index())
-                {
-                    self.peers_by_index.insert(
-                        (transport_id, new_our_index.as_u32()),
-                        node_addr,
-                    );
-                }
+                // New index was pre-registered in peers_by_index during
+                // msg2 handling (handshake.rs). Verify, don't duplicate.
+                debug_assert!(
+                    peer.transport_id().is_some()
+                        && peer.our_index().is_some()
+                        && self.peers_by_index.contains_key(
+                            &(peer.transport_id().unwrap(), peer.our_index().unwrap().as_u32())
+                        ),
+                    "peers_by_index should contain pre-registered new index after cutover"
+                );
                 info!(
                     peer = %self.peer_display_name(&node_addr),
                     "Rekey cutover complete (initiator), K-bit flipped"
@@ -281,10 +287,14 @@ impl Node {
                 continue;
             }
 
-            // 1. Initiator-side cutover: completed rekey, pending session ready
+            // 1. Initiator-side cutover: completed rekey, pending session ready.
+            //    Defer cutover until msg3 has had time to reach the responder.
+            //    Without this delay, K-bit-flipped data can arrive before
+            //    msg3, causing decryption failures on the responder.
             if entry.pending_new_session().is_some()
                 && !entry.has_rekey_in_progress()
                 && entry.is_rekey_initiator()
+                && now_ms.saturating_sub(entry.rekey_completed_ms()) >= FSP_CUTOVER_DELAY_MS
             {
                 sessions_to_cutover.push(*node_addr);
                 continue;

src/node/handlers/session.rs

@@ -595,6 +595,7 @@ impl Node {
             };
 
             entry.set_pending_session(session);
+            entry.set_rekey_completed_ms(Self::now_ms());
             self.sessions.insert(*src_addr, entry);
 
             debug!(

src/node/session.rs

@@ -107,6 +107,9 @@ pub(crate) struct SessionEntry {
     rekey_initiator: bool,
     /// Dampening: last time peer sent us a rekey msg1 (Unix ms).
     last_peer_rekey_ms: u64,
+    /// When the FSP rekey handshake completed (initiator sent msg3, Unix ms).
+    /// Used to defer cutover until msg3 has time to reach the responder.
+    rekey_completed_ms: u64,
 }
 
 impl SessionEntry {
@@ -142,6 +145,7 @@ impl SessionEntry {
             pending_new_session: None,
             rekey_initiator: false,
             last_peer_rekey_ms: 0,
+            rekey_completed_ms: 0,
         }
     }
 
@@ -369,6 +373,16 @@ impl SessionEntry {
         }
     }
 
+    /// When the FSP rekey handshake completed (initiator sent msg3).
+    pub(crate) fn rekey_completed_ms(&self) -> u64 {
+        self.rekey_completed_ms
+    }
+
+    /// Record when the FSP rekey handshake completed (initiator side).
+    pub(crate) fn set_rekey_completed_ms(&mut self, ms: u64) {
+        self.rekey_completed_ms = ms;
+    }
+
     /// Store a completed rekey session.
     pub(crate) fn set_pending_session(&mut self, session: NoiseSession) {
         self.pending_new_session = Some(session);
@@ -408,6 +422,12 @@ impl SessionEntry {
         self.session_start_ms = now_ms;
         self.rekey_state = None;
         self.rekey_initiator = false;
+        self.rekey_completed_ms = 0;
+
+        // Reset MMP counters to avoid metric discontinuity
+        if let Some(mmp) = &mut self.mmp {
+            mmp.reset_for_rekey();
+        }
         true
     }
 
@@ -430,6 +450,11 @@ impl SessionEntry {
         self.session_start_ms = now_ms;
         self.rekey_state = None;
         self.rekey_initiator = false;
+
+        // Reset MMP counters to avoid metric discontinuity
+        if let Some(mmp) = &mut self.mmp {
+            mmp.reset_for_rekey();
+        }
         true
     }