diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs index ec86f8c..fab03e3 100644 --- a/src/node/dataplane/peer_actions.rs +++ b/src/node/dataplane/peer_actions.rs @@ -6,52 +6,33 @@ //! 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** — no live -//! handler path drives the machine and nothing inserts into `peer_machines`, so -//! every method here is `#[allow(dead_code)]` and behaviorally inert. +//! The executor is wired incrementally. Live today: the outbound msg2 promote +//! (`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`), and the +//! connection-oriented dial (`OpenTransport` performs the non-blocking +//! `transport.connect`; `TransportConnected` drives the connect-resolution msg1 +//! send from `poll_pending_connects`). //! -//! The arm bodies that a later reap/rekey/timer commit will drive for real are -//! ported 1:1 from the IK-lineage executor and adapted to next's Node API — they -//! reproduce next's inline shell bodies exactly: +//! Inbound establish is not machine-driven here: `handle_msg1` builds and sends +//! msg2 inline, so `PeerEvent::InboundMsg1` is never dispatched and the +//! `SendHandshake` `their_index == Some` (msg2) branch stays dormant. //! -//! - `SwapSendState` / `CompleteDrain` mirror `handlers/rekey.rs`'s `check_rekey` -//! `Cutover` / `Drain` bodies (`peer.cutover_to_new_session()` + -//! `register_decrypt_worker_session` gated on `did_cutover`; -//! `peer.complete_drain()` → `peers_by_index.remove` + -//! `unregister_decrypt_worker_session` + `index_allocator.free`). -//! - `InvalidateSendState` → `remove_active_peer` (the full teardown). -//! - `ReportLost` → `note_link_dead` (the reconciler loss reflex). -//! - `UnregisterDecryptSession` / `FreeIndex` index-plane cleanups. -//! - `RegisterDecryptSession` is a deliberate no-op — the decrypt-worker register -//! for the rekey cutover relocates into the driven `SwapSendState` site here; -//! for the establish promote it stays INSIDE `promote_connection` (unlike the -//! IK lineage), so `PromoteToActive` does NOT re-register. -//! -//! Actions not yet exercised are inert stubs: the establish send-path -//! (`SendHandshake` framing) lands when the inbound -//! establish send is wired; `SendRekey`/`SendLinkMessage`, the timers -//! (`SetTimer`/`CancelTimer`), the outbound dial (`OpenTransport`), and the -//! connected-UDP plane are inert here. -//! -//! `PromoteToActive` is LIVE: it drives the inbound establish promote -//! (`promote_connection` + the msg2/tree-announce/bloom follow-ups), transcribing -//! `handle_msg3`'s shared promote block. Two XX-semantic arms remain explicit -//! **deferred stubs** whose real bodies land when the inbound cross-connection / -//! rekey-responder paths are wired: `SwapToInboundSession` and -//! `RekeyRespondTrigger`. Each carries a `debug_assert!(false, …)` guard + a -//! benign no-op — they are unreachable now (those decisions stay inline in -//! `handle_msg3`). +//! 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::reject::{HandshakeReject, RejectReason}; use crate::node::{Node, NodeError}; -use crate::peer::machine::{PeerAction, PeerEvent}; +use crate::peer::machine::{LostKind, PeerAction, PeerEvent}; use crate::proto::fmp::PromotionResult; +use crate::proto::fmp::wire::build_msg2; use crate::transport::{LinkId, TransportAddr, TransportId}; use crate::utils::index::SessionIndex; -use crate::{NodeAddr, PeerIdentity}; use std::collections::VecDeque; use tracing::{debug, info, trace, warn}; @@ -134,16 +115,99 @@ impl Node { let mut queue: VecDeque = actions.into(); while let Some(action) = queue.pop_front() { match action { - PeerAction::OpenTransport { .. } => { - // Outbound dial (`initiate_connection`). 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: Some(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 { .. } => { - // Inbound establish send-path: frame the unframed Noise msg2 - // payload with our/their index (`build_msg2(our_index, - // their_index, &payload)`) and send, with the msg2-send-failure - // cleanup + queue abort. Lands with `PromoteToActive` when the - // inbound establish path is wired; inert in the shadow-only skeleton. + PeerAction::SendHandshake { bytes } => { + // 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) + { + let frame = build_msg2(sender_idx, receiver_idx, &bytes); + // Surface the send Result. A missing transport skips + // the send and continues (mirrors `handle_msg1`'s + // `if let Some(transport)` guard); a send *error* runs the + // pre-refactor msg2-send-failure cleanup (`handle_msg1` + // L494-503) and ABORTS the remaining queue so the queued + // `PromoteToActive` never runs. + let send_err = match self.transports.get(&ambient.transport_id) { + Some(transport) => { + transport.send(&ambient.remote_addr, &frame).await.err() + } + None => None, + }; + if let Some(e) = send_err { + // Restored pre-refactor msg2-send-failure warn! + // (`handle_msg1` L665): the send error text is surfaced + // at the executor point where the failure is now handled. + warn!(link_id = %link, error = %e, "Failed to send msg2"); + self.connections.remove(&link); + self.links.remove(&link); + self.addr_to_link + .remove(&(ambient.transport_id, ambient.remote_addr.clone())); + if let Some(idx) = ambient.our_index { + let _ = self.index_allocator.free(idx); + } + self.peer_machines.remove(&link); + self.stats_mut() + .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 { .. } => { // Rekey msg framing (`build_msg2(our_new_index, …)`) + send. @@ -513,19 +577,22 @@ impl Node { // the legacy tick timers still run, so driving these would // double-schedule. } - PeerAction::ReportLost { peer } => { - // The single loss token → the reconciler reflex. - self.report_peer_lost(peer, ambient.now_ms); + PeerAction::ReportLost { peer, kind } => { + // The single loss token, routed to the reconciler reflex the + // `kind` names: an un-promoted handshake attempt takes the + // connected-guarded `note_handshake_timeout` (`driver.rs:28`), + // an established peer's link-death takes the unconditional + // `note_link_dead` (`driver.rs:48`). + match kind { + LostKind::HandshakeTimeout => { + self.note_handshake_timeout(peer, ambient.now_ms); + } + LostKind::LinkDead => { + self.note_link_dead(peer, ambient.now_ms); + } + } } } } } - - /// `ReportLost` → `note_link_dead` (kept as a named seam so the ambient clock - /// source is explicit and the reconciler-computed backoff can be threaded - /// later). - #[allow(dead_code)] - fn report_peer_lost(&mut self, peer: NodeAddr, now_ms: u64) { - self.note_link_dead(peer, now_ms); - } } diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 4f34609..78fea02 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -597,6 +597,8 @@ impl Node { { self.pending_outbound.remove(&key); self.connections.remove(&link_id); + // Drop the machine persisted at dial — this leg never promotes. + self.peer_machines.remove(&link_id); self.remove_link(&link_id); self.stats_mut() .record_reject(RejectReason::Handshake(HandshakeReject::BadState)); @@ -660,6 +662,12 @@ impl Node { // This ensures both nodes use the same Noise handshake (the winner's // outbound = the loser's inbound). if self.peers.contains_key(&peer_node_addr) { + // The dial-persisted outbound machine is not consumed by the inline + // Swap/Keep resolution below (which mutates the existing promoted peer + // directly, with no machine). Drop it on entry so none of this block's + // exits leave a dangling machine — matching the pre-persistence path, + // which created no machine for a cross-connection. + self.peer_machines.remove(&link_id); let our_outbound_wins = cross_connection_winner( self.identity().node_addr(), &peer_node_addr, @@ -787,20 +795,42 @@ impl Node { // an existing peer), so this is the net-new path: promote_connection hits // its normal-promotion branch and returns Promoted. // - // The transient machine is NOT inserted into peer_machines — the - // persistent Established machine is created inside promote_connection. - // With no outbound decision core to run, the machine emits PromoteToActive - // unconditionally. The promote tail (info log, tree/bloom/backoff, and the - // pending_outbound removal) lives in the executor's PromoteToActive arm. - let mut machine = PeerMachine::new_outbound(link_id, peer_identity, packet.timestamp_ms); - let actions = machine.step( - PeerEvent::OutboundMsg2 { - their_index: header.sender_idx, - }, - packet.timestamp_ms, - &mut self.index_allocator, + // Net-new outbound promote via the per-peer machine. Look up the machine + // persisted at DIAL for an identified leg and step `OutboundMsg2 → + // [PromoteToActive]` in place; an anonymous-discovery leg never persisted + // a machine at dial and falls to the transient in the `None` arm. The + // transient is NOT inserted into `peer_machines` — the persistent + // Established machine is created inside `promote_connection` (next's + // invariant: `promote_connection` owns the Established machine). With no + // outbound decision core to run, the machine emits `PromoteToActive` + // unconditionally (the net-new-vs-cross-connection decision was the + // `peers.contains_key` test above, which returns on an existing peer). The + // promote tail (info log, tree/bloom/backoff, `pending_outbound` removal) + // lives in the executor's `PromoteToActive` arm. + let promote_actions = match self.peer_machines.get_mut(&link_id) { + Some(machine) => machine.step( + PeerEvent::OutboundMsg2 { + their_index: header.sender_idx, + }, + packet.timestamp_ms, + &mut self.index_allocator, + ), + None => { + let mut machine = + PeerMachine::new_outbound(link_id, peer_identity, packet.timestamp_ms); + machine.step( + PeerEvent::OutboundMsg2 { + their_index: header.sender_idx, + }, + packet.timestamp_ms, + &mut self.index_allocator, + ) + } + }; + debug_assert_eq!( + promote_actions, + vec![PeerAction::PromoteToActive { link: link_id }] ); - debug_assert_eq!(actions, vec![PeerAction::PromoteToActive { link: link_id }]); let ambient = PeerActionCtx { verified_identity: peer_identity, @@ -812,7 +842,8 @@ impl Node { is_outbound: true, pending_outbound_key: Some(key), }; - self.execute_peer_actions(link_id, &ambient, actions).await; + self.execute_peer_actions(link_id, &ambient, promote_actions) + .await; } /// Handle handshake message 3 (phase 0x3). diff --git a/src/node/handlers/timeout.rs b/src/node/handlers/timeout.rs index 7e8ec1c..2aa3ff7 100644 --- a/src/node/handlers/timeout.rs +++ b/src/node/handlers/timeout.rs @@ -123,6 +123,11 @@ impl Node { Some(c) => c, None => return, }; + // A dial-persisted outbound control machine shares the connection's + // `link_id` and lifetime; drop it here so a reaped handshake leg leaves + // no dangling machine. A no-op for promoted peers — `promote_connection` + // already removed their connection, so this reaper never runs for them. + self.peer_machines.remove(&link_id); let transport_id = conn.transport_id(); // Free session index and pending_outbound/pending_inbound if allocated diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs index 0ae52e2..e2fdd7b 100644 --- a/src/node/lifecycle/mod.rs +++ b/src/node/lifecycle/mod.rs @@ -12,9 +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::{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}; @@ -482,54 +484,185 @@ impl Node { self.addr_to_link .insert((transport_id, remote_addr.clone()), link_id); - 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(()) => { - if let Some(ref id) = peer_identity { - debug!( - peer = %self.peer_display_name(id.node_addr()), - transport_id = %transport_id, - remote_addr = %remote_addr, - link_id = %link_id, - "Transport connect initiated (non-blocking)" - ); - } else { - debug!( - transport_id = %transport_id, - remote_addr = %remote_addr, - link_id = %link_id, - "Transport connect initiated (anonymous discovery)" - ); + match peer_identity { + Some(identity) => { + // Identified dial: persist the outbound control machine at dial, + // keyed by the same `link_id` as the (soon-to-be-built) + // connection. It parks in `Discovered` (inert to reap and rekey — + // absent from `peers`, never established) until `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 on every + // failure path in the dial window (`prepare_outbound_msg1` / + // `poll_pending_connects`), mirroring the connection's lifetime. + let machine = PeerMachine::new_outbound(link_id, identity, Self::now_ms()); + self.peer_machines.insert(link_id, machine); + + if !is_connection_oriented { + // 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 — then drive the machine + // to send it (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, identity)?; + } + + // Drive the machine: connection-oriented dials emit `OpenTransport` + // (the executor connects and pushes `PendingConnect`); connectionless + // dials emit `SendHandshake` (msg1) for the wire armed above. + let now = Self::now_ms(); + let ambient = PeerActionCtx { + verified_identity: identity, + transport_id, + remote_addr: remote_addr.clone(), + our_index: None, + their_index: None, + now_ms: now, + is_outbound: true, + pending_outbound_key: None, + }; + self.advance_peer_machine( + link_id, + PeerEvent::Dial { + transport_id, + remote_addr, + peer_identity: identity, + connection_oriented: is_connection_oriented, + }, + now, + &ambient, + ) + .await; + Ok(()) + } + None => { + // Anonymous-discovery dial: next's inline path, no control machine + // (the transient is born at `handle_msg2` when the peer's identity + // crystallizes from `conn.expected_identity()`). + 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!( + transport_id = %transport_id, + remote_addr = %remote_addr, + link_id = %link_id, + "Transport connect initiated (anonymous discovery)" + ); + self.peering.pending_connects.push(super::PendingConnect { + link_id, + transport_id, + remote_addr, + peer_identity: None, + }); + } + Err(e) => { + // Clean up link + self.links.remove(&link_id); + self.addr_to_link.remove(&(transport_id, remote_addr)); + return Err(NodeError::TransportError(e.to_string())); + } } - self.peering.pending_connects.push(super::PendingConnect { - link_id, - transport_id, - remote_addr, - peer_identity, - }); - } - Err(e) => { - // Clean up link - self.links.remove(&link_id); - self.addr_to_link.remove(&(transport_id, remote_addr)); - return Err(NodeError::TransportError(e.to_string())); } + Ok(()) + } else { + // Connectionless: proceed with immediate handshake. + self.start_handshake(link_id, transport_id, remote_addr, None) + .await } } - Ok(()) - } else { - // Connectionless: proceed with immediate handshake - self.start_handshake(link_id, transport_id, remote_addr, peer_identity) - .await } } - /// 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. + /// Prepare an outbound Noise msg1 at dial for an IDENTIFIED leg: 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 index-allocation or Noise failure (cleaning the partial leg, + /// including the dial-time control machine), 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. Anonymous-discovery legs use the monolithic `start_handshake` + /// instead; only identified legs persist a machine, so this is the only + /// caller path that cleans up `peer_machines`. + 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(); + + // Create connection in handshake phase (outbound knows expected identity) + let current_time_ms = Self::now_ms(); + let mut connection = PeerConnection::outbound(link_id, peer_identity, current_time_ms); + + // Allocate a session index for this handshake + let our_index = match self.index_allocator.allocate() { + Ok(idx) => idx, + Err(e) => { + // Clean up the link and dial-time machine we just created + self.links.remove(&link_id); + self.addr_to_link + .remove(&(transport_id, remote_addr.clone())); + self.peer_machines.remove(&link_id); + return Err(NodeError::IndexAllocationFailed(e.to_string())); + } + }; + + // Start the Noise handshake and get message 1 + let our_keypair = self.identity().keypair(); + let noise_msg1 = + match connection.start_handshake(our_keypair, self.startup_epoch(), current_time_ms) { + Ok(msg) => msg, + Err(e) => { + // Clean up the index, link, and dial-time machine + let _ = self.index_allocator.free(our_index); + self.links.remove(&link_id); + self.addr_to_link + .remove(&(transport_id, remote_addr.clone())); + self.peer_machines.remove(&link_id); + return Err(NodeError::HandshakeFailed(e.to_string())); + } + }; + + // 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] + let wire_msg1 = build_msg1(our_index, &noise_msg1); + + debug!( + peer = %self.peer_display_name(&peer_node_addr), + transport_id = %transport_id, + remote_addr = %remote_addr, + link_id = %link_id, + our_index = %our_index, + "Connection initiated" + ); + + // 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, current_time_ms + resend_interval); + + // Track in pending_outbound for msg2 dispatch + self.pending_outbound + .insert((transport_id, our_index.as_u32()), link_id); + self.connections.insert(link_id, connection); + + Ok(()) + } + + /// Start an outbound Noise handshake inline (allocate index, run the Noise + /// leaf, arm the resend, track `pending_outbound`, and send msg1). Used by + /// the anonymous-discovery dial paths (connectionless dial and the + /// connection-oriented connect-resolution), which drive no control machine. + /// Anonymous discovery (no `peer_identity`) leaves identity to be learned + /// from the XX msg2. pub(super) async fn start_handshake( &mut self, link_id: LinkId, @@ -636,6 +769,57 @@ impl Node { 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 — 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, + 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 { + Ok(bytes) => { + 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!( + link_id = %link_id, + error = %e, + "Failed to send handshake message" + ); + // Mark connection as failed but don't remove it yet + // The event loop can handle retry logic + if let Some(conn) = self.connections.get_mut(&link_id) { + conn.mark_failed(); + } + } + } + } + } + /// Poll all transports for discovered peers and auto-connect. /// /// Called from the tick handler. Iterates operational transports, @@ -1187,23 +1371,77 @@ 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 - { - warn!( - link_id = %pending.link_id, - error = %e, - "Failed to start handshake after transport connect" - ); - // Clean up link on handshake failure - self.remove_link(&pending.link_id); + match pending.peer_identity { + Some(identity) => { + // Identified leg: prepare msg1 now that the transport is + // connected, then drive the dial-persisted 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, + identity, + ) { + warn!( + link_id = %pending.link_id, + error = %e, + "Failed to start handshake after transport connect" + ); + // 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: identity, + transport_id: pending.transport_id, + remote_addr: pending.remote_addr.clone(), + our_index: None, + their_index: None, + now_ms: now, + is_outbound: true, + pending_outbound_key: None, + }; + self.advance_peer_machine( + pending.link_id, + PeerEvent::TransportConnected, + now, + &ambient, + ) + .await; + } + } + None => { + // Anonymous-discovery leg: next's inline handshake, no + // control machine. + if let Err(e) = self + .start_handshake( + pending.link_id, + pending.transport_id, + pending.remote_addr.clone(), + None, + ) + .await + { + warn!( + link_id = %pending.link_id, + error = %e, + "Failed to start handshake after transport connect" + ); + // Clean up link on handshake failure + self.remove_link(&pending.link_id); + } + } } } else { let reason = reason.unwrap_or_default(); @@ -1215,12 +1453,14 @@ impl Node { "Transport connect failed" ); - // Clean up link and schedule retry. Anonymous discovery - // connections (no expected identity) don't retry — - // they'll be rediscovered via the shared-medium beacon. + // Clean up link and, for identified legs, the dial-time machine, + // then schedule retry. Anonymous discovery connections (no expected + // identity) don't retry — they'll be rediscovered via the + // shared-medium beacon. self.remove_link(&pending.link_id); self.links.remove(&pending.link_id); if let Some(id) = &pending.peer_identity { + self.peer_machines.remove(&pending.link_id); self.note_handshake_timeout(*id.node_addr(), Self::now_ms()); } } diff --git a/src/node/tests/tcp.rs b/src/node/tests/tcp.rs index f120afa..0255b59 100644 --- a/src/node/tests/tcp.rs +++ b/src/node/tests/tcp.rs @@ -286,3 +286,113 @@ async fn test_tcp_reconnection_after_link_death() { cleanup_nodes(&mut nodes).await; } + +/// Connection-oriented outbound connect succeeds on loopback and reaches the +/// handshake. +/// +/// `initiate_connection` opens a real non-blocking TCP connect to a live +/// listener (node 1's `TcpTransport`) and queues a `PendingConnect`; +/// `poll_pending_connects` promotes the resolved connection to `Connected` and +/// runs `start_handshake`. The observable is `pending_outbound`: `start_handshake` +/// → `prepare_outbound_msg1` tracks the dispatched msg1 there (keyed by the +/// allocated session index), which is empty until the connect resolves. +#[tokio::test] +async fn test_tcp_oriented_connect_success_reaches_handshake() { + let mut nodes = vec![make_test_node_tcp().await, make_test_node_tcp().await]; + + // Target: node 1's live TCP listener + its verified identity. + let target_addr = nodes[1].addr.clone(); + let target_identity = PeerIdentity::from_pubkey_full(nodes[1].node.identity().pubkey_full()); + let transport_id = nodes[0].transport_id; + + // Kick off the non-blocking oriented connect. + nodes[0] + .node + .initiate_connection(transport_id, target_addr, Some(target_identity)) + .await + .expect("initiate_connection should queue a pending connect"); + + // One pending connect queued, no msg1 dispatched yet. + assert_eq!(nodes[0].node.peering.pending_connects.len(), 1); + assert!(nodes[0].node.pending_outbound.is_empty()); + + // Drive the tick-side poll until the background connect resolves. + for _ in 0..100 { + nodes[0].node.poll_pending_connects().await; + if nodes[0].node.peering.pending_connects.is_empty() { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + + assert!( + nodes[0].node.peering.pending_connects.is_empty(), + "connect should have resolved" + ); + // Reaching start_handshake -> prepare_outbound_msg1 tracks the outbound msg1 + // in pending_outbound: proof the connect-success path dispatched the handshake. + assert_eq!( + nodes[0].node.pending_outbound.len(), + 1, + "connect-success path should have reached start_handshake and dispatched msg1" + ); + + cleanup_nodes(&mut nodes).await; +} + +/// Connection-oriented outbound connect to a closed port fails and tears down +/// the link. +/// +/// The dial targets a guaranteed-closed loopback port (bind then drop the +/// listener). `initiate_connection` is non-blocking, so it still queues a +/// `PendingConnect` and a `Connecting` link; `poll_pending_connects` observes the +/// background task's `Failed` result and routes through the failure arm +/// (`remove_link` + `note_handshake_timeout`). The observable is the removed link +/// and the absent `pending_outbound` entry — the failure path never reaches +/// `start_handshake`. +#[tokio::test] +async fn test_tcp_oriented_connect_failure_tears_down_link() { + let mut nodes = vec![make_test_node_tcp().await]; + + // A guaranteed-closed loopback port: bind to grab a port, then drop it. + let closed = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let closed_addr = closed.local_addr().unwrap(); + drop(closed); + let target_addr = TransportAddr::from_string(&closed_addr.to_string()); + let target_identity = make_peer_identity(); + let transport_id = nodes[0].transport_id; + + nodes[0] + .node + .initiate_connection(transport_id, target_addr, Some(target_identity)) + .await + .expect("initiate_connection is non-blocking; it queues a pending connect"); + + // One pending connect + the in-flight dial's link. + assert_eq!(nodes[0].node.peering.pending_connects.len(), 1); + assert_eq!(nodes[0].node.links.len(), 1); + + for _ in 0..100 { + nodes[0].node.poll_pending_connects().await; + if nodes[0].node.peering.pending_connects.is_empty() { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + + assert!( + nodes[0].node.peering.pending_connects.is_empty(), + "failed connect should have been drained" + ); + // Failure path removes the link and never reaches start_handshake. + assert!( + nodes[0].node.links.is_empty(), + "connect-failure path should have torn down the link" + ); + assert!( + nodes[0].node.pending_outbound.is_empty(), + "connect-failure path must not dispatch msg1" + ); + + cleanup_nodes(&mut nodes).await; +} diff --git a/src/peer/machine.rs b/src/peer/machine.rs index 8e177bf..6fa73d8 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -213,11 +213,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, @@ -280,6 +284,21 @@ pub(crate) enum PeerEvent { Tick, } +/// Why a peer was reported lost. Selects the reconciler reflex the executor +/// routes the `ReportLost` token to: an un-promoted handshake attempt that +/// failed (`HandshakeTimeout`, connected-guarded like the old `schedule_retry`) +/// versus an established peer whose link died (`LinkDead`, unconditional like +/// the old `schedule_reconnect`). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum LostKind { + /// An outbound handshake attempt timed out or its dial failed before the + /// peer promoted — routes to the connected-guarded reflex. + HandshakeTimeout, + /// An established peer's link went dead or is being replaced — routes to + /// the unconditional reconnect reflex. + LinkDead, +} + /// An effect the driver executes on the machine's behalf. Runtime-agnostic /// plain data — no tokio handles, time only as `at_ms` fields. #[derive(Clone, Debug, PartialEq, Eq)] @@ -351,8 +370,9 @@ pub(crate) enum PeerAction { /// Cancel a scheduled timer. CancelTimer { kind: TimerKind }, /// Report the peer lost to the reconciler (the single loss token — there is - /// deliberately no `ScheduleRetry` machine action). - ReportLost { peer: NodeAddr }, + /// deliberately no `ScheduleRetry` machine action). `kind` selects the + /// reflex (handshake-timeout vs link-dead) the executor routes to. + ReportLost { peer: NodeAddr, kind: LostKind }, } // ============================================================================ @@ -540,8 +560,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 } => self.on_inbound_msg1(link, now, index_allocator), @@ -587,19 +608,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. - 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 { @@ -615,7 +645,13 @@ impl PeerMachine { } let mut actions = Vec::new(); if let Some(peer) = self.addr() { - actions.push(PeerAction::ReportLost { peer }); + // Dial failure on an un-promoted leg routes like a handshake timeout + // (the connected-guarded reflex). Dormant today — no `TransportFailed` + // event is dispatched until the connection-oriented cutover (C5). + actions.push(PeerAction::ReportLost { + peer, + kind: LostKind::HandshakeTimeout, + }); } self.state = PeerState::Closed { backoff_deadline_ms: now + CLOSED_BACKOFF_MS, @@ -740,7 +776,10 @@ impl PeerMachine { // connection in the stale peer's place. vec![ PeerAction::InvalidateSendState, - PeerAction::ReportLost { peer }, + PeerAction::ReportLost { + peer, + kind: LostKind::LinkDead, + }, PeerAction::PromoteToActive { link: self.link }, ] } @@ -1071,7 +1110,12 @@ impl PeerMachine { PeerAction::TeardownConnectedUdp, ]; if let Some(peer) = self.addr() { - actions.push(PeerAction::ReportLost { peer }); + // An established peer whose link died — the unconditional reconnect + // reflex (the live liveness-reap producer). + actions.push(PeerAction::ReportLost { + peer, + kind: LostKind::LinkDead, + }); } self.state = PeerState::Closed { backoff_deadline_ms: now + CLOSED_BACKOFF_MS, @@ -1145,7 +1189,13 @@ impl PeerMachine { for act in Fmp::new().poll_timeouts(vec![snap]) { match act { ConnAction::ScheduleRetry { peer } => { - lost.push(PeerAction::ReportLost { peer }); + // Handshake timeout on an un-promoted leg — the connected- + // guarded reflex. Dormant today (no `Timeout` event is + // dispatched until the timeout fold in C5). + lost.push(PeerAction::ReportLost { + peer, + kind: LostKind::HandshakeTimeout, + }); } ConnAction::Teardown { .. } => { if let Some(idx) = self.conn.our_index() { @@ -1435,7 +1485,10 @@ mod tests { PeerAction::CancelTimer { kind: TimerKind::Liveness, }, - PeerAction::ReportLost { peer }, + PeerAction::ReportLost { + peer, + kind: LostKind::LinkDead, + }, PeerAction::SwapToInboundSession { peer, our_index: SessionIndex::new(8), @@ -1708,7 +1761,10 @@ mod tests { actions, vec![ PeerAction::InvalidateSendState, - PeerAction::ReportLost { peer: peer_addr }, + PeerAction::ReportLost { + peer: peer_addr, + kind: LostKind::LinkDead, + }, PeerAction::PromoteToActive { link: LinkId::new(1) }, @@ -1957,6 +2013,207 @@ mod tests { ); } + // ---- Test 7b: dial-persisted outbound promote leaves our_index unset --- + // An outbound machine persisted at DIAL (`new_outbound`, `Discovered`, with + // `conn.our_index` UNSET — the shell owns the index on its own + // `PeerConnection`, never on the machine) must, on promote via msg2, end with + // `our_index == None`, exactly as the pre-persistence transient did. The + // guard: a subsequent inbound restart then emits NO + // `UnregisterDecryptSession` (contrast `restart_override`, whose machine has + // `our_index == Some`). A leaked `Some(dial_index)` here would wrongly + // unregister — on index reuse, ANOTHER peer's — worker session; keeping the + // field `None` is the Model-B dial-persistence neutrality property. + #[test] + fn dial_persisted_outbound_promote_no_restart_unregister() { + let mut alloc = IndexAllocator::new(); + let peer = peer_identity(); + let peer_addr = *peer.node_addr(); + let our = *peer_identity().node_addr(); + + // Persisted at dial: Discovered, conn.our_index deliberately NOT set. + let mut m = PeerMachine::new_outbound(LinkId::new(1), peer, 0); + assert_eq!(m.our_index(), None); + + // Promote via msg2 from Discovered (the production path — the former + // transient was likewise stepped from `new_outbound` without a state set). + let promote = m.step( + PeerEvent::OutboundMsg2 { + their_index: SessionIndex::new(0x77), + }, + 300, + &mut alloc, + ); + assert_eq!( + promote, + vec![PeerAction::PromoteToActive { + link: LinkId::new(1) + }] + ); + assert_eq!( + m.our_index(), + None, + "outbound promote must leave our_index unset" + ); + + // Drive promotion to Established (from Discovered, as in production). + let _ = m.step( + PeerEvent::PromotionResolved { + result: PromotionResult::Promoted(peer_addr), + }, + 300, + &mut alloc, + ); + assert_eq!(m.state(), PeerState::Established { addr: peer_addr }); + assert_eq!(m.our_index(), None); + + // A subsequent inbound restart (peer restart, new epoch) must NOT emit + // UnregisterDecryptSession, because our_index is None. + let mut est = est_new_peer(our); + est.has_existing_peer = true; + est.existing_peer_epoch = Some([1u8; 8]); + let wire = wire_outcome(peer_addr, Some([2u8; 8])); + let restart = m.step(PeerEvent::InboundMsg3 { wire, est }, 1_000, &mut alloc); + assert!( + !restart + .iter() + .any(|a| matches!(a, PeerAction::UnregisterDecryptSession { .. })), + "no UnregisterDecryptSession when the promoted outbound machine's our_index is None" + ); + assert!( + restart.iter().any(|a| matches!( + a, + PeerAction::ReportLost { + kind: LostKind::LinkDead, + .. + } + )), + "restart still reports the loss via the link-dead reconnect reflex" + ); + } + + // ---- 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 promote = m.step( + PeerEvent::OutboundMsg2 { + their_index: SessionIndex::new(0x77), + }, + 200, + &mut alloc, + ); + assert_eq!( + promote, + vec![PeerAction::PromoteToActive { + link: LinkId::new(1) + }] + ); + assert_eq!(m.our_index(), None); + } + + // ---- Test 7d: connection-oriented dial opens transport first ---------- + // The connection-oriented cutover drives the outbound machine + // Discovered -> (Dial, connection_oriented=true) -> Connecting + // (emitting ONLY OpenTransport, no msg1 yet), then TransportConnected -> + // Handshaking{SentMsg1} with the same SendHandshake + two SetTimer that + // `start_outbound_handshake` emits. Covers the oriented reach into + // `start_outbound_handshake` via `on_transport_connected` (the connectionless + // reach via `on_dial` is already covered by the test above). + #[test] + fn connection_oriented_dial_opens_transport_then_connected_handshakes() { + let mut alloc = IndexAllocator::new(); + let peer = peer_identity(); + + let mut m = PeerMachine::new_outbound(LinkId::new(1), peer, 0); + // Connection-oriented dial: open the transport first, no msg1 yet. + 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: true, + }, + 100, + &mut alloc, + ); + assert_eq!( + m.state(), + PeerState::Connecting { + link: LinkId::new(1) + } + ); + assert_eq!( + dial, + vec![PeerAction::OpenTransport { + transport_id: TransportId::new(1), + remote_addr: TransportAddr::from_string("127.0.0.1:9999"), + }], + "connection-oriented dial emits exactly one OpenTransport and no msg1" + ); + + // Transport connected: now send msg1 and arm the handshake timers. + let connected = m.step(PeerEvent::TransportConnected, 200, &mut alloc); + assert!(matches!( + m.state(), + PeerState::Handshaking { + phase: HandshakePhase::SentMsg1, + .. + } + )); + assert_eq!( + connected, + vec![ + PeerAction::SendHandshake { bytes: Vec::new() }, + PeerAction::SetTimer { + kind: TimerKind::HandshakeRetransmit, + at_ms: 200 + HANDSHAKE_RETRANSMIT_INTERVAL_MS, + }, + PeerAction::SetTimer { + kind: TimerKind::HandshakeTimeout, + at_ms: 200 + HANDSHAKE_TIMEOUT_MS, + }, + ] + ); + } + // ---- Test 8: liveness -> LinkDeadSuspected -> ReportLost -------------- #[test] fn liveness_to_link_dead() { @@ -1987,7 +2244,10 @@ mod tests { vec![ PeerAction::InvalidateSendState, PeerAction::TeardownConnectedUdp, - PeerAction::ReportLost { peer: addr }, + PeerAction::ReportLost { + peer: addr, + kind: LostKind::LinkDead, + }, ] ); assert!(matches!(m.state(), PeerState::Closed { .. }));