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

Forward-merge the per-peer FMP decomposition from the master line onto
the next line. The establishment and rekey machine-drive on refactor-node
is built on the IK establishment cores (establish_outbound, identity
learned at msg1) that the next line replaced with the XX msg3 model, so
it cannot be carried textually; it is re-derived against next's XX cores
in follow-on commits on this branch.

This merge commit carries only the two changes that are neutral and
next-compatible as-is:

- OwnedFd hygiene for the connected-UDP socket (RawFd -> OwnedFd, drop the
  hand-rolled unsafe Drop; OwnedFd's own Drop closes the descriptor).
- The two-tier send-state boundary: the send-critical subset of ActivePeer
  is regrouped into a PeerSendState struct, with accessors kept stable so
  no caller changes. next's XX establishment, rekey, and liveness-reap
  behavior is unchanged (verified: exact field partition, identical
  accessor signatures and semantics, identical init values, 3-arg mmp
  construction).

The per-peer machine, its action executor, and the peer_machines map are
not carried here; they are re-derived against next's XX establishment
surface in subsequent commits on this branch.

src/peer/active.rs

@@ -75,6 +75,125 @@ impl fmt::Display for ConnectivityState {
     }
 }
 
+/// Published active-send-state for a peer (the two-tier boundary).
+///
+/// This is the send-critical subset of an `ActivePeer` that the data plane
+/// reads (and, on roam/responder-cutover, writes) directly by plain borrow
+/// with no FSM dispatch: the three epoch slots (current / previous-draining /
+/// pending), the K-bit + session-relative time base, the transport target,
+/// the connected-UDP handles, and the hot per-packet counters. Grouping these
+/// draws the control/published-send-state boundary inside the peer entry.
+///
+/// Co-located, not behind `Arc`/`ArcSwap` — the data plane is not sharded, so
+/// the hot path reads this by plain borrow. Publishing behind `Arc`/`ArcSwap`
+/// is later-increment plumbing for a sharded data plane.
+///
+/// Like `ActivePeer`, this does not implement `Clone` because it contains
+/// `NoiseSession`, which cannot be safely cloned (cloning would risk nonce
+/// reuse, a catastrophic security failure).
+#[derive(Debug)]
+struct PeerSendState {
+    // === Current epoch slot ===
+    /// Noise session for encryption/decryption (None if legacy peer).
+    noise_session: Option<NoiseSession>,
+    /// Our session index (they include this when sending TO us).
+    our_index: Option<SessionIndex>,
+    /// Their session index (we include this when sending TO them).
+    their_index: Option<SessionIndex>,
+
+    // === Previous / draining epoch slot ===
+    /// Previous session kept alive during drain window after cutover.
+    previous_session: Option<NoiseSession>,
+    /// Previous session's our_index (for peers_by_index cleanup on drain expiry).
+    previous_our_index: Option<SessionIndex>,
+    /// When the drain window started (None = no drain in progress).
+    drain_started: Option<Instant>,
+
+    // === Pending epoch slot ===
+    /// Pending new session from completed rekey (before K-bit cutover).
+    pending_new_session: Option<NoiseSession>,
+    /// Pending new session's our_index.
+    pending_our_index: Option<SessionIndex>,
+    /// Pending new session's their_index.
+    pending_their_index: Option<SessionIndex>,
+
+    // === Epoch bit + session-relative time base ===
+    /// Current K-bit epoch value (alternates each rekey).
+    current_k_bit: bool,
+    /// Session start time for computing session-relative timestamps.
+    /// Used as the epoch for the 4-byte inner header timestamp field.
+    session_start: Instant,
+
+    // === Transport target ===
+    /// Transport ID for this peer's link.
+    transport_id: Option<TransportId>,
+    /// Current transport address (for roaming support).
+    current_addr: Option<TransportAddr>,
+    /// Link used to reach this peer.
+    link_id: LinkId,
+
+    // === Connected-UDP handles ===
+    /// Unix UDP fast-path: per-peer `connect()`-ed socket (paired with
+    /// the listen socket via `SO_REUSEPORT`). The kernel demux prefers
+    /// the connected 5-tuple, so inbound packets land here; the
+    /// encrypt-worker send path sends with `msg_name = NULL`, skipping
+    /// per-packet sockaddr handling + route lookup. Behind an `Arc` so
+    /// in-flight worker jobs survive rekey/address-change rotations.
+    #[cfg(any(target_os = "linux", target_os = "macos"))]
+    connected_udp: Option<std::sync::Arc<crate::peer::connected_udp::ConnectedPeerSocket>>,
+
+    /// Per-peer recv drain thread. Always paired with `connected_udp`:
+    /// the kernel routes inbound packets from this peer to the
+    /// connected socket, so it *must* be drained or the kernel recv
+    /// buffer fills. Drop signals shutdown via self-pipe.
+    #[cfg(any(target_os = "linux", target_os = "macos"))]
+    peer_recv_drain: Option<crate::peer::connected_udp::PeerRecvDrain>,
+
+    // === Hot counters ===
+    /// Link statistics.
+    link_stats: LinkStats,
+    /// When this peer was last seen (any activity, Unix milliseconds).
+    last_seen: u64,
+    /// Number of replay detections suppressed since last session reset.
+    replay_suppressed_count: u32,
+    /// Consecutive decryption failures (reset on any successful decrypt).
+    consecutive_decrypt_failures: u32,
+    /// Per-peer MMP state (None for legacy peers without Noise sessions).
+    mmp: Option<MmpPeerState>,
+}
+
+impl PeerSendState {
+    /// Empty send-state for a peer with no Noise session yet. Mirrors the
+    /// send-critical portion of `ActivePeer::new`.
+    fn new(link_id: LinkId, session_start: Instant, last_seen: u64) -> Self {
+        Self {
+            noise_session: None,
+            our_index: None,
+            their_index: None,
+            previous_session: None,
+            previous_our_index: None,
+            drain_started: None,
+            pending_new_session: None,
+            pending_our_index: None,
+            pending_their_index: None,
+            current_k_bit: false,
+            session_start,
+            transport_id: None,
+            current_addr: None,
+            link_id,
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
+            connected_udp: None,
+            #[cfg(any(target_os = "linux", target_os = "macos"))]
+            peer_recv_drain: None,
+            link_stats: LinkStats::new(),
+            last_seen,
+            replay_suppressed_count: 0,
+            consecutive_decrypt_failures: 0,
+            mmp: None,
+        }
+    }
+}
+
 /// A fully authenticated remote FIPS node.
 ///
 /// Created only after successful Noise KK handshake. The identity is
