diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 785d0d0..20f2ae6 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -334,6 +334,11 @@ impl Node { // variants complete the rate-limiter and return here. The local machine // enters `peer_machines` only at the promote tails. let mut machine = PeerMachine::new_inbound(link_id, packet.timestamp_ms); + // The inbound leg carries the transport ID from msg1, but the machine's + // carrier is only written on the outbound dial. Seed it here so the + // promotion hand-off reads it from the surviving carrier, matching the + // leg's inbound seed. + machine.set_conn_transport_id(packet.transport_id); let (decision, actions) = machine.inbound_msg1(link_id, &wire, est, packet.timestamp_ms); match decision { InboundDecision::Reject { @@ -616,7 +621,6 @@ impl Node { // already freed by `remove_active_peer` above, BEFORE this fresh // allocation — matching the pre-refactor allocation sequence. conn.set_our_index(our_index); - conn.set_their_index(their_index); let link = Link::connectionless( link_id, packet.transport_id, @@ -765,7 +769,6 @@ impl Node { // set indices on the shell connection, insert link / reverse map / // connection, then build + store the framed msg2. conn.set_our_index(our_index); - conn.set_their_index(their_index); let link = Link::connectionless( link_id, packet.transport_id, @@ -1021,7 +1024,6 @@ impl Node { return; } - conn.set_their_index(header.sender_idx); conn.set_source_addr(packet.remote_addr.clone()); let peer_identity = match conn.expected_identity() { @@ -1388,14 +1390,18 @@ impl Node { link_id, reason: "missing our_index".into(), })?; - let their_index = connection - .their_index() + let their_index = self + .peer_machines + .get(&link_id) + .and_then(|machine| machine.conn_their_index()) .ok_or_else(|| NodeError::PromotionFailed { link_id, reason: "missing their_index".into(), })?; - let transport_id = connection - .transport_id() + let transport_id = self + .peer_machines + .get(&link_id) + .and_then(|machine| machine.conn_transport_id()) .ok_or_else(|| NodeError::PromotionFailed { link_id, reason: "missing transport_id".into(), @@ -1407,7 +1413,11 @@ impl Node { reason: "missing source_addr".into(), })? .clone(); - let link_stats = connection.link_stats().clone(); + let link_stats = self + .peer_machines + .get(&link_id) + .map(|machine| machine.conn_link_stats().clone()) + .unwrap_or_default(); let remote_epoch = connection.remote_epoch(); let peer_node_addr = *verified_identity.node_addr(); diff --git a/src/node/handlers/timeout.rs b/src/node/handlers/timeout.rs index 09ea623..741c702 100644 --- a/src/node/handlers/timeout.rs +++ b/src/node/handlers/timeout.rs @@ -124,12 +124,17 @@ impl Node { Some(c) => c, None => return, }; + // Read the transport ID off the surviving carrier before disposing the + // machine (the leg no longer projects it). + let transport_id = self + .peer_machines + .get(&link_id) + .and_then(|machine| machine.conn_transport_id()); self.remove_peer_machine(link_id); - let transport_id = conn.transport_id(); // Free session index and pending_outbound if allocated if let Some(idx) = conn.our_index() { - if let Some(tid) = conn.transport_id() { + if let Some(tid) = transport_id { self.pending_outbound.remove(&(tid, idx.as_u32())); } let _ = self.index_allocator.free(idx); @@ -310,7 +315,12 @@ impl Node { }; let (transport_id, remote_addr) = match self.leg(&link) { - Some(conn) => match (conn.transport_id(), conn.source_addr()) { + Some(conn) => match ( + self.peer_machines + .get(&link) + .and_then(|machine| machine.conn_transport_id()), + conn.source_addr(), + ) { (Some(tid), Some(addr)) => (tid, addr.clone()), _ => continue, }, diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs index 5a90c0c..c1789e9 100644 --- a/src/node/lifecycle/mod.rs +++ b/src/node/lifecycle/mod.rs @@ -385,12 +385,14 @@ impl Node { transport_id: TransportId, remote_addr: &TransportAddr, ) -> bool { - self.connections().any(|conn| { - conn.expected_identity() - .map(|id| id.node_addr() == peer_node_addr) - .unwrap_or(false) - && conn.transport_id() == Some(transport_id) - && conn.source_addr() == Some(remote_addr) + self.peer_machines.values().any(|machine| { + machine.leg().is_some_and(|conn| { + conn.expected_identity() + .map(|id| id.node_addr() == peer_node_addr) + .unwrap_or(false) + && machine.conn_transport_id() == Some(transport_id) + && conn.source_addr() == Some(remote_addr) + }) }) || self.peering.pending_connects.iter().any(|pending| { pending.peer_identity.node_addr() == peer_node_addr && pending.transport_id == transport_id @@ -611,7 +613,6 @@ impl Node { // Set index and transport info on the connection connection.set_our_index(our_index); - connection.set_transport_id(transport_id); connection.set_source_addr(remote_addr.clone()); // Build wire format msg1: [0x01][sender_idx:4 LE][noise_msg1:82] @@ -653,6 +654,10 @@ impl Node { // round-trip separates dial from msg1 preparation. machine.set_conn_started_at(current_time_ms); machine.touch_conn(current_time_ms); + // Record the transport ID on the surviving carrier (the leg no longer + // projects it to the promotion hand-off); holds even if a direct caller + // reached here without the dial-time `on_dial` write. + machine.set_conn_transport_id(transport_id); // Store the msg1 wire on the surviving carrier (the leg no longer holds // the resend source); the retransmit driver reads it from here. machine.set_conn_handshake_msg1(wire_msg1, first_resend_at_ms); diff --git a/src/node/mod.rs b/src/node/mod.rs index f711c78..6b8d2f4 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -2367,9 +2367,9 @@ impl Node { .links .values() .any(|link| link.transport_id() == transport_id) - || self - .connections() - .any(|conn| conn.transport_id() == Some(transport_id)) + || self.peer_machines.values().any(|machine| { + machine.leg().is_some() && machine.conn_transport_id() == Some(transport_id) + }) || self .peers .values() @@ -2441,18 +2441,25 @@ impl Node { }); } - self.peer_machines - .entry(link_id) - .or_insert_with(|| { - let now = connection.started_at(); - match connection.expected_identity() { - Some(identity) if connection.is_outbound() => { - PeerMachine::new_outbound(link_id, *identity, now) - } - _ => PeerMachine::new_inbound(link_id, now), + let machine = self.peer_machines.entry(link_id).or_insert_with(|| { + let now = connection.started_at(); + match connection.expected_identity() { + Some(identity) if connection.is_outbound() => { + PeerMachine::new_outbound(link_id, *identity, now) } - }) - .set_leg(connection); + _ => PeerMachine::new_inbound(link_id, now), + } + }); + // Seed the surviving carrier's peer index and transport from the + // pre-built leg so the promotion hand-off reads them from the machine, + // matching the establish paths that write them on the machine directly. + if let Some(their) = connection.their_index() { + machine.set_conn_their_index(their); + } + if let Some(tid) = connection.transport_id() { + machine.set_conn_transport_id(tid); + } + machine.set_leg(connection); Ok(()) } diff --git a/src/peer/connection.rs b/src/peer/connection.rs index 4c7e211..ea689bd 100644 --- a/src/peer/connection.rs +++ b/src/peer/connection.rs @@ -10,7 +10,7 @@ use crate::PeerIdentity; use crate::noise::{self, NoiseError, NoiseSession}; use crate::proto::fmp::ConnectionState; -use crate::transport::{LinkDirection, LinkId, LinkStats, TransportAddr, TransportId}; +use crate::transport::{LinkDirection, LinkId, TransportAddr, TransportId}; use crate::utils::index::SessionIndex; use secp256k1::Keypair; use std::fmt; @@ -131,16 +131,6 @@ impl PeerConnection { self.state.idle_time(current_time_ms) } - /// Get link statistics. - pub fn link_stats(&self) -> &LinkStats { - self.state.link_stats() - } - - /// Get mutable link statistics. - pub fn link_stats_mut(&mut self) -> &mut LinkStats { - self.state.link_stats_mut() - } - // === Index Accessors === /// Get our session index (if set). diff --git a/src/peer/machine.rs b/src/peer/machine.rs index fa0801e..1912d2e 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -63,7 +63,7 @@ use crate::proto::fmp::{ RekeyResendSnapshot, WireOutcome, }; use crate::proto::link::LinkMessageType; -use crate::transport::{LinkId, TransportAddr, TransportId}; +use crate::transport::{LinkId, LinkStats, TransportAddr, TransportId}; use crate::utils::index::{IndexAllocator, SessionIndex}; use crate::{NodeAddr, PeerIdentity}; @@ -620,6 +620,39 @@ impl PeerMachine { self.conn.set_handshake_msg2(msg2); } + /// Peer session index of the surviving carrier — the source for the + /// promotion hand-off now that the leg no longer projects it. + pub(crate) fn conn_their_index(&self) -> Option { + self.conn.their_index() + } + + /// Transport ID of the surviving carrier — the source for the promotion + /// hand-off, the stale-connection cleanup, and the msg1 resend send. + pub(crate) fn conn_transport_id(&self) -> Option { + self.conn.transport_id() + } + + /// Link statistics of the surviving carrier — the seed copied into the + /// active peer at promotion. + pub(crate) fn conn_link_stats(&self) -> &LinkStats { + self.conn.link_stats() + } + + /// Record the peer session index on the surviving carrier. Seeds the + /// carrier from a pre-built leg (`Node::add_connection`) so the promotion + /// hand-off matches the establish paths that write it on the machine. + pub(crate) fn set_conn_their_index(&mut self, index: SessionIndex) { + self.conn.set_their_index(index); + } + + /// Record the transport ID on the surviving carrier. Populated on the + /// inbound establish path (the leg seeds it at msg1, but the machine's + /// carrier is only written on the outbound dial) and when seeding the + /// carrier from a pre-built leg (`Node::add_connection`). + pub(crate) fn set_conn_transport_id(&mut self, id: TransportId) { + self.conn.set_transport_id(id); + } + /// 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. diff --git a/src/proto/fmp/state.rs b/src/proto/fmp/state.rs index 7e9dcbc..a951373 100644 --- a/src/proto/fmp/state.rs +++ b/src/proto/fmp/state.rs @@ -93,9 +93,6 @@ pub struct ConnectionState { /// Number of resends performed so far. resend_count: u32, - - /// When the next resend should fire (Unix ms). 0 = no resend scheduled. - next_resend_at_ms: u64, } impl ConnectionState { @@ -122,7 +119,6 @@ impl ConnectionState { handshake_msg1: None, handshake_msg2: None, resend_count: 0, - next_resend_at_ms: 0, } } @@ -146,7 +142,6 @@ impl ConnectionState { handshake_msg1: None, handshake_msg2: None, resend_count: 0, - next_resend_at_ms: 0, } } @@ -174,7 +169,6 @@ impl ConnectionState { handshake_msg1: None, handshake_msg2: None, resend_count: 0, - next_resend_at_ms: 0, } } @@ -230,11 +224,6 @@ impl ConnectionState { &self.link_stats } - /// Get mutable link statistics. - pub fn link_stats_mut(&mut self) -> &mut LinkStats { - &mut self.link_stats - } - // === Index Accessors === /// Get our session index (if set). @@ -300,11 +289,12 @@ impl ConnectionState { // === Handshake Resend === - /// Store the wire-format msg1 bytes for resend and schedule the first resend. - pub fn set_handshake_msg1(&mut self, msg1: Vec, first_resend_at_ms: u64) { + /// Store the wire-format msg1 bytes for resend and reset the resend counter. + /// The first-resend deadline is scheduled by the shell timer driver, not + /// tracked here. + pub fn set_handshake_msg1(&mut self, msg1: Vec, _first_resend_at_ms: u64) { self.handshake_msg1 = Some(msg1); self.resend_count = 0; - self.next_resend_at_ms = first_resend_at_ms; } /// Store the wire-format msg2 bytes for resend on duplicate msg1. @@ -327,16 +317,10 @@ impl ConnectionState { self.resend_count } - /// When the next resend is scheduled (Unix ms). - #[cfg(test)] - pub fn next_resend_at_ms(&self) -> u64 { - self.next_resend_at_ms - } - - /// Record a resend and schedule the next one. - pub fn record_resend(&mut self, next_resend_at_ms: u64) { + /// Record a resend. The next-resend deadline is scheduled by the shell + /// timer driver, not tracked here. + pub fn record_resend(&mut self, _next_resend_at_ms: u64) { self.resend_count += 1; - self.next_resend_at_ms = next_resend_at_ms; } // === Activity / Timeout === diff --git a/src/proto/fmp/tests/state.rs b/src/proto/fmp/tests/state.rs index abd8418..2792806 100644 --- a/src/proto/fmp/tests/state.rs +++ b/src/proto/fmp/tests/state.rs @@ -27,7 +27,6 @@ fn outbound_initializes_pure_fields() { assert!(state.source_addr().is_none()); assert!(state.remote_epoch().is_none()); assert_eq!(state.resend_count(), 0); - assert_eq!(state.next_resend_at_ms(), 0); } #[test] @@ -102,21 +101,17 @@ fn resend_bookkeeping() { state.set_handshake_msg1(vec![1, 2, 3], 500); assert_eq!(state.handshake_msg1(), Some(&[1u8, 2, 3][..])); assert_eq!(state.resend_count(), 0); - assert_eq!(state.next_resend_at_ms(), 500); state.record_resend(900); assert_eq!(state.resend_count(), 1); - assert_eq!(state.next_resend_at_ms(), 900); state.record_resend(1300); assert_eq!(state.resend_count(), 2); - assert_eq!(state.next_resend_at_ms(), 1300); // set_handshake_msg1 resets the resend counter. state.set_handshake_msg1(vec![4, 5], 100); assert_eq!(state.handshake_msg1(), Some(&[4u8, 5][..])); assert_eq!(state.resend_count(), 0); - assert_eq!(state.next_resend_at_ms(), 100); state.set_handshake_msg2(vec![6, 7, 8]); assert_eq!(state.handshake_msg2(), Some(&[6u8, 7, 8][..]));