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

node: source connection index/transport/stats from the peer machine

Browse Source 
The connection leg's their_index, transport_id and link_stats duplicated the peer
machine's own connection-state copy. Route every production reader through the
machine copy: promotion, the stale-connection reaper, the outbound msg1 resend, and
the in-use / connecting-path checks. Seed the inbound machine's transport_id from the
msg1 packet (the same source the leg used), matching the outbound dial seed, so the
hard-required promote read never sees a missing transport. link_stats is a
never-mutated zero seed, so its leg accessors are removed outright; the leg
their_index/transport_id getters/setters remain only for the test-only connection
builder. Also delete the write-only next_resend_at_ms field from the connection
state (the resend deadline is carried by the machine-armed retransmit timer, not this
field). Byte-identical.
This commit is contained in:
Johnathan Corgan
2026-07-18 16:43:22 +00:00
parent 60b8acf716
commit 6a80790742
8 changed files with 106 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/handlers/handshake.rs
+18 -8
View File
@@ -334,6 +334,11 @@ impl Node {
// variants complete the rate-limiter and return here. The local machine
// enters `peer_machines` only at the promote tails.
let mut machine = PeerMachine::new_inbound(link_id, packet.timestamp_ms);
// The inbound leg carries the transport ID from msg1, but the machine's
// carrier is only written on the outbound dial. Seed it here so the
// promotion hand-off reads it from the surviving carrier, matching the
// leg's inbound seed.
machine.set_conn_transport_id(packet.transport_id);
let (decision, actions) = machine.inbound_msg1(link_id, &wire, est, packet.timestamp_ms);
match decision {
InboundDecision::Reject {
@@ -616,7 +621,6 @@ impl Node {
// already freed by `remove_active_peer` above, BEFORE this fresh
// allocation — matching the pre-refactor allocation sequence.
conn.set_our_index(our_index);
conn.set_their_index(their_index);
let link = Link::connectionless(
link_id,
packet.transport_id,
@@ -765,7 +769,6 @@ impl Node {
// set indices on the shell connection, insert link / reverse map /
// connection, then build + store the framed msg2.
conn.set_our_index(our_index);
conn.set_their_index(their_index);
let link = Link::connectionless(
link_id,
packet.transport_id,
@@ -1021,7 +1024,6 @@ impl Node {
return;
}
conn.set_their_index(header.sender_idx);
conn.set_source_addr(packet.remote_addr.clone());
let peer_identity = match conn.expected_identity() {
@@ -1388,14 +1390,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(),
@@ -1407,7 +1413,11 @@ 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_node_addr = *verified_identity.node_addr();
src/node/handlers/timeout.rs
+13 -3
View File
@@ -124,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 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()));
}
let _ = self.index_allocator.free(idx);
@@ -310,7 +315,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,
},
src/node/lifecycle/mod.rs
+12 -7
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.node_addr() == peer_node_addr
&& pending.transport_id == transport_id
@@ -611,7 +613,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]
@@ -653,6 +654,10 @@ impl Node {
// 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);
// 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);
src/node/mod.rs
+21 -14
View File
@@ -2367,9 +2367,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()
@@ -2441,18 +2441,25 @@ impl Node {
});
}
self.peer_machines
.entry(link_id)
.or_insert_with(|| {
let now = connection.started_at();
match connection.expected_identity() {
Some(identity) if connection.is_outbound() => {
PeerMachine::new_outbound(link_id, *identity, now)
}
_ => PeerMachine::new_inbound(link_id, now),
let machine = self.peer_machines.entry(link_id).or_insert_with(|| {
let now = connection.started_at();
match connection.expected_identity() {
Some(identity) if connection.is_outbound() => {
PeerMachine::new_outbound(link_id, *identity, now)
}
})
.set_leg(connection);
_ => 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(())
}
src/peer/connection.rs
+1 -11
View File
@@ -10,7 +10,7 @@
use crate::PeerIdentity;
use crate::noise::{self, NoiseError, NoiseSession};
use crate::proto::fmp::ConnectionState;
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;
@@ -131,16 +131,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).
src/peer/machine.rs
+34 -1
View File
@@ -63,7 +63,7 @@ use crate::proto::fmp::{
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};
@@ -620,6 +620,39 @@ impl PeerMachine {
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()
}
/// 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);
}
/// 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.
src/proto/fmp/state.rs
+7 -23
View File
@@ -93,9 +93,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 {
@@ -122,7 +119,6 @@ impl ConnectionState {
handshake_msg1: None,
handshake_msg2: None,
resend_count: 0,
next_resend_at_ms: 0,
}
}
@@ -146,7 +142,6 @@ impl ConnectionState {
handshake_msg1: None,
handshake_msg2: None,
resend_count: 0,
next_resend_at_ms: 0,
}
}
@@ -174,7 +169,6 @@ impl ConnectionState {
handshake_msg1: None,
handshake_msg2: None,
resend_count: 0,
next_resend_at_ms: 0,
}
}
@@ -230,11 +224,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).
@@ -300,11 +289,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.
@@ -327,16 +317,10 @@ 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 ===
src/proto/fmp/tests/state.rs
-5
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][..]));