diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 31bab0f..785d0d0 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -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> { - // 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()); } diff --git a/src/node/handlers/timeout.rs b/src/node/handlers/timeout.rs index db43646..09ea623 100644 --- a/src/node/handlers/timeout.rs +++ b/src/node/handlers/timeout.rs @@ -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, diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs index 2f107f9..5a90c0c 100644 --- a/src/node/lifecycle/mod.rs +++ b/src/node/lifecycle/mod.rs @@ -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, }; diff --git a/src/node/tests/establish_chartests.rs b/src/node/tests/establish_chartests.rs index 02cae39..3acc53e 100644 --- a/src/node/tests/establish_chartests.rs +++ b/src/node/tests/establish_chartests.rs @@ -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(); diff --git a/src/node/tests/handshake.rs b/src/node/tests/handshake.rs index e54a397..e5438bc 100644 --- a/src/node/tests/handshake.rs +++ b/src/node/tests/handshake.rs @@ -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( diff --git a/src/peer/machine.rs b/src/peer/machine.rs index 3b84949..fa0801e 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -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, 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) { + 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.