laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-08-09 08:14:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'refactor-node' into refactor-node-next

Browse Source 
Bring the outbound-handshake-through-the-machine series onto the next-branch
Noise-XX cores. Re-expressed rather than transcribed: next has no outbound
decision core, so the promote drives PeerEvent::OutboundMsg2 (not the IK Msg2
snapshot event) and cross-connection resolution stays inline. Hybrid provenance
-- identified outbound legs persist the control machine at dial and drive
through it; anonymous-discovery legs (identity unknown until XX msg2) retain the
inline transient-at-msg2 path. ReportLost{kind} routing, the connection-oriented
dial via the machine (OpenTransport + TransportConnected), and the prepare/send
msg1 split all carried over.

Behavior-neutral on next: the XX msg2 body (Noise completion, FMP, ACL, msg3) is
untouched; anonymous legs are unchanged; the dial-persisted machine promotes
with our_index unset.
This commit is contained in:
Johnathan Corgan
2026-07-15 15:52:29 +00:00
parent 4620d9f82e 3e7ca90212
commit 8abcd6ff8c
6 changed files with 864 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/dataplane/peer_actions.rs
+124 -57
View File
@@ -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<PeerAction> = 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);
}
}
src/node/handlers/handshake.rs
+45 -14
View File
@@ -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).
src/node/handlers/timeout.rs
+5
View File
@@ -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
src/node/lifecycle/mod.rs
+301 -61
View File
@@ -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());
}
}
src/node/tests/tcp.rs
+110
View File
@@ -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;
}
src/peer/machine.rs
+279 -19
View File
@@ -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<PeerAction> {
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<PeerAction> {
@@ -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 { .. }));