From b3f2018fce5fed3d414ae9e1f7d9d869defd0b3e Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Sat, 18 Jul 2026 15:11:43 +0000 Subject: [PATCH] node: source connection start/activity timestamps from the peer machine The connection leg's started_at and last_activity duplicated the peer machine's own connection-state copy. Make the machine copy the sole live telemetry and timeout carrier: re-stamp it with the leg's provenance when the leg is attached at handshake start (the msg1-prep clock, not the earlier dial-time constructor value) and mirror the completion touch, then delete the leg's last_activity/touch accessors and source the connection-row projection, show_connections, and the timeout readers from the machine. Byte-identical for all normal paths. --- src/control/queries.rs | 4 ++-- src/node/handlers/handshake.rs | 7 +++++++ src/node/handlers/timeout.rs | 16 ++++++++++++---- src/node/lifecycle/mod.rs | 13 ++++++++++--- src/node/mod.rs | 25 +++++++++++++++++++++++-- src/peer/connection.rs | 15 ++++----------- src/peer/machine.rs | 30 ++++++++++++++++++++++++++++++ src/proto/fmp/state.rs | 7 +++++++ 8 files changed, 95 insertions(+), 22 deletions(-) diff --git a/src/control/queries.rs b/src/control/queries.rs index 4982ffb..da123db 100644 --- a/src/control/queries.rs +++ b/src/control/queries.rs @@ -1308,8 +1308,8 @@ pub fn show_connections(node: &Node) -> Value { "link_id": conn.link_id().as_u64(), "direction": format!("{}", conn.direction()), "handshake_state": node.connection_handshake_state(conn.link_id()), - "started_at_ms": conn.started_at(), - "idle_ms": now.saturating_sub(conn.last_activity()), + "started_at_ms": node.connection_started_at(conn.link_id()), + "idle_ms": now.saturating_sub(node.connection_last_activity(conn.link_id())), "resend_count": node.connection_resend_count(conn.link_id()), }); diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index a9e8bba..31bab0f 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -1030,6 +1030,13 @@ impl Node { (peer_identity, conn.our_index()) }; + // Mirror the leg's completion `touch` on the surviving carrier so the + // connection's last-activity advances at msg2 completion, matching the + // leg's clock. + if let Some(machine) = self.peer_machines.get_mut(&link_id) { + machine.touch_conn(packet.timestamp_ms); + } + if self .authorize_peer( &peer_identity, diff --git a/src/node/handlers/timeout.rs b/src/node/handlers/timeout.rs index d47eccb..db43646 100644 --- a/src/node/handlers/timeout.rs +++ b/src/node/handlers/timeout.rs @@ -20,9 +20,9 @@ impl LifecycleView for Node { self.peer_machines .iter() .filter_map(|(link_id, machine)| machine.leg().map(|conn| (link_id, machine, conn))) - .filter(|(link_id, machine, conn)| { + .filter(|(link_id, machine, _conn)| { machine.is_failed() - || (conn.is_timed_out(now_ms, timeout_ms) + || (machine.conn_is_timed_out(now_ms, timeout_ms) && !self.peer_timers.get(*link_id).is_some_and(|timers| { timers.contains_key(&TimerKind::HandshakeTimeout) })) @@ -89,7 +89,8 @@ impl Node { debug!( link_id = %link, direction = %direction, - idle_secs = conn.idle_time(now_ms) / 1000, + idle_secs = + now_ms.saturating_sub(self.connection_last_activity(link)) / 1000, "Stale handshake connection timed out" ); } @@ -182,8 +183,15 @@ impl Node { .map(|(link, _)| *link) .collect(); for link in timer_links { + // The idle-timeout threshold reads the survivor carrier's + // last-activity (the leg no longer projects it); the leg still + // supplies direction/identity for the retry decision below. + let timed_out = self + .peer_machines + .get(&link) + .is_some_and(|machine| machine.conn_is_timed_out(now_ms, timeout_ms)); let (reap, retry_peer) = match self.leg(&link) { - Some(conn) if conn.is_timed_out(now_ms, timeout_ms) => { + Some(conn) if timed_out => { let retry_peer = if conn.is_outbound() { conn.expected_identity().map(|id| *id.node_addr()) } else { diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs index 5bfbaa3..2f107f9 100644 --- a/src/node/lifecycle/mod.rs +++ b/src/node/lifecycle/mod.rs @@ -642,10 +642,17 @@ impl Node { self.peer_machines.contains_key(&link_id), "outbound msg1 prepared for link {link_id} with no dial-time machine" ); - self.peer_machines + let machine = self + .peer_machines .entry(link_id) - .or_insert_with(|| PeerMachine::new_outbound(link_id, peer_identity, current_time_ms)) - .set_leg(connection); + .or_insert_with(|| PeerMachine::new_outbound(link_id, peer_identity, current_time_ms)); + // The dial-born machine carrier was stamped at dial; re-stamp it with the + // leg's msg1-prep clock so the surviving `started_at`/`last_activity` + // carry the leg's provenance. The two clocks differ when a connect + // round-trip separates dial from msg1 preparation. + machine.set_conn_started_at(current_time_ms); + machine.touch_conn(current_time_ms); + machine.set_leg(connection); Ok(()) } diff --git a/src/node/mod.rs b/src/node/mod.rs index 458aca2..602a80c 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -1982,8 +1982,8 @@ impl Node { link_id: conn.link_id().as_u64(), direction: format!("{}", conn.direction()), handshake_state: self.connection_handshake_state(conn.link_id()).to_string(), - started_at_ms: conn.started_at(), - last_activity_ms: conn.last_activity(), + started_at_ms: self.connection_started_at(conn.link_id()), + last_activity_ms: self.connection_last_activity(conn.link_id()), resend_count: self.connection_resend_count(conn.link_id()), expected_peer: conn.expected_identity().map(|id| id.npub()), }) @@ -2309,6 +2309,27 @@ impl Node { .map_or(0, |machine| machine.resend_count()) } + /// Operator-visible connection-start timestamp for a pending handshake + /// `link`, read from the per-peer machine carrier (the timing's home now + /// that the leg no longer projects it). A link with no machine reports 0; + /// every leg surfaced by `connections()` is embedded in a machine, so the + /// lookup resolves. + pub(crate) fn connection_started_at(&self, link: LinkId) -> u64 { + self.peer_machines + .get(&link) + .map_or(0, |machine| machine.conn_started_at()) + } + + /// Operator-visible last-activity timestamp for a pending handshake `link`, + /// read from the per-peer machine carrier. A link with no machine reports 0; + /// every leg surfaced by `connections()` is embedded in a machine, so the + /// lookup resolves. + pub(crate) fn connection_last_activity(&self, link: LinkId) -> u64 { + self.peer_machines + .get(&link) + .map_or(0, |machine| machine.conn_last_activity()) + } + /// Operator-visible handshake-state string for a pending handshake `link`, /// derived from the per-peer control machine (the phase's home now that the /// leg no longer carries it). Every leg surfaced by `connections()` is diff --git a/src/peer/connection.rs b/src/peer/connection.rs index e8c4aee..4c7e211 100644 --- a/src/peer/connection.rs +++ b/src/peer/connection.rs @@ -113,16 +113,14 @@ impl PeerConnection { self.state.is_inbound() } - /// When the connection started. + /// When the connection started. Retained only to seed a control machine's + /// carrier from a pre-built leg (`Node::add_connection`); the operator-facing + /// `started_at_ms`/`last_activity_ms` telemetry now reads the machine carrier, + /// not the leg. pub fn started_at(&self) -> u64 { self.state.started_at() } - /// When the last activity occurred. - pub fn last_activity(&self) -> u64 { - self.state.last_activity() - } - /// Connection duration so far. pub fn duration(&self, current_time_ms: u64) -> u64 { self.state.duration(current_time_ms) @@ -356,11 +354,6 @@ impl PeerConnection { self.noise_handshake = None; } - /// Update last activity timestamp. - pub fn touch(&mut self, current_time_ms: u64) { - self.state.touch(current_time_ms); - } - // === Validation === /// Check if the connection has timed out. diff --git a/src/peer/machine.rs b/src/peer/machine.rs index 3343d7a..02c5576 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -570,6 +570,36 @@ impl PeerMachine { self.conn.record_resend(next_resend_at_ms); } + /// Connection-start timestamp of the surviving carrier — the home for the + /// operator-visible `started_at_ms` now that the leg no longer projects it. + pub(crate) fn conn_started_at(&self) -> u64 { + self.conn.started_at() + } + + /// Last-activity timestamp of the surviving carrier — the home for the + /// operator-visible `last_activity_ms`/`idle_ms` and the idle-timeout check. + pub(crate) fn conn_last_activity(&self) -> u64 { + self.conn.last_activity() + } + + /// Whether the carrier has been idle past `timeout_ms`. + pub(crate) fn conn_is_timed_out(&self, now_ms: u64, timeout_ms: u64) -> bool { + self.conn.is_timed_out(now_ms, timeout_ms) + } + + /// Adopt an explicit connection-start timestamp on the carrier, so the + /// surviving state keeps the leg's start provenance rather than the + /// dial-time construction default. + pub(crate) fn set_conn_started_at(&mut self, started_at_ms: u64) { + self.conn.set_started_at(started_at_ms); + } + + /// Advance the carrier's last-activity timestamp — the machine-tier paired + /// write for the leg's handshake `touch`. + pub(crate) fn touch_conn(&mut self, now_ms: u64) { + self.conn.touch(now_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. diff --git a/src/proto/fmp/state.rs b/src/proto/fmp/state.rs index 817c640..7e9dcbc 100644 --- a/src/proto/fmp/state.rs +++ b/src/proto/fmp/state.rs @@ -341,6 +341,13 @@ impl ConnectionState { // === Activity / Timeout === + /// Overwrite the connection-start timestamp. Used when the surviving + /// control-machine carrier adopts the leg's start provenance instead of the + /// machine-construction default. + pub fn set_started_at(&mut self, started_at_ms: u64) { + self.started_at = started_at_ms; + } + /// Update last activity timestamp. pub fn touch(&mut self, current_time_ms: u64) { self.last_activity = current_time_ms;