diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs index fcfa376..74b99e8 100644 --- a/src/node/dataplane/peer_actions.rs +++ b/src/node/dataplane/peer_actions.rs @@ -6,18 +6,19 @@ //! stands for (`build_msg2` + `transport.send`, `promote_connection`, //! `remove_active_peer`, `index_allocator.free`, `note_link_dead`, …). //! -//! ## Shadow-only skeleton +//! ## Progressive cutover //! -//! The machine home (`Node.peer_machines`), the -//! executor, and the disjoint-borrow advance helper. It is **unwired** — the live -//! `handle_msg1`/`handle_msg2` path does not drive it yet, so every method here is -//! `#[allow(dead_code)]`. The inbound cutover (`handle_msg1` → `step(InboundMsg1)`) -//! and the outbound cutover (`handle_msg2` / dial) are wired later. +//! The executor is wired incrementally. Live today: the inbound establish +//! (`handle_msg1` → `step(InboundMsg1)`), the outbound msg2 promote +//! (`handle_msg2` looks up the dial-persisted machine), and the connectionless +//! outbound msg1 send (`SendHandshake` with `their_index == None` → +//! `send_stored_msg1`, driven from `initiate_connection`). //! -//! Arms not yet exercised are inert stubs (outbound dial, rekey/crypto installs, -//! link-control frames, timers, and the connected-UDP plane are inert stubs -//! realized as those planes are wired). -//! `RegisterDecryptSession` is a deliberate no-op — see its arm for the note. +//! Not yet driven, so their arms stay inert stubs: `OpenTransport` (the +//! connection-oriented dial), rekey/crypto installs, link-control frames, the +//! timers (`SetTimer`/`CancelTimer` — the legacy tick still runs them), and the +//! connected-UDP plane. `RegisterDecryptSession` is a deliberate no-op — see its +//! arm for the note. use crate::PeerIdentity; use crate::node::Node; @@ -114,11 +115,16 @@ impl Node { // yet; inert in the shadow-only skeleton. } PeerAction::SendHandshake { bytes } => { - // The machine payload is the UNFRAMED Noise msg2 payload; - // frame it with our/their index (mirrors `handshake.rs:472`'s - // `build_msg2(our_index, their_index, &payload)`) before the - // wire send. A fresh-outbound msg1 (empty payload → build msg1 - // from indices) is framed differently and is wired later. + // Two outbound directions share this action, discriminated by + // `their_index`: + // msg2 (`their_index == Some`): the machine payload is the + // UNFRAMED Noise msg2; frame it with our/their index + // (`build_msg2`) and send. + // msg1 (`their_index == None`): a fresh outbound handshake; + // the machine's empty payload is ignored — the shell already + // allocated the index, ran the Noise leaf, and armed the + // wire at dial (`prepare_outbound_msg1`); this just sends the + // stored wire (see `send_stored_msg1`). if let (Some(sender_idx), Some(receiver_idx)) = (ambient.our_index, ambient.their_index) { @@ -152,6 +158,14 @@ impl Node { .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); return; } + } else { + // msg1: the shell already allocated the index, ran the + // Noise leaf, and armed the wire on the connection at dial + // (`prepare_outbound_msg1`); send the stored wire. The + // machine's empty payload is ignored. + let _ = bytes; + self.send_stored_msg1(link, ambient.transport_id, &ambient.remote_addr) + .await; } } PeerAction::SendRekey { .. } => { diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 345dfc9..3f78903 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -1140,10 +1140,12 @@ impl Node { // loser-link surgery is wired later). Direct analog of the inbound // net-new arm — no ordering constraint, lowest risk. // - // Look up the outbound machine persisted at DIAL (`start_handshake`), - // parked in `Discovered`, and step `Msg2 → [PromoteToActive]` in place. - // `on_msg2` is state-independent — it decides from the outbound snapshot, - // not the machine's state — and reads the machine's unset `conn.our_index` + // Look up the outbound machine persisted at DIAL, and step `Msg2 → + // [PromoteToActive]` in place. It is in `Discovered` (connection-oriented + // dial, not yet cut over) or `Handshaking{SentMsg1}` (connectionless dial, + // which drives the machine to send msg1) — either way `on_msg2` is + // state-independent: it decides from the outbound snapshot, not the + // machine's state, and reads the machine's unset `conn.our_index` // as `None`, so stepping the persisted (vs the former transient) machine // is byte-identical: same `Promote` decision (`has_existing_peer == false`), // same `our_index == None`. The machine is already in `peer_machines` diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs index 6c84e45..2f32a14 100644 --- a/src/node/lifecycle/mod.rs +++ b/src/node/lifecycle/mod.rs @@ -12,10 +12,11 @@ use super::peering::retry::MAX_RETRY_CONNECTIONS_PER_TICK; use crate::config::{ConnectPolicy, PeerAddress, PeerConfig}; use crate::node::acl::PeerAclContext; +use crate::node::dataplane::PeerActionCtx; use crate::nostr::{BootstrapEvent, NostrRendezvous}; use crate::nostr::{BootstrapHandoffResult, EstablishedTraversal}; use crate::peer::PeerConnection; -use crate::peer::machine::PeerMachine; +use crate::peer::machine::{PeerEvent, PeerMachine}; use crate::proto::fmp::wire::build_msg1; use crate::proto::fmp::{Disconnect, DisconnectReason}; use crate::transport::{Link, LinkDirection, LinkId, TransportAddr, TransportId, packet_channel}; @@ -509,22 +510,71 @@ impl Node { } Ok(()) } else { - // Connectionless: proceed with immediate handshake - self.start_handshake(link_id, transport_id, remote_addr, peer_identity) - .await + // Connectionless: no connect step. Prepare msg1 in the shell — the + // index alloc, Noise leaf, and framing can fail and the error must + // propagate to the caller (matching the pre-cutover path) — then drive + // the machine to send it. The machine goes straight to + // `start_outbound_handshake`; the executor's `SendHandshake` msg1 + // branch sends the wire `prepare_outbound_msg1` armed on the + // connection. + self.prepare_outbound_msg1(link_id, transport_id, &remote_addr, peer_identity)?; + let now = Self::now_ms(); + let ambient = PeerActionCtx { + verified_identity: peer_identity, + transport_id, + remote_addr: remote_addr.clone(), + our_index: None, + their_index: None, + now_ms: now, + is_outbound: true, + }; + self.advance_peer_machine( + link_id, + PeerEvent::Dial { + transport_id, + remote_addr, + peer_identity, + connection_oriented: false, + }, + now, + &ambient, + ) + .await; + Ok(()) } } /// Start the Noise handshake on a link and send msg1. /// - /// Called immediately for connectionless transports, or after the - /// transport connection is established for connection-oriented transports. + /// Called after a connection-oriented transport connects. (Connectionless + /// dials `prepare_outbound_msg1` + drive the machine to send in + /// `initiate_connection`.) pub(super) async fn start_handshake( &mut self, link_id: LinkId, transport_id: TransportId, remote_addr: TransportAddr, peer_identity: PeerIdentity, + ) -> Result<(), NodeError> { + self.prepare_outbound_msg1(link_id, transport_id, &remote_addr, peer_identity)?; + self.send_stored_msg1(link_id, transport_id, &remote_addr) + .await; + Ok(()) + } + + /// Prepare an outbound Noise msg1 at dial: allocate the session index, run + /// the Noise leaf, frame the wire, arm the shell-side resend, track + /// `pending_outbound`, and persist the connection + control machine (parked + /// in `Discovered`). Returns `Err` on index-allocation or Noise failure + /// (cleaning the partial leg), leaving the armed wire on the connection for + /// `send_stored_msg1` to transmit. Does NOT send — so the fallible setup can + /// propagate its error synchronously before any machine drive. + pub(in crate::node) fn prepare_outbound_msg1( + &mut self, + link_id: LinkId, + transport_id: TransportId, + remote_addr: &TransportAddr, + peer_identity: PeerIdentity, ) -> Result<(), NodeError> { let peer_node_addr = *peer_identity.node_addr(); @@ -538,7 +588,8 @@ impl Node { Err(e) => { // Clean up the link we just created self.links.remove(&link_id); - self.addr_to_link.remove(&(transport_id, remote_addr)); + self.addr_to_link + .remove(&(transport_id, remote_addr.clone())); return Err(NodeError::IndexAllocationFailed(e.to_string())); } }; @@ -552,7 +603,8 @@ impl Node { // Clean up the index and link let _ = self.index_allocator.free(our_index); self.links.remove(&link_id); - self.addr_to_link.remove(&(transport_id, remote_addr)); + self.addr_to_link + .remove(&(transport_id, remote_addr.clone())); return Err(NodeError::HandshakeFailed(e.to_string())); } }; @@ -576,7 +628,7 @@ impl Node { // Store msg1 for resend and schedule first resend let resend_interval = self.config().node.rate_limit.handshake_resend_interval_ms; - connection.set_handshake_msg1(wire_msg1.clone(), current_time_ms + resend_interval); + connection.set_handshake_msg1(wire_msg1, current_time_ms + resend_interval); // Track in pending_outbound for msg2 dispatch self.pending_outbound @@ -586,7 +638,8 @@ impl Node { // Persist the outbound control machine at dial, keyed by the same // `link_id` as the connection. It parks in `Discovered` (inert to reap // and rekey — it is absent from `peers`, never established) until - // `handle_msg2` looks it up to drive the promote. Its `our_index` is + // `handle_msg2` looks it up to drive the promote, or a connectionless + // dial drives it to `Handshaking` to send. Its `our_index` is // deliberately left unset so a later inbound restart does not emit a // spurious `UnregisterDecryptSession`. It is removed wherever the // connection is torn down without promoting (the stale reaper and the @@ -595,16 +648,42 @@ impl Node { let machine = PeerMachine::new_outbound(link_id, peer_identity, current_time_ms); self.peer_machines.insert(link_id, machine); + Ok(()) + } + + /// Send the msg1 wire that `prepare_outbound_msg1` armed on the connection. + /// On send error, marks the connection failed and RETAINS it (the legacy + /// resend tick retries); a missing wire or transport is a no-op. This is the + /// body of the executor's `SendHandshake` msg1 action and the send tail of + /// the connection-oriented `start_handshake`. + pub(in crate::node) async fn send_stored_msg1( + &mut self, + link_id: LinkId, + transport_id: TransportId, + remote_addr: &TransportAddr, + ) { + let wire_msg1 = match self + .connections + .get(&link_id) + .and_then(|c| c.handshake_msg1()) + { + Some(w) => w.to_vec(), + None => return, + }; + let our_index = self.connections.get(&link_id).and_then(|c| c.our_index()); + // Send the wire format handshake message if let Some(transport) = self.transports.get(&transport_id) { - match transport.send(&remote_addr, &wire_msg1).await { + match transport.send(remote_addr, &wire_msg1).await { Ok(bytes) => { - debug!( - link_id = %link_id, - our_index = %our_index, - bytes, - "Sent Noise handshake message 1 (wire format)" - ); + if let Some(idx) = our_index { + debug!( + link_id = %link_id, + our_index = %idx, + bytes, + "Sent Noise handshake message 1 (wire format)" + ); + } } Err(e) => { warn!( @@ -620,8 +699,6 @@ impl Node { } } } - - Ok(()) } /// Poll all transports for discovered peers and auto-connect. diff --git a/src/peer/machine.rs b/src/peer/machine.rs index e4f0ba8..02e1990 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -188,11 +188,15 @@ pub(crate) enum TimerKind { /// /// Not `Debug`/`PartialEq`: the reused core snapshot payloads derive neither. pub(crate) enum PeerEvent { - /// Reconciler dial intent. + /// Reconciler dial intent. `connection_oriented` selects the outbound + /// path: connection-oriented transports open the transport first + /// (`OpenTransport` → `Connecting`); connectionless ones send msg1 + /// immediately (`start_outbound_handshake` → `Handshaking`). Dial { transport_id: TransportId, remote_addr: TransportAddr, peer_identity: PeerIdentity, + connection_oriented: bool, }, /// Connection-oriented transport connected. TransportConnected, @@ -459,8 +463,9 @@ impl PeerMachine { PeerEvent::Dial { transport_id, remote_addr, + connection_oriented, .. - } => self.on_dial(transport_id, remote_addr, now), + } => self.on_dial(transport_id, remote_addr, connection_oriented, now), PeerEvent::TransportConnected => self.on_transport_connected(now), PeerEvent::TransportFailed => self.on_transport_failed(now), PeerEvent::InboundMsg1 { link, wire, est } => { @@ -513,21 +518,28 @@ impl PeerMachine { &mut self, transport_id: TransportId, remote_addr: TransportAddr, - _now: u64, + connection_oriented: bool, + now: u64, ) -> Vec { if !matches!(self.state, PeerState::Discovered) { return Vec::new(); } self.conn.set_transport_id(transport_id); - // Connection-oriented transports open the transport first; connectionless - // ones send msg1 immediately. This models the connection-oriented arm - // (OpenTransport) — the reconciler's candidate carries the transport - // kind at wiring time; connectionless dial reuses `start_handshake`. - self.state = PeerState::Connecting { link: self.link }; - vec![PeerAction::OpenTransport { - transport_id, - remote_addr, - }] + if connection_oriented { + // Connection-oriented transports open the transport first; the + // executor's `OpenTransport` arm connects, then feeds + // `TransportConnected` → `start_outbound_handshake`. + self.state = PeerState::Connecting { link: self.link }; + vec![PeerAction::OpenTransport { + transport_id, + remote_addr, + }] + } else { + // Connectionless transports have no connect step — send msg1 + // immediately (the executor's `SendHandshake` msg1 branch performs + // the Noise leaf, framing, index alloc, and send). + self.start_outbound_handshake(now) + } } fn on_transport_connected(&mut self, now: u64) -> Vec { @@ -2093,6 +2105,69 @@ mod tests { ); } + // ---- Test 7c: connectionless dial reaches Handshaking, Msg2 neutral ---- + // The connectionless cutover drives the outbound machine + // Discovered -> (Dial, connection_oriented=false) -> Handshaking{SentMsg1} + // BEFORE msg2, whereas the pre-cutover path stepped Msg2 while still in + // Discovered. `on_msg2` is state-independent, so both must yield the + // identical `[PromoteToActive]` and leave `our_index == None`. + #[test] + fn connectionless_dial_then_msg2_promotes_from_handshaking() { + let mut alloc = IndexAllocator::new(); + let peer = peer_identity(); + + let mut m = PeerMachine::new_outbound(LinkId::new(1), peer, 0); + // Connectionless dial: no OpenTransport, straight to Handshaking{SentMsg1}. + let dial = m.step( + PeerEvent::Dial { + transport_id: TransportId::new(1), + remote_addr: TransportAddr::from_string("127.0.0.1:9999"), + peer_identity: peer, + connection_oriented: false, + }, + 100, + &mut alloc, + ); + assert!(matches!( + m.state(), + PeerState::Handshaking { + phase: HandshakePhase::SentMsg1, + .. + } + )); + assert!( + dial.iter() + .any(|a| matches!(a, PeerAction::SendHandshake { .. })) + ); + assert!( + !dial + .iter() + .any(|a| matches!(a, PeerAction::OpenTransport { .. })), + "connectionless dial emits no OpenTransport" + ); + + // Step Msg2 from Handshaking — identical promote to the Discovered path. + let out = OutboundSnapshot { + has_existing_peer: false, + our_outbound_wins: false, + }; + let promote = m.step( + PeerEvent::Msg2 { + their_index: SessionIndex::new(0x77), + out, + }, + 200, + &mut alloc, + ); + assert_eq!( + promote, + vec![PeerAction::PromoteToActive { + link: LinkId::new(1) + }] + ); + assert_eq!(m.our_index(), None); + } + // ---- Test 8: liveness -> LinkDeadSuspected -> ReportLost -------------- #[test] fn liveness_to_link_dead() {