fmp: bound rekey msg1 retransmission and make link-dead heartbeat rekey-aware

The FMP rekey msg1 resend driver retransmitted indefinitely with no cap
and no abandon, so a rekey that never completed kept resending msg1
forever. Give it a retransmission budget: cap resends at
handshake_max_resends with exponential backoff and abandon the rekey
cycle cleanly once the budget is exhausted, mirroring the FSP session
rekey msg3 driver.

With the cap in place the link-dead heartbeat can safely become
rekey-aware: check_link_heartbeats now suppresses teardown while a rekey
is in progress with msg1 budget remaining, instead of reaping a link
that is still actively carrying rekey-handshake traffic. The suppression
terminates deterministically (the budget abandons on exhaustion, cutover
clears the in-progress flag), so a genuinely dead link is still reaped on
the next cycle.

Adds a rekey_msg1_resend_count counter on ActivePeer reset at every
rekey-clear and cutover site, msg1 resend-budget unit tests, and two-node
heartbeat suppression/resume/regression integration tests.

src/node/handlers/mmp.rs

@@ -521,6 +521,7 @@ impl Node {
         let now = Instant::now();
         let heartbeat_interval = Duration::from_secs(self.config.node.heartbeat_interval_secs);
         let dead_timeout = Duration::from_secs(self.config.node.link_dead_timeout_secs);
+        let max_resends = self.config.node.rate_limit.handshake_max_resends;
         let heartbeat_msg = [LinkMessageType::Heartbeat.to_byte()];
 
         // Collect heartbeats to send and dead peers to remove
@@ -530,7 +531,7 @@ impl Node {
         for (node_addr, peer) in self.peers.iter() {
             // Check liveness via MMP receiver last_recv_time.
             // Fall back to session_start for peers that never sent data.
-            let is_dead = if let Some(mmp) = peer.mmp() {
+            let time_dead = if let Some(mmp) = peer.mmp() {
                 let reference_time = mmp
                     .receiver
                     .last_recv_time()
@@ -539,6 +540,17 @@ impl Node {
             } else {
                 false
             };
+
+            // Suppress teardown while an FMP rekey is genuinely in flight with
+            // budget left: a rekey-handshake link is not silent. The msg1
+            // resend cap guarantees this terminates (abandon on exhaustion or
+            // cutover on completion clears `rekey_in_progress`), so a truly
+            // dead link is reaped on the next cycle.
+            let rekey_active = peer.rekey_in_progress()
+                && peer.rekey_msg1_resend_count() < max_resends
+                && peer.rekey_msg1().is_some();
+
+            let is_dead = time_dead && !rekey_active;
             if is_dead {
                 dead_peers.push(*node_addr);
                 continue;

src/node/handlers/rekey.rs

@@ -235,19 +235,37 @@ impl Node {
         }
 
         let interval_ms = self.config.node.rate_limit.handshake_resend_interval_ms;
+        let backoff = self.config.node.rate_limit.handshake_resend_backoff;
+        let max_resends = self.config.node.rate_limit.handshake_max_resends;
 
         // Collect peers needing action
         let mut to_resend: Vec<(NodeAddr, Vec<u8>)> = Vec::new();
+        let mut to_abandon: Vec<NodeAddr> = Vec::new();
 
         for (node_addr, peer) in &self.peers {
             if !peer.rekey_in_progress() || peer.rekey_msg1().is_none() {
                 continue;
             }
+            if peer.rekey_msg1_resend_count() >= max_resends {
+                to_abandon.push(*node_addr);
+                continue;
+            }
             if peer.needs_msg1_resend(now_ms) {
                 to_resend.push((*node_addr, peer.rekey_msg1().unwrap().to_vec()));
             }
         }
 
+        // Abandon rekey cycles that exhausted their retransmission budget.
+        for node_addr in to_abandon {
+            if let Some(peer) = self.peers.get_mut(&node_addr) {
+                peer.abandon_rekey();
+            }
+            warn!(
+                peer = %self.peer_display_name(&node_addr),
+                "FMP rekey aborted: msg1 unconfirmed after max retransmissions, abandoning cycle"
+            );
+        }
+
         for (node_addr, msg1_bytes) in to_resend {
             let (transport_id, remote_addr) = match self.peers.get(&node_addr) {
                 Some(p) => match (p.transport_id(), p.current_addr()) {
@@ -263,12 +281,13 @@ impl Node {
                 false
             };
 
-            if sent {
-                if let Some(peer) = self.peers.get_mut(&node_addr) {
-                    peer.set_msg1_next_resend(now_ms + interval_ms);
-                }
+            if sent && let Some(peer) = self.peers.get_mut(&node_addr) {
+                let count = peer.rekey_msg1_resend_count() + 1;
+                let next = now_ms + (interval_ms as f64 * backoff.powi(count as i32)) as u64;
+                peer.record_rekey_msg1_resend(next);
                 trace!(
                     peer = %self.peer_display_name(&node_addr),
+                    resend = count,
                     "Resent rekey msg1"
                 );
             }

src/node/tests/heartbeat.rs

@@ -0,0 +1,114 @@
+//! Link-dead heartbeat rekey-awareness integration tests.
+//!
+//! `check_link_heartbeats()` reaps a peer after the link-dead timeout,
+//! but suppresses teardown while an FMP rekey is genuinely in flight with
+//! its msg1 retransmission budget unexhausted. These tests drive a real
+//! two-node UDP peering, inject rekey state on the peer, and verify the
+//! suppress / resume / regression behaviors. `link_dead_timeout_secs` is
+//! set to 0 so the elapsed-time predicate is always satisfied and the only
+//! variable is the rekey-active guard.
+
+use super::spanning_tree::*;
+use super::*;
+use crate::Identity;
+use crate::noise::HandshakeState;
+use crate::utils::index::SessionIndex;
+
+/// Arm a real (initiator) FMP rekey on the peer the given node holds for
+/// `peer_addr`, so the msg1 resend budget can be exercised.
+fn arm_rekey(node: &mut crate::node::Node, peer_addr: &NodeAddr) {
+    let remote = Identity::generate();
+    let local = Identity::generate();
+    let hs = HandshakeState::new_initiator(local.keypair(), remote.pubkey_full());
+    let peer = node.get_peer_mut(peer_addr).expect("peer present");
+    peer.set_rekey_state(hs, SessionIndex::new(7), vec![0xAB; 64], 0);
+}
+
+/// A peer past the link-dead timeout is NOT reaped while an FMP rekey is in
+/// progress with its msg1 budget unexhausted.
+#[tokio::test]
+async fn heartbeat_suppressed_during_rekey() {
+    let mut nodes = run_tree_test(2, &[(0, 1)], false).await;
+    verify_tree_convergence(&nodes);
+
+    let addr_1 = *nodes[1].node.node_addr();
+    assert!(nodes[0].node.get_peer(&addr_1).is_some());
+
+    // Force every link to read as dead on elapsed time alone.
+    nodes[0].node.config.node.link_dead_timeout_secs = 0;
+
+    // Arm a rekey with budget left (count 0 < max_resends default 5).
+    arm_rekey(&mut nodes[0].node, &addr_1);
+    assert!(nodes[0].node.get_peer(&addr_1).unwrap().rekey_in_progress());
+
+    nodes[0].node.check_link_heartbeats().await;
+
+    assert!(
+        nodes[0].node.get_peer(&addr_1).is_some(),
+        "peer reaped despite an in-flight rekey with budget remaining"
+    );
+
+    cleanup_nodes(&mut nodes).await;
+}
+
+/// Once the msg1 budget is exhausted the rekey-active guard no longer
+/// holds, so a peer past the link-dead timeout IS reaped.
+#[tokio::test]
+async fn heartbeat_resumes_after_budget_exhausted() {
+    let mut nodes = run_tree_test(2, &[(0, 1)], false).await;
+    verify_tree_convergence(&nodes);
+
+    let addr_1 = *nodes[1].node.node_addr();
+    assert!(nodes[0].node.get_peer(&addr_1).is_some());
+
+    nodes[0].node.config.node.link_dead_timeout_secs = 0;
+    let max_resends = nodes[0].node.config.node.rate_limit.handshake_max_resends;
+
+    arm_rekey(&mut nodes[0].node, &addr_1);
+
+    // Exhaust the budget: count reaches max_resends, guard goes false.
+    let peer = nodes[0].node.get_peer_mut(&addr_1).unwrap();
+    for i in 0..max_resends {
+        peer.record_rekey_msg1_resend(1000 + i as u64 * 100);
+    }
+    assert_eq!(
+        nodes[0]
+            .node
+            .get_peer(&addr_1)
+            .unwrap()
+            .rekey_msg1_resend_count(),
+        max_resends
+    );
+
+    nodes[0].node.check_link_heartbeats().await;
+
+    assert!(
+        nodes[0].node.get_peer(&addr_1).is_none(),
+        "peer not reaped after its rekey budget was exhausted"
+    );
+
+    cleanup_nodes(&mut nodes).await;
+}
+
+/// Regression guard: with no rekey in flight, a peer past the link-dead
+/// timeout is reaped exactly as before.
+#[tokio::test]
+async fn heartbeat_unaffected_without_rekey() {
+    let mut nodes = run_tree_test(2, &[(0, 1)], false).await;
+    verify_tree_convergence(&nodes);
+
+    let addr_1 = *nodes[1].node.node_addr();
+    assert!(nodes[0].node.get_peer(&addr_1).is_some());
+    assert!(!nodes[0].node.get_peer(&addr_1).unwrap().rekey_in_progress());
+
+    nodes[0].node.config.node.link_dead_timeout_secs = 0;
+
+    nodes[0].node.check_link_heartbeats().await;
+
+    assert!(
+        nodes[0].node.get_peer(&addr_1).is_none(),
+        "dead peer with no rekey in flight should be reaped"
+    );
+
+    cleanup_nodes(&mut nodes).await;
+}

src/node/tests/mod.rs

@@ -17,6 +17,7 @@ mod discovery;
 mod ethernet;
 mod forwarding;
 mod handshake;
+mod heartbeat;
 mod routing;
 mod session;
 mod spanning_tree;