From 765819f52b383a81a4f17dd417511aaf354b1384 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Wed, 15 Jul 2026 20:23:49 +0000 Subject: [PATCH] node: reap the handshake timeout through the per-peer machine timer Move the outbound handshake-timeout reap off the unconditional check_timeouts scan onto the machine-armed HandshakeTimeout timer. A new drive_handshake_timeouts (run before the retransmit drive, so a timed-out leg is reaped rather than resent on the same tick) reaps the outbound legs that carry a HandshakeTimeout timer and have idle-timed-out this tick. The timer's presence selects the leg (only outbound legs arm one; IK inbound arms none); the reap threshold is the shell is_timed_out(now, config) predicate, not the timer's stored deadline. The machine arms the timer from a hardcoded constant at dial, which is not authoritative for an operator-tuned handshake_timeout_secs, so reading the threshold from config each tick keeps the reap neutral for any timeout value and on the last_activity clock exactly as before. check_timeouts keeps reaping everything else: failed connections (all of them, promptly) and the idle-timeout of legs without a machine timer (inbound legs, and any machine-less connection). Total coverage is unchanged. check_timeouts runs before the timer drive, so a failed outbound leg is reaped there first, its timers dropped, and the timeout drive never double-fires on it. Split drive_peer_timers into the timeout drive and the (byte-identical) retransmit drive. The dormant machine timeout handler is not dispatched, so the session index is freed once, by cleanup_stale_connection. A regression test drives a timed-out outbound leg to reap; the existing check_timeouts tests continue to anchor the residual sweep. Test count 1631 to 1632. --- src/node/dataplane/peer_actions.rs | 17 +++-- src/node/handlers/timeout.rs | 104 ++++++++++++++++++++++++----- src/node/tests/handshake.rs | 77 +++++++++++++++++++++ 3 files changed, 172 insertions(+), 26 deletions(-) diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs index 01520b5..bd171f6 100644 --- a/src/node/dataplane/peer_actions.rs +++ b/src/node/dataplane/peer_actions.rs @@ -22,10 +22,10 @@ //! a deliberate no-op — see its arm for the note. //! //! The timer arms (`SetTimer`/`CancelTimer`) populate/clear the per-peer timer -//! store (`peer_timers`). The `HandshakeRetransmit` deadline is read and fired -//! by `drive_peer_timers` (the msg1-resend home). The `HandshakeTimeout` and -//! rekey/liveness kinds are still SHADOW — driven by the legacy `check_timeouts` -//! / rekey shell drivers — so populating them stays behavior-neutral. +//! store (`peer_timers`). The `HandshakeRetransmit` and `HandshakeTimeout` +//! deadlines are read and fired by `drive_peer_timers` (the handshake resend + +//! reap home). The rekey/liveness kinds are still SHADOW — driven by their own +//! shell drivers — so populating them stays behavior-neutral. use crate::PeerIdentity; use crate::node::Node; @@ -501,11 +501,10 @@ impl Node { } PeerAction::SetTimer { kind, at_ms } => { // Populate the per-peer timer store (overwrite = reschedule). - // The `HandshakeRetransmit` deadline is now read + fired by - // `drive_peer_timers` (the msg1-resend home). Other kinds are - // still SHADOW here — `HandshakeTimeout` is driven by the - // legacy `check_timeouts`, and rekey/liveness keep their own - // shell drivers — so populating them stays behavior-neutral. + // The `HandshakeRetransmit` and `HandshakeTimeout` deadlines + // are read + fired by `drive_peer_timers`. Rekey/liveness kinds + // are still SHADOW here — they keep their own shell drivers — + // so populating them stays behavior-neutral. self.peer_timers .entry(link) .or_default() diff --git a/src/node/handlers/timeout.rs b/src/node/handlers/timeout.rs index d892717..42e6e7c 100644 --- a/src/node/handlers/timeout.rs +++ b/src/node/handlers/timeout.rs @@ -11,9 +11,21 @@ use tracing::{debug, info}; impl LifecycleView for Node { fn stale_connections(&self, now_ms: u64, timeout_ms: u64) -> Vec { + // `is_failed()` legs are always reaped here (~1s), as before. The + // idle-timeout is reaped here ONLY for legs whose timeout is not already + // driven by a machine `HandshakeTimeout` timer — i.e. inbound legs (IK + // inbound arms none) and machine-less test connections. Outbound legs with + // an armed timer are reaped by `drive_handshake_timeouts`, so excluding + // them here avoids a double reap. self.connections .iter() - .filter(|(_, conn)| conn.is_timed_out(now_ms, timeout_ms) || conn.is_failed()) + .filter(|(link_id, conn)| { + conn.is_failed() + || (conn.is_timed_out(now_ms, timeout_ms) + && !self.peer_timers.get(*link_id).is_some_and(|timers| { + timers.contains_key(&TimerKind::HandshakeTimeout) + })) + }) .map(|(link_id, conn)| ConnSnapshot { link: *link_id, is_outbound: conn.is_outbound(), @@ -115,29 +127,87 @@ impl Node { } } - /// Drive the per-peer machine timers that have come due on this tick. + /// Act on the per-peer machine timers this tick. /// /// The sans-IO machine arms `SetTimer`/`CancelTimer` actions into - /// [`peer_timers`](Node::peer_timers); this is the shell driver that fires - /// the due ones. At this rung only the msg1-retransmit timer is driven (the - /// handshake-timeout timer stays on the legacy [`check_timeouts`](Self::check_timeouts) - /// path until the next fold); the rekey/liveness kinds keep their own shell - /// drivers, so this deliberately kind-filters to `HandshakeRetransmit`. - /// - /// Retransmit is the pre-fold `resend_pending_handshakes` logic, re-homed: - /// the *due* signal is now the machine-armed timer (not the connection's - /// `next_resend_at_ms`), and the resend counter lives on the machine (the - /// operator-visible count reads from there). The wire bytes and transport - /// target still come from the shell connection, the pure core computes the - /// backoff schedule, and — matching the old shell exactly — the count and - /// reschedule advance only on a successful send; a failed send neither - /// advances the count nor marks the connection failed, it just retries next - /// tick. + /// [`peer_timers`](Node::peer_timers); this is the shell driver that acts on + /// them: timeout reaps idle-timed-out outbound legs, retransmit resends the + /// due msg1s. Handshake-TIMEOUT is driven before handshake-RETRANSMIT so a + /// timed-out leg is reaped rather than resent on the same tick. The + /// rekey/liveness kinds keep their own shell drivers, so only the two + /// handshake kinds are driven here. pub(in crate::node) async fn drive_peer_timers(&mut self, now_ms: u64) { if self.peer_timers.is_empty() { return; } + self.drive_handshake_timeouts(now_ms); + self.drive_handshake_retransmits(now_ms).await; + } + /// Reap the outbound legs whose machine `HandshakeTimeout` timer marks them + /// as machine-timeout-owned and which have idle-timed-out this tick. + /// + /// The timer's PRESENCE selects the leg (only OUTBOUND legs arm one — IK + /// inbound arms none); the reap THRESHOLD is the shell `is_timed_out(now, + /// config)` predicate, NOT the timer's stored deadline. This matters because + /// the machine arms the timer from a hardcoded constant at dial, which is not + /// authoritative for an operator-tuned `handshake_timeout_secs` — reading the + /// threshold from config each tick keeps the reap neutral for any config, and + /// off the `last_activity` clock exactly as the old `check_timeouts` did. A + /// timed-out leg is reaped by the old Teardown path: the outbound retry + /// reflex, then `cleanup_stale_connection` (which drops the machine + timers). + /// + /// `check_timeouts` keeps reaping everything else — `is_failed()` legs and the + /// idle-timeout of legs without a machine timer (inbound / machine-less). + fn drive_handshake_timeouts(&mut self, now_ms: u64) { + let timeout_ms = self.config().node.rate_limit.handshake_timeout_secs * 1000; + let timer_links: Vec = self + .peer_timers + .iter() + .filter(|(_, timers)| timers.contains_key(&TimerKind::HandshakeTimeout)) + .map(|(link, _)| *link) + .collect(); + for link in timer_links { + let (reap, retry_peer) = match self.connections.get(&link) { + Some(conn) if conn.is_timed_out(now_ms, timeout_ms) => { + let retry_peer = if conn.is_outbound() { + conn.expected_identity().map(|id| *id.node_addr()) + } else { + None + }; + (true, retry_peer) + } + // Not yet idle-timed-out: leave the timer for a later tick. + Some(_) => (false, None), + None => { + // Orphan timer (connection already reaped elsewhere) — drop it. + if let Some(timers) = self.peer_timers.get_mut(&link) { + timers.remove(&TimerKind::HandshakeTimeout); + } + (false, None) + } + }; + if reap { + if let Some(peer) = retry_peer { + self.note_handshake_timeout(peer, now_ms); + } + debug!(link_id = %link, "Handshake connection timed out"); + self.cleanup_stale_connection(link, now_ms); + } + } + } + + /// Fire due handshake-retransmit timers: resend the stored msg1. + /// + /// The pre-fold `resend_pending_handshakes` logic, re-homed: the *due* signal + /// is the machine-armed timer (not the connection's `next_resend_at_ms`), and + /// the resend counter lives on the machine (the operator-visible count reads + /// from there). The wire bytes and transport target still come from the shell + /// connection, the pure core computes the backoff schedule, and — matching the + /// old shell exactly — the count and reschedule advance only on a successful + /// send; a failed send neither advances the count nor marks the connection + /// failed, it just retries next tick. + async fn drive_handshake_retransmits(&mut self, now_ms: u64) { let max_resends = self.config().node.rate_limit.handshake_max_resends; let interval_ms = self.config().node.rate_limit.handshake_resend_interval_ms; let backoff = self.config().node.rate_limit.handshake_resend_backoff; diff --git a/src/node/tests/handshake.rs b/src/node/tests/handshake.rs index ddabf11..9dcc0a2 100644 --- a/src/node/tests/handshake.rs +++ b/src/node/tests/handshake.rs @@ -903,6 +903,83 @@ async fn test_resend_scheduling() { ); } +/// Test that the timer driver reaps an outbound leg whose machine +/// `HandshakeTimeout` timer has come due (the timeout fold). The reap re-checks +/// the shell `is_timed_out` predicate, then tears the connection down exactly as +/// the old `check_timeouts` Teardown path did. +#[tokio::test] +async fn test_handshake_timeout_drive() { + let mut node = make_node(); + let transport_id = TransportId::new(1); + let peer_identity = make_peer_identity(); + let remote_addr = TransportAddr::from_string("10.0.0.2:2121"); + + let dial_ms = 1000u64; + let link_id = node.allocate_link_id(); + let mut conn = PeerConnection::outbound(link_id, peer_identity, dial_ms); + let our_index = node.index_allocator.allocate().unwrap(); + let our_keypair = node.identity().keypair(); + let _ = conn + .start_handshake(our_keypair, node.startup_epoch(), dial_ms) + .unwrap(); + conn.set_our_index(our_index); + conn.set_transport_id(transport_id); + conn.set_source_addr(remote_addr.clone()); + + let link = Link::connectionless( + link_id, + transport_id, + remote_addr.clone(), + LinkDirection::Outbound, + Duration::from_millis(100), + ); + node.links.insert(link_id, link); + node.addr_to_link + .insert((transport_id, remote_addr.clone()), link_id); + node.connections.insert(link_id, conn); + node.pending_outbound + .insert((transport_id, our_index.as_u32()), link_id); + + // Machine in SentMsg1 with a HandshakeTimeout timer armed at dial + 30s. + let mut machine = + crate::peer::machine::PeerMachine::new_outbound(link_id, peer_identity, dial_ms); + let _ = machine.step( + crate::peer::machine::PeerEvent::Dial { + transport_id, + remote_addr: remote_addr.clone(), + peer_identity, + connection_oriented: false, + }, + dial_ms, + &mut node.index_allocator, + ); + node.peer_machines.insert(link_id, machine); + node.peer_timers.entry(link_id).or_default().insert( + crate::peer::machine::TimerKind::HandshakeTimeout, + dial_ms + 30_000, + ); + + assert_eq!(node.connection_count(), 1); + + // Well past dial + 30s: the timer is due and the leg is idle-timed-out. + node.drive_peer_timers(dial_ms + 100_000).await; + + assert_eq!( + node.connection_count(), + 0, + "Timed-out leg reaped by the timer drive" + ); + assert_eq!(node.index_allocator.count(), 0, "Session index freed"); + assert!( + !node.peer_machines.contains_key(&link_id), + "Control machine dropped with the reaped connection" + ); + assert!( + !node.peer_timers.contains_key(&link_id), + "Timer store dropped with the reaped connection" + ); +} + /// Test that msg2 is stored on PeerConnection for responder resend. #[test] fn test_msg2_stored_on_connection() {