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

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

Browse Source 
Forward-merge the state-carrier collapse (connection start/activity timestamps,
peer-identity telemetry, handshake resend buffers, index/transport bookkeeping,
and the session-index shadow all moved onto the peer machine's own connection
state) onto the XX handshake.

Next-specific resolutions folded into the merge: the responder learns its transport
id and peer index at msg1 and they are written directly onto the machine there (the
XX machine has no msg1 step, so the promote read would otherwise be unset); the
negotiated peer profile is mirrored onto the machine carrier at both the initiator
(msg2) and responder (msg3) negotiation sites; and the outbound carrier writes land
on both the identified and anonymous dial paths. The leg's transport/index/buffer
copies remain as dead duplicates until the leg dissolves; the machine copy is the
sole live source.
This commit is contained in:
Johnathan Corgan
2026-07-18 18:24:44 +00:00
parent 91ba660213 791b35c221
commit df52ce459c
10 changed files with 403 additions and 174 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/control/queries.rs
View File

@@ -1321,12 +1321,12 @@ pub fn show_connections(node: &Node) -> Value {
"link_id": conn.link_id().as_u64(),
"direction": format!("{}", conn.direction()),
"handshake_state": node.connection_handshake_state(conn.link_id()),
"started_at_ms": conn.started_at(),
"idle_ms": now.saturating_sub(conn.last_activity()),
"started_at_ms": node.connection_started_at(conn.link_id()),
"idle_ms": now.saturating_sub(node.connection_last_activity(conn.link_id())),
"resend_count": node.connection_resend_count(conn.link_id()),
});
if let Some(identity) = conn.expected_identity() {
if let Some(identity) = node.connection_expected_identity(conn.link_id()) {
conn_json["expected_peer"] = json!(identity.npub());
}

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

@@ -316,10 +316,10 @@ impl Node {
self.links.insert(link_id, link);
self.addr_to_link.insert(addr_key, link_id);
// Build the msg2 response, storing it on the connection for potential
// resend before the connection is embedded on the machine below.
// Build the msg2 response, storing it on the surviving carrier for
// potential resend before the connection is embedded on the machine
// below.
let wire_msg2 = build_msg2(our_index, header.sender_idx, &msg2_response);
conn.set_handshake_msg2(wire_msg2.clone());
// The leg's persistent control machine is born carrying its pending
// connection, parked at `SentMsg2` awaiting msg3 (identity is unknown
@@ -328,6 +328,15 @@ impl Node {
// steps this same machine; every teardown path disposes it with the
// embedded leg.
let mut machine = PeerMachine::inbound_msg2_sent(link_id, our_index, packet.timestamp_ms);
// The inbound leg carries the transport ID and peer index from msg1, but
// the machine's carrier is only written on the outbound dial and at msg3.
// Seed both here so the promotion hand-off reads them from the surviving
// carrier, matching the leg's inbound seeds.
machine.set_conn_transport_id(packet.transport_id);
machine.set_conn_their_index(header.sender_idx);
// Store the framed msg2 on the surviving carrier for duplicate-msg1
// resend while the connection is still pending.
machine.set_conn_handshake_msg2(wire_msg2.clone());
machine.set_leg(conn);
self.peer_machines.insert(link_id, machine);
@@ -373,12 +382,15 @@ impl Node {
/// Find stored msg2 bytes for a given link (pre- or post-promotion).
///
/// Checks the PeerConnection (if still pending) and then the ActivePeer
/// (if already promoted).
/// Checks the control machine's carrier (if still pending) and then the
/// ActivePeer (if already promoted).
fn find_stored_msg2(&self, link_id: LinkId) -> Option<Vec<u8>> {
// Check pending connection first
if let Some(conn) = self.leg(&link_id)
&& let Some(msg2) = conn.handshake_msg2()
// Check pending connection first (its stored msg2 lives on the control
// machine's carrier).
if let Some(msg2) = self
.peer_machines
.get(&link_id)
.and_then(|machine| machine.conn_handshake_msg2())
{
return Some(msg2.to_vec());
}
@@ -700,6 +712,18 @@ impl Node {
let peer_node_addr = *peer_identity.node_addr();
// Mirror the leg's completion `touch` and the negotiated peer profile
// onto the surviving carrier so the connection's last-activity advances
// at msg2 completion (matching the leg's clock) and the promotion
// hand-off reads the profile from the carrier.
let negotiated_profile = self.leg(&link_id).and_then(|conn| conn.peer_profile());
if let Some(machine) = self.peer_machines.get_mut(&link_id) {
machine.touch_conn(packet.timestamp_ms);
if let Some(profile) = negotiated_profile {
machine.set_conn_peer_profile(profile);
}
}
// ACL check: with XX, this is the first point where the initiator
// knows the responder's identity.
if self
@@ -1158,6 +1182,15 @@ impl Node {
let peer_node_addr = *peer_identity.node_addr();
// Mirror the negotiated peer profile onto the surviving carrier so the
// promotion hand-off reads it from the carrier, not the leg.
let negotiated_profile = self.leg(&link_id).and_then(|conn| conn.peer_profile());
if let Some(profile) = negotiated_profile
&& let Some(machine) = self.peer_machines.get_mut(&link_id)
{
machine.set_conn_peer_profile(profile);
}
// ACL check: with XX, this is the first point where the responder
// knows the initiator's identity.
if self
@@ -1492,14 +1525,18 @@ impl Node {
link_id,
reason: "missing our_index".into(),
})?;
let their_index = connection
.their_index()
let their_index = self
.peer_machines
.get(&link_id)
.and_then(|machine| machine.conn_their_index())
.ok_or_else(|| NodeError::PromotionFailed {
link_id,
reason: "missing their_index".into(),
})?;
let transport_id = connection
.transport_id()
let transport_id = self
.peer_machines
.get(&link_id)
.and_then(|machine| machine.conn_transport_id())
.ok_or_else(|| NodeError::PromotionFailed {
link_id,
reason: "missing transport_id".into(),
@@ -1511,10 +1548,16 @@ impl Node {
reason: "missing source_addr".into(),
})?
.clone();
let link_stats = connection.link_stats().clone();
let link_stats = self
.peer_machines
.get(&link_id)
.map(|machine| machine.conn_link_stats().clone())
.unwrap_or_default();
let remote_epoch = connection.remote_epoch();
let peer_profile = connection
.peer_profile()
let peer_profile = self
.peer_machines
.get(&link_id)
.and_then(|machine| machine.conn_peer_profile())
.unwrap_or(crate::proto::fmp::NodeProfile::Full);
let peer_node_addr = *verified_identity.node_addr();

38
src/node/handlers/timeout.rs
View File

@@ -20,9 +20,9 @@ impl LifecycleView for Node {
self.peer_machines
.iter()
.filter_map(|(link_id, machine)| machine.leg().map(|conn| (link_id, machine, conn)))
.filter(|(link_id, machine, conn)| {
.filter(|(link_id, machine, _conn)| {
machine.is_failed()
|| (conn.is_timed_out(now_ms, timeout_ms)
|| (machine.conn_is_timed_out(now_ms, timeout_ms)
&& !self.peer_timers.get(*link_id).is_some_and(|timers| {
timers.contains_key(&TimerKind::HandshakeTimeout)
}))
@@ -89,7 +89,8 @@ impl Node {
debug!(
link_id = %link,
direction = %direction,
idle_secs = conn.idle_time(now_ms) / 1000,
idle_secs =
now_ms.saturating_sub(self.connection_last_activity(link)) / 1000,
"Stale handshake connection timed out"
);
}
@@ -123,12 +124,17 @@ impl Node {
Some(c) => c,
None => return,
};
// Read the transport ID off the surviving carrier before disposing the
// machine (the leg no longer projects it).
let transport_id = self
.peer_machines
.get(&link_id)
.and_then(|machine| machine.conn_transport_id());
self.remove_peer_machine(link_id);
let transport_id = conn.transport_id();
// Free session index and pending_outbound/pending_inbound if allocated
if let Some(idx) = conn.our_index() {
if let Some(tid) = conn.transport_id() {
if let Some(tid) = transport_id {
self.pending_outbound.remove(&(tid, idx.as_u32()));
self.pending_inbound.remove(&(tid, idx.as_u32()));
}
@@ -183,8 +189,15 @@ impl Node {
.map(|(link, _)| *link)
.collect();
for link in timer_links {
// The idle-timeout threshold reads the survivor carrier's
// last-activity (the leg no longer projects it); the leg still
// supplies direction/identity for the retry decision below.
let timed_out = self
.peer_machines
.get(&link)
.is_some_and(|machine| machine.conn_is_timed_out(now_ms, timeout_ms));
let (reap, retry_peer) = match self.leg(&link) {
Some(conn) if conn.is_timed_out(now_ms, timeout_ms) => {
Some(conn) if timed_out => {
let retry_peer = if conn.is_outbound() {
conn.expected_identity().map(|id| *id.node_addr())
} else {
@@ -278,7 +291,11 @@ impl Node {
continue;
}
};
match self.leg(&link).and_then(|c| c.handshake_msg1()) {
match self
.peer_machines
.get(&link)
.and_then(|machine| machine.conn_handshake_msg1())
{
// Armed but the stored wire isn't there yet — leave the timer and
// retry next tick (matches the old candidate filter skipping it).
None => continue,
@@ -311,7 +328,12 @@ impl Node {
};
let (transport_id, remote_addr) = match self.leg(&link) {
Some(conn) => match (conn.transport_id(), conn.source_addr()) {
Some(conn) => match (
self.peer_machines
.get(&link)
.and_then(|machine| machine.conn_transport_id()),
conn.source_addr(),
) {
(Some(tid), Some(addr)) => (tid, addr.clone()),
_ => continue,
},

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

@@ -385,12 +385,14 @@ impl Node {
transport_id: TransportId,
remote_addr: &TransportAddr,
) -> bool {
self.connections().any(|conn| {
conn.expected_identity()
.map(|id| id.node_addr() == peer_node_addr)
.unwrap_or(false)
&& conn.transport_id() == Some(transport_id)
&& conn.source_addr() == Some(remote_addr)
self.peer_machines.values().any(|machine| {
machine.leg().is_some_and(|conn| {
conn.expected_identity()
.map(|id| id.node_addr() == peer_node_addr)
.unwrap_or(false)
&& machine.conn_transport_id() == Some(transport_id)
&& conn.source_addr() == Some(remote_addr)
})
}) || self.peering.pending_connects.iter().any(|pending| {
pending
.peer_identity
@@ -491,11 +493,14 @@ impl Node {
// 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.
// drives it to `Handshaking` to send. Its carrier index is
// written at msg1 preparation (`prepare_outbound_msg1` /
// `start_handshake`); a later inbound msg1 for the same peer is
// handled on a FRESH `new_inbound` machine (`handle_msg1`), never
// by driving this map-resident machine through the inbound path.
// 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, Some(identity), Self::now_ms());
self.peer_machines.insert(link_id, machine);
@@ -637,7 +642,6 @@ impl Node {
// 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]
@@ -652,9 +656,10 @@ impl Node {
"Connection initiated"
);
// Store msg1 for resend and schedule first resend
// Schedule the first msg1 resend; the wire itself is stored on the
// surviving carrier once the machine is in hand below.
let resend_interval = self.config().node.rate_limit.handshake_resend_interval_ms;
connection.set_handshake_msg1(wire_msg1, current_time_ms + resend_interval);
let first_resend_at_ms = current_time_ms + resend_interval;
// Track in pending_outbound for msg2 dispatch
self.pending_outbound
@@ -668,12 +673,27 @@ impl Node {
self.peer_machines.contains_key(&link_id),
"outbound msg1 prepared for link {link_id} with no dial-time machine"
);
self.peer_machines
.entry(link_id)
.or_insert_with(|| {
PeerMachine::new_outbound(link_id, Some(peer_identity), current_time_ms)
})
.set_leg(connection);
let machine = self.peer_machines.entry(link_id).or_insert_with(|| {
PeerMachine::new_outbound(link_id, Some(peer_identity), current_time_ms)
});
// The dial-born machine carrier was stamped at dial; re-stamp it with the
// leg's msg1-prep clock so the surviving `started_at`/`last_activity`
// carry the leg's provenance. The two clocks differ when a connect
// round-trip separates dial from msg1 preparation.
machine.set_conn_started_at(current_time_ms);
machine.touch_conn(current_time_ms);
// Record the transport ID on the surviving carrier (the leg no longer
// projects it to the promotion hand-off); holds even if a direct caller
// reached here without the dial-time `on_dial` write.
machine.set_conn_transport_id(transport_id);
// Record our session index on the surviving carrier — the same index just
// written on the leg above — so the carrier is the single index home on
// the outbound path (the inbound path writes it at authorize).
machine.set_conn_our_index(our_index);
// Store the msg1 wire on the surviving carrier (the leg no longer holds
// the resend source); the retransmit driver reads it from here.
machine.set_conn_handshake_msg1(wire_msg1, first_resend_at_ms);
machine.set_leg(connection);
Ok(())
}
@@ -755,7 +775,8 @@ 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);
let first_resend_at_ms = current_time_ms + resend_interval;
connection.set_handshake_msg1(wire_msg1.clone(), first_resend_at_ms);
// Track in pending_outbound for msg2 dispatch
self.pending_outbound
@@ -773,6 +794,15 @@ impl Node {
// are armed and no event is dispatched: this path sends msg1 inline,
// so the machine parks at `Discovered` until msg2.
let mut machine = PeerMachine::new_outbound(link_id, peer_identity, current_time_ms);
// Seed the surviving carrier with the leg's msg1-prep provenance so the
// started_at/last_activity timestamps, the transport ID, our index, and
// the msg1 resend wire all live on the carrier — the promotion hand-off
// and the retransmit driver read them there, not the leg.
machine.set_conn_started_at(current_time_ms);
machine.touch_conn(current_time_ms);
machine.set_conn_transport_id(transport_id);
machine.set_conn_our_index(our_index);
machine.set_conn_handshake_msg1(wire_msg1.clone(), first_resend_at_ms);
machine.set_leg(connection);
self.peer_machines.insert(link_id, machine);
@@ -828,7 +858,11 @@ impl Node {
remote_addr: &TransportAddr,
now_ms: u64,
) {
let wire_msg1 = match self.leg(&link_id).and_then(|c| c.handshake_msg1()) {
let wire_msg1 = match self
.peer_machines
.get(&link_id)
.and_then(|machine| machine.conn_handshake_msg1())
{
Some(w) => w.to_vec(),
None => return,
};

75
src/node/mod.rs
View File

@@ -2022,10 +2022,12 @@ impl Node {
link_id: conn.link_id().as_u64(),
direction: format!("{}", conn.direction()),
handshake_state: self.connection_handshake_state(conn.link_id()).to_string(),
started_at_ms: conn.started_at(),
last_activity_ms: conn.last_activity(),
started_at_ms: self.connection_started_at(conn.link_id()),
last_activity_ms: self.connection_last_activity(conn.link_id()),
resend_count: self.connection_resend_count(conn.link_id()),
expected_peer: conn.expected_identity().map(|id| id.npub()),
expected_peer: self
.connection_expected_identity(conn.link_id())
.map(|id| id.npub()),
})
.collect();
@@ -2348,6 +2350,38 @@ impl Node {
.map_or(0, |machine| machine.resend_count())
}
/// Operator-visible connection-start timestamp for a pending handshake
/// `link`, read from the per-peer machine carrier (the timing's home now
/// that the leg no longer projects it). A link with no machine reports 0;
/// every leg surfaced by `connections()` is embedded in a machine, so the
/// lookup resolves.
pub(crate) fn connection_started_at(&self, link: LinkId) -> u64 {
self.peer_machines
.get(&link)
.map_or(0, |machine| machine.conn_started_at())
}
/// Operator-visible last-activity timestamp for a pending handshake `link`,
/// read from the per-peer machine carrier. A link with no machine reports 0;
/// every leg surfaced by `connections()` is embedded in a machine, so the
/// lookup resolves.
pub(crate) fn connection_last_activity(&self, link: LinkId) -> u64 {
self.peer_machines
.get(&link)
.map_or(0, |machine| machine.conn_last_activity())
}
/// Operator-visible expected peer identity for a pending handshake `link`,
/// read from the per-peer machine carrier (the identity's telemetry home now
/// that the leg no longer projects it). A link with no machine reports
/// `None`; every leg surfaced by `connections()` is embedded in a machine, so
/// the lookup resolves.
pub(crate) fn connection_expected_identity(&self, link: LinkId) -> Option<PeerIdentity> {
self.peer_machines
.get(&link)
.and_then(|machine| machine.conn_expected_identity().copied())
}
/// Operator-visible handshake-state string for a pending handshake `link`,
/// derived from the per-peer control machine (the phase's home now that the
/// leg no longer carries it). Every leg surfaced by `connections()` is
@@ -2372,9 +2406,9 @@ impl Node {
.links
.values()
.any(|link| link.transport_id() == transport_id)
|| self
.connections()
.any(|conn| conn.transport_id() == Some(transport_id))
|| self.peer_machines.values().any(|machine| {
machine.leg().is_some() && machine.conn_transport_id() == Some(transport_id)
})
|| self
.peers
.values()
@@ -2446,17 +2480,24 @@ impl Node {
});
}
self.peer_machines
.entry(link_id)
.or_insert_with(|| {
let now = connection.started_at();
if connection.is_outbound() {
PeerMachine::new_outbound(link_id, connection.expected_identity().copied(), now)
} else {
PeerMachine::new_inbound(link_id, now)
}
})
.set_leg(connection);
let machine = self.peer_machines.entry(link_id).or_insert_with(|| {
let now = connection.started_at();
if connection.is_outbound() {
PeerMachine::new_outbound(link_id, connection.expected_identity().copied(), now)
} else {
PeerMachine::new_inbound(link_id, now)
}
});
// Seed the surviving carrier's peer index and transport from the
// pre-built leg so the promotion hand-off reads them from the machine,
// matching the establish paths that write them on the machine directly.
if let Some(their) = connection.their_index() {
machine.set_conn_their_index(their);
}
if let Some(tid) = connection.transport_id() {
machine.set_conn_transport_id(tid);
}
machine.set_leg(connection);
Ok(())
}

5
src/node/tests/handshake.rs
View File

@@ -903,7 +903,7 @@ async fn test_resend_scheduling() {
// Store msg1 with first resend at now + 1000ms
let wire_msg1 = crate::proto::fmp::wire::build_msg1(our_index, &noise_msg1);
conn.set_handshake_msg1(wire_msg1, now_ms + 1000);
conn.set_handshake_msg1(wire_msg1.clone(), now_ms + 1000);
let link = Link::connectionless(
link_id,
@@ -934,6 +934,9 @@ async fn test_resend_scheduling() {
now_ms,
&mut node.index_allocator,
);
// The msg1 wire lives on the machine's carrier (the retransmit driver's
// resend source), mirroring `prepare_outbound_msg1`.
machine.set_conn_handshake_msg1(wire_msg1, now_ms + 1000);
machine.set_leg(conn);
node.peer_machines.insert(link_id, machine);
node.peer_timers.entry(link_id).or_default().insert(

27
src/peer/connection.rs
View File

@@ -10,7 +10,7 @@
use crate::PeerIdentity;
use crate::noise::{self, NoiseError, NoiseSession};
use crate::proto::fmp::{ConnectionState, NodeProfile};
use crate::transport::{LinkDirection, LinkId, LinkStats, TransportAddr, TransportId};
use crate::transport::{LinkDirection, LinkId, TransportAddr, TransportId};
use crate::utils::index::SessionIndex;
use secp256k1::Keypair;
use std::fmt;
@@ -126,16 +126,14 @@ impl PeerConnection {
self.state.is_inbound()
}
/// When the connection started.
/// When the connection started. Retained only to seed a control machine's
/// carrier from a pre-built leg (`Node::add_connection`); the operator-facing
/// `started_at_ms`/`last_activity_ms` telemetry now reads the machine carrier,
/// not the leg.
pub fn started_at(&self) -> u64 {
self.state.started_at()
}
/// When the last activity occurred.
pub fn last_activity(&self) -> u64 {
self.state.last_activity()
}
/// Connection duration so far.
pub fn duration(&self, current_time_ms: u64) -> u64 {
self.state.duration(current_time_ms)
@@ -146,16 +144,6 @@ impl PeerConnection {
self.state.idle_time(current_time_ms)
}
/// Get link statistics.
pub fn link_stats(&self) -> &LinkStats {
self.state.link_stats()
}
/// Get mutable link statistics.
pub fn link_stats_mut(&mut self) -> &mut LinkStats {
self.state.link_stats_mut()
}
// === Index Accessors ===
/// Get our session index (if set).
@@ -488,11 +476,6 @@ impl PeerConnection {
self.noise_handshake = None;
}
/// Update last activity timestamp.
pub fn touch(&mut self, current_time_ms: u64) {
self.state.touch(current_time_ms);
}
// === Validation ===
/// Check if the connection has timed out.

227
src/peer/machine.rs
View File

@@ -86,11 +86,11 @@
use crate::peer::PeerConnection;
use crate::proto::fmp::{
ConnAction, ConnSnapshot, ConnectionState, EstablishSnapshot, Fmp, InboundDecision,
InboundReject, OutboundDecision, OutboundSnapshot, PeerSnapshot, PromotionResult, RekeyCfg,
RekeyResendSnapshot, WireOutcome,
InboundReject, NodeProfile, OutboundDecision, OutboundSnapshot, PeerSnapshot, PromotionResult,
RekeyCfg, RekeyResendSnapshot, WireOutcome,
};
use crate::proto::link::LinkMessageType;
use crate::transport::{LinkId, TransportAddr, TransportId};
use crate::transport::{LinkId, LinkStats, TransportAddr, TransportId};
use crate::utils::index::{IndexAllocator, SessionIndex};
use crate::{NodeAddr, PeerIdentity};
@@ -541,9 +541,12 @@ pub(crate) struct PeerMachine {
rekey_jitter_secs: i64,
last_heartbeat_sent_ms: u64,
// --- decrypt-registration shadow ---
/// The currently-registered decrypt index (post-establish).
our_index: Option<SessionIndex>,
// --- decrypt-registration drain window ---
// The machine owns decrypt-worker register/unregister via actions. The
// currently-registered index lives on the surviving carrier (`conn`); this
// field tracks the previous index held open during a post-cutover drain
// window (to later unregister/free). Control knowledge of the registration
// lifecycle, distinct from the hot send-state slots.
/// The previous index held open during a post-cutover drain window.
draining_index: Option<SessionIndex>,
}
@@ -578,7 +581,6 @@ impl PeerMachine {
authenticated_at_ms: 0,
rekey_jitter_secs: 0,
last_heartbeat_sent_ms: 0,
our_index: None,
draining_index: None,
}
}
@@ -607,7 +609,6 @@ impl PeerMachine {
authenticated_at_ms: 0,
rekey_jitter_secs: 0,
last_heartbeat_sent_ms: 0,
our_index: None,
draining_index: None,
}
}
@@ -624,7 +625,6 @@ impl PeerMachine {
pub(crate) fn inbound_msg2_sent(link: LinkId, our_index: SessionIndex, now: u64) -> Self {
let mut machine = Self::new_inbound(link, now);
machine.conn.set_our_index(our_index);
machine.our_index = Some(our_index);
machine.state = PeerState::Handshaking {
link,
phase: HandshakePhase::SentMsg2,
@@ -676,7 +676,6 @@ impl PeerMachine {
authenticated_at_ms: now,
rekey_jitter_secs: 0,
last_heartbeat_sent_ms: 0,
our_index: Some(our_index),
draining_index: None,
}
}
@@ -708,12 +707,13 @@ impl PeerMachine {
self.leg = Some(leg);
}
/// The index we allocated for this peer's inbound session at msg1. `None`
/// before allocation. The inbound cross-connection / rekey-respond triggers
/// and the executor read this to perform the shell registry surgery with the
/// machine-owned index.
/// The session index we allocated for this peer, read from the surviving
/// carrier. Populated once the index is allocated on either establish path
/// (inbound at msg1, outbound at msg1 preparation). `None` before
/// allocation (and after a rejected/unauthorized msg1). The inbound cutover
/// reads this to perform the shell registry surgery.
pub(crate) fn our_index(&self) -> Option<SessionIndex> {
self.our_index
self.conn.our_index()
}
/// The peer identity this machine was born with (outbound identified dial)
@@ -756,6 +756,122 @@ impl PeerMachine {
self.conn.record_resend(next_resend_at_ms);
}
/// Connection-start timestamp of the surviving carrier — the home for the
/// operator-visible `started_at_ms` now that the leg no longer projects it.
pub(crate) fn conn_started_at(&self) -> u64 {
self.conn.started_at()
}
/// Last-activity timestamp of the surviving carrier — the home for the
/// operator-visible `last_activity_ms`/`idle_ms` and the idle-timeout check.
pub(crate) fn conn_last_activity(&self) -> u64 {
self.conn.last_activity()
}
/// Whether the carrier has been idle past `timeout_ms`.
pub(crate) fn conn_is_timed_out(&self, now_ms: u64, timeout_ms: u64) -> bool {
self.conn.is_timed_out(now_ms, timeout_ms)
}
/// Expected peer identity of the surviving carrier — the home for the
/// operator-visible `expected_peer` now that the leg no longer projects it.
/// Outbound carries the dial identity from construction; inbound stays `None`
/// in this view (the identity learned mid-handshake never rests here, as the
/// leg is consumed by promotion within the same message-handling step).
pub(crate) fn conn_expected_identity(&self) -> Option<&PeerIdentity> {
self.conn.expected_identity()
}
/// Stored wire-format msg1 of the surviving carrier — the resend source for
/// the outbound handshake retransmit, now that the leg no longer carries it.
pub(crate) fn conn_handshake_msg1(&self) -> Option<&[u8]> {
self.conn.handshake_msg1()
}
/// Stored wire-format msg2 of the surviving carrier — the resend source for a
/// duplicate msg1 while the inbound handshake is still pending.
pub(crate) fn conn_handshake_msg2(&self) -> Option<&[u8]> {
self.conn.handshake_msg2()
}
/// Store the wire-format msg1 for resend on the surviving carrier and record
/// the first resend deadline, mirroring the leg's start-of-handshake write.
pub(crate) fn set_conn_handshake_msg1(&mut self, msg1: Vec<u8>, first_resend_at_ms: u64) {
self.conn.set_handshake_msg1(msg1, first_resend_at_ms);
}
/// Store the wire-format msg2 for duplicate-msg1 resend on the surviving
/// carrier, mirroring the leg's responder write.
pub(crate) fn set_conn_handshake_msg2(&mut self, msg2: Vec<u8>) {
self.conn.set_handshake_msg2(msg2);
}
/// Peer session index of the surviving carrier — the source for the
/// promotion hand-off now that the leg no longer projects it.
pub(crate) fn conn_their_index(&self) -> Option<SessionIndex> {
self.conn.their_index()
}
/// Transport ID of the surviving carrier — the source for the promotion
/// hand-off, the stale-connection cleanup, and the msg1 resend send.
pub(crate) fn conn_transport_id(&self) -> Option<TransportId> {
self.conn.transport_id()
}
/// Link statistics of the surviving carrier — the seed copied into the
/// active peer at promotion.
pub(crate) fn conn_link_stats(&self) -> &LinkStats {
self.conn.link_stats()
}
/// Negotiated peer profile of the surviving carrier — the source for the
/// promotion hand-off now that the leg no longer projects it.
pub(crate) fn conn_peer_profile(&self) -> Option<NodeProfile> {
self.conn.peer_profile()
}
/// Record the negotiated peer profile on the surviving carrier, mirroring
/// the leg's `set_negotiation_results` write.
pub(crate) fn set_conn_peer_profile(&mut self, profile: NodeProfile) {
self.conn.set_negotiation_results(profile);
}
/// Record the peer session index on the surviving carrier. Seeds the
/// carrier from a pre-built leg (`Node::add_connection`) so the promotion
/// hand-off matches the establish paths that write it on the machine.
pub(crate) fn set_conn_their_index(&mut self, index: SessionIndex) {
self.conn.set_their_index(index);
}
/// Record the transport ID on the surviving carrier. Populated on the
/// inbound establish path (the leg seeds it at msg1, but the machine's
/// carrier is only written on the outbound dial) and when seeding the
/// carrier from a pre-built leg (`Node::add_connection`).
pub(crate) fn set_conn_transport_id(&mut self, id: TransportId) {
self.conn.set_transport_id(id);
}
/// Record our session index on the surviving carrier. The outbound establish
/// path allocates the index shell-side and writes it on the leg; this paired
/// write keeps the carrier the single index home once the leg dissolves (the
/// inbound path writes the carrier directly at authorize).
pub(crate) fn set_conn_our_index(&mut self, index: SessionIndex) {
self.conn.set_our_index(index);
}
/// Adopt an explicit connection-start timestamp on the carrier, so the
/// surviving state keeps the leg's start provenance rather than the
/// dial-time construction default.
pub(crate) fn set_conn_started_at(&mut self, started_at_ms: u64) {
self.conn.set_started_at(started_at_ms);
}
/// Advance the carrier's last-activity timestamp — the machine-tier paired
/// write for the leg's handshake `touch`.
pub(crate) fn touch_conn(&mut self, now_ms: u64) {
self.conn.touch(now_ms);
}
/// Whether this is an outbound leg parked at `SentMsg1` — the only state in
/// which a msg1 resend is due. Mirrors `on_handshake_retransmit`'s guard so
/// the shell timer driver can gate without reaching into machine state.
@@ -1004,7 +1120,6 @@ impl PeerMachine {
}
};
self.conn.set_our_index(our_index);
self.our_index = Some(our_index);
self.state = PeerState::Handshaking {
link,
phase: HandshakePhase::SentMsg2,
@@ -1023,7 +1138,7 @@ impl PeerMachine {
/// stays a driver mechanism (the stored msg2 bytes are replayed as-is), so
/// this method emits no `SendHandshake`. The ACL gate and the Noise finalize
/// already ran shell-side before this event; our index was allocated at msg1
/// (`self.our_index`).
/// (`self.conn.our_index()`).
pub(crate) fn inbound_msg3(
&mut self,
wire: WireOutcome,
@@ -1043,7 +1158,6 @@ impl PeerMachine {
// it to return the index. Without this seed those arms would emit nothing
// and leak the index.
self.conn.set_our_index(our_index);
self.our_index = Some(our_index);
let decision = Fmp::new().establish_inbound(&est, &wire);
let actions = match &decision {
@@ -1230,8 +1344,9 @@ impl PeerMachine {
}
PromotionResult::CrossConnectionLost { .. } => {
let mut actions = Vec::new();
if let Some(idx) = self.our_index.take() {
if let Some(idx) = self.conn.our_index() {
actions.push(PeerAction::FreeIndex { index: idx });
self.conn.clear_our_index();
}
self.state = PeerState::Failed {
reason: FailReason::HandshakeFailed,
@@ -1242,7 +1357,7 @@ impl PeerMachine {
}
fn register_current_index(&self) -> Vec<PeerAction> {
match self.our_index {
match self.conn.our_index() {
Some(idx) => vec![PeerAction::RegisterDecryptSession { index: idx }],
None => Vec::new(),
}
@@ -1329,9 +1444,13 @@ impl PeerMachine {
match act {
ConnAction::Cutover { peer } => {
// Initiator cutover: swap to the pending epoch, register the new
// index, open the drain window.
self.draining_index = self.our_index;
self.our_index = self.rekey_our_index.take();
// index, open the drain window. Slot-rotation mechanics stay in
// active.rs; the machine emits the action sequence.
self.draining_index = self.conn.our_index();
match self.rekey_our_index.take() {
Some(idx) => self.conn.set_our_index(idx),
None => self.conn.clear_our_index(),
}
self.rekey_in_progress = false;
self.state = PeerState::Maintaining {
addr: peer,
@@ -1340,7 +1459,7 @@ impl PeerMachine {
let mut actions = vec![PeerAction::SwapSendState {
epoch: self.remote_epoch.unwrap_or_default(),
}];
if let Some(idx) = self.our_index {
if let Some(idx) = self.conn.our_index() {
actions.push(PeerAction::RegisterDecryptSession { index: idx });
}
actions.push(PeerAction::SetTimer {
@@ -1594,8 +1713,9 @@ impl PeerMachine {
msg: disconnect_frame(reason),
}];
actions.push(PeerAction::InvalidateSendState);
if let Some(idx) = self.our_index.take() {
if let Some(idx) = self.conn.our_index() {
actions.push(PeerAction::UnregisterDecryptSession { index: idx });
self.conn.clear_our_index();
}
actions.push(PeerAction::TeardownConnectedUdp);
// No ReportLost on operator Requested.
@@ -1876,7 +1996,6 @@ mod tests {
phase: HandshakePhase::SentMsg1,
};
m2.conn.set_our_index(SessionIndex::new(0x2222)); // outbound index
m2.our_index = Some(SessionIndex::new(0x1111)); // old inbound index
let swap = m2.step(
PeerEvent::OutboundMsg2 {
their_index: SessionIndex::new(0x99),
@@ -1897,7 +2016,7 @@ mod tests {
PeerAction::FreeIndex { .. } | PeerAction::RegisterDecryptSession { .. }
)));
// The decision arm leaves the machine untouched: still Handshaking,
// our_index unchanged.
// the carrier's outbound index unchanged.
assert_eq!(
m2.state(),
PeerState::Handshaking {
@@ -1905,7 +2024,7 @@ mod tests {
phase: HandshakePhase::SentMsg1,
}
);
assert_eq!(m2.our_index(), Some(SessionIndex::new(0x1111)));
assert_eq!(m2.our_index(), Some(SessionIndex::new(0x2222)));
// Cross-connection KEEP: our outbound loses -> decision only; the
// shell's inline resolution frees the unused outbound index.
@@ -1941,7 +2060,7 @@ mod tests {
phase: HandshakePhase::SentMsg1,
}
);
assert_eq!(m3.our_index(), None);
assert_eq!(m3.our_index(), Some(SessionIndex::new(0x3333)));
}
// ---- Contract: actions are runtime-agnostic data ----------------------
@@ -2078,7 +2197,7 @@ mod tests {
kind: MaintainKind::Rekey(RekeyPhase::PendingCutover),
};
m.rekey_our_index = Some(SessionIndex::new(0x2222));
m.our_index = Some(SessionIndex::new(0x1111));
m.conn.set_our_index(SessionIndex::new(0x1111));
m.remote_epoch = Some([9u8; 8]);
m.session_established_at_ms = 0;
@@ -2176,7 +2295,6 @@ mod tests {
};
m.rekey_in_progress = true;
m.conn.set_our_index(SessionIndex::new(0x55));
m.our_index = Some(SessionIndex::new(0x55));
let mut est = est_new_peer(our);
est.has_existing_peer = true;
est.existing_peer_epoch = Some([1u8; 8]);
@@ -2224,7 +2342,6 @@ mod tests {
};
m.rekey_in_progress = true;
m.conn.set_our_index(SessionIndex::new(0x55));
m.our_index = Some(SessionIndex::new(0x55));
let mut est = est_new_peer(our);
est.has_existing_peer = true;
est.existing_peer_epoch = Some([1u8; 8]);
@@ -2711,29 +2828,29 @@ 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 7b: dial-persisted outbound promote keeps the carrier index ---
// An outbound machine persisted at DIAL carries the allocated index on its
// surviving carrier (the shell writes it at msg1 preparation). On promote via
// msg2 the machine keeps that index, so `our_index()` reads `Some(dial_index)`
// through Established. A subsequent inbound restart (peer restart, new epoch)
// still emits NO separate `UnregisterDecryptSession`: the restart teardown is
// the full `InvalidateSendState` (remove_active_peer), which owns the index
// cleanup, and the msg3 restart arm never reads the carrier index.
#[test]
fn dial_persisted_outbound_promote_no_restart_unregister() {
fn dial_persisted_outbound_promote_keeps_carrier_index() {
let mut alloc = IndexAllocator::new();
let peer = peer_identity();
let peer_addr = *peer.node_addr();
let our = *peer_identity().node_addr();
let dial_index = SessionIndex::new(0x55AA);
// Persisted at dial: Discovered, conn.our_index deliberately NOT set.
// Persisted at dial: Discovered, with the carrier index written by the
// shell at msg1 preparation.
let mut m = PeerMachine::new_outbound(LinkId::new(1), Some(peer), 0);
assert_eq!(m.our_index(), None);
m.conn.set_our_index(dial_index);
assert_eq!(m.our_index(), Some(dial_index));
// Promote via msg2 from Discovered (the production path — the former
// transient was likewise stepped from `new_outbound` without a state set).
// Promote via msg2 from Discovered.
let promote = m.step(
PeerEvent::OutboundMsg2 {
their_index: SessionIndex::new(0x77),
@@ -2759,13 +2876,9 @@ mod tests {
}
]
);
assert_eq!(
m.our_index(),
None,
"outbound promote must leave our_index unset"
);
assert_eq!(m.our_index(), Some(dial_index));
// Drive promotion to Established (from Discovered, as in production).
// Drive promotion to Established.
let _ = m.step(
PeerEvent::PromotionResolved {
result: PromotionResult::Promoted(peer_addr),
@@ -2774,7 +2887,7 @@ mod tests {
&mut alloc,
);
assert_eq!(m.state(), PeerState::Established { addr: peer_addr });
assert_eq!(m.our_index(), None);
assert_eq!(m.our_index(), Some(dial_index));
// A subsequent inbound restart (peer restart, new epoch) must NOT emit a
// separate UnregisterDecryptSession: the restart teardown is the full
@@ -2796,7 +2909,7 @@ mod tests {
!restart
.iter()
.any(|a| matches!(a, PeerAction::UnregisterDecryptSession { .. })),
"no UnregisterDecryptSession when the promoted outbound machine's our_index is None"
"no separate UnregisterDecryptSession: the msg3 restart teardown is InvalidateSendState"
);
assert!(
restart.iter().any(|a| matches!(
@@ -3079,7 +3192,7 @@ mod tests {
let addr = *id.node_addr();
let mut m = PeerMachine::new_outbound(LinkId::new(1), Some(id), 0);
m.state = PeerState::Active { addr };
m.our_index = Some(SessionIndex::new(0x4242));
m.conn.set_our_index(SessionIndex::new(0x4242));
let hb = m.step(PeerEvent::HeartbeatDue, 1_000, &mut alloc);
assert_eq!(
@@ -3127,7 +3240,7 @@ mod tests {
kind: MaintainKind::Rekey(RekeyPhase::PendingCutover),
};
m.rekey_our_index = Some(SessionIndex::new(0x2222));
m.our_index = Some(SessionIndex::new(0x1111));
m.conn.set_our_index(SessionIndex::new(0x1111));
m.remote_epoch = Some([9u8; 8]);
// Consume the shell-decided Cutover.

43
src/proto/fmp/state.rs
View File

@@ -98,9 +98,6 @@ pub struct ConnectionState {
/// Number of resends performed so far.
resend_count: u32,
/// When the next resend should fire (Unix ms). 0 = no resend scheduled.
next_resend_at_ms: u64,
}
impl ConnectionState {
@@ -128,7 +125,6 @@ impl ConnectionState {
handshake_msg1: None,
handshake_msg2: None,
resend_count: 0,
next_resend_at_ms: 0,
}
}
@@ -155,7 +151,6 @@ impl ConnectionState {
handshake_msg1: None,
handshake_msg2: None,
resend_count: 0,
next_resend_at_ms: 0,
}
}
@@ -180,7 +175,6 @@ impl ConnectionState {
handshake_msg1: None,
handshake_msg2: None,
resend_count: 0,
next_resend_at_ms: 0,
}
}
@@ -209,7 +203,6 @@ impl ConnectionState {
handshake_msg1: None,
handshake_msg2: None,
resend_count: 0,
next_resend_at_ms: 0,
}
}
@@ -265,11 +258,6 @@ impl ConnectionState {
&self.link_stats
}
/// Get mutable link statistics.
pub fn link_stats_mut(&mut self) -> &mut LinkStats {
&mut self.link_stats
}
// === Index Accessors ===
/// Get our session index (if set).
@@ -282,6 +270,11 @@ impl ConnectionState {
self.our_index = Some(index);
}
/// Clear our session index (back to unset).
pub fn clear_our_index(&mut self) {
self.our_index = None;
}
/// Get their session index (if known).
pub fn their_index(&self) -> Option<SessionIndex> {
self.their_index
@@ -347,11 +340,12 @@ impl ConnectionState {
// === Handshake Resend ===
/// Store the wire-format msg1 bytes for resend and schedule the first resend.
pub fn set_handshake_msg1(&mut self, msg1: Vec<u8>, first_resend_at_ms: u64) {
/// Store the wire-format msg1 bytes for resend and reset the resend counter.
/// The first-resend deadline is scheduled by the shell timer driver, not
/// tracked here.
pub fn set_handshake_msg1(&mut self, msg1: Vec<u8>, _first_resend_at_ms: u64) {
self.handshake_msg1 = Some(msg1);
self.resend_count = 0;
self.next_resend_at_ms = first_resend_at_ms;
}
/// Store the wire-format msg2 bytes for resend on duplicate msg1.
@@ -374,20 +368,21 @@ impl ConnectionState {
self.resend_count
}
/// When the next resend is scheduled (Unix ms).
#[cfg(test)]
pub fn next_resend_at_ms(&self) -> u64 {
self.next_resend_at_ms
}
/// Record a resend and schedule the next one.
pub fn record_resend(&mut self, next_resend_at_ms: u64) {
/// Record a resend. The next-resend deadline is scheduled by the shell
/// timer driver, not tracked here.
pub fn record_resend(&mut self, _next_resend_at_ms: u64) {
self.resend_count += 1;
self.next_resend_at_ms = next_resend_at_ms;
}
// === Activity / Timeout ===
/// Overwrite the connection-start timestamp. Used when the surviving
/// control-machine carrier adopts the leg's start provenance instead of the
/// machine-construction default.
pub fn set_started_at(&mut self, started_at_ms: u64) {
self.started_at = started_at_ms;
}
/// Update last activity timestamp.
pub fn touch(&mut self, current_time_ms: u64) {
self.last_activity = current_time_ms;

5
src/proto/fmp/tests/state.rs
View File

@@ -27,7 +27,6 @@ fn outbound_initializes_pure_fields() {
assert!(state.source_addr().is_none());
assert!(state.remote_epoch().is_none());
assert_eq!(state.resend_count(), 0);
assert_eq!(state.next_resend_at_ms(), 0);
}
#[test]
@@ -102,21 +101,17 @@ fn resend_bookkeeping() {
state.set_handshake_msg1(vec![1, 2, 3], 500);
assert_eq!(state.handshake_msg1(), Some(&[1u8, 2, 3][..]));
assert_eq!(state.resend_count(), 0);
assert_eq!(state.next_resend_at_ms(), 500);
state.record_resend(900);
assert_eq!(state.resend_count(), 1);
assert_eq!(state.next_resend_at_ms(), 900);
state.record_resend(1300);
assert_eq!(state.resend_count(), 2);
assert_eq!(state.next_resend_at_ms(), 1300);
// set_handshake_msg1 resets the resend counter.
state.set_handshake_msg1(vec![4, 5], 100);
assert_eq!(state.handshake_msg1(), Some(&[4u8, 5][..]));
assert_eq!(state.resend_count(), 0);
assert_eq!(state.next_resend_at_ms(), 100);
state.set_handshake_msg2(vec![6, 7, 8]);
assert_eq!(state.handshake_msg2(), Some(&[6u8, 7, 8][..]));