diff --git a/src/node/dataplane/dispatch.rs b/src/node/dataplane/dispatch.rs index e150942..8238fa5 100644 --- a/src/node/dataplane/dispatch.rs +++ b/src/node/dataplane/dispatch.rs @@ -195,7 +195,7 @@ impl Node { // ever touches the in-flight establish's (distinct) `link_id`; no reader // depends on a stale entry, so removal changes no behavior — it only bounds // the map. - self.peer_machines.remove(&link_id); + self.remove_peer_machine(link_id); if let Some(transport_id) = transport_id { self.cleanup_bootstrap_transport_if_unused(transport_id); } diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs index 8dd2eaf..2b71a6c 100644 --- a/src/node/dataplane/peer_actions.rs +++ b/src/node/dataplane/peer_actions.rs @@ -18,9 +18,14 @@ //! send from `poll_pending_connects`). //! //! Not yet driven, so their arms stay inert stubs: rekey/crypto installs, -//! link-control frames, the timers (`SetTimer`/`CancelTimer` — the legacy tick -//! still runs them), and the connected-UDP plane. `RegisterDecryptSession` is a -//! deliberate no-op — see its arm for the note. +//! link-control frames, and the connected-UDP plane. `RegisterDecryptSession` is +//! a deliberate no-op — see its arm for the note. +//! +//! The timer arms (`SetTimer`/`CancelTimer`) are wired to populate/clear the +//! per-peer timer store (`peer_timers`), but that store is a SHADOW: the legacy +//! tick (`check_timeouts`/`resend_pending_handshakes`) still does all real timer +//! work and no `PeerEvent::Timeout` is fed, so the arms are behavior-neutral +//! until the handshake-kind driver fold cuts the legacy paths over. use crate::PeerIdentity; use crate::node::Node; @@ -145,7 +150,7 @@ impl Node { Err(_e) => { self.links.remove(&link); self.addr_to_link.remove(&(transport_id, remote_addr)); - self.peer_machines.remove(&link); + self.remove_peer_machine(link); return; } } @@ -190,7 +195,7 @@ impl Node { if let Some(idx) = ambient.our_index { let _ = self.index_allocator.free(idx); } - self.peer_machines.remove(&link); + self.remove_peer_machine(link); self.stats_mut() .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); return; @@ -356,7 +361,7 @@ impl Node { self.stats_mut().record_reject(RejectReason::Handshake( HandshakeReject::BadState, )); - self.peer_machines.remove(&promote_link); + self.remove_peer_machine(promote_link); } else { // OLD inbound (`handle_msg1` L587-591): drop the link // + reverse map, free our index, discard the machine, @@ -372,7 +377,7 @@ impl Node { if let Some(idx) = ambient.our_index { let _ = self.index_allocator.free(idx); } - self.peer_machines.remove(&promote_link); + self.remove_peer_machine(promote_link); self.stats_mut().record_reject(RejectReason::Handshake( HandshakeReject::BadState, )); @@ -494,10 +499,23 @@ impl Node { PeerAction::ActivateConnectedUdp | PeerAction::TeardownConnectedUdp => { // Connected-UDP plane ownership (`connected_udp.rs`). } - PeerAction::SetTimer { .. } | PeerAction::CancelTimer { .. } => { - // Timers become actions on the existing quantized tick. - // INERT — the legacy tick timers still run, so driving - // these would double-schedule. + PeerAction::SetTimer { kind, at_ms } => { + // Populate the driver's per-peer timer store (overwrite = + // reschedule). SHADOW ONLY at this rung: the legacy tick + // (`check_timeouts`/`resend_pending_handshakes`) still does + // all real work, and no `PeerEvent::Timeout` is fed yet, so + // this is behavior-neutral. The handshake-kind fold (C5-2b/c) + // adds the driver that reads this store and deletes the + // overlapping legacy paths. + self.peer_timers + .entry(link) + .or_default() + .insert(kind, at_ms); + } + PeerAction::CancelTimer { kind } => { + if let Some(timers) = self.peer_timers.get_mut(&link) { + timers.remove(&kind); + } } PeerAction::ReportLost { peer, kind } => { // The single loss token, routed to the reconciler reflex the diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 87563ab..ee02865 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -7,7 +7,7 @@ use crate::node::dataplane::PeerActionCtx; use crate::node::reject::{HandshakeReject, RejectReason}; use crate::node::{Node, NodeError}; use crate::peer::machine::{ - FailReason, HandshakePhase, PeerAction, PeerEvent, PeerMachine, PeerState, + FailReason, HandshakePhase, PeerAction, PeerEvent, PeerMachine, PeerState, TimerKind, }; use crate::peer::{ActivePeer, PeerConnection}; use crate::proto::fmp::wire::{Msg1Header, Msg2Header, build_msg2}; @@ -974,7 +974,7 @@ impl Node { } self.connections.remove(&link_id); // Drop the machine persisted at dial — this leg never promotes. - self.peer_machines.remove(&link_id); + self.remove_peer_machine(link_id); self.remove_link(&link_id); if let Some(idx) = our_index { let _ = self.index_allocator.free(idx); @@ -1014,7 +1014,7 @@ impl Node { // directly, with no machine). Drop it on entry so none of this block's // exits leave a dangling machine — matching the pre-persistence path, // which created no machine for a cross-connection. - self.peer_machines.remove(&link_id); + self.remove_peer_machine(link_id); // Extract the outbound connection let mut conn = match self.connections.remove(&link_id) { Some(c) => c, @@ -1186,9 +1186,22 @@ impl Node { actions } }; + // The outbound Msg2 Promote step cancels the two dial-armed handshake + // timers (the machine survives promotion, so they would otherwise linger + // in the driver's shadow store) and then promotes. The cancels are + // behavior-neutral at this rung — `peer_timers` is written but not yet + // driven — and `PromoteToActive` is still what performs the promotion. debug_assert_eq!( promote_actions, - vec![PeerAction::PromoteToActive { link: link_id }] + vec![ + PeerAction::CancelTimer { + kind: TimerKind::HandshakeRetransmit + }, + PeerAction::CancelTimer { + kind: TimerKind::HandshakeTimeout + }, + PeerAction::PromoteToActive { link: link_id }, + ] ); // Execute `[PromoteToActive]`. The executor calls `promote_connection`, diff --git a/src/node/handlers/timeout.rs b/src/node/handlers/timeout.rs index c1af290..907df03 100644 --- a/src/node/handlers/timeout.rs +++ b/src/node/handlers/timeout.rs @@ -119,7 +119,7 @@ impl Node { // `link_id` and lifetime; drop it here so a reaped handshake leg leaves // no dangling machine. A no-op for promoted peers — `promote_connection` // already removed their connection, so this reaper never runs for them. - self.peer_machines.remove(&link_id); + self.remove_peer_machine(link_id); let transport_id = conn.transport_id(); // Free session index and pending_outbound if allocated diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs index ec1c683..b3f35fb 100644 --- a/src/node/lifecycle/mod.rs +++ b/src/node/lifecycle/mod.rs @@ -588,7 +588,7 @@ impl Node { self.links.remove(&link_id); self.addr_to_link .remove(&(transport_id, remote_addr.clone())); - self.peer_machines.remove(&link_id); + self.remove_peer_machine(link_id); return Err(NodeError::IndexAllocationFailed(e.to_string())); } }; @@ -604,7 +604,7 @@ impl Node { self.links.remove(&link_id); self.addr_to_link .remove(&(transport_id, remote_addr.clone())); - self.peer_machines.remove(&link_id); + self.remove_peer_machine(link_id); return Err(NodeError::HandshakeFailed(e.to_string())); } }; @@ -1225,7 +1225,7 @@ impl Node { ); // Clean up link and dial-time machine on handshake failure self.remove_link(&pending.link_id); - self.peer_machines.remove(&pending.link_id); + self.remove_peer_machine(pending.link_id); } else { // Drive the dial-persisted machine: `Connecting` → // `on_transport_connected` → `start_outbound_handshake`, @@ -1264,7 +1264,7 @@ impl Node { // Clean up link and dial-time machine, then schedule retry self.remove_link(&pending.link_id); self.links.remove(&pending.link_id); - self.peer_machines.remove(&pending.link_id); + self.remove_peer_machine(pending.link_id); self.note_handshake_timeout(*pending.peer_identity.node_addr(), Self::now_ms()); } } diff --git a/src/node/mod.rs b/src/node/mod.rs index cb34989..b490e07 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -37,7 +37,7 @@ use self::reloadable::Reloadable; pub(crate) const REKEY_JITTER_SECS: i64 = 15; use crate::cache::CoordCache; use crate::node::session::SessionEntry; -use crate::peer::machine::PeerMachine; +use crate::peer::machine::{PeerMachine, TimerKind}; use crate::peer::{ActivePeer, PeerConnection}; use crate::proto::bloom::{BloomFilter, BloomState}; use crate::proto::fmp::Fmp; @@ -364,6 +364,15 @@ pub struct Node { #[allow(dead_code)] peer_machines: HashMap, + /// Per-peer timer store, keyed by `LinkId` then `TimerKind`, holding each + /// armed timer's absolute deadline (ms). The sans-IO time-as-input backing + /// for `PeerEvent::Timeout`: populated/cleared by the machine's + /// `SetTimer`/`CancelTimer` actions (`dataplane/peer_actions.rs`) and dropped + /// alongside the machine through the `remove_peer_machine` choke-point. At + /// this rung it is a SHADOW of the legacy tick timers — written but not yet + /// read by any driver (the handshake-kind fold wires the reader). + peer_timers: HashMap>, + // === Peers (Active Phase) === /// Authenticated peers. /// Indexed by NodeAddr (verified identity). @@ -631,6 +640,7 @@ impl Node { child_exit_rx: None, connections: HashMap::new(), peer_machines: HashMap::new(), + peer_timers: HashMap::new(), peers: HashMap::new(), sessions: HashMap::new(), identity_cache: HashMap::new(), @@ -779,6 +789,7 @@ impl Node { child_exit_rx: None, connections: HashMap::new(), peer_machines: HashMap::new(), + peer_timers: HashMap::new(), peers: HashMap::new(), sessions: HashMap::new(), identity_cache: HashMap::new(), @@ -2248,6 +2259,15 @@ impl Node { } } + /// Single choke-point for dropping a per-peer control machine. Also drops the + /// machine's timer store so no armed `SetTimer` outlives it (the store and the + /// machine share the `LinkId` lifetime). Every teardown path routes machine + /// removal here rather than calling `peer_machines.remove` directly. + pub(in crate::node) fn remove_peer_machine(&mut self, link: LinkId) { + self.peer_machines.remove(&link); + self.peer_timers.remove(&link); + } + pub(crate) fn cleanup_bootstrap_transport_if_unused(&mut self, transport_id: TransportId) { if !self .supervisor diff --git a/src/peer/machine.rs b/src/peer/machine.rs index 5305400..b7b9c4d 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -172,7 +172,14 @@ pub(crate) enum FailReason { } /// A timer the machine schedules on the driver's quantized tick. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] +/// +/// `Hash` lets it key the driver's per-peer timer store; `Ord` lets the driver +/// collect due kinds deterministically. Note the driver must fire +/// `HandshakeTimeout` before `HandshakeRetransmit` on a same-tick coincidence +/// (a reaped leg must not be resent), which is the reverse of this declaration +/// order — the driver orders explicitly rather than relying on the derived +/// ascending `Ord`. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] pub(crate) enum TimerKind { HandshakeRetransmit, HandshakeTimeout, @@ -607,7 +614,20 @@ impl PeerMachine { // Net-new: our outbound index (allocated at dial) is the one we // register once promotion resolves. self.our_index = self.conn.our_index(); - vec![PeerAction::PromoteToActive { link: self.link }] + // The machine survives promotion (it becomes the active peer's + // control machine), so cancel the outbound handshake timers here + // or they would linger in the driver's store. A late fire would + // no-op against the non-`Handshaking` state, but leaving them + // armed is a timer leak. + vec![ + PeerAction::CancelTimer { + kind: TimerKind::HandshakeRetransmit, + }, + PeerAction::CancelTimer { + kind: TimerKind::HandshakeTimeout, + }, + PeerAction::PromoteToActive { link: self.link }, + ] } OutboundDecision::CrossConnectionSwap => { // Our outbound wins: swap the peer to the outbound session, @@ -1935,9 +1955,17 @@ mod tests { ); assert_eq!( actions, - vec![PeerAction::PromoteToActive { - link: LinkId::new(1) - }] + vec![ + PeerAction::CancelTimer { + kind: TimerKind::HandshakeRetransmit + }, + PeerAction::CancelTimer { + kind: TimerKind::HandshakeTimeout + }, + PeerAction::PromoteToActive { + link: LinkId::new(1) + } + ] ); actions = m.step( PeerEvent::PromotionResolved { @@ -2051,9 +2079,17 @@ mod tests { ); assert_eq!( promote, - vec![PeerAction::PromoteToActive { - link: LinkId::new(1) - }] + vec![ + PeerAction::CancelTimer { + kind: TimerKind::HandshakeRetransmit + }, + PeerAction::CancelTimer { + kind: TimerKind::HandshakeTimeout + }, + PeerAction::PromoteToActive { + link: LinkId::new(1) + } + ] ); assert_eq!( m.our_index(), @@ -2161,9 +2197,17 @@ mod tests { ); assert_eq!( promote, - vec![PeerAction::PromoteToActive { - link: LinkId::new(1) - }] + vec![ + PeerAction::CancelTimer { + kind: TimerKind::HandshakeRetransmit + }, + PeerAction::CancelTimer { + kind: TimerKind::HandshakeTimeout + }, + PeerAction::PromoteToActive { + link: LinkId::new(1) + } + ] ); assert_eq!(m.our_index(), None); }