node: source handshake resend buffers from the peer machine

The connection leg's stored msg1/msg2 handshake-resend buffers duplicated the
peer machine's own connection-state copy. Write the machine copy at the outbound
msg1-prep and the two inbound authorize sites, then source the retransmit
resend-bytes reader, send_stored_msg1, and the pending tier of find_stored_msg2
from the machine. The post-promote active-peer msg2 copy is written from the same
wire bytes, so the pending tier now matching after promotion is value-identical.
Byte-identical resend behavior.
This commit is contained in:
Johnathan Corgan
2026-07-18 16:00:15 +00:00
parent 4fc295d90a
commit 60b8acf716
6 changed files with 65 additions and 14 deletions
+14 -7
View File
@@ -627,7 +627,9 @@ impl Node {
self.links.insert(link_id, link);
self.addr_to_link.insert(addr_key, link_id);
let wire_msg2 = build_msg2(our_index, their_index, &msg2_payload);
conn.set_handshake_msg2(wire_msg2.clone());
// 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());
// Register the machine, carrying the connection
// (Promote/Restart tail only).
@@ -774,7 +776,9 @@ impl Node {
self.links.insert(link_id, link);
self.addr_to_link.insert(addr_key, link_id);
let wire_msg2 = build_msg2(our_index, their_index, &msg2_payload);
conn.set_handshake_msg2(wire_msg2.clone());
// 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());
// Register the machine, carrying the connection (Promote tail
// only — discarded on every reject/resend/rekey arm per the
@@ -829,12 +833,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());
}
+5 -1
View File
@@ -273,7 +273,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,
+11 -3
View File
@@ -626,9 +626,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
@@ -652,6 +653,9 @@ impl Node {
// round-trip separates dial from msg1 preparation.
machine.set_conn_started_at(current_time_ms);
machine.touch_conn(current_time_ms);
// 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(())
@@ -671,7 +675,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,
};
+7 -2
View File
@@ -192,10 +192,9 @@ async fn chartest_msg1_duplicate_pending_resends_stored_msg2() {
// A pending inbound connection with a stored msg2, keyed in addr_to_link,
// NOT promoted to an active peer.
let link_id = node.allocate_link_id();
let mut conn =
let conn =
PeerConnection::inbound_with_transport(link_id, transport_id, peer_addr.clone(), 1000);
let stored_msg2 = vec![0xC1, 0xC2, 0xC3, 0xC4, 0xC5];
conn.set_handshake_msg2(stored_msg2.clone());
let link = Link::connectionless(
link_id,
transport_id,
@@ -207,6 +206,12 @@ async fn chartest_msg1_duplicate_pending_resends_stored_msg2() {
node.addr_to_link
.insert((transport_id, peer_addr.clone()), link_id);
node.add_connection(conn).unwrap();
// The stored msg2 lives on the control machine's carrier (the resend source
// for a duplicate msg1 while pending), mirroring the inbound establish path.
node.peer_machines
.get_mut(&link_id)
.unwrap()
.set_conn_handshake_msg2(stored_msg2.clone());
assert_eq!(node.peer_count(), 0);
let before_pending = node.msg1_rate_limiter.pending_count();
+4 -1
View File
@@ -862,7 +862,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,
@@ -893,6 +893,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(
+24
View File
@@ -596,6 +596,30 @@ impl PeerMachine {
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);
}
/// 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.