Free the abandoned rekey index and reschedule an ACL-rejected dial

Two handshake reject arms on master dropped state that the next line
already handles correctly, so bring both to the same shape.

The tick-loop `AbandonRekey` arm called `ActivePeer::abandon_rekey` in
statement position and threw away the `Option<SessionIndex>` the callee
hands back for freeing. The index was leaked from the allocator and its
`pending_outbound` entry, seeded when rekey msg1 went out, was left
pointing at the peer's live link. Bind the return, clear both registry
entries keyed by it, and free it, matching what the dual-initiation
loser and the rekey-msg2 failure arm in `handshake.rs` already do.

The outbound ACL reject in `handle_msg2` disposed of the machine, the
link, the `pending_outbound` entry and the index, but re-armed nothing.
That disposal takes the leg out of the stuck-leg sweep, which is the
only thing that otherwise reaches `note_handshake_timeout`, and
`close_connection` only drops the transport's pool entry. A configured
peer denied once was therefore never dialed again, even after the ACL
was relaxed. Reschedule from the arm itself.

Both arms get a discriminating test that asserts the state is present
before the call and gone after, with a control assertion so neither can
pass vacuously and a limb checking the live session and the disposed leg
are unaffected. The ACL test needed the responder listed as a configured
peer, so the existing denied-msg2 setup is extracted into a helper that
takes a config builder rather than being duplicated.

(cherry picked from commit 62506f1c0bac50ca0d502eb44b3ee7fbbafa6c17)

src/node/handlers/handshake.rs

