node: free the inbound session index on the msg3 reject and resend paths

On two inbound msg3 outcomes the msg1-allocated inbound session index
was dropped without returning it to the allocator: losing the dual-rekey
tie-break (we keep our in-progress rekey and drop their msg3), and a
duplicate handshake that resolves to a msg2 resend. Under repeated hits
this slowly leaks index space. Free the index on both arms, matching the
sibling reject arms.

Add a test-only helper to backdate a peer's session-established instant,
and two regression tests that drive a real second handshake into each
arm and assert the index is returned to the allocator (both fail if the
free is removed).

src/node/handlers/handshake.rs

@@ -1079,6 +1079,9 @@ impl Node {
                     pending_new_session = snap.pending_new_session,
                     "rekey-msg3 tie-break: we win (smaller addr), drop their msg3"
                 );
+                // We keep our in-progress rekey and drop their msg3; return the
+                // msg1-allocated inbound index rather than orphaning it.
+                let _ = self.index_allocator.free(our_index);
                 self.connections.remove(&link_id);
                 self.links.remove(&link_id);
                 self.stats_mut()
@@ -1103,6 +1106,9 @@ impl Node {
                         ),
                     }
                 }
+                // Duplicate handshake: the active peer is untouched, so return the
+                // msg1-allocated inbound index rather than orphaning it.
+                let _ = self.index_allocator.free(our_index);
                 self.connections.remove(&link_id);
                 self.links.remove(&link_id);
                 return;

src/node/tests/handshake.rs

@@ -1302,3 +1302,268 @@ async fn test_should_admit_msg1_admits_rekey_when_addr_form_differs() {
         "fresh msg1 from unknown source must still be rejected"
     );
 }
