diff --git a/src/peer/active.rs b/src/peer/active.rs index e49c60e..bd4ce29 100644 --- a/src/peer/active.rs +++ b/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, + /// Our session index (they include this when sending TO us). + our_index: Option, + /// Their session index (we include this when sending TO them). + their_index: Option, + + // === Previous / draining epoch slot === + /// Previous session kept alive during drain window after cutover. + previous_session: Option, + /// Previous session's our_index (for peers_by_index cleanup on drain expiry). + previous_our_index: Option, + /// When the drain window started (None = no drain in progress). + drain_started: Option, + + // === Pending epoch slot === + /// Pending new session from completed rekey (before K-bit cutover). + pending_new_session: Option, + /// Pending new session's our_index. + pending_our_index: Option, + /// Pending new session's their_index. + pending_their_index: Option, + + // === 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, + /// Current transport address (for roaming support). + current_addr: Option, + /// 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>, + + /// 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, + + // === 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, +} + +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, - /// Our session index (they include this when sending TO us). - our_index: Option, - /// Their session index (we include this when sending TO them). - their_index: Option, - /// Transport ID for this peer's link. - transport_id: Option, - /// Current transport address (for roaming support). - current_addr: Option, - // === Spanning Tree === /// Their latest parent declaration. declaration: Option, @@ -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, - // === Heartbeat === /// When we last sent a heartbeat to this peer. last_heartbeat_sent: Option, @@ -169,12 +261,6 @@ pub struct ActivePeer { /// Cleared after the handshake timeout window. handshake_msg2: Option>, - // === 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, - /// Previous session's our_index (for peers_by_index cleanup on drain expiry). - previous_our_index: Option, - /// When the drain window started (None = no drain in progress). - drain_started: Option, - /// Pending new session from completed rekey (before K-bit cutover). - pending_new_session: Option, - /// Pending new session's our_index. - pending_our_index: Option, - /// Pending new session's their_index. - pending_their_index: Option, /// 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>, - - /// 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, + // === 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> { - 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 { - self.our_index + self.send.our_index } /// Get their session index (we use this to send TO them). pub fn their_index(&self) -> Option { - 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 { 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 { - 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 { - self.pending_our_index + self.send.pending_our_index } /// Get the pending new session's their_index. pub fn pending_their_index(&self) -> Option { - 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 { - 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 { - 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 { - 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 { - 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() }) } diff --git a/src/peer/connected_udp/socket.rs b/src/peer/connected_udp/socket.rs index 85b709a..8b54e54 100644 --- a/src/peer/connected_udp/socket.rs +++ b/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() } }