diff --git a/src/control/queries.rs b/src/control/queries.rs index 53f276e..9c82847 100644 --- a/src/control/queries.rs +++ b/src/control/queries.rs @@ -1303,18 +1303,18 @@ pub fn show_connections(node: &Node) -> Value { let now = now_ms(); let connections: Vec = node .connections() - .filter_map(|(_, machine)| machine.leg()) - .map(|conn| { + .map(|(_, machine)| { + let link_id = machine.link_id(); let mut conn_json = json!({ - "link_id": conn.link_id().as_u64(), - "direction": format!("{}", conn.direction()), - "handshake_state": node.connection_handshake_state(conn.link_id()), - "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()), + "link_id": link_id.as_u64(), + "direction": format!("{}", machine.conn_direction()), + "handshake_state": node.connection_handshake_state(link_id), + "started_at_ms": node.connection_started_at(link_id), + "idle_ms": now.saturating_sub(node.connection_last_activity(link_id)), + "resend_count": node.connection_resend_count(link_id), }); - if let Some(identity) = node.connection_expected_identity(conn.link_id()) { + if let Some(identity) = node.connection_expected_identity(link_id) { conn_json["expected_peer"] = json!(identity.npub()); } diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index c5ebd3c..3ea26f8 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -283,6 +283,9 @@ impl Node { // so the promotion hand-off reads it from the surviving carrier, // matching the connection's own inbound seed. machine.set_conn_transport_id(packet.transport_id); + // The inbound connection is constructed carrying the peer's address; + // seed the surviving carrier with it at the same point. + machine.set_conn_source_addr(packet.remote_addr.clone()); machine.set_leg(conn); let our_keypair = self.identity().keypair(); @@ -1041,10 +1044,10 @@ impl Node { return; } + machine.set_conn_source_addr(packet.remote_addr.clone()); let conn = machine .leg_mut() .expect("pending connection present for msg2 completion"); - conn.set_source_addr(packet.remote_addr.clone()); let peer_identity = match conn.expected_identity() { Some(id) => *id, @@ -1398,6 +1401,8 @@ impl Node { .ok_or(NodeError::ConnectionNotFound(link_id))?; let carrier_their_index = machine.conn_their_index(); let carrier_transport_id = machine.conn_transport_id(); + let carrier_source_addr = machine.conn_source_addr().cloned(); + let carrier_is_outbound = machine.conn_is_outbound(); let link_stats = machine.conn_link_stats().clone(); // Verify handshake is complete and extract session @@ -1424,17 +1429,14 @@ impl Node { link_id, reason: "missing transport_id".into(), })?; - let current_addr = connection - .source_addr() - .ok_or_else(|| NodeError::PromotionFailed { - link_id, - reason: "missing source_addr".into(), - })? - .clone(); + let current_addr = carrier_source_addr.ok_or_else(|| NodeError::PromotionFailed { + link_id, + reason: "missing source_addr".into(), + })?; let remote_epoch = connection.remote_epoch(); let peer_node_addr = *verified_identity.node_addr(); - let is_outbound = connection.is_outbound(); + let is_outbound = carrier_is_outbound; // Check for cross-connection if let Some(existing_peer) = self.peers.get(&peer_node_addr) { @@ -1566,13 +1568,14 @@ impl Node { // the 30s handshake timeout. let pending_to_same_peer: Vec = self .connections() - .filter_map(|(_, machine)| machine.leg()) - .filter(|conn| { - conn.expected_identity() + .filter(|(_, machine)| { + machine + .leg() + .and_then(|conn| conn.expected_identity()) .map(|id| *id.node_addr() == peer_node_addr) .unwrap_or(false) }) - .map(|conn| conn.link_id()) + .map(|(_, machine)| machine.link_id()) .collect(); for pending_link_id in &pending_to_same_peer { diff --git a/src/node/handlers/timeout.rs b/src/node/handlers/timeout.rs index 741c702..7599cc2 100644 --- a/src/node/handlers/timeout.rs +++ b/src/node/handlers/timeout.rs @@ -27,9 +27,9 @@ impl LifecycleView for Node { timers.contains_key(&TimerKind::HandshakeTimeout) })) }) - .map(|(link_id, _machine, conn)| ConnSnapshot { + .map(|(link_id, machine, conn)| ConnSnapshot { link: *link_id, - is_outbound: conn.is_outbound(), + is_outbound: machine.conn_is_outbound(), retry_addr: conn.expected_identity().map(|id| *id.node_addr()), resend_count: 0, msg1: Vec::new(), @@ -77,8 +77,12 @@ impl Node { .peer_machines .get(&link) .is_some_and(|machine| machine.is_failed()); - if let Some(conn) = self.leg(&link) { - let direction = conn.direction(); + if let Some(machine) = self + .peer_machines + .get(&link) + .filter(|machine| machine.leg().is_some()) + { + let direction = machine.conn_direction(); if is_failed { debug!( link_id = %link, @@ -197,7 +201,11 @@ impl Node { .is_some_and(|machine| machine.conn_is_timed_out(now_ms, timeout_ms)); let (reap, retry_peer) = match self.leg(&link) { Some(conn) if timed_out => { - let retry_peer = if conn.is_outbound() { + let retry_peer = if self + .peer_machines + .get(&link) + .is_some_and(|machine| machine.conn_is_outbound()) + { conn.expected_identity().map(|id| *id.node_addr()) } else { None @@ -314,17 +322,14 @@ impl Node { continue; }; - let (transport_id, remote_addr) = match self.leg(&link) { - 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, - }, - None => continue, + let (transport_id, remote_addr) = match self.peer_machines.get(&link) { + Some(machine) if machine.leg().is_some() => { + match (machine.conn_transport_id(), machine.conn_source_addr()) { + (Some(tid), Some(addr)) => (tid, addr.clone()), + _ => continue, + } + } + _ => continue, }; let sent = if let Some(transport) = self.transports.get(&transport_id) { diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs index f8652d6..78c747c 100644 --- a/src/node/lifecycle/mod.rs +++ b/src/node/lifecycle/mod.rs @@ -393,7 +393,7 @@ impl Node { .map(|id| id.node_addr() == peer_node_addr) .unwrap_or(false) && machine.conn_transport_id() == Some(transport_id) - && conn.source_addr() == Some(remote_addr) + && machine.conn_source_addr() == Some(remote_addr) }) }) || self.peering.pending_connects.iter().any(|pending| { pending.peer_identity.node_addr() == peer_node_addr @@ -642,8 +642,11 @@ impl Node { .and_then(|machine| machine.leg_mut()) .expect("dial-time machine carries the connection"); conn.set_our_index(our_index); - conn.set_source_addr(remote_addr.clone()); } + self.peer_machines + .get_mut(&link_id) + .expect("dial-time machine carries the connection") + .set_conn_source_addr(remote_addr.clone()); // Build wire format msg1: [0x01][sender_idx:4 LE][noise_msg1:82] let wire_msg1 = build_msg1(our_index, &noise_msg1); @@ -947,13 +950,14 @@ impl Node { let now_ms = Self::now_ms(); let stale: Vec = self .connections() - .filter_map(|(_, machine)| machine.leg()) - .filter(|conn| { - conn.expected_identity() + .filter(|(_, machine)| { + machine + .leg() + .and_then(|conn| conn.expected_identity()) .map(|id| id.node_addr() == &peer_addr) .unwrap_or(false) }) - .map(|conn| conn.link_id()) + .map(|(_, machine)| machine.link_id()) .collect(); for link_id in stale { self.cleanup_stale_connection(link_id, now_ms); diff --git a/src/node/mod.rs b/src/node/mod.rs index bb7d2dc..3287337 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -1978,16 +1978,17 @@ impl Node { // --- connections (show_connections) --- let connection_rows: Vec = self .connections() - .filter_map(|(_, machine)| machine.leg()) - .map(|conn| snap::ConnectionRow { - 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: 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()), + .map(|(_, machine)| snap::ConnectionRow { + link_id: machine.link_id().as_u64(), + direction: format!("{}", machine.conn_direction()), + handshake_state: self + .connection_handshake_state(machine.link_id()) + .to_string(), + started_at_ms: self.connection_started_at(machine.link_id()), + last_activity_ms: self.connection_last_activity(machine.link_id()), + resend_count: self.connection_resend_count(machine.link_id()), expected_peer: self - .connection_expected_identity(conn.link_id()) + .connection_expected_identity(machine.link_id()) .map(|id| id.npub()), }) .collect(); @@ -2426,7 +2427,7 @@ impl Node { /// than through the dial or inbound-msg1 paths, which build the machine /// themselves. pub fn add_connection(&mut self, connection: PeerConnection) -> Result<(), NodeError> { - let link_id = connection.link_id(); + let link_id = connection.state().link_id(); if self .peer_machines @@ -2444,8 +2445,8 @@ impl Node { 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() => { + match connection.state().expected_identity() { + Some(identity) if connection.state().is_outbound() => { PeerMachine::new_outbound(link_id, *identity, now) } _ => PeerMachine::new_inbound(link_id, now), @@ -2460,6 +2461,9 @@ impl Node { if let Some(tid) = connection.transport_id() { machine.set_conn_transport_id(tid); } + if let Some(addr) = connection.state().source_addr() { + machine.set_conn_source_addr(addr.clone()); + } machine.set_leg(connection); Ok(()) } @@ -2492,9 +2496,7 @@ impl Node { if let Some(id) = seed.transport_id { connection.set_transport_id(id); } - if let Some(addr) = seed.source_addr { - connection.set_source_addr(addr); - } + let seeded_source_addr = seed.source_addr.clone(); if let Some(index) = seed.our_index { connection.set_our_index(index); } @@ -2518,8 +2520,8 @@ impl Node { 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() => { + match connection.state().expected_identity() { + Some(identity) if connection.state().is_outbound() => { PeerMachine::new_outbound(link_id, *identity, now) } _ => PeerMachine::new_inbound(link_id, now), @@ -2531,6 +2533,9 @@ impl Node { if let Some(tid) = connection.transport_id() { machine.set_conn_transport_id(tid); } + if let Some(addr) = seeded_source_addr { + machine.set_conn_source_addr(addr); + } machine.set_leg(connection); Ok(()) } diff --git a/src/node/tests/handshake.rs b/src/node/tests/handshake.rs index dae87c6..7367b1e 100644 --- a/src/node/tests/handshake.rs +++ b/src/node/tests/handshake.rs @@ -868,7 +868,7 @@ async fn test_msg1_stored_for_resend() { .unwrap(); conn.leg_mut().unwrap().set_our_index(our_index); conn.leg_mut().unwrap().set_transport_id(transport_id); - conn.leg_mut().unwrap().set_source_addr(remote_addr.clone()); + conn.set_conn_source_addr(remote_addr.clone()); // Build wire msg1 and store it (as initiate_peer_connection does) let wire_msg1 = build_msg1(our_index, &noise_msg1); @@ -901,7 +901,7 @@ async fn test_resend_scheduling() { .unwrap(); conn.leg_mut().unwrap().set_our_index(our_index); conn.leg_mut().unwrap().set_transport_id(transport_id); - conn.leg_mut().unwrap().set_source_addr(remote_addr.clone()); + conn.set_conn_source_addr(remote_addr.clone()); // Store msg1 with first resend at now + 1000ms let wire_msg1 = crate::proto::fmp::wire::build_msg1(our_index, &noise_msg1); @@ -988,7 +988,7 @@ async fn test_handshake_timeout_drive() { .unwrap(); conn.leg_mut().unwrap().set_our_index(our_index); conn.leg_mut().unwrap().set_transport_id(transport_id); - conn.leg_mut().unwrap().set_source_addr(remote_addr.clone()); + conn.set_conn_source_addr(remote_addr.clone()); let link = Link::connectionless( link_id, diff --git a/src/node/tests/unit.rs b/src/node/tests/unit.rs index adccce9..404596d 100644 --- a/src/node/tests/unit.rs +++ b/src/node/tests/unit.rs @@ -138,8 +138,7 @@ async fn test_try_peer_addresses_races_all_concrete_udp_candidates() { let mut addrs = node .connections() - .filter_map(|(_, machine)| machine.leg()) - .filter_map(|conn| conn.source_addr().and_then(|addr| addr.as_str())) + .filter_map(|(_, machine)| machine.conn_source_addr().and_then(|addr| addr.as_str())) .collect::>(); addrs.sort(); assert_eq!(addrs, vec!["127.0.0.1:10", "127.0.0.1:9"]); @@ -1346,9 +1345,8 @@ async fn update_peers_races_new_alternative_without_dropping_active_peer() { assert_eq!(node.connection_count(), 1); assert_eq!( node.connections() - .filter_map(|(_, machine)| machine.leg()) .next() - .and_then(|conn| conn.source_addr()), + .and_then(|(_, machine)| machine.conn_source_addr()), Some(&new_addr) ); let active = node.get_peer(&peer_node_addr).unwrap(); @@ -2475,3 +2473,78 @@ async fn test_failed_msg1_preparation_unwinds_the_dial_machine() { ); assert_eq!(node.connection_count(), 0); } + +/// The link, direction, and peer address that promotion and the operator view +/// read now come from the control machine rather than the pending connection. +/// A machine's direction is seeded at construction and must match the side that +/// actually opened the link, and its address must be populated by the time +/// promotion needs it. +/// +/// This covers the two shapes that seed a carrier independently: the dial and +/// an accepted inbound message 1. The cross-connection winner derives both +/// values from a carrier one of those two already seeded, so it has nothing +/// separate to pin. +#[tokio::test] +async fn test_machine_carries_link_direction_and_address_on_dial_and_inbound() { + // Outbound: the dial builds the machine, msg1 preparation fills it in. + let mut node = make_node(); + let link_id = LinkId::new(1); + let transport_id = TransportId::new(1); + let remote_addr = TransportAddr::from_string("127.0.0.1:5000"); + let peer_identity = make_peer_identity(); + node.peer_machines.insert( + link_id, + PeerMachine::new_outbound(link_id, peer_identity, 1000), + ); + node.prepare_outbound_msg1(link_id, transport_id, &remote_addr, peer_identity) + .unwrap(); + + let machine = node.peer_machines.get(&link_id).unwrap(); + assert_eq!(machine.link_id(), link_id); + assert!(machine.conn_is_outbound(), "a dial is outbound"); + assert!(!machine.conn_is_inbound()); + assert_eq!(machine.conn_direction(), LinkDirection::Outbound); + assert_eq!( + machine.conn_source_addr(), + Some(&remote_addr), + "the dialled address must reach the surviving carrier" + ); + + // Inbound: msg1 builds the machine from the packet. + let mut responder = make_node(); + let initiator = make_node(); + let responder_identity = PeerIdentity::from_pubkey_full(responder.identity().pubkey_full()); + let mut initiator_leg = outbound_leg(LinkId::new(9), responder_identity, 1000); + let noise_msg1 = initiator_leg + .start_handshake( + initiator.identity().keypair(), + initiator.startup_epoch(), + 1000, + ) + .unwrap(); + let inbound_addr = TransportAddr::from_string("127.0.0.1:6000"); + let packet = ReceivedPacket::with_timestamp( + TransportId::new(1), + inbound_addr.clone(), + crate::proto::fmp::wire::build_msg1(SessionIndex::new(7), &noise_msg1), + 1000, + ); + responder.handle_msg1(packet).await; + + // The responder completes at msg1 and promotes, so the machine survives as + // the active peer's control machine — the carrier outlives the connection. + let (link, machine) = responder + .peer_machines + .iter() + .next() + .expect("msg1 leaves a control machine behind"); + assert!(machine.conn_is_inbound(), "an accepted msg1 is inbound"); + assert!(!machine.conn_is_outbound()); + assert_eq!(machine.conn_direction(), LinkDirection::Inbound); + assert_eq!( + machine.conn_source_addr(), + Some(&inbound_addr), + "the sender's address must reach the surviving carrier" + ); + assert_eq!(machine.link_id(), *link); +} diff --git a/src/peer/connection.rs b/src/peer/connection.rs index 6eb00d0..c77670a 100644 --- a/src/peer/connection.rs +++ b/src/peer/connection.rs @@ -10,7 +10,7 @@ use crate::PeerIdentity; use crate::noise::{self, NoiseSession}; use crate::proto::fmp::ConnectionState; -use crate::transport::{LinkDirection, LinkId, TransportAddr, TransportId}; +use crate::transport::{LinkId, TransportAddr, TransportId}; use crate::utils::index::SessionIndex; use std::fmt; @@ -91,31 +91,11 @@ impl PeerConnection { // === Accessors (delegated to the pure ConnectionState) === - /// Get the link ID. - pub fn link_id(&self) -> LinkId { - self.state.link_id() - } - - /// Get the connection direction. - pub fn direction(&self) -> LinkDirection { - self.state.direction() - } - /// Get the expected/learned peer identity, if known. pub fn expected_identity(&self) -> Option<&PeerIdentity> { self.state.expected_identity() } - /// Check if this is an outbound connection. - pub fn is_outbound(&self) -> bool { - self.state.is_outbound() - } - - /// Check if this is an inbound connection. - pub fn is_inbound(&self) -> bool { - self.state.is_inbound() - } - /// 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, @@ -166,16 +146,6 @@ impl PeerConnection { self.state.set_transport_id(id); } - /// Get the source address (if known). - pub fn source_addr(&self) -> Option<&TransportAddr> { - self.state.source_addr() - } - - /// Set the source address. - pub fn set_source_addr(&mut self, addr: TransportAddr) { - self.state.set_source_addr(addr); - } - // === Epoch Accessors === /// Get the remote peer's startup epoch (available after handshake). @@ -210,6 +180,10 @@ impl PeerConnection { /// Mutable access to the pure bookkeeping, so the control machine's /// handshake operations can record their results here as well as on the /// surviving carrier. + pub(crate) fn state(&self) -> &ConnectionState { + &self.state + } + pub(crate) fn state_mut(&mut self) -> &mut ConnectionState { &mut self.state } diff --git a/src/peer/machine.rs b/src/peer/machine.rs index 6c3157c..5add30f 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -573,9 +573,10 @@ impl PeerMachine { current_time_ms: u64, ) -> Result, NoiseError> { let msg1 = { + let direction = self.conn.direction(); let leg = self.leg.as_mut().ok_or_else(no_pending_connection)?; - if leg.direction() != LinkDirection::Outbound { + if direction != LinkDirection::Outbound { return Err(NoiseError::WrongState { expected: "outbound connection".to_string(), got: "inbound connection".to_string(), @@ -613,9 +614,10 @@ impl PeerMachine { current_time_ms: u64, ) -> Result, NoiseError> { let (msg2, learned_identity, remote_epoch) = { + let direction = self.conn.direction(); let leg = self.leg.as_mut().ok_or_else(no_pending_connection)?; - if leg.direction() != LinkDirection::Inbound { + if direction != LinkDirection::Inbound { return Err(NoiseError::WrongState { expected: "inbound connection".to_string(), got: "outbound connection".to_string(), @@ -805,6 +807,39 @@ impl PeerMachine { self.conn.set_handshake_msg2(msg2); } + /// The link this machine controls. + pub(crate) fn link_id(&self) -> LinkId { + self.link + } + + /// Which side opened this connection, read from the surviving carrier. + /// Seeded at construction: an outbound machine carries an outbound + /// connection state and an inbound machine an inbound one. + pub(crate) fn conn_direction(&self) -> LinkDirection { + self.conn.direction() + } + + /// Whether we opened this connection. + pub(crate) fn conn_is_outbound(&self) -> bool { + self.conn.is_outbound() + } + + /// Whether the peer opened this connection. + pub(crate) fn conn_is_inbound(&self) -> bool { + self.conn.is_inbound() + } + + /// The peer's address on this link, read from the surviving carrier. + /// Written at the same three points the connection's own copy was: the + /// inbound seed, the outbound dial, and message-2 completion. + pub(crate) fn conn_source_addr(&self) -> Option<&TransportAddr> { + self.conn.source_addr() + } + + pub(crate) fn set_conn_source_addr(&mut self, addr: TransportAddr) { + self.conn.set_source_addr(addr); + } + /// 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 { @@ -3202,8 +3237,8 @@ mod tests { let identity = make_peer_identity(); let conn = outbound_leg(LinkId::new(1), identity, 1000); - assert!(conn.leg().unwrap().is_outbound()); - assert!(!conn.leg().unwrap().is_inbound()); + assert!(conn.conn_is_outbound()); + assert!(!conn.conn_is_inbound()); assert!(!conn.has_session()); assert!(conn.leg().unwrap().expected_identity().is_some()); assert_eq!(conn.leg().unwrap().started_at(), 1000); @@ -3213,8 +3248,8 @@ mod tests { fn test_inbound_connection() { let conn = inbound_leg(LinkId::new(2), 2000); - assert!(conn.leg().unwrap().is_inbound()); - assert!(!conn.leg().unwrap().is_outbound()); + assert!(conn.conn_is_inbound()); + assert!(!conn.conn_is_outbound()); assert!(!conn.has_session()); assert!(conn.leg().unwrap().expected_identity().is_none()); assert_eq!(conn.leg().unwrap().started_at(), 2000);