+
+// ===========================================================================
+// Regression: `handle_msg3` must return the msg1-allocated session index to
+// the allocator on the two inbound-establish arms that abandon the pending
+// inbound leg without promoting it — the `Reject{DualRekeyWon}` tie-break
+// (dual-init rekey we win) and the `ResendMsg2` duplicate-handshake arm. Both
+// tear the pending connection/link down; neither must orphan the index.
+// ===========================================================================
+
+/// A node bundled with its UDP transport, receive channel, and bound address,
+/// used to drive real msg1/msg2/msg3 exchanges below.
+struct HsNode {
+    node: Node,
+    transport_id: TransportId,
+    packet_rx: crate::transport::PacketRx,
+    addr: TransportAddr,
+}
+
+/// Build an `HsNode` on an ephemeral localhost UDP port from an explicit config.
+async fn make_hs_node(config: Config) -> HsNode {
+    use crate::config::UdpConfig;
+    use crate::transport::udp::UdpTransport;
+
+    let mut node = make_node_with(config);
+    let transport_id = TransportId::new(1);
+    let udp_config = UdpConfig {
+        bind_addr: Some("127.0.0.1:0".to_string()),
+        mtu: Some(1280),
+        ..Default::default()
+    };
+    let (packet_tx, packet_rx) = packet_channel(64);
+    let mut transport = UdpTransport::new(transport_id, None, udp_config, packet_tx);
+    transport.start_async().await.unwrap();
+    let addr = TransportAddr::from_string(&transport.local_addr().unwrap().to_string());
+    node.transports
+        .insert(transport_id, TransportHandle::Udp(transport));
+
+    HsNode {
+        node,
+        transport_id,
+        packet_rx,
+        addr,
+    }
+}
+
+async fn stop_hs(n: &mut HsNode) {
+    for (_, t) in n.node.transports.iter_mut() {
+        t.stop().await.ok();
+    }
+}
+
+/// Receive the next packet whose handshake phase matches `phase` (the low
+/// nibble of the wire type byte: 1=msg1, 2=msg2, 3=msg3), skipping unrelated
+/// traffic. Post-promotion tree/filter announces (phase 0, encrypted data)
+/// share these channels, so a phase filter keeps the hand-driven exchange in
+/// step.
+async fn recv_phase(rx: &mut crate::transport::PacketRx, phase: u8, what: &str) -> ReceivedPacket {
+    use std::time::Duration;
+    use tokio::time::timeout;
+
+    loop {
+        let pkt = timeout(Duration::from_secs(1), rx.recv())
+            .await
+            .unwrap_or_else(|_| panic!("timeout waiting for {}", what))
+            .expect("channel closed");
+        if pkt.data.first().is_some_and(|b| b & 0x0f == phase) {
+            return pkt;
+        }
+    }
+}
+
+/// Drive one initiator -> responder XX exchange (msg1 then msg2) and return the
+/// responder's inbound msg3 packet, left unhandled for the caller. The msg3 is
+/// produced by the initiator's real `handle_msg2`, so it carries a valid Noise
+/// payload.
+async fn drive_to_msg3(
+    initiator: &mut HsNode,
+    responder: &mut HsNode,
+    now_ms: u64,
+) -> ReceivedPacket {
+    use crate::proto::fmp::wire::build_msg1;
+    use std::time::Duration;
+
+    let peer_identity = PeerIdentity::from_pubkey_full(responder.node.identity().pubkey_full());
+
+    let link_id = initiator.node.allocate_link_id();
+    let mut conn = PeerConnection::outbound(link_id, peer_identity, now_ms);
+    let our_index = initiator.node.index_allocator.allocate().unwrap();
+    let our_keypair = initiator.node.identity().keypair();
+    let noise_msg1 = conn
+        .start_handshake(our_keypair, initiator.node.startup_epoch(), now_ms)
+        .unwrap();
+    conn.set_our_index(our_index);
+    conn.set_transport_id(initiator.transport_id);
+    conn.set_source_addr(responder.addr.clone());
+
+    let wire_msg1 = build_msg1(our_index, &noise_msg1);
+    let link = Link::connectionless(
+        link_id,
+        initiator.transport_id,
+        responder.addr.clone(),
+        LinkDirection::Outbound,
+        Duration::from_millis(100),
+    );
+    initiator.node.links.insert(link_id, link);
+    initiator
+        .node
+        .addr_to_link
+        .insert((initiator.transport_id, responder.addr.clone()), link_id);
+    initiator.node.connections.insert(link_id, conn);
+    initiator
+        .node
+        .pending_outbound
+        .insert((initiator.transport_id, our_index.as_u32()), link_id);
+
+    initiator
+        .node
+        .transports
+        .get(&initiator.transport_id)
+        .unwrap()
+        .send(&responder.addr, &wire_msg1)
+        .await
+        .expect("send msg1");
+
+    // Responder processes msg1 and emits msg2 back to the initiator.
+    let msg1_pkt = recv_phase(&mut responder.packet_rx, 1, "msg1").await;
+    responder.node.handle_msg1(msg1_pkt).await;
+
+    // Initiator processes msg2 and emits msg3 to the responder.
+    let msg2_pkt = recv_phase(&mut initiator.packet_rx, 2, "msg2").await;
+    initiator.node.handle_msg2(msg2_pkt).await;
+
+    // Capture the responder's inbound msg3, left unhandled for the caller.
+    recv_phase(&mut responder.packet_rx, 3, "msg3").await
+}
+
+#[tokio::test]
+async fn test_msg3_dual_rekey_won_frees_index() {
+    // Rekey enabled with a tiny interval so the rekey age floor collapses to
+    // its 5s minimum; the peer session is then backdated past it.
+    let make_config = || {
+        let mut c = Config::new();
+        c.node.rekey.enabled = true;
+        c.node.rekey.after_secs = 1;
+        c
+    };
+
+    let mut initiator = make_hs_node(Config::new()).await;
+    // The DualRekeyWon tie-break is won by the numerically smaller node addr,
+    // so the responder (whose handle_msg3 we exercise) must be the smaller.
+    let mut responder = loop {
+        let cand = make_hs_node(make_config()).await;
+        if cand.node.node_addr() < initiator.node.node_addr() {
+            break cand;
+        }
+    };
+
+    // First handshake: the responder promotes the initiator to a healthy active
+    // peer holding exactly one allocated session index.
+    let msg3 = drive_to_msg3(&mut initiator, &mut responder, 1000).await;
+    responder.node.handle_msg3(msg3).await;
+    assert_eq!(responder.node.peer_count(), 1);
+    let baseline = responder.node.index_allocator.count();
+    assert_eq!(baseline, 1, "responder holds exactly the peer's index");
+
+    // Age the session past the rekey floor and mark a rekey in progress so a
+    // fresh inbound msg3 classifies as the dual-init rekey we win.
+    let peer_addr =
+        *PeerIdentity::from_pubkey_full(initiator.node.identity().pubkey_full()).node_addr();
+    {
+        let peer = responder.node.get_peer_mut(&peer_addr).unwrap();
+        peer.test_backdate_session_established(std::time::Duration::from_secs(6));
+        peer.set_rekey_in_progress();
+    }
+
+    // Second handshake: the new inbound msg1 allocates a fresh index, then the
+    // msg3 lands on the DualRekeyWon reject arm.
+    let msg3b = drive_to_msg3(&mut initiator, &mut responder, 2000).await;
+    assert_eq!(
+        responder.node.index_allocator.count(),
+        baseline + 1,
+        "second msg1 allocated a fresh index"
+    );
+    responder.node.handle_msg3(msg3b).await;
+
+    // The rejected msg3 must return its index and leave the active peer intact.
+    assert_eq!(
+        responder.node.index_allocator.count(),
+        baseline,
+        "DualRekeyWon must free the msg1-allocated index"
+    );
+    assert_eq!(responder.node.peer_count(), 1, "active peer untouched");
+    assert!(
+        responder
+            .node
+            .get_peer(&peer_addr)
+            .unwrap()
+            .pending_new_session()
+            .is_none(),
+        "reject arm must not store rekey-responder state"
+    );
+
+    stop_hs(&mut initiator).await;
+    stop_hs(&mut responder).await;
+}
+
+#[tokio::test]
+async fn test_msg3_resend_msg2_frees_index() {
+    // Rekey disabled so an aged-session inbound msg3 classifies as a duplicate
+    // handshake (ResendMsg2), not a rekey; the tiny interval keeps the
+    // cross-connection age bound (the rekey floor) at its 5s minimum so the
+    // aged session skips the cross-connection arm too.
+    let mut config = Config::new();
+    config.node.rekey.enabled = false;
+    config.node.rekey.after_secs = 1;
+
+    let mut initiator = make_hs_node(Config::new()).await;
+    let mut responder = make_hs_node(config).await;
+
+    // First handshake establishes the active peer.
+    let msg3 = drive_to_msg3(&mut initiator, &mut responder, 1000).await;
+    responder.node.handle_msg3(msg3).await;
+    assert_eq!(responder.node.peer_count(), 1);
+    let baseline = responder.node.index_allocator.count();
+    assert_eq!(baseline, 1, "responder holds exactly the peer's index");
+
+    // Age the session past the cross-connection bound so the duplicate inbound
+    // msg3 resolves to ResendMsg2 rather than CrossConnect.
+    let peer_addr =
+        *PeerIdentity::from_pubkey_full(initiator.node.identity().pubkey_full()).node_addr();
+    responder
+        .node
+        .get_peer_mut(&peer_addr)
+        .unwrap()
+        .test_backdate_session_established(std::time::Duration::from_secs(6));
+
+    // Second (duplicate) handshake: fresh index allocated at msg1, then freed
+    // on the ResendMsg2 arm.
+    let msg3b = drive_to_msg3(&mut initiator, &mut responder, 2000).await;
+    assert_eq!(
+        responder.node.index_allocator.count(),
+        baseline + 1,
+        "second msg1 allocated a fresh index"
+    );
+    responder.node.handle_msg3(msg3b).await;
+
+    assert_eq!(
+        responder.node.index_allocator.count(),
+        baseline,
+        "ResendMsg2 must free the msg1-allocated index"
+    );
+    assert_eq!(responder.node.peer_count(), 1, "active peer untouched");
+    assert!(
+        responder
+            .node
+            .get_peer(&peer_addr)
+            .unwrap()
+            .pending_new_session()
+            .is_none(),
+        "duplicate-handshake arm must not store rekey-responder state"
+    );
+
+    stop_hs(&mut initiator).await;
+    stop_hs(&mut responder).await;
+}

src/peer/active.rs

@@ -1015,6 +1015,20 @@ impl ActivePeer {
             .unwrap_or_else(Instant::now);
     }
 
+    /// Test-only seam: backdate the session-established instant so a test can
+    /// make `session_established_at().elapsed()` read as `age`-old (needed to
+    /// drive the inbound establish paths that gate on session age past the
+    /// rekey age floor). This only shifts the private timestamp field; it
+    /// changes no decision logic, no threshold, and is compiled out of release
+    /// builds.
+    #[cfg(test)]
+    pub(crate) fn test_backdate_session_established(&mut self, age: std::time::Duration) {
+        self.session_established_at = self
+            .session_established_at
+            .checked_sub(age)
+            .unwrap_or_else(Instant::now);
+    }
+
     /// Per-session symmetric rekey-timer jitter offset (seconds).
     ///
     /// Drawn at session construction and at each rekey cutover; uniform