node: drive the handshake msg1 resend from the per-peer machine timer

Move the outbound msg1 resend off the unconditional tick function onto the
machine-armed retransmit timer. The per-peer machine already arms a
HandshakeRetransmit deadline at dial; a new drive_peer_timers fires the due ones
(kind-filtered to the retransmit timer) and homes the resend counter on the
machine, where the operator-visible count now reads from.

The resend decision reads the same operator config as before (interval, backoff,
max), the wire bytes and transport target still come from the shell connection,
and the pure core computes the backoff schedule. As with the deleted
resend_pending_handshakes, 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 on the next tick. The handshake-timeout timer stays on the legacy
check_timeouts path, and the rekey/liveness timers keep their own shell drivers,
so drive_peer_timers deliberately fires only the retransmit kind.

The show_connections resend count is relocated to read from the machine (the
counter's new home) via connection_resend_count; machine-less and inbound
connections report 0, matching what the shell connection reported before. Delete
resend_pending_handshakes and resend_candidates (the latter also from the
LifecycleView trait, its only user). The resend unit test is re-expressed against
the machine + timer path, covering the due check and the record-on-success
semantics.

Known limitation: the first resend interval is armed from the machine's hardcoded
1000ms constant, which equals the config default. Under an operator override of
handshake_resend_interval_ms the first resend diverges from the pre-change
dial+interval; subsequent resends and the cap remain config-driven. Neutralizing
the first-resend override needs the interval threaded into the sync core (the
shell arm would be clobbered by the machine's own timer arming at dial), deferred
to when the connection and machine entities merge. Test count unchanged at 1631.

src/control/queries.rs

@@ -1310,7 +1310,7 @@ pub fn show_connections(node: &Node) -> Value {
                 "handshake_state": format!("{}", conn.handshake_state()),
                 "started_at_ms": conn.started_at(),
                 "idle_ms": now.saturating_sub(conn.last_activity()),
-                "resend_count": conn.resend_count(),
+                "resend_count": node.connection_resend_count(conn.link_id()),
             });
 
             if let Some(identity) = conn.expected_identity() {

src/node/dataplane/peer_actions.rs

@@ -21,11 +21,11 @@
 //! 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.
+//! 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.
 
 use crate::PeerIdentity;
 use crate::node::Node;
@@ -500,13 +500,12 @@ impl Node {
                     // Connected-UDP plane ownership (`connected_udp.rs`).
                 }
                 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.
+                    // 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.
                     self.peer_timers
                         .entry(link)
                         .or_default()

src/node/dataplane/rx_loop.rs

@@ -334,7 +334,7 @@ impl Node {
                     self.poll_pending_connects().await;
                     self.poll_nostr_rendezvous().await;
                     self.poll_lan_rendezvous().await;
-                    self.resend_pending_handshakes(now_ms).await;
+                    self.drive_peer_timers(now_ms).await;
                     self.resend_pending_rekeys(now_ms).await;
                     self.resend_pending_session_handshakes(now_ms).await;
                     self.resend_pending_session_msg3(now_ms).await;

src/node/handlers/timeout.rs

@@ -2,7 +2,7 @@
 //! and handshake message resend scheduling.
 
 use crate::node::Node;
-use crate::peer::HandshakeState;
+use crate::peer::machine::TimerKind;
 use crate::proto::fmp::{
     ConnAction, ConnSnapshot, LifecycleView, PeerSnapshot, RekeyResendSnapshot,
 };
@@ -24,28 +24,6 @@ impl LifecycleView for Node {
             .collect()
     }
 
-    fn resend_candidates(&self, now_ms: u64, max_resends: u32) -> Vec<ConnSnapshot> {
-        self.connections
-            .iter()
-            .filter(|(_, conn)| {
-                conn.is_outbound()
-                    && conn.handshake_state() == HandshakeState::SentMsg1
-                    && conn.resend_count() < max_resends
-                    && conn.next_resend_at_ms() > 0
-                    && now_ms >= conn.next_resend_at_ms()
-            })
-            .filter_map(|(link_id, conn)| {
-                conn.handshake_msg1().map(|msg1| ConnSnapshot {
-                    link: *link_id,
-                    is_outbound: true,
-                    retry_addr: None,
-                    resend_count: conn.resend_count(),
-                    msg1: msg1.to_vec(),
-                })
-            })
-            .collect()
-    }
-
     fn rekey_peers(&self) -> Vec<PeerSnapshot> {
         // The snapshot builder lives in `rekey` beside its drain/dampening
         // constants; the read-seam unifies here.
@@ -137,12 +115,26 @@ impl Node {
         }
     }
 
-    /// Resend handshake messages for pending connections.
+    /// Drive the per-peer machine timers that have come due on this tick.
     ///
-    /// For outbound connections in SentMsg1 state, resends the stored msg1
-    /// with exponential backoff. Called periodically from the RX event loop.
-    pub(in crate::node) async fn resend_pending_handshakes(&mut self, now_ms: u64) {
-        if self.connections.is_empty() {
+    /// 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.
+    pub(in crate::node) async fn drive_peer_timers(&mut self, now_ms: u64) {
+        if self.peer_timers.is_empty() {
             return;
         }
 
@@ -150,9 +142,64 @@ impl Node {
         let interval_ms = self.config().node.rate_limit.handshake_resend_interval_ms;
         let backoff = self.config().node.rate_limit.handshake_resend_backoff;
 
-        // The shell resolves the resend-candidate predicate and copies the
-        // opaque msg1 bytes; the core computes the backoff schedule.
-        let candidates = self.resend_candidates(now_ms, max_resends);
+        // Collect due retransmit timers (kind-filtered).
+        let due: Vec<LinkId> = self
+            .peer_timers
+            .iter()
+            .filter(|(_, timers)| {
+                timers
+                    .get(&TimerKind::HandshakeRetransmit)
+                    .is_some_and(|&at_ms| now_ms >= at_ms)
+            })
+            .map(|(link, _)| *link)
+            .collect();
+        if due.is_empty() {
+            return;
+        }
+
+        // Classify each due link against the machine + connection. A timer whose
+        // machine has left `SentMsg1` (promoted/gone) or has hit the resend cap
+        // is dropped — no more resends, exactly as the old shell stopped
+        // selecting a capped/settled connection; the handshake-timeout reaper
+        // takes it from there.
+        let mut candidates: Vec<ConnSnapshot> = Vec::new();
+        let mut drop_timers: Vec<LinkId> = Vec::new();
+        for link in due {
+            let armed = match self.peer_machines.get(&link) {
+                Some(machine)
+                    if machine.is_handshaking_sent_msg1()
+                        && machine.resend_count() < max_resends =>
+                {
+                    machine.resend_count()
+                }
+                Some(_) => {
+                    drop_timers.push(link);
+                    continue;
+                }
+                None => {
+                    drop_timers.push(link);
+                    continue;
+                }
+            };
+            match self.connections.get(&link).and_then(|c| c.handshake_msg1()) {
+                // Armed but the stored wire isn't there yet — leave the timer and
+                // retry next tick (matches the old candidate filter skipping it).
+                None => continue,
+                Some(msg1) => candidates.push(ConnSnapshot {
+                    link,
+                    is_outbound: true,
+                    retry_addr: None,
+                    resend_count: armed,
+                    msg1: msg1.to_vec(),
+                }),
+            }
+        }
+        for link in drop_timers {
+            if let Some(timers) = self.peer_timers.get_mut(&link) {
+                timers.remove(&TimerKind::HandshakeRetransmit);
+            }
+        }
+
         for action in self
             .fmp
             .poll_resends(candidates, now_ms, interval_ms, backoff)
@@ -166,7 +213,6 @@ impl Node {
                 continue;
             };
 
-            // Get transport and address info from the connection
             let (transport_id, remote_addr) = match self.connections.get(&link) {
                 Some(conn) => match (conn.transport_id(), conn.source_addr()) {
                     (Some(tid), Some(addr)) => (tid, addr.clone()),
@@ -175,7 +221,6 @@ impl Node {
                 None => continue,
             };
 
-            // Send the stored msg1
             let sent = if let Some(transport) = self.transports.get(&transport_id) {
                 match transport.send(&remote_addr, &bytes).await {
                     Ok(_) => true,
@@ -192,14 +237,27 @@ impl Node {
                 false
             };
 
-            if sent && let Some(conn) = self.connections.get_mut(&link) {
-                conn.record_resend(next_resend_at_ms);
+            if sent {
+                if let Some(machine) = self.peer_machines.get_mut(&link) {
+                    machine.record_resend(next_resend_at_ms);
                     debug!(
                         link_id = %link,
-                    resend = conn.resend_count(),
+                        resend = machine.resend_count(),
                         "Resent handshake msg1"
                     );
                 }
+                self.peer_timers
+                    .entry(link)
+                    .or_default()
+                    .insert(TimerKind::HandshakeRetransmit, next_resend_at_ms);
+            } else {
+                // Failed send: keep retrying at the tick cadence (the old shell
+                // left next_resend_at_ms unchanged so the connection stayed due).
+                self.peer_timers
+                    .entry(link)
+                    .or_default()
+                    .insert(TimerKind::HandshakeRetransmit, now_ms);
+            }
         }
     }
 

src/node/mod.rs

@@ -1984,7 +1984,7 @@ impl Node {
                 handshake_state: format!("{}", conn.handshake_state()),
                 started_at_ms: conn.started_at(),
                 last_activity_ms: conn.last_activity(),
-                resend_count: conn.resend_count(),
+                resend_count: self.connection_resend_count(conn.link_id()),
                 expected_peer: conn.expected_identity().map(|id| id.npub()),
             })
             .collect();
@@ -2268,6 +2268,17 @@ impl Node {
         self.peer_timers.remove(&link);
     }
 
+    /// Operator-visible msg1 resend count for a pending handshake `link`, read
+    /// from the per-peer machine (the counter's home once the resend drive moved
+    /// off the shell connection). Machine-less connections (inbound legs that
+    /// never resend, and test-created connections) report 0, matching what the
+    /// shell connection reported before the counter moved.
+    pub(crate) fn connection_resend_count(&self, link: LinkId) -> u32 {
+        self.peer_machines
+            .get(&link)
+            .map_or(0, |machine| machine.resend_count())
+    }
+
     pub(crate) fn cleanup_bootstrap_transport_if_unused(&mut self, transport_id: TransportId) {
         if !self
             .supervisor

src/node/tests/handshake.rs

@@ -858,28 +858,48 @@ async fn test_resend_scheduling() {
     );
     node.links.insert(link_id, link);
     node.addr_to_link
-        .insert((transport_id, remote_addr), link_id);
+        .insert((transport_id, remote_addr.clone()), link_id);
     node.pending_outbound
         .insert((transport_id, our_index.as_u32()), link_id);
     node.connections.insert(link_id, conn);
 
-    // Before resend time: nothing should happen (no transport = can't send,
-    // but the filter should exclude it because now < next_resend_at)
-    node.resend_pending_handshakes(now_ms + 500).await;
-    let conn = node.connections.get(&link_id).unwrap();
-    assert_eq!(conn.resend_count(), 0, "No resend before scheduled time");
+    // The msg1-resend counter and its due timer live on the per-peer machine.
+    // Dial it to `SentMsg1` (connectionless: no connect step) and arm its
+    // retransmit timer at now + 1000ms, mirroring what a real dial arms.
+    let mut machine =
+        crate::peer::machine::PeerMachine::new_outbound(link_id, peer_identity, now_ms);
+    let _ = machine.step(
+        crate::peer::machine::PeerEvent::Dial {
+            transport_id,
+            remote_addr: remote_addr.clone(),
+            peer_identity,
+            connection_oriented: false,
+        },
+        now_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::HandshakeRetransmit,
+        now_ms + 1000,
+    );
 
-    // At resend time: would resend if transport existed. Without transport,
-    // the send fails silently and resend_count stays at 0.
-    // This tests the filtering logic — the connection IS a candidate.
-    node.resend_pending_handshakes(now_ms + 1000).await;
-    // No transport registered, so send fails — count stays 0.
-    // That's the expected behavior (transport absence is a transient condition).
-    let conn = node.connections.get(&link_id).unwrap();
+    // Before the scheduled time the timer isn't due, so nothing fires.
+    node.drive_peer_timers(now_ms + 500).await;
     assert_eq!(
-        conn.resend_count(),
+        node.connection_resend_count(link_id),
         0,
-        "No transport means no resend recorded"
+        "No resend before scheduled time"
+    );
+
+    // At the scheduled time the timer is due, but no transport is registered so
+    // the send fails. Record-on-success: the count does NOT advance (and the
+    // connection is not marked failed) — a failed resend just retries next tick.
+    node.drive_peer_timers(now_ms + 1000).await;
+    assert_eq!(
+        node.connection_resend_count(link_id),
+        0,
+        "Failed send records no resend"
     );
 }
 

src/peer/machine.rs

@@ -446,6 +446,36 @@ impl PeerMachine {
         self.our_index
     }
 
+    /// The msg1 resend count for this outbound handshake leg. The per-peer
+    /// machine is the home for this counter — the timer driver advances it via
+    /// [`record_resend`](Self::record_resend) on each successful resend, and the
+    /// control-socket connection snapshot reads it here so the operator-visible
+    /// count follows the machine rather than the (now inert) shell connection.
+    pub(crate) fn resend_count(&self) -> u32 {
+        self.conn.resend_count()
+    }
+
+    /// Record a successful msg1 resend: advance the count and store the next
+    /// backoff deadline. The driver calls this only after the resend actually
+    /// went out (record-on-success — a failed send neither advances the count
+    /// nor reschedules), matching the pre-fold shell semantics.
+    pub(crate) fn record_resend(&mut self, next_resend_at_ms: u64) {
+        self.conn.record_resend(next_resend_at_ms);
+    }
+
+    /// Whether this is an outbound leg parked at `SentMsg1` — the only state in
+    /// which a msg1 resend is due. Mirrors `on_handshake_retransmit`'s guard so
+    /// the shell timer driver can gate without reaching into machine state.
+    pub(crate) fn is_handshaking_sent_msg1(&self) -> bool {
+        matches!(
+            self.state,
+            PeerState::Handshaking {
+                phase: HandshakePhase::SentMsg1,
+                ..
+            }
+        )
+    }
+
     /// The crystallized node address, if identity is known.
     fn addr(&self) -> Option<NodeAddr> {
         self.identity.map(|id| *id.node_addr())

src/proto/fmp/core.rs

@@ -328,13 +328,6 @@ pub(crate) trait LifecycleView {
     /// timeout/failed predicate; the core decides retry-then-teardown.
     fn stale_connections(&self, now_ms: u64, timeout_ms: u64) -> Vec<ConnSnapshot>;
 
-    /// Snapshot every outbound connection whose stored msg1 is due for a
-    /// resend as of `now_ms` and still under `max_resends`. The shell resolves
-    /// the "outbound, in `SentMsg1`, has stored msg1, under budget, past the
-    /// scheduled time" predicate and copies the opaque msg1 bytes; the core
-    /// computes the backoff schedule.
-    fn resend_candidates(&self, now_ms: u64, max_resends: u32) -> Vec<ConnSnapshot>;
-
     /// Snapshot every active peer with a session that is healthy, pre-computing
     /// its rekey-relevant ages and timer predicates (see [`PeerSnapshot`]). The
     /// shell resolves every clock read here; the core applies the thresholds.