peer: add a per-peer timer store populated by the machine timer actions

The control machine already emits SetTimer/CancelTimer actions when it arms the
handshake retransmit and timeout deadlines, but their executor arms were a
single no-op stub and the machine's Timeout event was never dispatched -- the
real work still runs on the legacy tick (check_timeouts,
resend_pending_handshakes).

Add the storage the time-as-input driver will read: a peer_timers map keyed by
LinkId then TimerKind, holding each armed timer's absolute deadline. The SetTimer
arm now inserts (overwrite = reschedule) and CancelTimer removes; the outbound
msg2 promote cancels the two dial-armed handshake timers, since the machine
survives promotion and the entries would otherwise linger in the store.

This store is a shadow: it is written and cleared but no driver reads it yet, so
behavior is unchanged. The legacy tick stays authoritative until the driver that
feeds Timeout and deletes the overlapping tick paths lands.

Route every machine removal through a new remove_peer_machine(link) choke-point
that drops the timer store alongside the machine, replacing the twelve direct
peer_machines.remove sites so no armed timer outlives its machine. TimerKind
gains Hash (to key the store) and Ord (for deterministic driver collection); it
is internal and never serialized. Test count unchanged at 1631.

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);
         }

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

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`,

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

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());
             }
         }

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<LinkId, PeerMachine>,
 
+    /// 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<LinkId, HashMap<TimerKind, u64>>,
+
     // === 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

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);
     }