@@ -90,23 +209,9 @@ pub struct ActivePeer {
     identity: PeerIdentity,
 
     // === Connection ===
-    /// Link used to reach this peer.
-    link_id: LinkId,
     /// Current connectivity state.
     connectivity: ConnectivityState,
 
-    // === Session (Wire Protocol) ===
-    /// Noise session for encryption/decryption (None if legacy peer).
-    noise_session: Option<NoiseSession>,
-    /// Our session index (they include this when sending TO us).
-    our_index: Option<SessionIndex>,
-    /// Their session index (we include this when sending TO them).
-    their_index: Option<SessionIndex>,
-    /// Transport ID for this peer's link.
-    transport_id: Option<TransportId>,
-    /// Current transport address (for roaming support).
-    current_addr: Option<TransportAddr>,
-
     // === Spanning Tree ===
     /// Their latest parent declaration.
     declaration: Option<ParentDeclaration>,
@@ -131,18 +236,9 @@ pub struct ActivePeer {
     /// Whether we owe them a filter update.
     pending_filter_update: bool,
 
-    // === Timing ===
-    /// Session start time for computing session-relative timestamps.
-    /// Used as the epoch for the 4-byte inner header timestamp field.
-    session_start: Instant,
-
     // === Statistics ===
-    /// Link statistics.
-    link_stats: LinkStats,
     /// When this peer was authenticated (Unix milliseconds).
     authenticated_at: u64,
-    /// When this peer was last seen (any activity, Unix milliseconds).
-    last_seen: u64,
 
     // === Epoch (Restart Detection) ===
     /// Remote peer's startup epoch (from handshake). Used to detect restarts.
@@ -156,10 +252,6 @@ pub struct ActivePeer {
     /// Whether to send receiver reports to this peer (our provides_rr AND peer wants_rr).
     send_rr: bool,
 
-    // === MMP ===
-    /// Per-peer MMP state (None for legacy peers without Noise sessions).
-    mmp: Option<MmpPeerState>,
-
     // === Heartbeat ===
     /// When we last sent a heartbeat to this peer.
     last_heartbeat_sent: Option<Instant>,
@@ -169,12 +261,6 @@ pub struct ActivePeer {
     /// Cleared after the handshake timeout window.
     handshake_msg2: Option<Vec<u8>>,
 
-    // === Replay Detection Suppression ===
-    /// Number of replay detections suppressed since last session reset.
-    replay_suppressed_count: u32,
-    /// Consecutive decryption failures (reset on any successful decrypt).
-    consecutive_decrypt_failures: u32,
-
     // === Rekey (Key Rotation) ===
     /// When the current Noise session was established (for rekey timer).
     session_established_at: Instant,
@@ -184,20 +270,6 @@ pub struct ActivePeer {
     /// dual-initiation in symmetric-start meshes; mean interval is
     /// preserved.
     rekey_jitter_secs: i64,
-    /// Current K-bit epoch value (alternates each rekey).
-    current_k_bit: bool,
-    /// Previous session kept alive during drain window after cutover.
-    previous_session: Option<NoiseSession>,
-    /// Previous session's our_index (for peers_by_index cleanup on drain expiry).
-    previous_our_index: Option<SessionIndex>,
-    /// When the drain window started (None = no drain in progress).
-    drain_started: Option<Instant>,
-    /// Pending new session from completed rekey (before K-bit cutover).
-    pending_new_session: Option<NoiseSession>,
-    /// Pending new session's our_index.
-    pending_our_index: Option<SessionIndex>,
-    /// Pending new session's their_index.
-    pending_their_index: Option<SessionIndex>,
     /// Whether a rekey is currently in progress (handshake sent, not yet complete).
     rekey_in_progress: bool,
     /// When we last received a rekey msg1 from this peer (dampening).
@@ -232,21 +304,10 @@ pub struct ActivePeer {
     /// Number of msg3 retransmissions performed this rekey cycle.
     rekey_msg3_resend_count: u32,
 
-    /// Unix UDP fast-path: per-peer `connect()`-ed socket (paired with
-    /// the listen socket via `SO_REUSEPORT`). The kernel demux prefers
-    /// the connected 5-tuple, so inbound packets land here; the
-    /// encrypt-worker send path sends with `msg_name = NULL`, skipping
-    /// per-packet sockaddr handling + route lookup. Behind an `Arc` so
-    /// in-flight worker jobs survive rekey/address-change rotations.
-    #[cfg(any(target_os = "linux", target_os = "macos"))]
-    connected_udp: Option<std::sync::Arc<crate::peer::connected_udp::ConnectedPeerSocket>>,
-
-    /// Per-peer recv drain thread. Always paired with `connected_udp`:
-    /// the kernel routes inbound packets from this peer to the
-    /// connected socket, so it *must* be drained or the kernel recv
-    /// buffer fills. Drop signals shutdown via self-pipe.
-    #[cfg(any(target_os = "linux", target_os = "macos"))]
-    peer_recv_drain: Option<crate::peer::connected_udp::PeerRecvDrain>,
+    // === Published active-send-state (two-tier boundary) ===
+    /// The send-critical subset read (and, on roam/responder-cutover, written)
+    /// directly by the data plane. See `PeerSendState`.
+    send: PeerSendState,
 }
 
 impl ActivePeer {
@@ -258,13 +319,7 @@ impl ActivePeer {
         let now = Instant::now();
         Self {
             identity,
-            link_id,
             connectivity: ConnectivityState::Connected,
-            noise_session: None,
-            our_index: None,
-            their_index: None,
-            transport_id: None,
-            current_addr: None,
             declaration: None,
             ancestry: None,
             tree_announce_min_interval_ms: 500,
@@ -274,28 +329,15 @@ impl ActivePeer {
             filter_sequence: 0,
             filter_received_at: 0,
             pending_filter_update: true, // Send filter on new connection
-            session_start: now,
-            link_stats: LinkStats::new(),
             authenticated_at,
-            last_seen: authenticated_at,
             remote_epoch: None,
             peer_profile: NodeProfile::Full,
             send_sr: true,
             send_rr: true,
-            mmp: None,
             last_heartbeat_sent: None,
             handshake_msg2: None,
-            replay_suppressed_count: 0,
-            consecutive_decrypt_failures: 0,
             session_established_at: now,
             rekey_jitter_secs: draw_rekey_jitter(),
-            current_k_bit: false,
-            previous_session: None,
-            previous_our_index: None,
-            drain_started: None,
-            pending_new_session: None,
-            pending_our_index: None,
-            pending_their_index: None,
             rekey_in_progress: false,
             last_peer_rekey: None,
             rekey_handshake: None,
@@ -308,10 +350,7 @@ impl ActivePeer {
             rekey_msg3_payload: None,
             rekey_msg3_next_resend_ms: 0,
             rekey_msg3_resend_count: 0,
-            #[cfg(any(target_os = "linux", target_os = "macos"))]
-            connected_udp: None,
-            #[cfg(any(target_os = "linux", target_os = "macos"))]
-            peer_recv_drain: None,
+            send: PeerSendState::new(link_id, now, authenticated_at),
         }
     }
 
@@ -326,7 +365,7 @@ impl ActivePeer {
         link_stats: LinkStats,
     ) -> Self {
         let mut peer = Self::new(identity, link_id, authenticated_at);
-        peer.link_stats = link_stats;
+        peer.send.link_stats = link_stats;
         peer
     }
 
@@ -358,15 +397,21 @@ impl ActivePeer {
         let send_rr = our_neg.provides_rr() && their_neg.wants_rr();
 
         let now = Instant::now();
+        let mut send = PeerSendState::new(link_id, now, authenticated_at);
+        send.noise_session = Some(noise_session);
+        send.our_index = Some(our_index);
+        send.their_index = Some(their_index);
+        send.transport_id = Some(transport_id);
+        send.current_addr = Some(current_addr);
+        send.link_stats = link_stats;
+        send.mmp = Some(MmpPeerState::new(
+            mmp_config.mode,
+            mmp_config.log_interval_secs,
+            mmp_config.owd_window_size,
+        ));
         Self {
             identity,
-            link_id,
             connectivity: ConnectivityState::Connected,
-            noise_session: Some(noise_session),
-            our_index: Some(our_index),
-            their_index: Some(their_index),
-            transport_id: Some(transport_id),
-            current_addr: Some(current_addr),
             declaration: None,
             ancestry: None,
             tree_announce_min_interval_ms: 500,
@@ -376,32 +421,15 @@ impl ActivePeer {
             filter_sequence: 0,
             filter_received_at: 0,
             pending_filter_update: true,
-            session_start: now,
-            link_stats,
             authenticated_at,
-            last_seen: authenticated_at,
             remote_epoch,
             peer_profile,
             send_sr,
             send_rr,
-            mmp: Some(MmpPeerState::new(
-                mmp_config.mode,
-                mmp_config.log_interval_secs,
-                mmp_config.owd_window_size,
-            )),
             last_heartbeat_sent: None,
             handshake_msg2: None,
-            replay_suppressed_count: 0,
-            consecutive_decrypt_failures: 0,
             session_established_at: now,
             rekey_jitter_secs: draw_rekey_jitter(),
-            current_k_bit: false,
-            previous_session: None,
-            previous_our_index: None,
-            drain_started: None,
-            pending_new_session: None,
-            pending_our_index: None,
-            pending_their_index: None,
             rekey_in_progress: false,
             last_peer_rekey: None,
             rekey_handshake: None,
@@ -414,10 +442,7 @@ impl ActivePeer {
             rekey_msg3_payload: None,
             rekey_msg3_next_resend_ms: 0,
             rekey_msg3_resend_count: 0,
-            #[cfg(any(target_os = "linux", target_os = "macos"))]
-            connected_udp: None,
-            #[cfg(any(target_os = "linux", target_os = "macos"))]
-            peer_recv_drain: None,
+            send,
         }
     }
 
@@ -430,7 +455,7 @@ impl ActivePeer {
     pub(crate) fn connected_udp(
         &self,
     ) -> Option<std::sync::Arc<crate::peer::connected_udp::ConnectedPeerSocket>> {
-        self.connected_udp.clone()
+        self.send.connected_udp.clone()
     }
 
     /// Install a per-peer `connect()`-ed UDP socket with its paired
@@ -444,10 +469,10 @@ impl ActivePeer {
     ) {
         // Drop the old drain BEFORE the old socket so its last fd
         // reference is released cleanly.
-        self.peer_recv_drain = None;
-        self.connected_udp = None;
-        self.connected_udp = Some(socket);
-        self.peer_recv_drain = Some(drain);
+        self.send.peer_recv_drain = None;
+        self.send.connected_udp = None;
+        self.send.connected_udp = Some(socket);
+        self.send.peer_recv_drain = Some(drain);
     }
 
     /// Clear the per-peer connected UDP socket + drain. The drain
@@ -457,8 +482,8 @@ impl ActivePeer {
     #[cfg(any(target_os = "linux", target_os = "macos"))]
     #[allow(dead_code)] // called from session-deregister + rekey follow-up
     pub(crate) fn clear_connected_udp(&mut self) {
-        self.peer_recv_drain = None;
-        self.connected_udp = None;
+        self.send.peer_recv_drain = None;
+        self.send.connected_udp = None;
     }
 
     // === Identity Accessors ===
@@ -492,7 +517,7 @@ impl ActivePeer {
 
     /// Get the link ID.
     pub fn link_id(&self) -> LinkId {
-        self.link_id
+        self.send.link_id
     }
 
     /// Get the connectivity state.
@@ -519,34 +544,34 @@ impl ActivePeer {
 
     /// Check if this peer has a Noise session.
     pub fn has_session(&self) -> bool {
-        self.noise_session.is_some()
+        self.send.noise_session.is_some()
     }
 
     /// Get the Noise session, if present.
     pub fn noise_session(&self) -> Option<&NoiseSession> {
-        self.noise_session.as_ref()
+        self.send.noise_session.as_ref()
     }
 
     /// Get mutable access to the Noise session.
     pub fn noise_session_mut(&mut self) -> Option<&mut NoiseSession> {
-        self.noise_session.as_mut()
+        self.send.noise_session.as_mut()
     }
 
     /// Get our session index (they use this to send TO us).
     pub fn our_index(&self) -> Option<SessionIndex> {
-        self.our_index
+        self.send.our_index
     }
 
     /// Get their session index (we use this to send TO them).
     pub fn their_index(&self) -> Option<SessionIndex> {
-        self.their_index
+        self.send.their_index
     }
 
     /// Update their session index (used during cross-connection resolution
     /// when the losing node keeps its inbound session but needs the peer's
     /// outbound index).
     pub fn set_their_index(&mut self, index: SessionIndex) {
-        self.their_index = Some(index);
+        self.send.their_index = Some(index);
     }
 
     /// Replace the Noise session and indices during cross-connection resolution.
@@ -565,21 +590,21 @@ impl ActivePeer {
         new_their_index: SessionIndex,
     ) -> Option<SessionIndex> {
         self.reset_replay_suppressed();
-        let old_our_index = self.our_index;
-        self.noise_session = Some(new_session);
-        self.our_index = Some(new_our_index);
-        self.their_index = Some(new_their_index);
+        let old_our_index = self.send.our_index;
+        self.send.noise_session = Some(new_session);
+        self.send.our_index = Some(new_our_index);
+        self.send.their_index = Some(new_their_index);
         old_our_index
     }
 
     /// Get the transport ID for this peer.
     pub fn transport_id(&self) -> Option<TransportId> {
-        self.transport_id
+        self.send.transport_id
     }
 
     /// Get the current transport address.
     pub fn current_addr(&self) -> Option<&TransportAddr> {
-        self.current_addr.as_ref()
+        self.send.current_addr.as_ref()
     }
 
     /// Update the current address (for roaming support).
@@ -589,10 +614,10 @@ impl ActivePeer {
     /// use this to invalidate per-peer `connect(2)`-ed UDP sockets whose
     /// 5-tuple just went stale.
     pub fn set_current_addr(&mut self, transport_id: TransportId, addr: TransportAddr) -> bool {
-        let changed =
-            self.transport_id != Some(transport_id) || self.current_addr.as_ref() != Some(&addr);
-        self.transport_id = Some(transport_id);
-        self.current_addr = Some(addr);
+        let changed = self.send.transport_id != Some(transport_id)
+            || self.send.current_addr.as_ref() != Some(&addr);
+        self.send.transport_id = Some(transport_id);
+        self.send.current_addr = Some(addr);
         changed
     }
 
@@ -617,38 +642,38 @@ impl ActivePeer {
 
     /// Increment replay suppression counter. Returns the new count.
     pub fn increment_replay_suppressed(&mut self) -> u32 {
-        self.replay_suppressed_count += 1;
-        self.replay_suppressed_count
+        self.send.replay_suppressed_count += 1;
+        self.send.replay_suppressed_count
     }
 
     /// Reset replay suppression counter, returning previous count.
     pub fn reset_replay_suppressed(&mut self) -> u32 {
-        let count = self.replay_suppressed_count;
-        self.replay_suppressed_count = 0;
+        let count = self.send.replay_suppressed_count;
+        self.send.replay_suppressed_count = 0;
         count
     }
 
     /// Current replay suppression count.
     pub fn replay_suppressed_count(&self) -> u32 {
-        self.replay_suppressed_count
+        self.send.replay_suppressed_count
     }
 
     // === Decryption Failure Tracking ===
 
     /// Increment consecutive decryption failure counter, returning new count.
     pub fn increment_decrypt_failures(&mut self) -> u32 {
-        self.consecutive_decrypt_failures += 1;
-        self.consecutive_decrypt_failures
+        self.send.consecutive_decrypt_failures += 1;
+        self.send.consecutive_decrypt_failures
     }
 
     /// Reset consecutive decryption failure counter.
     pub fn reset_decrypt_failures(&mut self) {
-        self.consecutive_decrypt_failures = 0;
+        self.send.consecutive_decrypt_failures = 0;
     }
 
     /// Current consecutive decryption failure count.
     pub fn consecutive_decrypt_failures(&self) -> u32 {
-        self.consecutive_decrypt_failures
+        self.send.consecutive_decrypt_failures
     }
 
     // === Epoch Accessors ===
@@ -736,24 +761,24 @@ impl ActivePeer {
 
     /// Get link statistics.
     pub fn link_stats(&self) -> &LinkStats {
-        &self.link_stats
+        &self.send.link_stats
     }
 
     /// Get mutable link statistics.
     pub fn link_stats_mut(&mut self) -> &mut LinkStats {
-        &mut self.link_stats
+        &mut self.send.link_stats
     }
 
     // === MMP Accessors ===
 
     /// Get MMP state (None for legacy peers without sessions).
     pub fn mmp(&self) -> Option<&MmpPeerState> {
-        self.mmp.as_ref()
+        self.send.mmp.as_ref()
     }
 
     /// Get mutable MMP state.
     pub fn mmp_mut(&mut self) -> Option<&mut MmpPeerState> {
-        self.mmp.as_mut()
+        self.send.mmp.as_mut()
     }
 
     /// Link cost for routing decisions.
@@ -789,12 +814,12 @@ impl ActivePeer {
 
     /// When this peer was last seen.
     pub fn last_seen(&self) -> u64 {
-        self.last_seen
+        self.send.last_seen
     }
 
     /// Time since last activity.
     pub fn idle_time(&self, current_time_ms: u64) -> u64 {
-        current_time_ms.saturating_sub(self.last_seen)
+        current_time_ms.saturating_sub(self.send.last_seen)
     }
 
     /// Connection duration since authentication.
@@ -807,12 +832,12 @@ impl ActivePeer {
     /// Returns milliseconds since session establishment, truncated to u32.
     /// Wraps at ~49.7 days which is acceptable for session-relative timing.
     pub fn session_elapsed_ms(&self) -> u32 {
-        self.session_start.elapsed().as_millis() as u32
+        self.send.session_start.elapsed().as_millis() as u32
     }
 
     /// When this peer's session started (for link-dead fallback timing).
     pub fn session_start(&self) -> Instant {
-        self.session_start
+        self.send.session_start
     }
 
     // === Heartbeat ===
@@ -831,7 +856,7 @@ impl ActivePeer {
 
     /// Update last seen timestamp.
     pub fn touch(&mut self, current_time_ms: u64) {
-        self.last_seen = current_time_ms;
+        self.send.last_seen = current_time_ms;
         // If we were stale, receiving traffic makes us connected again
         if self.connectivity == ConnectivityState::Stale {
             self.connectivity = ConnectivityState::Connected;
@@ -858,12 +883,12 @@ impl ActivePeer {
     /// Mark peer as connected (e.g., after successful reconnect).
     pub fn mark_connected(&mut self, current_time_ms: u64) {
         self.connectivity = ConnectivityState::Connected;
-        self.last_seen = current_time_ms;
+        self.send.last_seen = current_time_ms;
     }
 
     /// Update the link ID (e.g., on reconnect).
     pub fn set_link_id(&mut self, link_id: LinkId) {
-        self.link_id = link_id;
+        self.send.link_id = link_id;
     }
 
     // === Tree Updates ===
@@ -877,7 +902,7 @@ impl ActivePeer {
     ) {
         self.declaration = Some(declaration);
         self.ancestry = Some(ancestry);
-        self.last_seen = current_time_ms;
+        self.send.last_seen = current_time_ms;
     }
 
     /// Clear peer's tree position.
@@ -931,7 +956,7 @@ impl ActivePeer {
         self.inbound_filter = Some(filter);
         self.filter_sequence = sequence;
         self.filter_received_at = current_time_ms;
-        self.last_seen = current_time_ms;
+        self.send.last_seen = current_time_ms;
     }
 
     /// Clear peer's inbound filter.
@@ -969,7 +994,7 @@ impl ActivePeer {
             mode,
             ..MmpConfig::default()
         };
-        self.mmp = Some(MmpPeerState::new(
+        self.send.mmp = Some(MmpPeerState::new(
             config.mode,
             config.log_interval_secs,
             config.owd_window_size,
@@ -983,7 +1008,8 @@ impl ActivePeer {
     /// is compiled out of release builds.
     #[cfg(test)]
     pub(crate) fn test_backdate_session_start(&mut self, age: std::time::Duration) {
-        self.session_start = self
+        self.send.session_start = self
+            .send
             .session_start
             .checked_sub(age)
             .unwrap_or_else(Instant::now);
@@ -1001,7 +1027,7 @@ impl ActivePeer {
 
     /// Current K-bit epoch value.
     pub fn current_k_bit(&self) -> bool {
-        self.current_k_bit
+        self.send.current_k_bit
     }
 
     /// Whether a rekey is currently in progress.
@@ -1029,38 +1055,38 @@ impl ActivePeer {
 
     /// Get the pending new session's our_index.
     pub fn pending_our_index(&self) -> Option<SessionIndex> {
-        self.pending_our_index
+        self.send.pending_our_index
     }
 
     /// Get the pending new session's their_index.
     pub fn pending_their_index(&self) -> Option<SessionIndex> {
-        self.pending_their_index
+        self.send.pending_their_index
     }
 
     /// Get the previous session's our_index (during drain).
     pub fn previous_our_index(&self) -> Option<SessionIndex> {
-        self.previous_our_index
+        self.send.previous_our_index
     }
 
     /// Get the previous session for decryption fallback.
     pub fn previous_session(&self) -> Option<&NoiseSession> {
-        self.previous_session.as_ref()
+        self.send.previous_session.as_ref()
     }
 
     /// Get mutable access to the previous session for decryption.
     pub fn previous_session_mut(&mut self) -> Option<&mut NoiseSession> {
-        self.previous_session.as_mut()
+        self.send.previous_session.as_mut()
     }
 
     /// Get the pending new session (completed rekey, not yet cut over).
     pub fn pending_new_session(&self) -> Option<&NoiseSession> {
-        self.pending_new_session.as_ref()
+        self.send.pending_new_session.as_ref()
     }
 
     /// Mutable access to the pending new session, for trial-decrypt of an
     /// inbound frame before promoting it on a peer K-bit flip.
     pub fn pending_new_session_mut(&mut self) -> Option<&mut NoiseSession> {
-        self.pending_new_session.as_mut()
+        self.send.pending_new_session.as_mut()
     }
 
     /// Store a completed rekey session and its indices.
@@ -1073,9 +1099,9 @@ impl ActivePeer {
         our_index: SessionIndex,
         their_index: SessionIndex,
     ) {
-        self.pending_new_session = Some(session);
-        self.pending_our_index = Some(our_index);
-        self.pending_their_index = Some(their_index);
+        self.send.pending_new_session = Some(session);
+        self.send.pending_our_index = Some(our_index);
+        self.send.pending_their_index = Some(their_index);
         self.rekey_in_progress = false;
         // Clear initiator handshake state (index now lives in pending_our_index)
         self.rekey_our_index = None;
@@ -1091,24 +1117,24 @@ impl ActivePeer {
     /// flips the K-bit. Returns the old our_index that should remain in peers_by_index
     /// during the drain window.
     pub fn cutover_to_new_session(&mut self) -> Option<SessionIndex> {
-        let new_session = self.pending_new_session.take()?;
-        let new_our_index = self.pending_our_index.take();
-        let new_their_index = self.pending_their_index.take();
+        let new_session = self.send.pending_new_session.take()?;
+        let new_our_index = self.send.pending_our_index.take();
+        let new_their_index = self.send.pending_their_index.take();
 
         // Demote current to previous
-        self.previous_session = self.noise_session.take();
-        self.previous_our_index = self.our_index;
-        self.drain_started = Some(Instant::now());
+        self.send.previous_session = self.send.noise_session.take();
+        self.send.previous_our_index = self.send.our_index;
+        self.send.drain_started = Some(Instant::now());
 
         // Promote pending to current
-        self.noise_session = Some(new_session);
-        self.our_index = new_our_index;
-        self.their_index = new_their_index;
+        self.send.noise_session = Some(new_session);
+        self.send.our_index = new_our_index;
+        self.send.their_index = new_their_index;
 
         // Flip K-bit and reset timing
-        self.current_k_bit = !self.current_k_bit;
+        self.send.current_k_bit = !self.send.current_k_bit;
         self.session_established_at = Instant::now();
-        self.session_start = Instant::now();
+        self.send.session_start = Instant::now();
         self.rekey_in_progress = false;
         self.rekey_msg1_resend_count = 0;
         self.rekey_jitter_secs = draw_rekey_jitter();
@@ -1116,11 +1142,11 @@ impl ActivePeer {
 
         // Reset MMP counters to avoid metric discontinuity
         let now_ms = crate::time::mono_ms();
-        if let Some(mmp) = &mut self.mmp {
+        if let Some(mmp) = &mut self.send.mmp {
             mmp.reset_for_rekey(now_ms);
         }
 
-        self.previous_our_index
+        self.send.previous_our_index
     }
 
     /// Handle receiving a K-bit flip from the peer (responder side).
@@ -1128,24 +1154,24 @@ impl ActivePeer {
     /// Promotes pending_new_session to current, demotes current to previous.
     /// Returns the old our_index for drain tracking.
     pub fn handle_peer_kbit_flip(&mut self) -> Option<SessionIndex> {
-        let new_session = self.pending_new_session.take()?;
-        let new_our_index = self.pending_our_index.take();
-        let new_their_index = self.pending_their_index.take();
+        let new_session = self.send.pending_new_session.take()?;
+        let new_our_index = self.send.pending_our_index.take();
+        let new_their_index = self.send.pending_their_index.take();
 
         // Demote current to previous
-        self.previous_session = self.noise_session.take();
-        self.previous_our_index = self.our_index;
-        self.drain_started = Some(Instant::now());
+        self.send.previous_session = self.send.noise_session.take();
+        self.send.previous_our_index = self.send.our_index;
+        self.send.drain_started = Some(Instant::now());
 
         // Promote pending to current
-        self.noise_session = Some(new_session);
-        self.our_index = new_our_index;
-        self.their_index = new_their_index;
+        self.send.noise_session = Some(new_session);
+        self.send.our_index = new_our_index;
+        self.send.their_index = new_their_index;
 
         // Match peer's K-bit
-        self.current_k_bit = !self.current_k_bit;
+        self.send.current_k_bit = !self.send.current_k_bit;
         self.session_established_at = Instant::now();
-        self.session_start = Instant::now();
+        self.send.session_start = Instant::now();
         self.rekey_in_progress = false;
         self.rekey_msg1_resend_count = 0;
         self.rekey_jitter_secs = draw_rekey_jitter();
@@ -1153,16 +1179,16 @@ impl ActivePeer {
 
         // Reset MMP counters to avoid metric discontinuity
         let now_ms = crate::time::mono_ms();
-        if let Some(mmp) = &mut self.mmp {
+        if let Some(mmp) = &mut self.send.mmp {
             mmp.reset_for_rekey(now_ms);
         }
 
-        self.previous_our_index
+        self.send.previous_our_index
     }
 
     /// Check if the drain window has expired.
     pub fn drain_expired(&self, drain_secs: u64) -> bool {
-        match self.drain_started {
+        match self.send.drain_started {
             Some(t) => t.elapsed().as_secs() >= drain_secs,
             None => false,
         }
@@ -1170,7 +1196,7 @@ impl ActivePeer {
 
     /// Whether a drain is in progress.
     pub fn is_draining(&self) -> bool {
-        self.drain_started.is_some()
+        self.send.drain_started.is_some()
     }
 
     /// Complete the drain: drop previous session and free its index.
@@ -1178,9 +1204,9 @@ impl ActivePeer {
     /// Returns the previous our_index so the caller can remove it from
     /// peers_by_index and free it from the IndexAllocator.
     pub fn complete_drain(&mut self) -> Option<SessionIndex> {
-        self.previous_session = None;
-        self.drain_started = None;
-        self.previous_our_index.take()
+        self.send.previous_session = None;
+        self.send.drain_started = None;
+        self.send.previous_our_index.take()
     }
 
     /// Abandon an in-progress rekey.
@@ -1197,9 +1223,9 @@ impl ActivePeer {
         self.clear_rekey_msg3_payload();
         // Return whichever index needs freeing
         self.rekey_our_index.take().or_else(|| {
-            self.pending_new_session = None;
-            self.pending_their_index = None;
-            self.pending_our_index.take()
+            self.send.pending_new_session = None;
+            self.send.pending_their_index = None;
+            self.send.pending_our_index.take()
         })
     }
 

src/peer/connected_udp/socket.rs

@@ -6,7 +6,7 @@
 #![allow(dead_code)]
 
 use std::net::SocketAddr;
-use std::os::unix::io::{AsRawFd, IntoRawFd, OwnedFd, RawFd};
+use std::os::unix::io::{AsRawFd, OwnedFd, RawFd};
 
 /// A `connect()`-ed UDP socket for one established peer.
 ///
@@ -26,7 +26,7 @@ use std::os::unix::io::{AsRawFd, IntoRawFd, OwnedFd, RawFd};
 ///   needs to be redone on the data path.
 #[derive(Debug)]
 pub(crate) struct ConnectedPeerSocket {
-    fd: RawFd,
+    fd: OwnedFd,
     peer_addr: SocketAddr,
     local_addr: SocketAddr,
 }
@@ -34,10 +34,10 @@ pub(crate) struct ConnectedPeerSocket {
 impl ConnectedPeerSocket {
     /// Adopt an already-opened, bound, and `connect()`-ed fd (from
     /// `crate::transport::udp::open_connected_fd`) into an owning
-    /// handle. Takes ownership of the fd; it is closed on drop.
+    /// handle. Takes ownership of the fd; the `OwnedFd` closes it on drop.
     pub(crate) fn from_fd(fd: OwnedFd, peer_addr: SocketAddr, local_addr: SocketAddr) -> Self {
         Self {
-            fd: fd.into_raw_fd(),
+            fd,
             peer_addr,
             local_addr,
         }
@@ -55,18 +55,7 @@ impl ConnectedPeerSocket {
 
 impl AsRawFd for ConnectedPeerSocket {
     fn as_raw_fd(&self) -> RawFd {
-        self.fd
-    }
-}
-
-impl Drop for ConnectedPeerSocket {
-    fn drop(&mut self) {
-        // Best-effort close. Ignore the result — if close fails the
-        // kernel has already done what it can; we don't want to panic
-        // in Drop.
-        unsafe {
-            libc::close(self.fd);
-        }
+        self.fd.as_raw_fd()
     }
 }