diff --git a/src/lib.rs b/src/lib.rs index c8724b1..720efb8 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -95,9 +95,7 @@ pub use cache::{CacheEntry, CacheError, CacheStats, CoordCache}; pub use proto::fmp::{PromotionResult, cross_connection_winner}; // Re-export peer types -pub use peer::{ - ActivePeer, ConnectivityState, HandshakeState, PeerConnection, PeerError, PeerSlot, -}; +pub use peer::{ActivePeer, ConnectivityState, HandshakeState, PeerConnection, PeerError}; // Re-export node types pub use node::{Node, NodeError, NodeState, UpdatePeersOutcome}; diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs index 1db6c42..aa21273 100644 --- a/src/node/dataplane/peer_actions.rs +++ b/src/node/dataplane/peer_actions.rs @@ -11,18 +11,24 @@ //! The executor is wired incrementally. Live today: the outbound msg2 promote //! (`handle_msg2` looks up the dial-persisted machine), the connectionless //! outbound msg1 send (`SendHandshake` with `their_index == None` → -//! `send_stored_msg1`, driven from `initiate_connection`), and the +//! `send_stored_msg1`, driven from `initiate_connection`), the //! connection-oriented dial (`OpenTransport` performs the non-blocking //! `transport.connect`; `TransportConnected` drives the connect-resolution msg1 -//! send from `poll_pending_connects`). +//! send from `poll_pending_connects`), the rekey cadence (`check_rekey` → +//! `route_rekey_cadence` → `RekeyConsume`, driving the `SwapSendState` and +//! `CompleteDrain` arms), and the liveness reap (`route_link_dead` → +//! `LinkDeadSuspected`, driving `InvalidateSendState` → `remove_active_peer`). //! -//! Inbound establish is not machine-driven here: `handle_msg1` builds and sends +//! Inbound msg1 is not machine-driven here: `handle_msg1` builds and sends //! msg2 inline, so `PeerEvent::InboundMsg1` is never dispatched and the -//! `SendHandshake` `their_index == Some` (msg2) branch stays dormant. +//! `SendHandshake` `their_index == Some` (msg2) branch stays dormant. Inbound +//! msg3 IS machine-driven: `handle_msg3` steps a throwaway decision machine and +//! this executor performs its verdict (`PromoteToActive`, +//! `SwapToInboundSession`, `RekeyRespondTrigger`). //! -//! Not yet driven, so their arms stay inert stubs: rekey/crypto installs, -//! link-control frames, and the connected-UDP plane. `RegisterDecryptSession` is -//! a deliberate no-op — see its arm for the note. +//! The genuine inert stubs remaining are `SendRekey`, `SendLinkMessage`, and +//! the connected-UDP arms. `RegisterDecryptSession` is a deliberate no-op — +//! see its arm for the note. //! //! The timer arms (`SetTimer`/`CancelTimer`) populate/clear the per-peer timer //! store (`peer_timers`). The `HandshakeRetransmit` and `HandshakeTimeout` @@ -430,13 +436,15 @@ impl Node { } } PeerAction::SwapSendState { .. } => { - // Initiator cutover. Reproduces the `ConnAction::Cutover` body - // in `handlers/rekey.rs`'s `check_rekey` EXACTLY. `addr` is - // resolved from the ambient verified identity. The decrypt - // re-register folds HERE, gated on `did_cutover` — the generic - // `RegisterDecryptSession` arm stays a no-op so a promote never - // double-registers. Shadow-only until the cadence fold routes - // here. + // Initiator cutover: the live authoritative rekey-cadence + // path, routed here from `check_rekey` via + // `route_rekey_cadence` → `PeerEvent::RekeyConsume`; the + // inline body survives only as `cutover_peer_inline`, a + // debug-assert release fallback. `addr` is resolved + // from the ambient verified identity (as `InvalidateSendState` + // does). The decrypt re-register folds HERE, gated on + // `did_cutover` — the generic `RegisterDecryptSession` arm stays a + // no-op so a promote never double-registers. let node_addr = *ambient.verified_identity.node_addr(); let did_cutover = if let Some(peer) = self.peers.get_mut(&node_addr) { if let Some(_old_our_index) = peer.cutover_to_new_session() { @@ -487,12 +495,14 @@ impl Node { let _ = did_cutover; } PeerAction::CompleteDrain { peer: node_addr } => { - // Initiator drain completion. Reproduces the `ConnAction::Drain` - // body in `handlers/rekey.rs`'s `check_rekey` EXACTLY. Extract - // the real previous index + transport_id under the peer borrow, - // drop the borrow, then run the cache_key cleanup (which takes - // &mut self for unregister_decrypt_worker_session). Shadow-only - // until the cadence fold routes here. + // Initiator drain completion: the live authoritative + // rekey-cadence path, routed here from `check_rekey` via + // `route_rekey_cadence` → `PeerEvent::RekeyConsume`; the + // inline body survives only as `drain_peer_inline`, a + // debug-assert release fallback. Extract the real previous + // index + transport_id under the peer borrow, drop the + // borrow, then run the cache_key cleanup (which takes + // &mut self for unregister_decrypt_worker_session). let drained = self.peers.get_mut(&node_addr).and_then(|peer| { peer.complete_drain().map(|idx| (idx, peer.transport_id())) }); diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index fc024cb..ad4b501 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -913,9 +913,10 @@ impl Node { }; // The outbound Msg2 Promote step cancels the two dial-armed handshake // timers (the machine survives promotion, so they would otherwise linger - // in the driver's shadow store) and then promotes. The cancels are - // behavior-neutral at this rung — `peer_timers` is written but not yet - // driven — and `PromoteToActive` is still what performs the promotion. + // in `peer_timers` until `drive_peer_timers` lazily discards them — the + // promoted leg's `connections` entry is gone and the machine has left + // `SentMsg1`, so they can no longer fire) and then promotes. + // `PromoteToActive` is still what performs the promotion. debug_assert_eq!( promote_actions, vec![ diff --git a/src/node/mod.rs b/src/node/mod.rs index b22178e..eb637db 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -371,22 +371,29 @@ pub struct Node { // === Per-Peer Control Machines === /// Per-peer lifecycle control FSMs, keyed by the stable `LinkId` that spans - /// the handshake→active lifetime. A NEW parallel structure introduced by the + /// the handshake→active lifetime. A parallel structure introduced by the /// node-runtime decomposition: `connections`/`peers` stay byte-unchanged (hot - /// path pristine) and are cut over to this machine home path-by-path. Unwired - /// initially: the executor (`dataplane/peer_actions.rs`) and advance helper exist - /// but no live handler path drives them yet and nothing inserts into this map; - /// the inbound establish cutover lands in a later commit. - #[allow(dead_code)] + /// path pristine) and are cut over to this machine home path-by-path. + /// Machines are inserted at the identified dial and born `established()` + /// inside `promote_connection` (msg3 decisions run through a throwaway + /// machine), and stepped in production by the handshake handlers, the + /// rekey-cadence and liveness-reap routers, and the lifecycle paths, with + /// the executor (`dataplane/peer_actions.rs`) performing the returned + /// actions. Timer FIRING decisions remain shell-side: `PeerEvent::Timeout` + /// is never dispatched in production. peer_machines: HashMap, /// Per-peer timer store, keyed by `LinkId` then `TimerKind`, holding each /// armed timer's absolute deadline (ms). The sans-IO time-as-input backing /// for `PeerEvent::Timeout`: populated/cleared by the machine's /// `SetTimer`/`CancelTimer` actions (`dataplane/peer_actions.rs`) and dropped - /// alongside the machine through the `remove_peer_machine` choke-point. At - /// this rung it is a SHADOW of the legacy tick timers — written but not yet - /// read by any driver (the handshake-kind fold wires the reader). + /// alongside the machine through the `remove_peer_machine` choke-point. The + /// `HandshakeRetransmit`/`HandshakeTimeout` kinds are driven by + /// `drive_peer_timers` (the retransmit fires on the stored deadline; the + /// timeout reap keys on the timer's presence, with the threshold read from + /// config); the rekey/liveness kinds are still SHADOWS of their + /// own shell drivers, and the machine's `on_timeout` handlers stay dormant + /// (`PeerEvent::Timeout` is never dispatched in production). peer_timers: HashMap>, // === Peers (Active Phase) === diff --git a/src/node/tests/handshake.rs b/src/node/tests/handshake.rs index b504673..8daeac4 100644 --- a/src/node/tests/handshake.rs +++ b/src/node/tests/handshake.rs @@ -860,8 +860,6 @@ async fn test_msg1_stored_for_resend() { // Verify stored msg1 matches what was built assert_eq!(conn.handshake_msg1().unwrap(), &wire_msg1); - assert_eq!(conn.resend_count(), 0); - assert!(conn.next_resend_at_ms() > now_ms); } /// Test that resend scheduling respects max_resends and backoff. @@ -1034,31 +1032,6 @@ fn test_msg2_stored_on_connection() { assert_eq!(conn.handshake_msg2().unwrap(), &msg2_bytes); } -/// Test that resend_count and next_resend_at_ms track correctly. -#[test] -fn test_resend_count_tracking() { - let peer_identity = make_peer_identity(); - let mut conn = PeerConnection::outbound(LinkId::new(1), peer_identity, 1000); - - assert_eq!(conn.resend_count(), 0); - assert_eq!(conn.next_resend_at_ms(), 0); - - // Simulate storing msg1 and scheduling first resend - conn.set_handshake_msg1(vec![0x01], 2000); - assert_eq!(conn.resend_count(), 0); - assert_eq!(conn.next_resend_at_ms(), 2000); - - // Record first resend - conn.record_resend(4000); // next at 4000 (2s backoff) - assert_eq!(conn.resend_count(), 1); - assert_eq!(conn.next_resend_at_ms(), 4000); - - // Record second resend - conn.record_resend(8000); // next at 8000 (4s backoff) - assert_eq!(conn.resend_count(), 2); - assert_eq!(conn.next_resend_at_ms(), 8000); -} - /// Test that duplicate msg2 is silently dropped when pending_outbound is already cleared. #[tokio::test] async fn test_duplicate_msg2_dropped() { diff --git a/src/peer/connection.rs b/src/peer/connection.rs index 25c07df..31dcf69 100644 --- a/src/peer/connection.rs +++ b/src/peer/connection.rs @@ -262,21 +262,6 @@ impl PeerConnection { self.state.handshake_msg2() } - /// Number of resends performed. - pub fn resend_count(&self) -> u32 { - self.state.resend_count() - } - - /// When the next resend is scheduled (Unix ms). - pub fn next_resend_at_ms(&self) -> u64 { - self.state.next_resend_at_ms() - } - - /// Record a resend and schedule the next one. - pub fn record_resend(&mut self, next_resend_at_ms: u64) { - self.state.record_resend(next_resend_at_ms); - } - // === Noise Handshake Operations === /// Start the handshake as initiator and generate message 1. diff --git a/src/peer/machine.rs b/src/peer/machine.rs index 074ec85..88ff76b 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -1,10 +1,15 @@ //! Per-peer FMP control FSM (sans-IO reducer) — XX re-derivation. //! //! The unified per-peer lifecycle state machine, ported onto next's XX cores -//! from the IK-lineage template. It provides the FSM types, -//! the machine struct (control-tier state only), and the pure `step` reducer, -//! plus its unit tests. It is **unwired** — nothing in the codebase calls it -//! yet; the action executor and the msg3/rekey/reap wiring land in later commits. +//! from the IK-lineage template. It provides the FSM types, the machine struct +//! (control-tier state only), and the pure `step` reducer, plus its unit +//! tests. `step` is driven in production by the handshake handlers (msg2 on +//! the dial-persisted machine; msg3 through a throwaway decision machine), the +//! rekey-cadence and liveness-reap routers, and the dial/lifecycle paths, with +//! the executor in `crate::node::dataplane::peer_actions` performing the +//! returned actions. Still dormant: `PeerEvent::Timeout` and +//! `PeerEvent::InboundMsg1` are never dispatched — timer FIRING decisions stay +//! with the shell drivers, and `handle_msg1` sends msg2 inline. //! //! ## Shape //! @@ -45,8 +50,9 @@ //! ([`SwapToInboundSession`](PeerAction::SwapToInboundSession) / //! [`RekeyRespondTrigger`](PeerAction::RekeyRespondTrigger)) carrying indices + //! the tie-break flag, and the executor performs the session/registry surgery -//! shell-side, matching next's inline `handle_msg3` verbatim. These trigger -//! variants are **provisional** (the wiring may refine them) and unwired. +//! shell-side, matching the former inline `handle_msg3` bodies verbatim. Both +//! triggers are live: `handle_msg3` steps the decision machine and the +//! executor performs its verdict. //! //! Likewise the outbound completion is NOT routed through the machine: next has //! no `establish_outbound` core — `handle_msg2` learns the identity from @@ -84,11 +90,14 @@ use crate::{NodeAddr, PeerIdentity}; // ============================================================================ // Timing placeholders // -// This module is unwired; the real intervals come from `NodeConfig` when the -// driver is wired. The `poll_*` cores already take the interval/backoff as -// arguments, so these are only used to compute `SetTimer{at_ms}` deadlines and -// the `Closed{backoff_deadline_ms}` park time. The unit tests assert on timer -// *kinds*, not exact deadlines. +// The `poll_*` cores already take the interval/backoff as arguments, so these +// are only used to compute `SetTimer{at_ms}` deadlines and the +// `Closed{backoff_deadline_ms}` park time. The handshake timers are armed +// live at dial time from these constants: the retransmit driver fires on the +// machine-armed deadline, while the timeout reaper keys on the timer's +// presence with its threshold read from `NodeConfig`, which also governs the +// reschedule cadence shell-side. The unit tests assert on timer *kinds*, not +// exact deadlines. // ============================================================================ const HANDSHAKE_RETRANSMIT_INTERVAL_MS: u64 = 1_000; @@ -1343,10 +1352,9 @@ impl PeerMachine { } fn on_tick(&mut self, now: u64) -> Vec { - // The driver evaluates due machine timers on the quantized tick and - // re-enters the Timeout{kind} handlers. This module is unwired; the deadline - // bookkeeping is threaded from the driver at wiring time, so Tick is a - // no-op here. + // Dormant no-op: `PeerEvent::Tick` is not dispatched in production. + // The shell drivers evaluate due timer deadlines themselves, so there + // is no machine-side bookkeeping to advance here. let _ = now; Vec::new() } diff --git a/src/peer/mod.rs b/src/peer/mod.rs index 87c7db9..65e2c33 100644 --- a/src/peer/mod.rs +++ b/src/peer/mod.rs @@ -3,9 +3,6 @@ //! Two-phase peer lifecycle: //! 1. **PeerConnection** - Handshake phase, before identity is verified //! 2. **ActivePeer** - Authenticated phase, after successful Noise handshake -//! -//! The PeerSlot enum represents either phase, enabling unified storage -//! while maintaining type safety for phase-specific operations. mod active; #[cfg(any(target_os = "linux", target_os = "macos"))] @@ -18,7 +15,6 @@ pub use connection::{HandshakeState, PeerConnection}; use crate::NodeAddr; use crate::transport::LinkId; -use std::fmt; use thiserror::Error; // ============================================================================ @@ -62,127 +58,12 @@ pub enum PeerError { MaxPeersExceeded { max: usize }, } -// ============================================================================ -// PeerSlot -// ============================================================================ - -/// A slot in the peer table, representing either connection or active phase. -#[derive(Debug)] -pub enum PeerSlot { - /// Connection in handshake phase. - Connecting(Box), - /// Authenticated peer. - Active(Box), -} - -impl PeerSlot { - /// Create a new connecting slot (outbound). - pub fn outbound(conn: PeerConnection) -> Self { - PeerSlot::Connecting(Box::new(conn)) - } - - /// Create a new connecting slot (inbound). - pub fn inbound(conn: PeerConnection) -> Self { - PeerSlot::Connecting(Box::new(conn)) - } - - /// Create a new active slot. - pub fn active(peer: ActivePeer) -> Self { - PeerSlot::Active(Box::new(peer)) - } - - /// Check if this is a connecting slot. - pub fn is_connecting(&self) -> bool { - matches!(self, PeerSlot::Connecting(_)) - } - - /// Check if this is an active slot. - pub fn is_active(&self) -> bool { - matches!(self, PeerSlot::Active(_)) - } - - /// Get the link ID for this slot. - pub fn link_id(&self) -> LinkId { - match self { - PeerSlot::Connecting(conn) => conn.link_id(), - PeerSlot::Active(peer) => peer.link_id(), - } - } - - /// Get as connection reference, if connecting. - pub fn as_connection(&self) -> Option<&PeerConnection> { - match self { - PeerSlot::Connecting(conn) => Some(conn), - PeerSlot::Active(_) => None, - } - } - - /// Get as mutable connection reference, if connecting. - pub fn as_connection_mut(&mut self) -> Option<&mut PeerConnection> { - match self { - PeerSlot::Connecting(conn) => Some(conn), - PeerSlot::Active(_) => None, - } - } - - /// Get as active peer reference, if active. - pub fn as_active(&self) -> Option<&ActivePeer> { - match self { - PeerSlot::Active(peer) => Some(peer), - PeerSlot::Connecting(_) => None, - } - } - - /// Get as mutable active peer reference, if active. - pub fn as_active_mut(&mut self) -> Option<&mut ActivePeer> { - match self { - PeerSlot::Active(peer) => Some(peer), - PeerSlot::Connecting(_) => None, - } - } - - /// Get the known node_addr, if any. - /// - /// For connections, this is the expected identity (may be None for inbound). - /// For active peers, this is always known. - pub fn node_addr(&self) -> Option<&NodeAddr> { - match self { - PeerSlot::Connecting(conn) => conn.expected_identity().map(|id| id.node_addr()), - PeerSlot::Active(peer) => Some(peer.node_addr()), - } - } -} - -impl fmt::Display for PeerSlot { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - PeerSlot::Connecting(conn) => { - write!( - f, - "connecting(link={}, state={})", - conn.link_id(), - conn.handshake_state() - ) - } - PeerSlot::Active(peer) => { - write!( - f, - "active(node={:?}, link={})", - peer.node_addr(), - peer.link_id() - ) - } - } - } -} - // ============================================================================ // Tests // ============================================================================ #[cfg(test)] mod tests { - use super::*; use crate::proto::fmp::PromotionResult; use crate::transport::LinkId; use crate::{Identity, PeerIdentity}; @@ -192,32 +73,6 @@ mod tests { PeerIdentity::from_pubkey(identity.pubkey()) } - #[test] - fn test_peer_slot_connecting() { - let identity = make_peer_identity(); - let conn = PeerConnection::outbound(LinkId::new(1), identity, 1000); - let slot = PeerSlot::Connecting(Box::new(conn)); - - assert!(slot.is_connecting()); - assert!(!slot.is_active()); - assert!(slot.as_connection().is_some()); - assert!(slot.as_active().is_none()); - assert_eq!(slot.link_id(), LinkId::new(1)); - } - - #[test] - fn test_peer_slot_active() { - let identity = make_peer_identity(); - let peer = ActivePeer::new(identity, LinkId::new(2), 2000); - let slot = PeerSlot::Active(Box::new(peer)); - - assert!(!slot.is_connecting()); - assert!(slot.is_active()); - assert!(slot.as_connection().is_none()); - assert!(slot.as_active().is_some()); - assert_eq!(slot.link_id(), LinkId::new(2)); - } - #[test] fn test_promotion_result_promoted() { let identity = make_peer_identity(); @@ -255,26 +110,4 @@ mod tests { assert!(!result.should_close_this_connection()); assert_eq!(result.link_to_close(), Some(LinkId::new(1))); } - - #[test] - fn test_peer_slot_node_addr() { - // Outbound connection knows expected identity - let identity = make_peer_identity(); - let expected_node_addr = *identity.node_addr(); - let conn = PeerConnection::outbound(LinkId::new(1), identity, 1000); - let slot = PeerSlot::Connecting(Box::new(conn)); - assert_eq!(slot.node_addr(), Some(&expected_node_addr)); - - // Inbound connection doesn't know identity yet - let conn_inbound = PeerConnection::inbound(LinkId::new(2), 2000); - let slot_inbound = PeerSlot::Connecting(Box::new(conn_inbound)); - assert!(slot_inbound.node_addr().is_none()); - - // Active peer always knows identity - let identity2 = make_peer_identity(); - let active_node_addr = *identity2.node_addr(); - let peer = ActivePeer::new(identity2, LinkId::new(3), 3000); - let slot_active = PeerSlot::Active(Box::new(peer)); - assert_eq!(slot_active.node_addr(), Some(&active_node_addr)); - } } diff --git a/src/proto/fmp/state.rs b/src/proto/fmp/state.rs index 821097c..d72c22e 100644 --- a/src/proto/fmp/state.rs +++ b/src/proto/fmp/state.rs @@ -469,6 +469,7 @@ impl ConnectionState { } /// When the next resend is scheduled (Unix ms). + #[cfg(test)] pub fn next_resend_at_ms(&self) -> u64 { self.next_resend_at_ms }