diff --git a/src/node/dataplane/rx_loop.rs b/src/node/dataplane/rx_loop.rs index 315a7f7..246eaa2 100644 --- a/src/node/dataplane/rx_loop.rs +++ b/src/node/dataplane/rx_loop.rs @@ -354,6 +354,11 @@ impl Node { self.sample_transport_congestion(); #[cfg(any(target_os = "linux", target_os = "macos"))] self.activate_connected_udp_sessions().await; + // Debug-build sweep of the peer-lifecycle map invariant + // (leaked machines / machine-less legs); two map scans, + // compiled out of release builds. + #[cfg(debug_assertions)] + self.debug_assert_peer_maps_coherent(); } // Shutdown signal → enter the bounded drain in place, ONCE. // Gated on `is_none()` so it only fires while serving; after diff --git a/src/node/mod.rs b/src/node/mod.rs index 2770b4b..d220085 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -2273,6 +2273,53 @@ impl Node { self.peer_timers.remove(&link); } + /// Gates the leg→machine direction of `debug_assert_peer_maps_coherent`. + /// Asserting it requires the convention that every `connections` insert is + /// paired with a `peer_machines` insert before the next await point — + /// which holds here: inbound msg1 births a machine alongside the window + /// leg, and dials birth one before the leg exists. Where handshake-window + /// legs legitimately run machine-less, flip this to `false` rather than + /// weakening the machine→carrier direction. + #[cfg(debug_assertions)] + const CHECK_LEGS_HAVE_MACHINES: bool = true; + + /// Debug-build coherence sweep over the peer-lifecycle maps, run once per + /// rx-loop tick and invoked directly by unit tests. + /// + /// Machine→carrier: every `peer_machines` entry must have a live carrier — + /// a pending connection on the same link, an active peer on it, or a + /// pending connect still resolving toward it. A machine with none of these + /// is unreachable by every teardown path and has leaked. + /// + /// Leg→machine: every pending connection carries a control machine (gated + /// by `CHECK_LEGS_HAVE_MACHINES` above). + #[cfg(debug_assertions)] + pub(in crate::node) fn debug_assert_peer_maps_coherent(&self) { + for link in self.peer_machines.keys() { + let has_carrier = self.connections.contains_key(link) + || self.peers.values().any(|peer| peer.link_id() == *link) + || self + .peering + .pending_connects + .iter() + .any(|pending| pending.link_id == *link); + assert!( + has_carrier, + "control machine for link {link} has no live carrier \ + (no pending connection, active peer, or pending connect)" + ); + } + + if Self::CHECK_LEGS_HAVE_MACHINES { + for link in self.connections.keys() { + assert!( + self.peer_machines.contains_key(link), + "pending connection on link {link} has no control machine" + ); + } + } + } + /// Operator-visible msg1 resend count for a pending handshake `link`, read /// from the per-peer machine (the counter's home once the resend drive moved /// off the shell connection). Machine-less connections (inbound legs that @@ -2335,6 +2382,11 @@ impl Node { // === Connection Management (Handshake Phase) === /// Add a pending connection. + /// + /// Also seeds a control machine for the leg when none exists yet, keeping + /// the leg→machine invariant (`debug_assert_peer_maps_coherent`) intact + /// for callers that insert a connection directly rather than through the + /// dial or inbound-msg1 paths, which pair the two inserts themselves. pub fn add_connection(&mut self, connection: PeerConnection) -> Result<(), NodeError> { let link_id = connection.link_id(); @@ -2348,6 +2400,16 @@ impl Node { }); } + self.peer_machines.entry(link_id).or_insert_with(|| { + let now = connection.started_at(); + match connection.expected_identity() { + Some(identity) if connection.is_outbound() => { + PeerMachine::new_outbound(link_id, *identity, now) + } + _ => PeerMachine::new_inbound(link_id, now), + } + }); + self.connections.insert(link_id, connection); Ok(()) } @@ -2362,8 +2424,10 @@ impl Node { self.connections.get_mut(link_id) } - /// Remove a connection. + /// Remove a connection, disposing its control machine alongside + /// (the disposal complement of `add_connection`'s machine seeding). pub fn remove_connection(&mut self, link_id: &LinkId) -> Option { + self.remove_peer_machine(*link_id); self.connections.remove(link_id) } diff --git a/src/node/tests/unit.rs b/src/node/tests/unit.rs index 150a0de..1d46941 100644 --- a/src/node/tests/unit.rs +++ b/src/node/tests/unit.rs @@ -364,6 +364,64 @@ fn test_node_connection_duplicate() { assert!(matches!(result, Err(NodeError::ConnectionAlreadyExists(_)))); } +#[cfg(debug_assertions)] +#[test] +fn test_peer_maps_coherent_after_add_connection() { + let mut node = make_node(); + + let identity = make_peer_identity(); + let link_id = LinkId::new(1); + let conn = PeerConnection::outbound(link_id, identity, 1000); + + node.add_connection(conn).unwrap(); + + assert!( + node.peer_machines.contains_key(&link_id), + "add_connection seeds a control machine for its leg" + ); + node.debug_assert_peer_maps_coherent(); +} + +#[cfg(debug_assertions)] +#[test] +fn test_peer_maps_coherent_through_establish() { + let mut node = make_node(); + let transport_id = TransportId::new(1); + + let link_id = LinkId::new(1); + let (conn, identity) = make_completed_connection(&mut node, link_id, transport_id, 1000); + + node.add_connection(conn).unwrap(); + node.debug_assert_peer_maps_coherent(); + + let result = node.promote_connection(link_id, identity, 2000).unwrap(); + assert!(matches!(result, PromotionResult::Promoted(_))); + + // The leg's connection entry is consumed by the promote; the active peer + // on the same link is now the machine's carrier. + node.debug_assert_peer_maps_coherent(); +} + +#[cfg(debug_assertions)] +#[test] +fn test_peer_maps_coherence_detects_orphaned_machine() { + let mut node = make_node(); + + // A machine with no connection, no active peer, and no pending connect + // has no live carrier; the check must panic on it. + let link_id = LinkId::new(7); + let machine = crate::peer::machine::PeerMachine::new_inbound(link_id, 1000); + node.peer_machines.insert(link_id, machine); + + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + node.debug_assert_peer_maps_coherent(); + })); + assert!( + result.is_err(), + "coherence check accepts a machine with no live carrier" + ); +} + #[test] fn test_node_promote_connection() { let mut node = make_node();