@@ -1317,6 +1317,19 @@ impl Node {
             if let Some(idx) = our_index {
                 let _ = self.index_allocator.free(idx);
             }
+            // Put the dial back on the retry schedule. The disposal above takes
+            // this leg out of the stuck-leg sweep — `has_pending_leg` reads false
+            // for it from here on, so the reap that normally reaches
+            // `note_handshake_timeout` never runs — and that reflex is the only
+            // thing that seeds `retry_pending` for a configured peer.
+            // `close_connection` only drops the transport's pool entry; it
+            // schedules nothing.
+            //
+            // `peer_identity` is safe to reschedule against here because this is
+            // an IK dial: the initiator's expected identity is fixed at
+            // `start_handshake` and `complete_handshake` never overwrites it, so
+            // it is still the peer we meant to dial and not whoever answered.
+            self.note_handshake_timeout(*peer_identity.node_addr(), packet.timestamp_ms);
             self.stats_mut()
                 .record_reject(RejectReason::Handshake(HandshakeReject::BadState));
             return;

src/node/handlers/rekey.rs

@@ -420,8 +420,38 @@ impl Node {
             match action {
                 // Abandon rekey cycles that exhausted their retransmission budget.
                 ConnAction::AbandonRekey { peer: node_addr } => {
-                    if let Some(peer) = self.peers.get_mut(&node_addr) {
-                        peer.abandon_rekey();
+                    // `abandon_rekey` hands back whichever index the abandoned
+                    // cycle owned; dropping the return value orphans it, and the
+                    // `pending_outbound` entry seeded at rekey msg1 with it. Same
+                    // shape as the dual-initiation loser and the rekey-msg2
+                    // failure arm in `handshake.rs`.
+                    //
+                    // The `peers_by_index` removal is parity with those two arms
+                    // and is a no-op at THIS call site: `AbandonRekey` is emitted
+                    // only from the msg1-resend-budget classification, so no rekey
+                    // msg2 ever arrived and nothing was inserted.
+                    //
+                    // Known exposure, kept for parity rather than closed here:
+                    // `transport_id()` is RE-READ, while the `pending_outbound`
+                    // entry was keyed by whatever it was when rekey msg1 went out.
+                    // A roam in between (`set_current_addr` overwrites
+                    // `send.transport_id`) makes the removal miss, so the index is
+                    // freed with a stale entry still pointing at the peer's live
+                    // link. Walked to its end: a later msg2 naming that index on
+                    // the old transport resolves the stale link, finds the
+                    // promoted peer's machine leg-less, finds no peer with a
+                    // matching `rekey_our_index` (this arm cleared it), and takes
+                    // the "not a rekey" arm, which removes the stale entry and
+                    // records a reject. No teardown, no wrong-peer effect.
+                    // Removing by index VALUE would close it outright.
+                    if let Some(peer) = self.peers.get_mut(&node_addr)
+                        && let Some(idx) = peer.abandon_rekey()
+                    {
+                        if let Some(tid) = peer.transport_id() {
+                            self.peers_by_index.remove(&(tid, idx.as_u32()));
+                            self.pending_outbound.remove(&(tid, idx.as_u32()));
+                        }
+                        let _ = self.index_allocator.free(idx);
                     }
                     debug!(
                         peer = %self.peer_display_name(&node_addr),

src/node/tests/acl.rs

@@ -75,13 +75,25 @@ async fn test_inbound_msg1_denied_by_acl() {
     assert_eq!(node_b.link_count(), 0);
 }
 
-#[tokio::test]
-async fn test_outbound_msg2_denied_after_acl_reload() {
-    let (dir, mut node_a) = make_acl_node();
+/// Drive a dialing node up to the instant a genuine msg2 arrives from a
+/// responder its denylist has just been reloaded to reject, and hand back the
+/// node, the responder's NodeAddr, and the msg2 packet ready to deliver.
+///
+/// `config_a` receives the responder's npub so a caller can list it as a
+/// configured peer, which is what makes the reject arm's retry reschedule
+/// observable.
+async fn outbound_denied_at_msg2(
+    config_a: impl FnOnce(&str) -> Config,
+) -> (tempfile::TempDir, Node, NodeAddr, ReceivedPacket) {
     let node_b = make_node();
+    let dir = tempfile::tempdir().unwrap();
+    let mut node_a = Node::new(config_a(&node_b.npub())).unwrap();
+    node_a.peer_acl = PeerAclReloader::with_paths(allow_path(&dir), deny_path(&dir));
+
     let transport_id = TransportId::new(1);
     let remote_addr = TransportAddr::from_string("127.0.0.1:5001");
     let peer_b_identity = PeerIdentity::from_pubkey_full(node_b.identity().pubkey_full());
+    let node_b_addr = *peer_b_identity.node_addr();
 
     let link_id_a = node_a.allocate_link_id();
     let our_index_a = node_a.index_allocator.allocate().unwrap();
@@ -134,6 +146,14 @@ async fn test_outbound_msg2_denied_after_acl_reload() {
     assert!(node_a.reload_peer_acl().await);
 
     let packet = ReceivedPacket::with_timestamp(transport_id, remote_addr, wire_msg2, 1100);
+    (dir, node_a, node_b_addr, packet)
+}
+
+#[tokio::test]
+async fn test_outbound_msg2_denied_after_acl_reload() {
+    let (_dir, mut node_a, _node_b_addr, packet) =
+        outbound_denied_at_msg2(|_npub| Config::new()).await;
+
     node_a.handle_msg2(packet).await;
 
     assert_eq!(node_a.peer_count(), 0);
@@ -142,6 +162,49 @@ async fn test_outbound_msg2_denied_after_acl_reload() {
     assert!(node_a.pending_outbound.is_empty());
 }
 
+/// The outbound ACL reject arm disposes of the whole leg — machine, link and
+/// index — which takes it out of the stuck-leg sweep that would otherwise reach
+/// `note_handshake_timeout`. For a configured peer the dial must therefore be
+/// put back on the retry schedule from the arm itself, or the node never dials
+/// again after the ACL is relaxed.
+#[tokio::test]
+async fn test_outbound_msg2_acl_reject_reschedules_a_configured_peer() {
+    let (_dir, mut node_a, node_b_addr, packet) = outbound_denied_at_msg2(|npub| {
+        let mut config = Config::new();
+        config.peers.push(crate::config::PeerConfig::new(
+            npub,
+            "udp",
+            "127.0.0.1:5001",
+        ));
+        config
+    })
+    .await;
+
+    assert!(
+        node_a.peering.reconciler.retry_pending.is_empty(),
+        "control: nothing is scheduled before the reject"
+    );
+
+    node_a.handle_msg2(packet).await;
+
+    let state = node_a
+        .peering
+        .reconciler
+        .retry_pending
+        .get(&node_b_addr)
+        .expect("the rejected dial is back on the retry schedule");
+    assert_eq!(state.retry_count, 1, "counted as a first failure");
+    assert!(
+        state.retry_after_ms > 1100,
+        "scheduled after the msg2 that triggered it"
+    );
+
+    // The reschedule must not resurrect any of the disposed leg.
+    assert_eq!(node_a.peer_count(), 0);
+    assert_eq!(node_a.link_count(), 0);
+    assert!(node_a.pending_outbound.is_empty());
+}
+
 #[tokio::test]
 async fn test_host_map_hot_reloads_from_tick() {
     let dir = tempfile::tempdir().unwrap();

src/node/tests/establish_chartests.rs

@@ -1032,3 +1032,107 @@ async fn chartest_msg1_rekey_dual_init_we_lose_becomes_responder() {
         "loser (now responder) emits a rekey msg2"
     );
 }
+
+// ===========================================================================
+// Rekey abandon (the tick-loop budget path). Not a `handle_msg1` branch, but it
+// reuses this module's rekey arming and establish helpers, so it lives here.
+// ===========================================================================
+
+/// A rekey cycle abandoned for exhausting its msg1 retransmission budget must
+/// return its session index to the allocator and clear both registry entries
+/// keyed by it, leaving the live session untouched.
+///
+/// `handshake_max_resends = 0` fires the abandon on the first
+/// `resend_pending_rekeys` call: the classifier tests `resend_count >=
+/// max_resends` before it consults the resend-due predicate, so the wall clock
+/// is out of the path entirely.
+#[tokio::test]
+async fn abandoned_rekey_frees_its_index_and_clears_both_registries() {
+    let mut config = Config::new();
+    config.node.rate_limit.handshake_max_resends = 0;
+    let mut node = make_node_with(config);
+    let transport_id = TransportId::new(1);
+    let (peer_sock, peer_addr) = register_udp_with_peer_socket(&mut node, transport_id).await;
+
+    let sender = Identity::generate();
+    let sender_addr = establish_active_peer_via_msg1(
+        &mut node,
+        &sender,
+        [9u8; 8],
+        transport_id,
+        &peer_addr,
+        &peer_sock,
+        1000,
+    )
+    .await;
+
+    let session_index = node
+        .get_peer(&sender_addr)
+        .expect("peer established")
+        .our_index()
+        .expect("the established peer holds a session index");
+    let rekey_index = arm_local_rekey(&mut node, &sender, &sender_addr, transport_id);
+    let allocated_before = node.index_allocator.count();
+
+    // Controls: without these the assertions after the abandon could pass
+    // against state that was never set up.
+    assert!(
+        node.index_allocator.is_allocated(rekey_index),
+        "control: the armed rekey holds an index"
+    );
+    assert!(
+        node.peers_by_index
+            .contains_key(&(transport_id, rekey_index.as_u32())),
+        "control: the rekey index is registered for dispatch"
+    );
+    assert!(
+        node.pending_outbound
+            .contains_key(&(transport_id, rekey_index.as_u32())),
+        "control: the rekey index is registered for msg2 dispatch"
+    );
+
+    // The rekey msg2 never arrives; the first poll spends the (zero) budget.
+    node.resend_pending_rekeys(2000).await;
+
+    let p = node.get_peer(&sender_addr).expect("peer still present");
+    assert!(
+        !p.rekey_in_progress(),
+        "control: the abandon actually fired"
+    );
+    assert!(p.rekey_our_index().is_none());
+
+    assert!(
+        !node.index_allocator.is_allocated(rekey_index),
+        "the abandoned cycle must return its index"
+    );
+    assert_eq!(
+        node.index_allocator.count(),
+        allocated_before - 1,
+        "and must free exactly one"
+    );
+    assert!(
+        !node
+            .peers_by_index
+            .contains_key(&(transport_id, rekey_index.as_u32())),
+        "the abandoned cycle must clear its dispatch registration"
+    );
+    assert!(
+        !node
+            .pending_outbound
+            .contains_key(&(transport_id, rekey_index.as_u32())),
+        "the abandoned cycle must clear its msg2 dispatch entry"
+    );
+
+    // The limb that catches a fix binding the wrong index: the live session is
+    // untouched.
+    assert_eq!(node.peer_count(), 1, "the session survives");
+    assert!(
+        node.index_allocator.is_allocated(session_index),
+        "the established session's own index must not be freed"
+    );
+    assert!(
+        node.peers_by_index
+            .contains_key(&(transport_id, session_index.as_u32())),
+        "the established session stays registered for dispatch"
+    );
+}