diff --git a/src/control/queries.rs b/src/control/queries.rs index fd1d6d3..b5bbe1c 100644 --- a/src/control/queries.rs +++ b/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() { diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs index 2b71a6c..01520b5 100644 --- a/src/node/dataplane/peer_actions.rs +++ b/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() diff --git a/src/node/dataplane/rx_loop.rs b/src/node/dataplane/rx_loop.rs index acf7efb..315a7f7 100644 --- a/src/node/dataplane/rx_loop.rs +++ b/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; diff --git a/src/node/handlers/timeout.rs b/src/node/handlers/timeout.rs index 907df03..d892717 100644 --- a/src/node/handlers/timeout.rs +++ b/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 { - 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 { // 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 = 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 = Vec::new(); + let mut drop_timers: Vec = 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,13 +237,26 @@ impl Node { false }; - if sent && let Some(conn) = self.connections.get_mut(&link) { - conn.record_resend(next_resend_at_ms); - debug!( - link_id = %link, - resend = conn.resend_count(), - "Resent handshake msg1" - ); + if sent { + if let Some(machine) = self.peer_machines.get_mut(&link) { + machine.record_resend(next_resend_at_ms); + debug!( + link_id = %link, + 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); } } } diff --git a/src/node/mod.rs b/src/node/mod.rs index b490e07..fb939b8 100644 --- a/src/node/mod.rs +++ b/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 diff --git a/src/node/tests/handshake.rs b/src/node/tests/handshake.rs index cc999c5..ddabf11 100644 --- a/src/node/tests/handshake.rs +++ b/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" ); } diff --git a/src/peer/machine.rs b/src/peer/machine.rs index b7b9c4d..f741256 100644 --- a/src/peer/machine.rs +++ b/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 { self.identity.map(|id| *id.node_addr()) diff --git a/src/proto/fmp/core.rs b/src/proto/fmp/core.rs index a3d1e07..4d46ce2 100644 --- a/src/proto/fmp/core.rs +++ b/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; - /// 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; - /// 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.