From 3e7ca90212e6473c2e93b108d695c8bfe09b4bc8 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Wed, 15 Jul 2026 14:40:58 +0000 Subject: [PATCH] peer: drive the connection-oriented outbound dial through the machine Route the connection-oriented (TCP/Tor) outbound path through the peer state machine, matching how the connectionless path already works. initiate_connection's oriented branch now drives PeerEvent::Dial with connection_oriented=true; the machine parks in Connecting and emits OpenTransport, whose executor arm performs the non-blocking transport.connect and pushes the PendingConnect. When the connect resolves, poll_pending_connects prepares msg1 in the shell and then drives PeerEvent::TransportConnected, which sends msg1 via the machine's SendHandshake arm. The msg1 prepare (index allocation, Noise leaf, wire arming) MUST run in the shell before the TransportConnected drive: send_stored_msg1 only transmits an already-armed wire, so a drive-only path would silently send nothing. The machine's our_index stays unset; the connect-failure path keeps its direct handshake-timeout handling (TransportFailed stays dormant). The now-unused Node::start_handshake helper is removed. Behavior-neutral: same transport.connect, same PendingConnect, same msg1 send and failure teardown as the removed inline path -- only the driver changes from inline shell code to the state machine. --- src/node/dataplane/peer_actions.rs | 59 +++++++++++--- src/node/handlers/handshake.rs | 8 +- src/node/lifecycle/mod.rs | 126 +++++++++++++++-------------- 3 files changed, 119 insertions(+), 74 deletions(-) diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs index 74b99e8..8dd2eaf 100644 --- a/src/node/dataplane/peer_actions.rs +++ b/src/node/dataplane/peer_actions.rs @@ -10,15 +10,17 @@ //! //! 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 +//! (`handle_msg2` looks up the dial-persisted machine), the connectionless //! outbound msg1 send (`SendHandshake` with `their_index == None` → -//! `send_stored_msg1`, driven from `initiate_connection`). +//! `send_stored_msg1`, driven from `initiate_connection`), and the +//! connection-oriented dial (`OpenTransport` performs the non-blocking +//! `transport.connect`; `TransportConnected` drives the connect-resolution msg1 +//! send from `poll_pending_connects`). //! -//! 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. +//! Not yet driven, so their arms stay inert stubs: 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; @@ -109,10 +111,45 @@ impl Node { let mut queue: VecDeque = actions.into(); while let Some(action) = queue.pop_front() { match action { - PeerAction::OpenTransport { .. } => { - // Outbound dial (`initiate_connection`, - // `lifecycle/mod.rs:470`). Outbound establish is not cut over - // yet; inert in the shadow-only skeleton. + PeerAction::OpenTransport { + transport_id, + remote_addr, + } => { + // Outbound connection-oriented dial. `initiate_connection`'s + // oriented branch drove the machine to `Connecting`, which + // emitted this action. Perform the non-blocking + // `transport.connect` and, on success, push the + // `PendingConnect` for `poll_pending_connects` to resolve. On + // connect error, tear down the dial-window state (link, + // reverse map, control machine) and abort the queue — the + // executor-local mirror of the old inline + // `initiate_connection` connect+push. + if let Some(transport) = self.transports.get(&transport_id) { + match transport.connect(&remote_addr).await { + Ok(()) => { + debug!( + transport_id = %transport_id, + remote_addr = %remote_addr, + link_id = %link, + "Transport connect initiated (non-blocking)" + ); + self.peering + .pending_connects + .push(crate::node::PendingConnect { + link_id: link, + transport_id, + remote_addr, + peer_identity: ambient.verified_identity, + }); + } + Err(_e) => { + self.links.remove(&link); + self.addr_to_link.remove(&(transport_id, remote_addr)); + self.peer_machines.remove(&link); + return; + } + } + } } PeerAction::SendHandshake { bytes } => { // Two outbound directions share this action, discriminated by diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 3f78903..87563ab 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -1141,10 +1141,10 @@ impl Node { // net-new arm — no ordering constraint, lowest risk. // // 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 + // [PromoteToActive]` in place. Both dial paths reach msg2 in + // `Handshaking{SentMsg1}` — each drives the machine to send msg1 before + // msg2 — and `on_msg2` is state-independent regardless: 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`), diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs index 927f465..ec1c683 100644 --- a/src/node/lifecycle/mod.rs +++ b/src/node/lifecycle/mod.rs @@ -439,8 +439,6 @@ impl Node { remote_addr: TransportAddr, peer_identity: PeerIdentity, ) -> Result<(), NodeError> { - let peer_node_addr = *peer_identity.node_addr(); - self.authorize_peer( &peer_identity, PeerAclContext::OutboundConnect, @@ -494,33 +492,35 @@ impl Node { self.peer_machines.insert(link_id, machine); if is_connection_oriented { - // Connection-oriented: start non-blocking connect, defer handshake - if let Some(transport) = self.transports.get(&transport_id) { - match transport.connect(&remote_addr).await { - Ok(()) => { - debug!( - peer = %self.peer_display_name(&peer_node_addr), - transport_id = %transport_id, - remote_addr = %remote_addr, - link_id = %link_id, - "Transport connect initiated (non-blocking)" - ); - self.peering.pending_connects.push(super::PendingConnect { - link_id, - transport_id, - remote_addr, - peer_identity, - }); - } - Err(e) => { - // Clean up link and the dial-time control machine - self.links.remove(&link_id); - self.addr_to_link.remove(&(transport_id, remote_addr)); - self.peer_machines.remove(&link_id); - return Err(NodeError::TransportError(e.to_string())); - } - } - } + // Connection-oriented: drive the machine to open the transport. The + // machine parks in `Connecting` and emits one `OpenTransport`; the + // executor performs the non-blocking `transport.connect` and pushes + // the `PendingConnect`. No index alloc or wire-arm happens here — + // that is deferred to `prepare_outbound_msg1` at connect-resolution + // time (`poll_pending_connects`), so `our_index` stays `None` on both + // the (not-yet-built) connection and the machine. + 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: true, + }, + now, + &ambient, + ) + .await; Ok(()) } else { // Connectionless: no connect step. Prepare msg1 in the shell — the @@ -557,24 +557,6 @@ impl Node { } } - /// Start the Noise handshake on a link and send msg1. - /// - /// 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. Returns `Err` on @@ -659,8 +641,9 @@ impl Node { /// 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`. + /// body of the executor's `SendHandshake` msg1 action — reached on both the + /// connectionless dial and the connection-oriented connect-resolution path, + /// after `prepare_outbound_msg1` has armed the wire. pub(in crate::node) async fn send_stored_msg1( &mut self, link_id: LinkId, @@ -1223,16 +1206,18 @@ impl Node { "Transport connected, starting handshake" ); - // Start the handshake now that the transport is connected - if let Err(e) = self - .start_handshake( - pending.link_id, - pending.transport_id, - pending.remote_addr.clone(), - pending.peer_identity, - ) - .await - { + // Prepare msg1 now that the transport is connected, then drive + // the machine to send it. The prepare (index alloc, Noise leaf, + // wire arm) MUST run BEFORE the `TransportConnected` drive: the + // executor's `SendHandshake` msg1 branch only transmits the wire + // this armed on the connection — a drive-only path would find no + // armed wire and silently send nothing. + if let Err(e) = self.prepare_outbound_msg1( + pending.link_id, + pending.transport_id, + &pending.remote_addr, + pending.peer_identity, + ) { warn!( link_id = %pending.link_id, error = %e, @@ -1241,6 +1226,29 @@ impl Node { // Clean up link and dial-time machine on handshake failure self.remove_link(&pending.link_id); self.peer_machines.remove(&pending.link_id); + } else { + // Drive the dial-persisted machine: `Connecting` → + // `on_transport_connected` → `start_outbound_handshake`, + // emitting `SendHandshake` (msg1) whose executor arm sends the + // wire just armed. `our_index`/`their_index` stay `None` so the + // executor takes the `their_index == None` msg1 branch. + let now = Self::now_ms(); + let ambient = PeerActionCtx { + verified_identity: pending.peer_identity, + transport_id: pending.transport_id, + remote_addr: pending.remote_addr.clone(), + our_index: None, + their_index: None, + now_ms: now, + is_outbound: true, + }; + self.advance_peer_machine( + pending.link_id, + PeerEvent::TransportConnected, + now, + &ambient, + ) + .await; } } else { let reason = reason.unwrap_or_default();