Catch a msg1 pending slot released at acquire time

The pending-slot guard's over-release direction is covered: a reject arm
that frees a slot it never took reds
msg1_reject_arms_do_not_release_another_handshakes_slot. The opposite
direction is covered by nothing. Rebinding handle_msg1's `let _slot` to
a bare `_` drops the guard the instant the slot is taken, so the
limiter's concurrency limb stops bounding anything, and the whole suite
stays green: `#[must_use]` does not fire, because the value is used, and
every counter a test can read after the handler returns is the same in
both worlds, since the slot comes back at the end either way. Until now
the invariant rested on the binding being spelled `_slot`.

The difference is only visible while the handler is on the stack, so
that is where the observation goes: a test-build assertion immediately
below the acquire, checking the count it just raised is still raised.
Add a test that drives two msg1 through the handler on different paths
past the acquire, each asserting the reject counter it must bump so a
msg1 refused before the acquire cannot pass this vacuously.

No behaviour change: the assertion is `#[cfg(test)]` and the guard
itself is untouched.

(cherry picked from commit 0693b81cf9fb1866a371b62a30e36497f8b484fb)

src/node/handlers/handshake.rs

@@ -339,8 +339,8 @@ impl Node {
         // function returns, and that is what releases the pending slot on
         // every exit path. Renaming it to a bare `_` drops the guard right
         // here instead, releasing the slot at acquire time — silently, with
-        // no test and no clippy lint catching the difference. Do not "tidy"
-        // this binding.
+        // no clippy lint catching the difference. Do not "tidy" this
+        // binding; the assertion below is what reds if it is tidied.
         let _slot = match self.msg1_rate_limiter.start_handshake(class) {
             Ok(slot) => slot,
             Err(reason) => {
@@ -354,6 +354,19 @@ impl Node {
             }
         };
 
+        // Test-build witness for the paragraph above, and the only thing that
+        // observes it. A guard released at acquire time leaves this msg1
+        // in flight with its slot already back in the pool, which no counter,
+        // log line or lint reports: by the time any test can look, the count
+        // has returned to its baseline either way. Sampling it here, on the
+        // handler's own stack, is what tells the two apart — see
+        // `msg1_handler_holds_its_pending_slot_while_the_handler_runs`.
+        #[cfg(test)]
+        assert!(
+            self.msg1_rate_limiter.pending_count() > 0,
+            "the msg1 pending slot was released before the handler ran"
+        );
+
         // accept_connections gate. Rekey/restart msg1 on an existing link
         // is always admitted; the gate only filters truly-fresh connections
         // from strangers. Without this carve-out, the dual-init tie-breaker

src/node/tests/unit.rs

@@ -2994,6 +2994,77 @@ async fn msg1_reject_arms_do_not_release_another_handshakes_slot() {
     assert_eq!(node.msg1_rate_limiter.pending_count(), 0);
 }
 
+/// The msg1 handler keeps its pending slot for as long as it is running.
+///
+/// The complement of `msg1_reject_arms_do_not_release_another_handshakes_slot`:
+/// that one covers releasing a slot the handler never took, this one covers
+/// releasing its own slot too early. Rebinding `handle_msg1`'s `let _slot` to
+/// a bare `_` drops the guard at acquire time, so the limiter's concurrency
+/// limb stops bounding anything — and every counter this test could read
+/// afterwards is identical either way, because the slot comes back at the end
+/// of the handler in both worlds. The difference exists only while the handler
+/// is on the stack, which is why the observation lives there: the
+/// `#[cfg(test)]` assertion in `handle_msg1` immediately below the acquire
+/// fires under the premature release and under nothing else.
+///
+/// Two packets, so the handler is entered twice on different paths past the
+/// acquire, and each arm asserts the reject counter it must bump. Without
+/// that, a msg1 refused before the acquire (an empty bucket, say) would leave
+/// this test passing while sampling nothing.
+#[tokio::test]
+async fn msg1_handler_holds_its_pending_slot_while_the_handler_runs() {
+    use crate::noise::HANDSHAKE_MSG1_SIZE;
+    use crate::proto::fmp::wire::build_msg1;
+    use crate::utils::index::SessionIndex;
+
+    // No transport is registered: both arms reject before any send, and the
+    // absent transport admits past the `accept_connections` gate.
+    let mut node = make_node();
+    let transport_id = TransportId::new(1);
+    let source = TransportAddr::from_string("198.51.100.9:4141");
+    let packet = |data: Vec<u8>| ReceivedPacket {
+        transport_id,
+        remote_addr: source.clone(),
+        data,
+        timestamp_ms: 1000,
+    };
+
+    assert_eq!(
+        node.msg1_rate_limiter.pending_count(),
+        0,
+        "baseline: no handshake in flight"
+    );
+
+    // Arm 1: rejected at the header parse, the shortest path past the acquire.
+    let before = node.stats().handshake.bad_state;
+    node.handle_msg1(packet(vec![0u8; 8])).await;
+    assert_eq!(
+        node.stats().handshake.bad_state,
+        before + 1,
+        "arm 1 must reach the invalid-header reject, not a rate-limit refusal"
+    );
+
+    // Arm 2: well-formed header, unusable Noise payload — rejected further in,
+    // after the duplicate short-circuit and the DH attempt.
+    let before = node.stats().handshake.bad_state;
+    node.handle_msg1(packet(build_msg1(
+        SessionIndex::new(0x4242),
+        &[0u8; HANDSHAKE_MSG1_SIZE],
+    )))
+    .await;
+    assert_eq!(
+        node.stats().handshake.bad_state,
+        before + 1,
+        "arm 2 must reach the receive_handshake_init reject"
+    );
+
+    assert_eq!(
+        node.msg1_rate_limiter.pending_count(),
+        0,
+        "each handler released its own slot exactly once on the way out"
+    );
+}
+
 /// The established-link bucket is wired from config at construction:
 /// derived from `max_peers` by default, overridden when the operator sets
 /// the key. This is the only test covering the config → limiter path.