laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-07-22 07:48:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

peer: drive the connectionless outbound msg1 send through the machine

Browse Source 
Split the outbound handshake setup out of start_handshake into
prepare_outbound_msg1 (allocate the index, run the Noise leaf, frame and arm
msg1 -- the fallible steps, returning an error the caller propagates) and
send_stored_msg1 (transmit the armed wire). A connectionless dial now runs
prepare in the shell, then drives the control machine, whose SendHandshake
action sends the wire via the executor. Connection-oriented dials keep calling
start_handshake, now prepare followed by the send inline.

The dial event gains a connection_oriented flag so the machine's on_dial sends
msg1 immediately for connectionless transports (no connect step) instead of
opening a transport first. Behavior is unchanged: the same index, Noise leaf,
wire bytes, maps, and send-error handling as before, with index-allocation and
Noise failures still propagated synchronously before the send. A regression
test covers that a promote from the post-dial handshaking state is identical to
the former discovered-state promote.
This commit is contained in:
Johnathan Corgan
2026-07-15 06:15:21 +00:00
parent 3b99a416ad
commit 1f765cfd8f
4 changed files with 218 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
src/node/dataplane/peer_actions.rs
View File

@@ -6,18 +6,19 @@
//! stands for (`build_msg2` + `transport.send`, `promote_connection`,
//! `remove_active_peer`, `index_allocator.free`, `note_link_dead`, …).
//!
//! ## Shadow-only skeleton
//! ## Progressive cutover
//!
//! The machine home (`Node.peer_machines`), the
//! executor, and the disjoint-borrow advance helper. It is **unwired** — the live
//! `handle_msg1`/`handle_msg2` path does not drive it yet, so every method here is
//! `#[allow(dead_code)]`. The inbound cutover (`handle_msg1` → `step(InboundMsg1)`)
//! and the outbound cutover (`handle_msg2` / dial) are wired later.
//! The executor is wired incrementally. Live today: the inbound establish
//! (`handle_msg1` → `step(InboundMsg1)`), the outbound msg2 promote
//! (`handle_msg2` looks up the dial-persisted machine), and the connectionless
//! outbound msg1 send (`SendHandshake` with `their_index == None` →
//! `send_stored_msg1`, driven from `initiate_connection`).
//!
//! Arms not yet exercised are inert stubs (outbound dial, rekey/crypto installs,
//! link-control frames, timers, and the connected-UDP plane are inert stubs
//! realized as those planes are wired).
//! `RegisterDecryptSession` is a deliberate no-op — see its arm for the note.
//! Not yet driven, so their arms stay inert stubs: `OpenTransport` (the
//! connection-oriented dial), rekey/crypto installs, link-control frames, the
//! timers (`SetTimer`/`CancelTimer` — the legacy tick still runs them), and the
//! connected-UDP plane. `RegisterDecryptSession` is a deliberate no-op — see its
//! arm for the note.
use crate::PeerIdentity;
use crate::node::Node;
@@ -114,11 +115,16 @@ impl Node {
// yet; inert in the shadow-only skeleton.
}
PeerAction::SendHandshake { bytes } => {
// The machine payload is the UNFRAMED Noise msg2 payload;
// frame it with our/their index (mirrors `handshake.rs:472`'s
// `build_msg2(our_index, their_index, &payload)`) before the
// wire send. A fresh-outbound msg1 (empty payload → build msg1
// from indices) is framed differently and is wired later.
// Two outbound directions share this action, discriminated by
// `their_index`:
// msg2 (`their_index == Some`): the machine payload is the
// UNFRAMED Noise msg2; frame it with our/their index
// (`build_msg2`) and send.
// msg1 (`their_index == None`): a fresh outbound handshake;
// the machine's empty payload is ignored — the shell already
// allocated the index, ran the Noise leaf, and armed the
// wire at dial (`prepare_outbound_msg1`); this just sends the
// stored wire (see `send_stored_msg1`).
if let (Some(sender_idx), Some(receiver_idx)) =
(ambient.our_index, ambient.their_index)
{
@@ -152,6 +158,14 @@ impl Node {
.record_reject(RejectReason::Handshake(HandshakeReject::BadState));
return;
}
} else {
// msg1: the shell already allocated the index, ran the
// Noise leaf, and armed the wire on the connection at dial
// (`prepare_outbound_msg1`); send the stored wire. The
// machine's empty payload is ignored.
let _ = bytes;
self.send_stored_msg1(link, ambient.transport_id, &ambient.remote_addr)
.await;
}
}
PeerAction::SendRekey { .. } => {

10
src/node/handlers/handshake.rs
View File

@@ -1140,10 +1140,12 @@ impl Node {
// loser-link surgery is wired later). Direct analog of the inbound
// net-new arm — no ordering constraint, lowest risk.
//
// Look up the outbound machine persisted at DIAL (`start_handshake`),
// parked in `Discovered`, and step `Msg2 → [PromoteToActive]` in place.
// `on_msg2` is state-independent — it decides from the outbound snapshot,
// not the machine's state — and reads the machine's unset `conn.our_index`
// Look up the outbound machine persisted at DIAL, and step `Msg2 →
// [PromoteToActive]` in place. It is in `Discovered` (connection-oriented
// dial, not yet cut over) or `Handshaking{SentMsg1}` (connectionless dial,
// which drives the machine to send msg1) — either way `on_msg2` is
// state-independent: it decides from the outbound snapshot, not the
// machine's state, and reads the machine's unset `conn.our_index`
// as `None`, so stepping the persisted (vs the former transient) machine
// is byte-identical: same `Promote` decision (`has_existing_peer == false`),
// same `our_index == None`. The machine is already in `peer_machines`

115
src/node/lifecycle/mod.rs
View File

@@ -12,10 +12,11 @@ use super::peering::retry::MAX_RETRY_CONNECTIONS_PER_TICK;
use crate::config::{ConnectPolicy, PeerAddress, PeerConfig};
use crate::node::acl::PeerAclContext;
use crate::node::dataplane::PeerActionCtx;
use crate::nostr::{BootstrapEvent, NostrRendezvous};
use crate::nostr::{BootstrapHandoffResult, EstablishedTraversal};
use crate::peer::PeerConnection;
use crate::peer::machine::PeerMachine;
use crate::peer::machine::{PeerEvent, PeerMachine};
use crate::proto::fmp::wire::build_msg1;
use crate::proto::fmp::{Disconnect, DisconnectReason};
use crate::transport::{Link, LinkDirection, LinkId, TransportAddr, TransportId, packet_channel};
@@ -509,22 +510,71 @@ impl Node {
}
Ok(())
} else {
// Connectionless: proceed with immediate handshake
self.start_handshake(link_id, transport_id, remote_addr, peer_identity)
.await
// Connectionless: no connect step. Prepare msg1 in the shell — the
// index alloc, Noise leaf, and framing can fail and the error must
// propagate to the caller (matching the pre-cutover path) — then drive
// the machine to send it. The machine goes straight to
// `start_outbound_handshake`; the executor's `SendHandshake` msg1
// branch sends the wire `prepare_outbound_msg1` armed on the
// connection.
self.prepare_outbound_msg1(link_id, transport_id, &remote_addr, peer_identity)?;
let now = Self::now_ms();
let ambient = PeerActionCtx {
verified_identity: peer_identity,
transport_id,
remote_addr: remote_addr.clone(),
our_index: None,
their_index: None,
now_ms: now,
is_outbound: true,
};
self.advance_peer_machine(
link_id,
PeerEvent::Dial {
transport_id,
remote_addr,
peer_identity,
connection_oriented: false,
},
now,
&ambient,
)
.await;
Ok(())
}
}
/// Start the Noise handshake on a link and send msg1.
///
/// Called immediately for connectionless transports, or after the
/// transport connection is established for connection-oriented transports.
/// Called after a connection-oriented transport connects. (Connectionless
/// dials `prepare_outbound_msg1` + drive the machine to send in
/// `initiate_connection`.)
pub(super) async fn start_handshake(
&mut self,
link_id: LinkId,
transport_id: TransportId,
remote_addr: TransportAddr,
peer_identity: PeerIdentity,
) -> Result<(), NodeError> {
self.prepare_outbound_msg1(link_id, transport_id, &remote_addr, peer_identity)?;
self.send_stored_msg1(link_id, transport_id, &remote_addr)
.await;
Ok(())
}
/// Prepare an outbound Noise msg1 at dial: allocate the session index, run
/// the Noise leaf, frame the wire, arm the shell-side resend, track
/// `pending_outbound`, and persist the connection + control machine (parked
/// in `Discovered`). Returns `Err` on index-allocation or Noise failure
/// (cleaning the partial leg), leaving the armed wire on the connection for
/// `send_stored_msg1` to transmit. Does NOT send — so the fallible setup can
/// propagate its error synchronously before any machine drive.
pub(in crate::node) fn prepare_outbound_msg1(
&mut self,
link_id: LinkId,
transport_id: TransportId,
remote_addr: &TransportAddr,
peer_identity: PeerIdentity,
) -> Result<(), NodeError> {
let peer_node_addr = *peer_identity.node_addr();
@@ -538,7 +588,8 @@ impl Node {
Err(e) => {
// Clean up the link we just created
self.links.remove(&link_id);
self.addr_to_link.remove(&(transport_id, remote_addr));
self.addr_to_link
.remove(&(transport_id, remote_addr.clone()));
return Err(NodeError::IndexAllocationFailed(e.to_string()));
}
};
@@ -552,7 +603,8 @@ impl Node {
// Clean up the index and link
let _ = self.index_allocator.free(our_index);
self.links.remove(&link_id);
self.addr_to_link.remove(&(transport_id, remote_addr));
self.addr_to_link
.remove(&(transport_id, remote_addr.clone()));
return Err(NodeError::HandshakeFailed(e.to_string()));
}
};
@@ -576,7 +628,7 @@ impl Node {
// Store msg1 for resend and schedule first resend
let resend_interval = self.config().node.rate_limit.handshake_resend_interval_ms;
connection.set_handshake_msg1(wire_msg1.clone(), current_time_ms + resend_interval);
connection.set_handshake_msg1(wire_msg1, current_time_ms + resend_interval);
// Track in pending_outbound for msg2 dispatch
self.pending_outbound
@@ -586,7 +638,8 @@ impl Node {
// Persist the outbound control machine at dial, keyed by the same
// `link_id` as the connection. It parks in `Discovered` (inert to reap
// and rekey — it is absent from `peers`, never established) until
// `handle_msg2` looks it up to drive the promote. Its `our_index` is
// `handle_msg2` looks it up to drive the promote, or a connectionless
// dial drives it to `Handshaking` to send. Its `our_index` is
// deliberately left unset so a later inbound restart does not emit a
// spurious `UnregisterDecryptSession`. It is removed wherever the
// connection is torn down without promoting (the stale reaper and the
@@ -595,16 +648,42 @@ impl Node {
let machine = PeerMachine::new_outbound(link_id, peer_identity, current_time_ms);
self.peer_machines.insert(link_id, machine);
Ok(())
}
/// Send the msg1 wire that `prepare_outbound_msg1` armed on the connection.
/// On send error, marks the connection failed and RETAINS it (the legacy
/// resend tick retries); a missing wire or transport is a no-op. This is the
/// body of the executor's `SendHandshake` msg1 action and the send tail of
/// the connection-oriented `start_handshake`.
pub(in crate::node) async fn send_stored_msg1(
&mut self,
link_id: LinkId,
transport_id: TransportId,
remote_addr: &TransportAddr,
) {
let wire_msg1 = match self
.connections
.get(&link_id)
.and_then(|c| c.handshake_msg1())
{
Some(w) => w.to_vec(),
None => return,
};
let our_index = self.connections.get(&link_id).and_then(|c| c.our_index());
// Send the wire format handshake message
if let Some(transport) = self.transports.get(&transport_id) {
match transport.send(&remote_addr, &wire_msg1).await {
match transport.send(remote_addr, &wire_msg1).await {
Ok(bytes) => {
debug!(
link_id = %link_id,
our_index = %our_index,
bytes,
"Sent Noise handshake message 1 (wire format)"
);
if let Some(idx) = our_index {
debug!(
link_id = %link_id,
our_index = %idx,
bytes,
"Sent Noise handshake message 1 (wire format)"
);
}
}
Err(e) => {
warn!(
@@ -620,8 +699,6 @@ impl Node {
}
}
}
Ok(())
}
/// Poll all transports for discovered peers and auto-connect.

99
src/peer/machine.rs
View File

@@ -188,11 +188,15 @@ pub(crate) enum TimerKind {
///
/// Not `Debug`/`PartialEq`: the reused core snapshot payloads derive neither.
pub(crate) enum PeerEvent {
/// Reconciler dial intent.
/// Reconciler dial intent. `connection_oriented` selects the outbound
/// path: connection-oriented transports open the transport first
/// (`OpenTransport` → `Connecting`); connectionless ones send msg1
/// immediately (`start_outbound_handshake` → `Handshaking`).
Dial {
transport_id: TransportId,
remote_addr: TransportAddr,
peer_identity: PeerIdentity,
connection_oriented: bool,
},
/// Connection-oriented transport connected.
TransportConnected,
@@ -459,8 +463,9 @@ impl PeerMachine {
PeerEvent::Dial {
transport_id,
remote_addr,
connection_oriented,
..
} => self.on_dial(transport_id, remote_addr, now),
} => self.on_dial(transport_id, remote_addr, connection_oriented, now),
PeerEvent::TransportConnected => self.on_transport_connected(now),
PeerEvent::TransportFailed => self.on_transport_failed(now),
PeerEvent::InboundMsg1 { link, wire, est } => {
@@ -513,21 +518,28 @@ impl PeerMachine {
&mut self,
transport_id: TransportId,
remote_addr: TransportAddr,
_now: u64,
connection_oriented: bool,
now: u64,
) -> Vec<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
// (OpenTransport) — the reconciler's candidate carries the transport
// kind at wiring time; connectionless dial reuses `start_handshake`.
self.state = PeerState::Connecting { link: self.link };
vec![PeerAction::OpenTransport {
transport_id,
remote_addr,
}]
if connection_oriented {
// Connection-oriented transports open the transport first; the
// executor's `OpenTransport` arm connects, then feeds
// `TransportConnected` → `start_outbound_handshake`.
self.state = PeerState::Connecting { link: self.link };
vec![PeerAction::OpenTransport {
transport_id,
remote_addr,
}]
} else {
// Connectionless transports have no connect step — send msg1
// immediately (the executor's `SendHandshake` msg1 branch performs
// the Noise leaf, framing, index alloc, and send).
self.start_outbound_handshake(now)
}
}
fn on_transport_connected(&mut self, now: u64) -> Vec<PeerAction> {
@@ -2093,6 +2105,69 @@ mod tests {
);
}
// ---- Test 7c: connectionless dial reaches Handshaking, Msg2 neutral ----
// The connectionless cutover drives the outbound machine
// Discovered -> (Dial, connection_oriented=false) -> Handshaking{SentMsg1}
// BEFORE msg2, whereas the pre-cutover path stepped Msg2 while still in
// Discovered. `on_msg2` is state-independent, so both must yield the
// identical `[PromoteToActive]` and leave `our_index == None`.
#[test]
fn connectionless_dial_then_msg2_promotes_from_handshaking() {
let mut alloc = IndexAllocator::new();
let peer = peer_identity();
let mut m = PeerMachine::new_outbound(LinkId::new(1), peer, 0);
// Connectionless dial: no OpenTransport, straight to Handshaking{SentMsg1}.
let dial = m.step(
PeerEvent::Dial {
transport_id: TransportId::new(1),
remote_addr: TransportAddr::from_string("127.0.0.1:9999"),
peer_identity: peer,
connection_oriented: false,
},
100,
&mut alloc,
);
assert!(matches!(
m.state(),
PeerState::Handshaking {
phase: HandshakePhase::SentMsg1,
..
}
));
assert!(
dial.iter()
.any(|a| matches!(a, PeerAction::SendHandshake { .. }))
);
assert!(
!dial
.iter()
.any(|a| matches!(a, PeerAction::OpenTransport { .. })),
"connectionless dial emits no OpenTransport"
);
// Step Msg2 from Handshaking — identical promote to the Discovered path.
let out = OutboundSnapshot {
has_existing_peer: false,
our_outbound_wins: false,
};
let promote = m.step(
PeerEvent::Msg2 {
their_index: SessionIndex::new(0x77),
out,
},
200,
&mut alloc,
);
assert_eq!(
promote,
vec![PeerAction::PromoteToActive {
link: LinkId::new(1)
}]
);
assert_eq!(m.our_index(), None);
}
// ---- Test 8: liveness -> LinkDeadSuspected -> ReportLost --------------
#[test]
fn liveness_to_link_dead() {