mirror of
https://github.com/jmcorgan/fips.git
synced 2026-07-30 19:46:15 +00:00
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.
This commit is contained in:
@@ -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()
|
||||
|
||||
@@ -11,9 +11,21 @@ use tracing::{debug, info};
|
||||
|
||||
impl LifecycleView for Node {
|
||||
fn stale_connections(&self, now_ms: u64, timeout_ms: u64) -> Vec<ConnSnapshot> {
|
||||
// `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<LinkId> = 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;
|
||||
|
||||
@@ -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() {
|
||||
|
||||
Reference in New Issue
Block a user