diff --git a/CHANGELOG.md b/CHANGELOG.md index 77fa5f74..25304e74 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -480,6 +480,22 @@ with v0.5.x or earlier peers. binder tearing down and rebinding every second while teardown silently declined to abort anything. +### Removed + +- **Source-breaking for consumers of the library crate**: `ActivePeer` no + longer stores a connectivity state. `ActivePeer::connectivity`, `can_send`, + `is_healthy`, `mark_stale`, `mark_reconnecting`, `mark_disconnected` and + `mark_connected` are gone. `ConnectivityState` stays public with only + `Connected` and `Stale`, the values `show_peers` reports, and loses + `can_send` and `is_healthy`. `ConnectivityState::is_terminal`, + `ActivePeer::is_disconnected`, `Node::sendable_peers` and + `Node::sendable_peer_count` keep their signatures and the results they + always had in the daemon: the first two return `false`, and the last two + cover every peer. Nothing in the daemon changed the stored state after a + peer was promoted, so every removed check was already true and the shipped + binaries behave as before. The peer wire and the control-socket response + shape are unchanged. + ### Fixed - A leaf-profile node no longer self-elects as tree root. A leaf holding the @@ -676,7 +692,15 @@ with v0.5.x or earlier peers. resynchronise from. BLE was the worst of the four: it awaited the L2CAP write while holding the connection-pool mutex, so one unresponsive peer froze every other BLE operation as well — connects, evictions, and each - receive loop's teardown. + receive loop's teardown. On TCP, Tor and Nym, a connection's writer and + receive loop act only on their own connection: one that outlives its + connection can no longer tear down a newer connection that has taken the + same address, and a receive loop that ends stops its writer rather than + leaving it writing to a peer that has gone. Closing one of their connections + on purpose, as a control-API disconnect does, now lets the writer finish the + frames already queued, within five seconds, instead of discarding them, so a + Disconnect sent just before the close reaches the peer. Stopping the + transport, or a connection that has failed, still discards them. - A per-peer `connect()`-ed UDP socket is no longer left pinned to an interface the host has moved off. Established UDP peers get their own socket for the @@ -712,6 +736,13 @@ with v0.5.x or earlier peers. platforms with the connected-socket fast path); elsewhere the heartbeat alone carries the new address. +- A peer reached by NAT traversal now gets its per-peer connected UDP socket. + The adopted traversal socket carried no address-reuse flags, so the connected + socket's bind to the same port was refused with `EADDRINUSE` on every tick and + the peer never left the unconnected path. The flags are now set when the + socket is adopted, after its bind, so the traversal bind still receives a + port no other socket holds. + #### Control socket - `show_links` (`fipsctl show links`) now reports the traffic a link has @@ -726,6 +757,16 @@ with v0.5.x or earlier peers. link was created with. The counters cover authenticated link frames only, so they are not expected to match the transport totals in `show_transports`. The response shape is unchanged. +- `show_peers` (`fipsctl show peers`) now reports a peer that has gone quiet + as `stale`. Its `connectivity` was read from a state that nothing outside + the tests ever changed, so every peer read `connected` until it was + removed, including one that had stopped answering tens of seconds earlier. + The value is now derived from how long the peer has been silent: `connected` + while its idle time is at or below `heartbeat_interval_secs`, and `stale` + above it, the same rule that decides whether discovery re-dials an active + peer on the path it already has. The `reconnecting` and `disconnected` + values the open-discovery tutorial described never occurred, and the + tutorial no longer lists them. The response shape is unchanged. #### Packaging diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index 7524597e..2967c39a 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -283,10 +283,15 @@ one would put a DNS lookup on the sample path; the address becomes numeric as soon as an authenticated packet arrives from the peer). A node holding no peers detects nothing, which is correct — it has nothing bound to the old path. -The cost is five non-blocking syscalls per peer per sample, read from the -probe's own code rather than measured: `socket(2)` and `bind(2)`, a `connect(2)` -that sends no packet, a `getsockname(2)`, and the `close(2)` the socket takes on -drop. Nothing goes on the wire and no name is resolved. +The cost is five non-blocking syscalls per peer per sample, counted with +`strace` on Linux: `socket(2)` and `bind(2)`, a `connect(2)` that sends no +packet, a `getsockname(2)`, and the `close(2)` the socket takes on drop. A peer +with no route costs four, because the lookup fails at `connect(2)`. Nothing goes +on the wire and no name is resolved. At the default `max_peers` of 128 that is +640 syscalls per sample. A detected change is resampled until it settles, so +with the default `debounce_ms` it costs between two samples and nine, which is +between 1280 and 5760 syscalls at 128 peers; the backstop timer also takes one +sample every `poll_interval_secs` whether or not anything moved. `node.limits.max_peers` bounds the per-sample total only where it is set: at `max_peers: 0`, which means unlimited, there is no bound and the cost tracks the live peer count instead. diff --git a/docs/tutorials/advertise-your-node.md b/docs/tutorials/advertise-your-node.md index 239880fa..f9f5bb6b 100644 --- a/docs/tutorials/advertise-your-node.md +++ b/docs/tutorials/advertise-your-node.md @@ -284,7 +284,7 @@ sudo fipsctl show peers In addition to your configured `test-us01` peer, you may see an entry for `test-us03` (the open-discovery test mesh node). -It will have `connectivity` active and its own +It will have `connectivity` `connected` and its own `transport_addr`. This peering appeared without you configuring anything — the test-mesh open-discovery node saw your advert, dialed the endpoint, and Noise XX established diff --git a/docs/tutorials/open-discovery.md b/docs/tutorials/open-discovery.md index 815897f8..0f13af04 100644 --- a/docs/tutorials/open-discovery.md +++ b/docs/tutorials/open-discovery.md @@ -200,11 +200,14 @@ You should see considerably more entries than before: Each entry has its own `connectivity` state, and every entry that appears here completed a handshake at least once: a peer whose advert was stale, or that NAT traversal never reached, -produces no entry at all rather than a failed one. Healthy links -read `connected`. A link not heard from recently reads `stale` -and still carries traffic; one that dropped and is being retried -reads `reconnecting`, and one explicitly torn down reads -`disconnected`. Neither of the last two can send. +produces no entry at all rather than a failed one. A link heard +from within the last heartbeat interval +(`node.heartbeat_interval_secs`, 10 seconds by default) reads +`connected`. One silent for longer reads `stale`; it still +carries traffic, and it reads `connected` again as soon as the +peer is heard from. A link that stays silent until it is declared +dead is removed, so its entry disappears rather than changing +state. To get a list of just the connected links: diff --git a/src/control/queries.rs b/src/control/queries.rs index c3613859..6f344c26 100644 --- a/src/control/queries.rs +++ b/src/control/queries.rs @@ -250,6 +250,7 @@ pub fn show_peers(node: &Node) -> Value { // start (no peer has SRTT) every peer uses the default link cost of 1.0. let any_peer_has_srtt = node.peers().any(|p| p.has_srtt()); + let now = now_ms(); let peers: Vec = node .peers() .map(|peer| { @@ -267,7 +268,7 @@ pub fn show_peers(node: &Node) -> Value { "npub": peer.npub(), "display_name": node.peer_display_name(&node_addr), "ipv6_addr": format!("{}", peer.address()), - "connectivity": format!("{}", peer.connectivity()), + "connectivity": format!("{}", node.peer_connectivity(peer, now)), "link_id": peer.link_id().as_u64(), "authenticated_at_ms": peer.authenticated_at(), "last_seen_ms": peer.last_seen(), diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index 80a3d74e..d922e3ff 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -1659,7 +1659,6 @@ impl Node { has_existing_peer: true, existing_peer_epoch: existing_peer.remote_epoch(), has_session: existing_peer.has_session(), - is_healthy: existing_peer.is_healthy(), pending_new_session: existing_peer.pending_new_session().is_some(), rekey_in_progress: existing_peer.rekey_in_progress(), existing_msg2: existing_peer.handshake_msg2().map(|m| m.to_vec()), @@ -1676,7 +1675,6 @@ impl Node { has_existing_peer: false, existing_peer_epoch: None, has_session: false, - is_healthy: false, pending_new_session: false, rekey_in_progress: false, existing_msg2: None, diff --git a/src/node/handlers/probe.rs b/src/node/handlers/probe.rs index 73071238..b37297a6 100644 --- a/src/node/handlers/probe.rs +++ b/src/node/handlers/probe.rs @@ -350,7 +350,7 @@ impl Node { session_established, session_is_ours: is_ours, session_error: job.session_error.take(), - target_is_direct_peer: self.peers.get(&target).is_some_and(|p| p.can_send()), + target_is_direct_peer: self.peers.contains_key(&target), counters, last_rtt_ms: mmp.and_then(|m| m.metrics.last_rtt_ms()), srtt_ms: mmp.and_then(|m| m.metrics.srtt_ms()), @@ -485,9 +485,7 @@ impl Node { if dest == self.node_addr() { return (None, Some(NoHopReason::Local)); } - if let Some(peer) = self.peers.get(dest) - && peer.can_send() - { + if self.peers.contains_key(dest) { return ( Some(NextHopFacts { node_addr: *dest, @@ -519,7 +517,7 @@ impl Node { let Some(hop) = selected else { return (None, Some(NoHopReason::NoCloserPeer)); }; - if !self.peers.get(&hop).is_some_and(|p| p.can_send()) { + if !self.peers.contains_key(&hop) { return (None, Some(NoHopReason::HopNotSendReady)); } let class = self.classify_forward(dest, &hop); diff --git a/src/node/handlers/rekey.rs b/src/node/handlers/rekey.rs index 02414ee9..96a970a1 100644 --- a/src/node/handlers/rekey.rs +++ b/src/node/handlers/rekey.rs @@ -94,7 +94,7 @@ impl Node { after_messages: self.config().node.rekey.after_messages, }; - // The shell snapshots each healthy peer's rekey ages/flags (every clock + // The shell snapshots each peer's rekey ages/flags (every clock // read resolved here); the core decides cutover/drain/trigger with no // clock, phase-grouped to preserve the pre-refactor execution order. let snapshots = self.rekey_peers(); @@ -285,7 +285,7 @@ impl Node { } } - /// Snapshot every healthy peer with a session for the rekey decision, + /// Snapshot every peer with a session for the rekey decision, /// pre-computing its monotonic ages and timer predicates so the pure core /// applies the thresholds without reading a clock (see [`PeerSnapshot`]). /// @@ -294,7 +294,7 @@ impl Node { pub(in crate::node) fn rekey_peer_snapshots(&self) -> Vec { self.peers .iter() - .filter(|(_, peer)| peer.has_session() && peer.is_healthy()) + .filter(|(_, peer)| peer.has_session()) .map(|(node_addr, peer)| PeerSnapshot { addr: *node_addr, has_pending: peer.pending_new_session().is_some(), diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs index 0106f4e2..bd9b4d08 100644 --- a/src/node/lifecycle/mod.rs +++ b/src/node/lifecycle/mod.rs @@ -2725,7 +2725,7 @@ impl Node { let peer_addrs: Vec = self .peers .iter() - .filter(|(_, peer)| peer.can_send() && peer.has_session()) + .filter(|(_, peer)| peer.has_session()) .map(|(addr, _)| *addr) .collect(); @@ -3541,13 +3541,7 @@ impl Node { let Some(peer) = self.peers.get(peer_node_addr) else { return false; }; - let stale_after_ms = self - .config() - .node - .heartbeat_interval_secs - .saturating_mul(1000) - .max(1000); - peer.idle_time(Self::now_ms()) > stale_after_ms + self.peer_link_is_stale(peer, Self::now_ms()) } pub(in crate::node) fn active_peer_matches_candidate( diff --git a/src/node/mod.rs b/src/node/mod.rs index 77109710..26f2b67a 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -59,8 +59,8 @@ use self::reloadable::Reloadable; pub(crate) const REKEY_JITTER_SECS: i64 = 15; use crate::cache::CoordCache; use crate::node::session::SessionEntry; -use crate::peer::ActivePeer; use crate::peer::machine::{PeerMachine, TimerKind}; +use crate::peer::{ActivePeer, ConnectivityState}; use crate::proto::bloom::{BloomFilter, BloomState}; use crate::proto::fmp::Fmp; use crate::proto::fmp::NodeProfile; @@ -2310,6 +2310,7 @@ impl Node { // SRTT) every peer falls back to the default link cost of 1.0. let any_peer_has_srtt = self.peers().any(|p| p.has_srtt()); + let now_ms = Self::now_ms(); let peer_rows: Vec = self .peers() .map(|peer| { @@ -2362,7 +2363,7 @@ impl Node { npub: peer.npub(), display_name: self.peer_display_name(&node_addr), ipv6_addr: format!("{}", peer.address()), - connectivity: format!("{}", peer.connectivity()), + connectivity: format!("{}", self.peer_connectivity(peer, now_ms)), link_id: peer.link_id().as_u64(), authenticated_at_ms: peer.authenticated_at(), last_seen_ms: peer.last_seen(), @@ -3109,6 +3110,37 @@ impl Node { self.peers.values() } + /// Whether an active peer has been silent at `now_ms` for longer than the + /// configured heartbeat interval, floored at one second. + /// + /// The one idle-time liveness rule: the control socket reports such a peer + /// as `stale`, the discovery dial gate + /// [`Self::active_peer_link_is_live`] no longer holds its link as live, and + /// discovery re-dials it on the path it already has. + pub(in crate::node) fn peer_link_is_stale(&self, peer: &ActivePeer, now_ms: u64) -> bool { + let stale_after_ms = self + .config() + .node + .heartbeat_interval_secs + .saturating_mul(1000) + .max(1000); + peer.idle_time(now_ms) > stale_after_ms + } + + /// Connectivity of an active peer as the control socket reports it: + /// `Stale` when [`Self::peer_link_is_stale`] holds at `now_ms`, otherwise + /// `Connected`. + /// + /// Derived from idle time rather than read from the state stored on the + /// peer, which nothing in production changes after promotion. + pub(crate) fn peer_connectivity(&self, peer: &ActivePeer, now_ms: u64) -> ConnectivityState { + if self.peer_link_is_stale(peer, now_ms) { + ConnectivityState::Stale + } else { + ConnectivityState::Connected + } + } + /// Reference to the Nostr discovery handle if discovery is enabled. /// Used by control queries (`show_peers` per-peer Nostr-traversal /// state) to read failure-state without taking shared ownership. @@ -3121,14 +3153,15 @@ impl Node { self.peers.keys() } - /// Iterate over peers that can send traffic. + /// Iterate over peers that can send traffic: every active peer, the same + /// peers as [`Self::peers`]. pub fn sendable_peers(&self) -> impl Iterator { - self.peers.values().filter(|p| p.can_send()) + self.peers.values() } - /// Number of peers that can send traffic. + /// Number of peers that can send traffic, the same as [`Self::peer_count`]. pub fn sendable_peer_count(&self) -> usize { - self.peers.values().filter(|p| p.can_send()).count() + self.peers.len() } // === End-to-End Sessions === @@ -3482,9 +3515,7 @@ impl Node { } // 2. Direct peer - if let Some(peer) = self.peers.get(dest_node_addr) - && peer.can_send() - { + if let Some(peer) = self.peers.get(dest_node_addr) { return Some(peer); } @@ -3501,7 +3532,7 @@ impl Node { // 3. Bloom filter candidates — requires dest_coords for loop-free selection. // If no candidate is strictly closer, fall through to tree routing. // The sans-IO core enumerates borrowed peers over the `RoutingView` - // seam, applies the bloom/send/progress filters, and tracks the + // seam, applies the bloom/progress filters, and tracks the // winner inline; the shell supplies only raw per-peer reads. let next_hop = { let view = NodeRoutingView { @@ -3525,7 +3556,7 @@ impl Node { let skip = self.non_full_peers(); let next_hop_id = self.tree_state.find_next_hop(&dest_coords, &skip)?; - self.peers.get(&next_hop_id).filter(|p| p.can_send()) + self.peers.get(&next_hop_id) } /// Classify a transit forward by route class from tree coordinates. @@ -4064,7 +4095,7 @@ impl Node { /// Shell-side [`routing::RoutingView`] seam over live `Node` state — the sole /// routing read adapter the shell retains. It hands the sans-IO routing core -/// borrowed peers plus raw `may_reach` / `can_send` / `link_cost` / `coords` +/// borrowed peers plus raw `may_reach` / `link_cost` / `coords` /// reads so selection and error synthesis live in `proto::routing::core`; no /// routing decision logic remains here. /// @@ -4113,10 +4144,6 @@ impl routing::RoutingView for NodeRoutingView<'_> { peer.1.may_reach(dest) } - fn peer_can_send<'a>(&'a self, peer: Self::Peer<'a>) -> bool { - peer.1.can_send() - } - fn peer_link_cost<'a>(&'a self, peer: Self::Peer<'a>) -> f64 { peer.1.link_cost() } diff --git a/src/node/netmon/mod.rs b/src/node/netmon/mod.rs index 0e45451c..8e3956f8 100644 --- a/src/node/netmon/mod.rs +++ b/src/node/netmon/mod.rs @@ -51,9 +51,9 @@ //! One local source address per peer: for every peer whose transport address is //! a numeric IP endpoint, the address the kernel would pick to reach *that //! peer*. A connected-but-never-sending UDP socket makes the kernel run its -//! route lookup and bind the source address it would use; five syscalls, read -//! off [`NetFingerprint::sample`] rather than measured, no packets, no name -//! resolution, and it works identically on every platform std supports. +//! route lookup and bind the source address it would use; five syscalls per +//! peer (see [`NetFingerprint::sample`] for the measured cost), no packets, no +//! name resolution, and it works identically on every platform std supports. //! //! Keying on peers bounds the *reaction* — only the peers a change names are //! acted on — and does not bound the *sampling*. One roaming peer still makes @@ -222,18 +222,26 @@ struct PeerPath { impl NetFingerprint { /// Probe every target and record the local address the kernel picks. /// - /// Five non-blocking syscalls per target: `socket(2)` and `bind(2)` behind - /// `UdpSocket::bind`, a `connect(2)` that sends no packet, a - /// `getsockname(2)`, and the `close(2)` the socket takes on drop. No I/O - /// wait, no name resolution, and no allocation beyond the map. + /// Five non-blocking syscalls per target, counted with `strace -f` on + /// Linux: `socket(2)` and `bind(2)` behind `UdpSocket::bind`, a + /// `connect(2)` that sends no packet, a `getsockname(2)`, and the + /// `close(2)` the socket takes on drop. A target with no route costs four, + /// because `connect(2)` fails and `getsockname(2)` is never reached. A + /// build with debug assertions on adds a sixth to each, the `fcntl(2)` std + /// uses to check a descriptor is still open before closing it, so count + /// against a release build. No I/O wait, no name resolution, and no + /// allocation beyond the map. /// - /// The count matters because a debounced handover resamples: up to - /// `MAX_DEBOUNCE_ROUNDS` rounds plus the settled sample, times the peers - /// held. `node.limits.max_peers` bounds that only where it is set — - /// the value 0 means unlimited, and there the cost tracks the live peer - /// count instead. It runs inline in the detector's own task rather than - /// through `spawn_blocking`, which is what keeps it off every other task - /// regardless. + /// The count matters because a debounced handover resamples: the sample + /// that saw the move, then up to `MAX_DEBOUNCE_ROUNDS` more until two + /// consecutive samples agree. At 128 peers, the `node.limits.max_peers` + /// default, that is 640 syscalls per sample, and a reported change under a + /// non-zero debounce costs from 1280 (settled on the first resample) to + /// 5760 (still moving after every round). `node.limits.max_peers` bounds + /// that only where it is set — the value 0 means unlimited, and there the + /// cost tracks the live peer count instead. It runs inline in the + /// detector's own task rather than through `spawn_blocking`, which is what + /// keeps it off every other task regardless. pub(in crate::node) fn sample(targets: &[ProbeTarget]) -> Self { Self { sources: targets @@ -541,7 +549,8 @@ struct WakeSource { /// either of which would otherwise leave the node noticing nothing at all. /// Keeping the period the poller would have used makes an event-driven /// backend a strict latency improvement rather than a replacement that can - /// regress, for the cost of a few syscalls per period. + /// regress, for the cost of one sample per period: five syscalls per probed + /// peer, or 640 at the default of 128 peers. timer: tokio::time::Interval, } diff --git a/src/node/netmon/tests.rs b/src/node/netmon/tests.rs index 30134a3a..e2ae35dd 100644 --- a/src/node/netmon/tests.rs +++ b/src/node/netmon/tests.rs @@ -251,6 +251,43 @@ async fn a_peer_joining_onto_a_settled_path_is_still_not_a_change() { .await; } +#[tokio::test(start_paused = true)] +async fn local_sources_trading_places_between_peers_is_reported_for_both() { + // The reason the fingerprint is a map from peer to source and not a set of + // sources. Two peers swap the local addresses they are reached from, so + // the host still leaves from exactly the same two addresses, yet each + // peer's connected socket is pinned to the source the other peer now + // uses. A set of addresses is unchanged across the swap and would report + // nothing while both sockets are stale. + let a = Some(v4(192, 168, 1, 10)); + let b = Some(v4(10, 40, 0, 7)); + let before = NetFingerprint::for_test(&[(peer(1), a), (peer(2), b)]); + let swapped = NetFingerprint::for_test(&[(peer(1), b), (peer(2), a)]); + let (sampler, _) = scripted(vec![before, swapped]); + let (tx, mut rx) = mpsc::channel(1); + + tokio::spawn(run_detector(tx, cfg(1, 0), sampler, timer_wake(1))); + + let change = expect_change(&mut rx).await; + assert_eq!(change.summary.moved.len(), 2, "{:?}", change.summary.moved); + assert_eq!( + change.summary.moved, + vec![ + PeerSourceMove { + peer: peer(1), + before: a, + after: b, + }, + PeerSourceMove { + peer: peer(2), + before: b, + after: a, + }, + ], + "each peer must be named with its own source before and after the swap" + ); +} + #[tokio::test(start_paused = true)] async fn churn_during_a_handover_does_not_mask_the_handover() { // Both at once: a peer leaves while the medium moves under the peer that diff --git a/src/node/tests/bootstrap.rs b/src/node/tests/bootstrap.rs index 840d85bb..0c876962 100644 --- a/src/node/tests/bootstrap.rs +++ b/src/node/tests/bootstrap.rs @@ -418,3 +418,132 @@ async fn test_adopted_udp_inherits_mtu_from_named_primary_config() { transport.stop().await.ok(); } } + +/// A peer reached through an adopted traversal socket gets a per-peer +/// connected UDP socket on that socket's own port, as a peer on a configured +/// listener does. The traversal socket comes from a plain bind with no reuse +/// flags, and the kernel refuses the connected socket's bind to a port whose +/// holder did not opt in to sharing it. +#[cfg(target_os = "linux")] +#[tokio::test] +async fn test_connected_udp_activates_on_an_adopted_traversal_transport_and_on_a_configured_one() { + let mut node_a = make_node(); + let mut node_b = make_node(); + + let transport_id_b = TransportId::new(1); + let udp_config = UdpConfig { + bind_addr: Some("127.0.0.1:0".to_string()), + mtu: Some(1280), + ..Default::default() + }; + + let (packet_tx_a, packet_rx_a) = packet_channel(64); + let (packet_tx_b, packet_rx_b) = packet_channel(64); + + node_a.supervisor.packet_tx = Some(packet_tx_a.clone()); + node_a.packet_rx = Some(packet_rx_a); + node_a.supervisor.state = NodeState::Running; + + let mut transport_b = UdpTransport::new(transport_id_b, None, udp_config, packet_tx_b.clone()); + transport_b.start_async().await.unwrap(); + + let addr_b = transport_b.local_addr().unwrap(); + node_b.supervisor.packet_tx = Some(packet_tx_b.clone()); + node_b.packet_rx = Some(packet_rx_b); + node_b.supervisor.state = NodeState::Running; + node_b + .transports + .insert(transport_id_b, TransportHandle::Udp(transport_b)); + + let adopted_socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + let handoff = + EstablishedTraversal::new("sess-connected", node_b.npub(), addr_b, adopted_socket) + .with_transport_name("nostr-punched"); + + let result = node_a.adopt_established_traversal(handoff).await.unwrap(); + + // XX three-way handshake, driven directly: node_a sent msg1 on adoption, + // node_b answers with msg2, node_a sends msg3, and node_b promotes node_a + // only once it has processed msg3. + let mut rx_a = node_a.packet_rx.take().expect("node_a packet_rx"); + let mut rx_b = node_b.packet_rx.take().expect("node_b packet_rx"); + + let pkt_at_b = timeout(Duration::from_secs(1), rx_b.recv()) + .await + .expect("timeout waiting for node_a -> node_b msg1") + .expect("node_b channel closed"); + assert_eq!(pkt_at_b.data[0] & 0x0f, PHASE_MSG1); + node_b.handle_msg1(pkt_at_b).await; + + let pkt_at_a = timeout(Duration::from_secs(1), rx_a.recv()) + .await + .expect("timeout waiting for node_b -> node_a msg2") + .expect("node_a channel closed"); + assert_eq!(pkt_at_a.data[0] & 0x0f, PHASE_MSG2); + node_a.handle_msg2(pkt_at_a).await; + + let pkt_at_b = timeout(Duration::from_secs(1), rx_b.recv()) + .await + .expect("timeout waiting for node_a -> node_b msg3") + .expect("node_b channel closed"); + assert_eq!(pkt_at_b.data[0] & 0x0f, PHASE_MSG3); + node_b.handle_msg3(pkt_at_b).await; + + let peer_a_node_addr = + *PeerIdentity::from_pubkey_full(node_a.identity().pubkey_full()).node_addr(); + let peer_b_node_addr = + *PeerIdentity::from_pubkey_full(node_b.identity().pubkey_full()).node_addr(); + + // Preconditions: both sides are peered, and node_a reaches node_b over + // the adopted transport rather than some other one. + assert_eq!(node_a.peer_count(), 1, "node_a should promote node_b"); + assert_eq!(node_b.peer_count(), 1, "node_b should promote node_a"); + assert!(node_b.get_peer(&peer_a_node_addr).unwrap().has_session()); + let peer_on_a = node_a.get_peer(&peer_b_node_addr).unwrap(); + assert!(peer_on_a.has_session()); + assert_eq!( + peer_on_a.transport_id(), + Some(result.transport_id), + "node_a's peer must be on the adopted transport", + ); + assert!(peer_on_a.current_addr().is_some()); + + node_b.activate_connected_udp_sessions().await; + node_a.activate_connected_udp_sessions().await; + + assert!( + node_b + .get_peer(&peer_a_node_addr) + .unwrap() + .connected_udp() + .is_some(), + "node_b's peer on its configured listener must get a connected UDP socket; \ + if it does not, check whether FIPS_CONNECTED_UDP turns the fast path off here", + ); + + let Some(connected) = node_a.get_peer(&peer_b_node_addr).unwrap().connected_udp() else { + let direct = + match crate::transport::udp::open_connected_fd(result.local_addr, addr_b, 65536, 65536) + { + Ok(_) => "succeeds".to_string(), + Err(e) => format!("fails with {e}"), + }; + panic!( + "node_a's peer on the adopted transport must get a connected UDP socket; \ + opening one on the adopted socket's address directly {direct}" + ); + }; + assert_eq!( + connected.local_addr(), + result.local_addr, + "the connected socket must join the adopted socket's port, not another transport's", + ); + drop(connected); + + for (_, transport) in node_a.transports.iter_mut() { + transport.stop().await.ok(); + } + for (_, transport) in node_b.transports.iter_mut() { + transport.stop().await.ok(); + } +} diff --git a/src/node/tests/control.rs b/src/node/tests/control.rs index 4bdf9fb8..dbb08b99 100644 --- a/src/node/tests/control.rs +++ b/src/node/tests/control.rs @@ -6,6 +6,7 @@ //! socket framing. use super::*; +use heartbeat::set_heartbeat_interval; use spanning_tree::{ TestNode, add_loopback_alias, cleanup_nodes, drain_all_packets, make_test_node, process_available_packets, run_tree_test, @@ -421,3 +422,110 @@ async fn show_links_reports_the_traffic_counters_of_the_peer_bound_to_each_link( cleanup_nodes(&mut nodes).await; } + +/// Insert an authenticated peer last heard from at `last_seen_ms` and return +/// its address. +fn insert_peer_last_seen_at(node: &mut Node, link: u64, last_seen_ms: u64) -> NodeAddr { + let identity = PeerIdentity::from_pubkey_full(Identity::generate().pubkey_full()); + let addr = *identity.node_addr(); + node.peers.insert( + addr, + ActivePeer::new(identity, LinkId::new(link), last_seen_ms), + ); + addr +} + +/// The `connectivity` string a `show_peers` response gives the peer at `addr`. +fn connectivity_of(peers: &serde_json::Value, addr: &NodeAddr) -> String { + let addr_hex = hex::encode(addr.as_bytes()); + peers["peers"] + .as_array() + .expect("show_peers returns a peers array") + .iter() + .find(|row| row["node_addr"] == addr_hex.as_str()) + .and_then(|row| row["connectivity"].as_str()) + .expect("show_peers lists the peer with a connectivity string") + .to_string() +} + +/// Render `show_peers` on the loop, then publish a tick and render it again +/// from the snapshot the control socket serves. +fn show_peers_both_renders(node: &mut Node) -> [(&'static str, serde_json::Value); 2] { + let on_loop = crate::control::queries::show_peers(node); + node.record_stats_history(); + let off_loop = crate::control::queries::show_peers_from_handle(&node.control_read_handle()); + [("on-loop", on_loop), ("snapshot", off_loop)] +} + +/// `show_peers` reports a peer silent for longer than the heartbeat interval +/// as `stale`, and a peer heard from just now as `connected`, on both the +/// on-loop render and the tick-published snapshot render. +#[test] +fn show_peers_reports_a_peer_idle_past_the_heartbeat_interval_as_stale() { + let mut node = make_node(); + let interval_ms = node.config().node.heartbeat_interval_secs * 1000; + assert_eq!(interval_ms, 10_000, "the default heartbeat interval"); + let now = Node::now_ms(); + let fresh = insert_peer_last_seen_at(&mut node, 1, now); + let idle = insert_peer_last_seen_at(&mut node, 2, now - interval_ms - 5_000); + + for (render, peers) in show_peers_both_renders(&mut node) { + assert_eq!( + connectivity_of(&peers, &idle), + "stale", + "{render} render, peer silent for 15 s" + ); + assert_eq!( + connectivity_of(&peers, &fresh), + "connected", + "{render} render, peer heard from just now" + ); + } +} + +/// The `stale` threshold is the configured heartbeat interval rather than a +/// fixed ten seconds: with a 30 s interval a peer silent for 15 s still reads +/// `connected`, and one silent for 35 s reads `stale`. +#[test] +fn show_peers_stale_threshold_follows_the_configured_heartbeat_interval() { + let mut node = make_node(); + set_heartbeat_interval(&mut node, 30); + let now = Node::now_ms(); + let quiet = insert_peer_last_seen_at(&mut node, 1, now - 15_000); + let idle = insert_peer_last_seen_at(&mut node, 2, now - 35_000); + + for (render, peers) in show_peers_both_renders(&mut node) { + assert_eq!( + connectivity_of(&peers, &idle), + "stale", + "{render} render, peer silent for 35 s" + ); + assert_eq!( + connectivity_of(&peers, &quiet), + "connected", + "{render} render, peer silent for 15 s" + ); + } +} + +/// The derived connectivity changes at the heartbeat interval exactly: a peer +/// silent for the whole interval still reads `connected`, and one millisecond +/// more reads `stale`. A zero interval is floored at one second, the floor the +/// discovery re-dial gate applies. +#[test] +fn peer_connectivity_turns_stale_one_millisecond_past_the_heartbeat_interval() { + let mut node = make_node(); + let seen = 1_000_000; + let addr = insert_peer_last_seen_at(&mut node, 1, seen); + let at = |node: &Node, now_ms: u64| { + let peer = node.get_peer(&addr).expect("the peer was inserted"); + node.peer_connectivity(peer, now_ms) + }; + + assert_eq!(at(&node, seen + 10_000), ConnectivityState::Connected); + assert_eq!(at(&node, seen + 10_001), ConnectivityState::Stale); + + set_heartbeat_interval(&mut node, 0); + assert_eq!(at(&node, seen + 1_000), ConnectivityState::Connected); + assert_eq!(at(&node, seen + 1_001), ConnectivityState::Stale); +} diff --git a/src/node/tests/handshake.rs b/src/node/tests/handshake.rs index cb7bb2f7..9814785e 100644 --- a/src/node/tests/handshake.rs +++ b/src/node/tests/handshake.rs @@ -217,13 +217,6 @@ async fn test_two_node_handshake_udp() { node_b.handle_encrypted_frame(encrypted_packet_b).await; - // Verify B's peer was touched (last_seen updated) - let peer_a = node_b.get_peer(&peer_a_node_addr).unwrap(); - assert!( - peer_a.is_healthy(), - "Peer A on B should still be healthy after receiving encrypted frame" - ); - // === Phase 5: Encrypted frame B → A === // Prepend inner header (timestamp + msg_type) as the real send path does @@ -251,13 +244,6 @@ async fn test_two_node_handshake_udp() { node_a.handle_encrypted_frame(encrypted_packet_a).await; - // Verify A's peer was touched - let peer_b = node_a.get_peer(&peer_b_node_addr).unwrap(); - assert!( - peer_b.is_healthy(), - "Peer B on A should still be healthy after receiving encrypted frame" - ); - // Clean up transports for (_, t) in node_a.transports.iter_mut() { t.stop().await.ok(); @@ -704,8 +690,6 @@ async fn test_cross_connection_both_initiate() { assert!(peer_b_on_a.has_session(), "Peer B on A should have session"); assert!(peer_a_on_b.has_session(), "Peer A on B should have session"); - assert!(peer_b_on_a.can_send(), "Peer B on A should be sendable"); - assert!(peer_a_on_b.can_send(), "Peer A on B should be sendable"); // The property the tie-break exists to produce: both ends kept the SAME // session, not merely a session each. The index pair is what makes that @@ -1916,13 +1900,35 @@ async fn test_msg3_dual_rekey_won_frees_index() { stop_hs(&mut responder).await; } +/// Complete a Noise XX handshake between two identities and return the +/// initiator's session, standing in for the fresh keys a cross-connection swap +/// installs. +fn replacement_session(ours: &Identity, theirs: &Identity) -> crate::noise::NoiseSession { + use crate::noise::HandshakeState; + + let mut initiator = HandshakeState::new_initiator(ours.keypair()); + let mut responder = HandshakeState::new_responder(theirs.keypair()); + initiator.set_local_epoch([0x11; 8]); + responder.set_local_epoch([0x22; 8]); + + let msg1 = initiator.write_message_1().unwrap(); + responder.read_message_1(&msg1).unwrap(); + let msg2 = responder.write_message_2().unwrap(); + initiator.read_message_2(&msg2).unwrap(); + let msg3 = initiator.write_message_3().unwrap(); + responder.read_message_3(&msg3).unwrap(); + + initiator.into_session().unwrap() +} + #[tokio::test] async fn test_msg3_resend_msg2_frees_index() { - // A declared rekey landing on a peer whose link is no longer healthy. The - // marker matches, so this is unambiguously a rekey and not a crossing dial; - // the classifier's health conjunct is what sends it to the duplicate arm, - // and the shell must then free the msg1-allocated index and leave the - // existing session alone. + // A declared rekey naming keys the responder no longer holds. Production + // reaches this when a cross-connection swap replaces the responder's keys + // for the peer while the peer's rekey is in flight: the msg3 marker then + // resolves to a mismatch, which is unambiguously a rekey and not a crossing + // dial, and the classifier sends it to the duplicate arm. The shell must + // then free the msg1-allocated index and leave the existing session alone. // // A bare second handshake will not reach this arm: it declares no rekey, and // an undeclared msg3 on a different link is a cross-connection, on which @@ -1960,9 +1966,6 @@ async fn test_msg3_resend_msg2_frees_index() { .unwrap() .test_backdate_session_established(std::time::Duration::from_secs(120)); - let before = responder.node.get_peer(&peer_addr).unwrap(); - let session_before = (before.our_index(), before.their_index()); - // The rekey's msg1 allocates a fresh index; its msg3 then frees it on the // duplicate arm. let msg3b = drive_rekey_to_msg3(&mut initiator, &mut responder).await; @@ -1972,20 +1975,67 @@ async fn test_msg3_resend_msg2_frees_index() { "the rekey msg1 allocated a fresh index" ); - // The link goes quiet past the heartbeat threshold while the rekey is in - // flight, exactly as the tick loop would mark it. An unhealthy peer is not a - // rekey candidate, so the declared rekey falls through to the duplicate arm. - // Marked after the msg1/msg2 exchange, since nothing on the handshake path - // re-marks a peer connected but a later `touch` would. + // The index the msg3 marker declares: the responder's index as the + // initiator knows it. + let declared = initiator + .node + .get_peer(&responder_addr) + .unwrap() + .their_index() + .expect("initiator holds the responder's index"); + + // While the rekey is in flight, replace the responder's keys for the peer + // exactly as the outbound cross-connection swap does: a fresh index and + // session on the peer, the index map moved to the new index, the old index + // freed, and the peer's control machine told of the swap. + let fresh_index = responder.node.index_allocator.allocate().unwrap(); + let session = replacement_session(responder.node.identity(), initiator.node.identity()); + let (old_index, their_index, transport_id, link) = { + let peer = responder.node.get_peer_mut(&peer_addr).unwrap(); + let their_index = peer + .their_index() + .expect("responder holds the peer's index"); + let old_index = peer.replace_session(session, fresh_index, their_index); + ( + old_index.expect("responder held an index before the swap"), + their_index, + peer.transport_id().expect("peer has a transport"), + peer.link_id(), + ) + }; responder .node - .get_peer_mut(&peer_addr) - .unwrap() - .mark_stale(); - assert!( - !responder.node.get_peer(&peer_addr).unwrap().is_healthy(), - "the peer must be unhealthy, or this reaches the rekey-responder arm" + .peers_by_index + .remove(&(transport_id, old_index.as_u32())); + let _ = responder.node.index_allocator.free(old_index); + responder + .node + .peers_by_index + .insert((transport_id, fresh_index.as_u32()), peer_addr); + let acts = responder.node.peer_machines.get_mut(&link).unwrap().step( + crate::peer::machine::PeerEvent::CrossConnResolved { + outcome: crate::peer::machine::CrossConnOutcome::Swap { + our_index: fresh_index, + their_index, + }, + }, + Node::now_ms(), + &mut responder.node.index_allocator, ); + assert!( + acts.is_empty(), + "cross-connection resolution is a pure observation" + ); + + assert_ne!( + responder.node.get_peer(&peer_addr).unwrap().our_index(), + Some(declared), + "the responder must no longer hold the keys the msg3 declares, or this \ + reaches the rekey-responder arm" + ); + + let before = responder.node.get_peer(&peer_addr).unwrap(); + let session_before = (before.our_index(), before.their_index()); responder.node.handle_msg3(msg3b).await; diff --git a/src/node/tests/heartbeat.rs b/src/node/tests/heartbeat.rs index f8bb2666..c9dee963 100644 --- a/src/node/tests/heartbeat.rs +++ b/src/node/tests/heartbeat.rs @@ -37,7 +37,7 @@ fn set_link_dead_timeout(node: &mut crate::node::Node, secs: u64) { /// Set `node.heartbeat_interval_secs` on an already-constructed node, the same /// way `set_link_dead_timeout` does. This is the knob the retry gate must not /// floor. -fn set_heartbeat_interval(node: &mut crate::node::Node, secs: u64) { +pub(super) fn set_heartbeat_interval(node: &mut crate::node::Node, secs: u64) { node.replace_context(|ctx| { let mut cfg = (*ctx.config).clone(); cfg.node.heartbeat_interval_secs = secs; diff --git a/src/node/tests/routing.rs b/src/node/tests/routing.rs index a9d51708..fb8fd58b 100644 --- a/src/node/tests/routing.rs +++ b/src/node/tests/routing.rs @@ -1698,42 +1698,6 @@ fn test_seam_bloom_hit_overrides_tree_tiebreak() { ); } -/// `NodeRoutingView::peer_can_send` keeps a down link out of the candidate set. -/// -/// Both peers hold a bloom hit, so the address tie-break would hand the route -/// to the low-address peer; that peer is the one marked reconnecting. Widening -/// `peer_can_send` to `true` lets it back in and hands a down link to the -/// forwarder. -#[test] -fn test_seam_unsendable_bloom_candidate_is_skipped() { - let mut node = make_node(); - let (near, far, dest) = seam_two_equidistant_peers(&mut node); - let low = near.min(far); - let high = near.max(far); - - seam_set_filter(&mut node, &low, &dest); - seam_set_filter(&mut node, &high, &dest); - node.get_peer_mut(&low).unwrap().mark_reconnecting(); - - assert_eq!(node.peers.len(), 2, "fixture: two peers"); - assert!( - !node.get_peer(&low).unwrap().can_send(), - "fixture: low is down" - ); - assert!( - node.get_peer(&high).unwrap().can_send(), - "fixture: high is up" - ); - - let hop = node.find_next_hop(&dest).expect("route exists"); - assert!(hop.can_send(), "a down link must never be returned"); - assert_eq!( - hop.node_addr(), - &high, - "the sendable peer must win despite losing the address tie-break" - ); -} - /// `NodeRoutingView::peer_link_cost` must carry the ETX factor. /// /// SRTT is equal on both peers, so ETX is the only thing that can order them, @@ -1847,12 +1811,11 @@ fn test_seam_peer_without_tree_coords_is_never_selected() { let (near, far, dest) = seam_two_equidistant_peers(&mut node); let tree_pick = near.min(far); - // A third peer: in the peer map, sendable, holding a bloom hit for dest, + // A third peer: in the peer map, holding a bloom hit for dest, // and absent from tree state. let ghost = seam_add_peer(&mut node, 3, TransportId::new(1)); seam_set_filter(&mut node, &ghost, &dest); - assert!(node.get_peer(&ghost).unwrap().can_send()); assert!(node.get_peer(&ghost).unwrap().may_reach(&dest)); assert!( node.tree_state().peer_coords(&ghost).is_none(), @@ -1897,7 +1860,6 @@ fn test_seam_routing_view_reads_match_live_peer_state() { // Make the two peers differ on every predicate, in opposite directions, so // no constant in either direction can satisfy the assertions below. seam_set_filter(&mut node, &near, &dest); - node.get_peer_mut(&far).unwrap().mark_reconnecting(); // Both cost factors off the multiplicative identity: with etx pinned at 1.0 // a cost that reads only the latency half is indistinguishable from the // real one, and this assertion would be blind to it. @@ -1949,10 +1911,6 @@ fn test_seam_routing_view_reads_match_live_peer_state() { "the filter is per-destination" ); - // peer_can_send: far was marked reconnecting. - assert!(view.peer_can_send(near_h)); - assert!(!view.peer_can_send(far_h)); - // peer_link_cost: etx 2.0 * (1.0 + 50ms/100) for near; far has no RTT // sample, so it takes the optimistic 1.0 default. Neither factor is at the // identity, so a cost reading only one half of the product is caught. @@ -1993,7 +1951,6 @@ fn test_seam_routing_view_reads_match_live_peer_state() { let addr = view.peer_addr(*peer); let live = node.peers.get(&addr).unwrap(); assert_eq!(view.peer_may_reach(*peer, &dest), live.may_reach(&dest)); - assert_eq!(view.peer_can_send(*peer), live.can_send()); assert_eq!(view.peer_link_cost(*peer), live.link_cost()); assert_eq!( view.peer_coords(*peer), diff --git a/src/node/tests/tcp.rs b/src/node/tests/tcp.rs index 2d2a80d4..841ac12f 100644 --- a/src/node/tests/tcp.rs +++ b/src/node/tests/tcp.rs @@ -459,3 +459,43 @@ async fn test_api_disconnect_closes_the_tcp_connection() { cleanup_nodes(&mut nodes).await; } + +/// The Disconnect `api_disconnect` sends must reach the peer before the TCP +/// connection closes, so the peer forgets this node at once. +/// +/// The send only queues the Disconnect for the connection's writer, and the +/// close follows in the same call. Nothing else removes the peer on node 1 +/// within the wait: its TCP EOF removes only the pool entry, and link-dead +/// detection takes far longer. +#[tokio::test] +async fn api_disconnect_delivers_the_disconnect_before_closing() { + let mut nodes = vec![make_test_node_tcp().await, make_test_node_tcp().await]; + + initiate_handshake(&mut nodes, 0, 1).await; + drain_all_packets(&mut nodes, false).await; + + let addr_0 = *nodes[0].node.node_addr(); + let node1_npub = nodes[1].node.npub(); + assert!( + nodes[1].node.get_peer(&addr_0).is_some(), + "node 1 should have node 0 as peer" + ); + + nodes[0] + .node + .api_disconnect(&node1_npub) + .await + .expect("api_disconnect should succeed"); + + let deadline = std::time::Instant::now() + Duration::from_secs(2); + while nodes[1].node.get_peer(&addr_0).is_some() && std::time::Instant::now() < deadline { + spanning_tree::process_available_packets(&mut nodes).await; + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert!( + nodes[1].node.get_peer(&addr_0).is_none(), + "node 1 never received the Disconnect sent before the close" + ); + + cleanup_nodes(&mut nodes).await; +} diff --git a/src/node/tests/unit.rs b/src/node/tests/unit.rs index 8fe1ca8a..2d364132 100644 --- a/src/node/tests/unit.rs +++ b/src/node/tests/unit.rs @@ -692,30 +692,30 @@ fn test_node_sendable_peers() { let mut node = make_node(); let transport_id = TransportId::new(1); - // Add a healthy peer + // Add a peer let link_id1 = LinkId::new(1); let identity1 = seed_completed_connection(&mut node, link_id1, transport_id, 1000); let node_addr1 = *identity1.node_addr(); node.promote_connection(link_id1, identity1, 2000).unwrap(); - // Add another peer and mark it stale (still sendable) + // Add another peer let link_id2 = LinkId::new(2); let identity2 = seed_completed_connection(&mut node, link_id2, transport_id, 1000); node.promote_connection(link_id2, identity2, 2000).unwrap(); - // Add a third peer and mark it disconnected (not sendable) + // Add a third peer let link_id3 = LinkId::new(3); let identity3 = seed_completed_connection(&mut node, link_id3, transport_id, 1000); let node_addr3 = *identity3.node_addr(); node.promote_connection(link_id3, identity3, 2000).unwrap(); - node.get_peer_mut(&node_addr3).unwrap().mark_disconnected(); assert_eq!(node.peer_count(), 3); - assert_eq!(node.sendable_peer_count(), 2); + assert_eq!(node.sendable_peer_count(), 3); let sendable: Vec<_> = node.sendable_peers().collect(); - assert_eq!(sendable.len(), 2); + assert_eq!(sendable.len(), 3); assert!(sendable.iter().any(|p| p.node_addr() == &node_addr1)); + assert!(sendable.iter().any(|p| p.node_addr() == &node_addr3)); } // === RX Loop Tests === diff --git a/src/peer/active.rs b/src/peer/active.rs index a773ae18..090e2b97 100644 --- a/src/peer/active.rs +++ b/src/peer/active.rs @@ -35,38 +35,24 @@ fn draw_rekey_jitter() -> i64 { rand::rng().random_range(-REKEY_JITTER_SECS..=REKEY_JITTER_SECS) } -/// Connectivity state for an active peer. +/// Connectivity of an active peer, as the control socket reports it. /// -/// This is simpler than the full PeerState since authentication is complete. +/// Not stored on the peer: the node derives it from how long the peer has +/// been silent, compared with the configured heartbeat interval. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum ConnectivityState { - /// Peer is fully connected and responsive. + /// Heard from within the heartbeat interval. Connected, - /// Peer hasn't been heard from recently (potential timeout). + /// Silent for longer than the heartbeat interval. Stale, - /// Connection lost, attempting to reconnect. - Reconnecting, - /// Peer has been explicitly disconnected. - Disconnected, } impl ConnectivityState { - /// Check if the peer is usable for sending traffic. - pub fn can_send(&self) -> bool { - matches!( - self, - ConnectivityState::Connected | ConnectivityState::Stale - ) - } - /// Check if this is a terminal state requiring cleanup. + /// + /// Always false: neither derived state is terminal. pub fn is_terminal(&self) -> bool { - matches!(self, ConnectivityState::Disconnected) - } - - /// Check if peer is fully healthy. - pub fn is_healthy(&self) -> bool { - matches!(self, ConnectivityState::Connected) + false } } @@ -75,8 +61,6 @@ impl fmt::Display for ConnectivityState { let s = match self { ConnectivityState::Connected => "connected", ConnectivityState::Stale => "stale", - ConnectivityState::Reconnecting => "reconnecting", - ConnectivityState::Disconnected => "disconnected", }; write!(f, "{}", s) } @@ -225,10 +209,6 @@ pub struct ActivePeer { /// Immutable for the same reason as [`ActivePeer::npub`]. short_npub: String, - // === Connection === - /// Current connectivity state. - connectivity: ConnectivityState, - // === Spanning Tree === /// Their latest parent declaration. declaration: Option, @@ -346,7 +326,6 @@ impl ActivePeer { npub: identity.npub(), short_npub: identity.short_npub(), identity, - connectivity: ConnectivityState::Connected, declaration: None, ancestry: None, tree_announce_min_interval_ms: 500, @@ -441,7 +420,6 @@ impl ActivePeer { npub: identity.npub(), short_npub: identity.short_npub(), identity, - connectivity: ConnectivityState::Connected, declaration: None, ancestry: None, tree_announce_min_interval_ms: 500, @@ -564,24 +542,12 @@ impl ActivePeer { self.send.link_id } - /// Get the connectivity state. - pub fn connectivity(&self) -> ConnectivityState { - self.connectivity - } - - /// Check if peer can receive traffic. - pub fn can_send(&self) -> bool { - self.connectivity.can_send() - } - - /// Check if peer is fully healthy. - pub fn is_healthy(&self) -> bool { - self.connectivity.is_healthy() - } - /// Check if peer is disconnected. + /// + /// Always false: the peer stores no connectivity state, and a peer that + /// goes away is removed from the node rather than marked. pub fn is_disconnected(&self) -> bool { - self.connectivity.is_terminal() + false } // === Session Accessors === @@ -920,33 +886,6 @@ impl ActivePeer { /// Update last seen timestamp. pub fn touch(&mut self, current_time_ms: u64) { 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; - } - } - - /// Mark peer as stale (no recent traffic). - pub fn mark_stale(&mut self) { - if self.connectivity == ConnectivityState::Connected { - self.connectivity = ConnectivityState::Stale; - } - } - - /// Mark peer as reconnecting. - pub fn mark_reconnecting(&mut self) { - self.connectivity = ConnectivityState::Reconnecting; - } - - /// Mark peer as disconnected. - pub fn mark_disconnected(&mut self) { - self.connectivity = ConnectivityState::Disconnected; - } - - /// Mark peer as connected (e.g., after successful reconnect). - pub fn mark_connected(&mut self, current_time_ms: u64) { - self.connectivity = ConnectivityState::Connected; - self.send.last_seen = current_time_ms; } /// Update the link ID (e.g., on reconnect). @@ -1633,16 +1572,8 @@ mod tests { #[test] fn test_connectivity_state_properties() { - assert!(ConnectivityState::Connected.can_send()); - assert!(ConnectivityState::Stale.can_send()); - assert!(!ConnectivityState::Reconnecting.can_send()); - assert!(!ConnectivityState::Disconnected.can_send()); - - assert!(ConnectivityState::Connected.is_healthy()); - assert!(!ConnectivityState::Stale.is_healthy()); - - assert!(ConnectivityState::Disconnected.is_terminal()); assert!(!ConnectivityState::Connected.is_terminal()); + assert!(!ConnectivityState::Stale.is_terminal()); } #[test] @@ -1652,8 +1583,7 @@ mod tests { assert_eq!(peer.identity().node_addr(), identity.node_addr()); assert_eq!(peer.link_id(), LinkId::new(1)); - assert!(peer.is_healthy()); - assert!(peer.can_send()); + assert!(!peer.is_disconnected()); assert_eq!(peer.authenticated_at(), 1000); assert!(peer.needs_filter_update()); // New peers need filter } @@ -1714,32 +1644,6 @@ mod tests { assert_eq!(short_first, short_second); } - #[test] - fn test_connectivity_transitions() { - let identity = make_peer_identity(); - let mut peer = ActivePeer::new(identity, LinkId::new(1), 1000); - - assert!(peer.is_healthy()); - - peer.mark_stale(); - assert_eq!(peer.connectivity(), ConnectivityState::Stale); - assert!(peer.can_send()); // Stale can still send - - // Traffic received brings back to connected - peer.touch(2000); - assert!(peer.is_healthy()); - - peer.mark_reconnecting(); - assert!(!peer.can_send()); - - peer.mark_connected(3000); - assert!(peer.is_healthy()); - - peer.mark_disconnected(); - assert!(peer.is_disconnected()); - assert!(!peer.can_send()); - } - #[test] fn test_tree_position() { let identity = make_peer_identity(); diff --git a/src/peer/machine.rs b/src/peer/machine.rs index dd68a076..ba352fc7 100644 --- a/src/peer/machine.rs +++ b/src/peer/machine.rs @@ -2303,7 +2303,6 @@ mod tests { has_existing_peer: false, existing_peer_epoch: None, has_session: false, - is_healthy: false, pending_new_session: false, rekey_in_progress: false, existing_msg2: None, @@ -2744,7 +2743,6 @@ mod tests { est.has_existing_peer = true; est.existing_peer_epoch = Some([1u8; 8]); est.has_session = true; - est.is_healthy = true; est.rekey_claim = RekeyClaim::Matches; // declared rekey of our session est.rekey_in_progress = true; let wire = wire_outcome(peer_addr, Some([1u8; 8])); @@ -2791,7 +2789,6 @@ mod tests { est.has_existing_peer = true; est.existing_peer_epoch = Some([1u8; 8]); est.has_session = true; - est.is_healthy = true; est.rekey_claim = RekeyClaim::Matches; est.rekey_in_progress = true; let wire = wire_outcome(peer_addr, Some([1u8; 8])); @@ -3112,7 +3109,6 @@ mod tests { est.has_existing_peer = true; est.existing_peer_epoch = Some([5u8; 8]); est.has_session = true; - est.is_healthy = true; est.different_link = true; est.rekey_claim = RekeyClaim::None; // no rekey declared -> cross-connection let wire = wire_outcome(peer_addr, Some([5u8; 8])); @@ -3159,7 +3155,6 @@ mod tests { est.has_existing_peer = true; est.existing_peer_epoch = Some([5u8; 8]); est.has_session = true; - est.is_healthy = true; est.different_link = true; est.rekey_claim = RekeyClaim::None; // no rekey declared -> cross-connection let wire = wire_outcome(peer_addr, Some([5u8; 8])); @@ -3199,13 +3194,12 @@ mod tests { assert_eq!(m.our_index(), None); let seed = SessionIndex::new(0xCD); - // Aged, healthy session, same epoch, same link, no rekey in progress -> + // Aged session, same epoch, same link, no rekey in progress -> // plain rekey responder (abandon_first: false). let mut est = est_new_peer(our); est.has_existing_peer = true; est.existing_peer_epoch = Some([7u8; 8]); est.has_session = true; - est.is_healthy = true; est.rekey_claim = RekeyClaim::Matches; // declared rekey of our session let wire = wire_outcome(peer_addr, Some([7u8; 8])); diff --git a/src/proto/fmp/core.rs b/src/proto/fmp/core.rs index d39d0d10..8ca5db61 100644 --- a/src/proto/fmp/core.rs +++ b/src/proto/fmp/core.rs @@ -265,8 +265,6 @@ pub(crate) struct EstablishSnapshot { pub existing_peer_epoch: Option<[u8; 8]>, /// The existing peer has an established Noise session. pub has_session: bool, - /// The existing peer's session is healthy. - pub is_healthy: bool, /// The existing peer already holds a pending post-rekey session awaiting /// K-bit cutover. On XX this is one of the two dual-init tie-break states /// (the widened window IK never reached) — NOT an unconditional reject. @@ -439,7 +437,7 @@ pub(crate) trait LifecycleView { /// timeout/failed predicate; the core decides retry-then-teardown. fn stale_connections(&self, now_ms: u64, timeout_ms: u64) -> Vec; - /// Snapshot every active peer with a session that is healthy, pre-computing + /// Snapshot every active peer with a session, pre-computing /// its rekey-relevant ages and timer predicates (see [`PeerSnapshot`]). The /// shell resolves every clock read here; the core applies the thresholds. fn rekey_peers(&self) -> Vec; @@ -474,7 +472,7 @@ pub(crate) enum InboundDecision { /// same promote sequence as [`Promote`](InboundDecision::Promote). `peer` is /// the teardown / reconnect target. RestartThenPromote { peer: NodeAddr }, - /// Same-epoch cross-connection resolved inline on `msg3`: a healthy session + /// Same-epoch cross-connection resolved inline on `msg3`: a session /// with no rekey of ours in flight received a concurrent `msg3` on a /// different link that declares no rekey — so it is a fresh dial crossing /// ours. `our_inbound_wins` (the larger-NodeAddr side) selects @@ -484,7 +482,7 @@ pub(crate) enum InboundDecision { peer: NodeAddr, our_inbound_wins: bool, }, - /// Same-epoch `msg3` on a healthy session, declared by its sender to replace + /// Same-epoch `msg3` on a session, declared by its sender to replace /// that very session: respond as the rekey responder. The shell extracts the /// fresh Noise session from the live /// connection, allocates a new index, and stores it as the peer's pending @@ -658,7 +656,7 @@ impl Fmp { .collect() } - /// Decide the per-tick rekey choreography for the healthy peers the shell + /// Decide the per-tick rekey choreography for the peers the shell /// snapshotted. Reproduces the pre-refactor priority and phase grouping /// exactly: /// @@ -773,7 +771,7 @@ impl Fmp { /// ([`Mismatch`](RekeyClaim::Mismatch)) → [`ResendMsg2`], not a reject: /// the sender has already committed to its pending session and the reject /// path sends nothing back. - /// 4. Same epoch, no marker, different link, healthy session → inline + /// 4. Same epoch, no marker, different link, with a session → inline /// [`CrossConnect`] (the XX widening: IK resolves this on `msg2`). /// `our_inbound_wins` is the larger-NodeAddr side, matching /// `cross_connection_winner(our, peer, /*outbound=*/ false)`. Taken @@ -781,7 +779,7 @@ impl Fmp { /// this tie-break cannot see that state, so declining here would diverge /// the pair; the executor abandons the displaced rekey instead. /// 5. Same epoch, marker naming the session we hold - /// ([`Matches`](RekeyClaim::Matches)), healthy session → [`RekeyRespond`], + /// ([`Matches`](RekeyClaim::Matches)), with a session → [`RekeyRespond`], /// whatever the local `rekey.enabled` says: that flag governs whether we /// initiate rekeys, never whether we accept one, or an asymmetric setting /// would diverge the two ends of the link. @@ -866,11 +864,7 @@ impl Fmp { // The `SwapToInboundSession` arm therefore abandons that rekey as // part of the swap, the same way the rekey-responder arm does when // it loses the dual-rekey tie-break. - if snap.rekey_claim == RekeyClaim::None - && snap.different_link - && snap.has_session - && snap.is_healthy - { + if snap.rekey_claim == RekeyClaim::None && snap.different_link && snap.has_session { // `cross_connection_winner(our, peer, this_is_outbound=false)`: // the smaller node prefers its outbound, so our *inbound* // wins iff we are the larger node (the exact negation of the @@ -892,7 +886,7 @@ impl Fmp { // timer. The sender's declaration matching the session we hold is // the authoritative signal, and it is a property of the wire, not // of our config. - if snap.rekey_claim == RekeyClaim::Matches && snap.has_session && snap.is_healthy { + if snap.rekey_claim == RekeyClaim::Matches && snap.has_session { // Widened dual-init tie-break: both the still-in-progress and // the already-pending states resolve by the smaller NodeAddr. if snap.rekey_in_progress || snap.pending_new_session { diff --git a/src/proto/fmp/tests/core.rs b/src/proto/fmp/tests/core.rs index d9aa360f..441d6ccd 100644 --- a/src/proto/fmp/tests/core.rs +++ b/src/proto/fmp/tests/core.rs @@ -546,7 +546,7 @@ fn establish_same_link_fresh_is_duplicate() { } } -/// Aged, healthy, same-epoch session with no dual-init in flight -> rekey +/// Aged, same-epoch session with no dual-init in flight -> rekey /// responder, no prior abandon. /// /// The snapshot carries no local rekey-trigger config, and deliberately: the @@ -720,13 +720,28 @@ fn establish_undeclared_during_our_rekey_still_cross_connects() { } } -/// An unhealthy or session-less aged peer is not a rekey candidate -> duplicate. +/// A declared rekey naming keys we do not hold, on the link the peer already +/// uses, is not a rekey -> duplicate. #[test] -fn establish_aged_unhealthy_is_duplicate() { +fn establish_mismatched_claim_on_the_same_link_is_duplicate() { + let fmp = Fmp::new(); + let mut snap = establish_snapshot(0x05); + snap.rekey_claim = RekeyClaim::Mismatch; + let wire = wire_outcome(0x02, SAME_EPOCH); + assert!(matches!( + fmp.establish_inbound(&snap, &wire), + InboundDecision::ResendMsg2 { .. } + )); +} + +/// A declared rekey naming our session, when we hold no session keys for the +/// peer, is not a rekey -> duplicate. +#[test] +fn establish_matching_claim_without_a_session_is_duplicate() { let fmp = Fmp::new(); let mut snap = establish_snapshot(0x05); snap.rekey_claim = RekeyClaim::Matches; - snap.is_healthy = false; + snap.has_session = false; let wire = wire_outcome(0x02, SAME_EPOCH); assert!(matches!( fmp.establish_inbound(&snap, &wire), diff --git a/src/proto/fmp/tests/util.rs b/src/proto/fmp/tests/util.rs index 41a72024..a4490a3b 100644 --- a/src/proto/fmp/tests/util.rs +++ b/src/proto/fmp/tests/util.rs @@ -22,7 +22,7 @@ pub(super) fn rekey_resend_snapshot( } } -/// Build a quiescent `PeerSnapshot` for `addr`: session-healthy but with no +/// Build a quiescent `PeerSnapshot` for `addr`, with no /// pending cutover, no drain, no dampening, zero ages/counter/jitter. Tests set /// only the fields the case exercises. pub(super) fn peer_snapshot(addr_byte: u8) -> PeerSnapshot { @@ -70,7 +70,7 @@ pub(super) fn resend_snapshot(link: LinkId, resend_count: u32, msg1: Vec) -> } } -/// Build an `EstablishSnapshot` describing an existing, healthy, same-epoch +/// Build an `EstablishSnapshot` describing an existing, same-epoch /// peer owned by node `our_byte`. /// The default is a quiescent session on the same link with no /// in-flight rekey / pending and no rekey declared by the sender. Tests override @@ -81,7 +81,6 @@ pub(super) fn establish_snapshot(our_byte: u8) -> EstablishSnapshot { has_existing_peer: true, existing_peer_epoch: Some([0x01; 8]), has_session: true, - is_healthy: true, pending_new_session: false, rekey_in_progress: false, existing_msg2: None, diff --git a/src/proto/routing/core.rs b/src/proto/routing/core.rs index d7168c66..8d30480d 100644 --- a/src/proto/routing/core.rs +++ b/src/proto/routing/core.rs @@ -48,8 +48,6 @@ pub(crate) trait RoutingView { /// Does `peer`'s bloom filter indicate it may reach `dest`? The raw /// per-peer predicate the core filters candidates on. fn peer_may_reach<'a>(&'a self, peer: Self::Peer<'a>, dest: &NodeAddr) -> bool; - /// Can `peer`'s session currently carry a forward? - fn peer_can_send<'a>(&'a self, peer: Self::Peer<'a>) -> bool; /// `peer`'s outgoing link cost (lower is preferred). fn peer_link_cost<'a>(&'a self, peer: Self::Peer<'a>) -> f64; /// `peer`'s tree coordinates, if known. @@ -333,15 +331,15 @@ impl RouteClass { /// Select the best next hop from the active peers that may reach `dest`. /// -/// Enumerates borrowed peers through [`RoutingView`], applies the bloom and -/// send-eligibility filters, and tracks the best hop inline without allocating -/// candidate vectors or cloning coordinates. Only peers strictly closer to the -/// destination than we are (`my_coords`) are eligible — the self-distance check -/// that prevents routing loops. +/// Enumerates borrowed peers through [`RoutingView`], applies the bloom filter, +/// and tracks the best hop inline without allocating candidate vectors or +/// cloning coordinates. Only peers strictly closer to the destination than we +/// are (`my_coords`) are eligible — the self-distance check that prevents +/// routing loops. /// /// Ordering: `(link_cost, distance_to_dest, node_addr)`. Returns the winning -/// peer's address, or `None` when no candidate is send-ready and strictly -/// closer to the destination than us. +/// peer's address, or `None` when no candidate is strictly closer to the +/// destination than us. pub(crate) fn select_best_candidate( rv: &impl RoutingView, dest: &NodeAddr, @@ -353,7 +351,7 @@ pub(crate) fn select_best_candidate( let mut best: Option<(NodeAddr, f64, usize)> = None; rv.for_each_peer(|peer| { - if !rv.peer_may_reach(peer, dest) || !rv.peer_is_full(peer) || !rv.peer_can_send(peer) { + if !rv.peer_may_reach(peer, dest) || !rv.peer_is_full(peer) { return; } diff --git a/src/proto/routing/tests/core.rs b/src/proto/routing/tests/core.rs index 19ee9733..6eeb945f 100644 --- a/src/proto/routing/tests/core.rs +++ b/src/proto/routing/tests/core.rs @@ -28,14 +28,12 @@ fn mock_peer( addr: u8, dest: NodeAddr, may_reach: bool, - can_send: bool, link_cost: f64, coords: Option<&[u8]>, ) -> MockPeer { MockPeer { addr: make_node_addr(addr), reach: may_reach.then_some(dest).into_iter().collect(), - can_send, link_cost, coords: coords.map(make_coords), is_full: true, @@ -48,7 +46,7 @@ fn mock_peer( fn mock_non_full_peer(addr: u8, dest: NodeAddr, link_cost: f64, coords: Option<&[u8]>) -> MockPeer { MockPeer { is_full: false, - ..mock_peer(addr, dest, true, true, link_cost, coords) + ..mock_peer(addr, dest, true, link_cost, coords) } } @@ -295,8 +293,8 @@ fn candidate_selection_is_independent_of_peer_enumeration_order() { let root = 0x00; let my_coords = make_coords(&[0x10, root]); let dest_coords = make_coords(&[0x50, root]); - let lower_addr = mock_peer(0x20, dest, true, true, 1.0, Some(&[root])); - let higher_addr = mock_peer(0x30, dest, true, true, 1.0, Some(&[root])); + let lower_addr = mock_peer(0x20, dest, true, 1.0, Some(&[root])); + let higher_addr = mock_peer(0x30, dest, true, 1.0, Some(&[root])); let forward = MockRoutingView { peers: vec![lower_addr.clone(), higher_addr.clone()], @@ -318,7 +316,7 @@ fn candidate_selection_is_independent_of_peer_enumeration_order() { } #[test] -fn candidate_selection_filters_bloom_unsendable_and_missing_coords() { +fn candidate_selection_filters_bloom_and_missing_coords() { let dest = make_node_addr(0x50); let root = 0x00; let my_coords = make_coords(&[0x10, root]); @@ -326,10 +324,9 @@ fn candidate_selection_filters_bloom_unsendable_and_missing_coords() { let eligible = make_node_addr(0x60); let rv = MockRoutingView { peers: vec![ - mock_peer(0x01, dest, true, false, 0.0, Some(&[0x50, root])), - mock_peer(0x02, dest, true, true, 0.0, None), - mock_peer(0x03, dest, false, true, 0.0, Some(&[0x50, root])), - mock_peer(0x60, dest, true, true, 10.0, Some(&[root])), + mock_peer(0x02, dest, true, 0.0, None), + mock_peer(0x03, dest, false, 0.0, Some(&[0x50, root])), + mock_peer(0x60, dest, true, 10.0, Some(&[root])), ], ..MockRoutingView::new(false) }; @@ -349,9 +346,9 @@ fn candidate_must_be_strictly_closer_than_self() { let rv = MockRoutingView { peers: vec![ // A sibling is exactly as far from dest as this node. - mock_peer(0x20, dest, true, true, 1.0, Some(&[0x20, root])), + mock_peer(0x20, dest, true, 1.0, Some(&[0x20, root])), // This descendant of a sibling is farther from dest. - mock_peer(0x21, dest, true, true, 0.5, Some(&[0x21, 0x20, root])), + mock_peer(0x21, dest, true, 0.5, Some(&[0x21, 0x20, root])), ], ..MockRoutingView::new(false) }; @@ -368,12 +365,12 @@ fn candidate_ordering_is_cost_then_distance_then_address() { let rv = MockRoutingView { peers: vec![ // Lowest address loses because distance precedes address. - mock_peer(0x01, dest, true, true, 1.0, Some(&[root])), + mock_peer(0x01, dest, true, 1.0, Some(&[root])), // Closest peer loses because cost is the primary key. - mock_peer(0x02, dest, true, true, 1.0, Some(&[0x50, root])), - mock_peer(0x04, dest, true, true, 0.5, Some(&[root])), + mock_peer(0x02, dest, true, 1.0, Some(&[0x50, root])), + mock_peer(0x04, dest, true, 0.5, Some(&[root])), // Same cost and distance: lower address wins. - mock_peer(0x03, dest, true, true, 0.5, Some(&[root])), + mock_peer(0x03, dest, true, 0.5, Some(&[root])), ], ..MockRoutingView::new(false) }; @@ -530,7 +527,7 @@ fn synth_mtu_exceeded_rate_limit_gate_suppresses_second_call() { } /// A non-Full peer is excluded even when its bloom filter may reach the -/// destination, it can send, it is strictly closer, and its link cost beats +/// destination, it is strictly closer, and its link cost beats /// every Full peer's. Only Full peers carry transit bloom filters, so trusting /// a non-Full peer's would route into a node that cannot forward. #[test] @@ -540,7 +537,7 @@ fn candidate_selection_excludes_non_full_peers() { let my_coords = make_coords(&[0x10, root]); let dest_coords = make_coords(&[0x50, root]); - let full = mock_peer(0x70, dest, true, true, 10.0, Some(&[root])); + let full = mock_peer(0x70, dest, true, 10.0, Some(&[root])); let non_full = mock_non_full_peer(0x71, dest, 0.5, Some(&[root])); let rv = MockRoutingView { diff --git a/src/proto/routing/tests/util.rs b/src/proto/routing/tests/util.rs index 9158f9f8..be7a69dd 100644 --- a/src/proto/routing/tests/util.rs +++ b/src/proto/routing/tests/util.rs @@ -6,12 +6,11 @@ use crate::testutil::make_node_addr; use crate::{NodeAddr, TreeCoordinate}; /// A mock peer for the routing seam: the set of destinations its bloom filter -/// reaches, its send state, link cost, tree coordinates and node profile. +/// reaches, its link cost, tree coordinates and node profile. #[derive(Clone)] pub(super) struct MockPeer { pub(super) addr: NodeAddr, pub(super) reach: Vec, - pub(super) can_send: bool, pub(super) link_cost: f64, pub(super) coords: Option, pub(super) is_full: bool, @@ -61,9 +60,6 @@ impl RoutingView for MockRoutingView { fn peer_may_reach<'a>(&'a self, peer: Self::Peer<'a>, dest: &NodeAddr) -> bool { peer.reach.contains(dest) } - fn peer_can_send<'a>(&'a self, peer: Self::Peer<'a>) -> bool { - peer.can_send - } fn peer_link_cost<'a>(&'a self, peer: Self::Peer<'a>) -> f64 { peer.link_cost } diff --git a/src/transport/mod.rs b/src/transport/mod.rs index fc650d9e..81a25fee 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -63,6 +63,7 @@ use tor::control::TorMonitoringInfo; use udp::UdpTransport; pub(crate) mod framing; +pub(crate) mod stream; mod stats_common; pub(crate) use stats_common::PoolCounters; diff --git a/src/transport/nym/mod.rs b/src/transport/nym/mod.rs index a468eb2b..ba496096 100644 --- a/src/transport/nym/mod.rs +++ b/src/transport/nym/mod.rs @@ -24,6 +24,7 @@ use crate::transport::socks5::{ Socks5Auth, Socks5Dialer, SocksTarget, poll_connecting, proxied_receive_loop, proxied_send_loop, }; +use crate::transport::stream::{ConnId, WRITER_DRAIN_TIMEOUT, drain_writer, next_conn_id}; use stats::NymStats; use std::collections::HashMap; @@ -358,12 +359,14 @@ impl NymTransport { let recv_stats = self.stats.clone(); let remote_addr = addr.clone(); let mtu = self.config.mtu(); + let id = next_conn_id(); let recv_task = tokio::spawn(async move { nym_receive_loop( read_half, transport_id, remote_addr.clone(), + id, packet_tx, pool, mtu, @@ -378,6 +381,7 @@ impl NymTransport { send_rx, transport_id, addr.clone(), + id, self.pool.clone(), self.stats.clone(), "Nym", @@ -391,6 +395,7 @@ impl NymTransport { mtu, established_at: Instant::now(), meta: (), + id, }; let mut pool = self.pool.lock().await; @@ -521,12 +526,14 @@ impl NymTransport { let pool = self.pool.clone(); let recv_stats = self.stats.clone(); let remote_addr = addr.clone(); + let id = next_conn_id(); let recv_task = tokio::spawn(async move { nym_receive_loop( read_half, transport_id, remote_addr.clone(), + id, packet_tx, pool, mtu, @@ -541,6 +548,7 @@ impl NymTransport { send_rx, transport_id, addr.clone(), + id, self.pool.clone(), self.stats.clone(), "Nym", @@ -554,6 +562,7 @@ impl NymTransport { mtu, established_at: Instant::now(), meta: (), + id, }; if let Ok(mut pool) = self.pool.try_lock() { @@ -576,11 +585,22 @@ impl NymTransport { } /// Close a specific connection asynchronously. + /// + /// Aborts the receive task and lets the writer finish the frames already + /// queued, within [`WRITER_DRAIN_TIMEOUT`], without waiting for it. This + /// mirrors `TcpTransport::close_connection_async`. pub async fn close_connection_async(&self, addr: &TransportAddr) { let mut pool = self.pool.lock().await; if let Some(conn) = pool.remove(addr) { - conn.recv_task.abort(); - conn.send_task.abort(); + let ProxiedConnection { + send_tx, + send_task, + recv_task, + .. + } = conn; + drop(send_tx); + recv_task.abort(); + drain_writer(send_task, WRITER_DRAIN_TIMEOUT); debug!( transport_id = %self.transport_id, remote_addr = %addr, @@ -680,10 +700,12 @@ fn parse_target_addr(addr: &TransportAddr) -> Result, mtu: u16, @@ -693,6 +715,7 @@ async fn nym_receive_loop( reader, transport_id, remote_addr.clone(), + id, packet_tx, pool, mtu, @@ -1029,4 +1052,191 @@ mod tests { nym.stop_async().await.unwrap(); dest.stop_async().await.unwrap(); } + + // ======================================================================== + // Connection identity and failure teardown + // ======================================================================== + + /// Poll `f` every 10ms until it holds or `limit` elapses. + async fn wait_until bool>(mut f: F, limit: Duration) -> bool { + let deadline = Instant::now() + limit; + loop { + if f() { + return true; + } + if Instant::now() >= deadline { + return false; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + } + + /// A destination TCP transport behind a mock SOCKS5 proxy, and a started + /// Nym transport dialing through it. + async fn nym_via_mock_proxy() -> ( + TcpTransport, + crate::transport::PacketRx, + NymTransport, + TransportAddr, + ) { + let (dest_tx, dest_rx) = packet_channel(32); + let dest_config = TcpConfig { + bind_addr: Some("127.0.0.1:0".to_string()), + ..Default::default() + }; + let mut dest = TcpTransport::new(TransportId::new(100), None, dest_config, dest_tx); + dest.start_async().await.unwrap(); + let dest_addr = dest.local_addr().unwrap(); + + let mock = MockSocks5Server::new(dest_addr).await.unwrap(); + let proxy_addr = mock.addr(); + let _proxy_handle = mock.spawn(); + + let (nym_tx, _nym_rx) = packet_channel(32); + let nym_config = NymConfig { + socks5_addr: Some(proxy_addr.to_string()), + startup_timeout_secs: Some(5), + connect_timeout_ms: Some(5000), + ..Default::default() + }; + let mut nym = NymTransport::new(TransportId::new(200), None, nym_config, nym_tx); + nym.start_async().await.unwrap(); + let target = TransportAddr::from_string(&dest_addr.to_string()); + (dest, dest_rx, nym, target) + } + + /// A connection displaced from the pool by a newer one at the same address + /// must not remove the newer one when its own receive loop ends. + /// + /// Both are built by `promote_connection`, and the MTU marks which entry + /// is pooled. The last step checks the newer connection still removes its + /// own entry. + #[tokio::test] + async fn nym_displaced_connection_cannot_remove_its_successor() { + let (tx, _rx) = packet_channel(32); + let nym = NymTransport::new(TransportId::new(1), None, make_config(), tx); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listen = listener.local_addr().unwrap(); + let remote = TransportAddr::from_string(&listen.to_string()); + + let a = TcpStream::connect(listen).await.unwrap(); + let (sa, _) = listener.accept().await.unwrap(); + let b = TcpStream::connect(listen).await.unwrap(); + let (sb, _) = listener.accept().await.unwrap(); + + nym.promote_connection(&remote, a, 1400); + nym.promote_connection(&remote, b, 1300); + { + let pool = nym.pool.lock().await; + assert_eq!(pool.len(), 1); + assert_eq!(pool.get(&remote).map(|c| c.mtu), Some(1300)); + } + + drop(sa); + assert!( + wait_until( + || nym.stats().snapshot().recv_errors == 1, + Duration::from_secs(2) + ) + .await, + "the displaced connection's receive loop should have read EOF" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + assert_eq!( + nym.pool.lock().await.get(&remote).map(|c| c.mtu), + Some(1300), + "the displaced connection's teardown removed its successor" + ); + + drop(sb); + assert!( + wait_until( + || nym.stats().snapshot().recv_errors == 2, + Duration::from_secs(2) + ) + .await, + "the newer connection's receive loop should have read EOF" + ); + assert!( + wait_until( + || nym.pool.try_lock().map(|p| p.is_empty()).unwrap_or(false), + Duration::from_secs(2) + ) + .await, + "the newer connection's teardown should remove its own entry" + ); + } + + /// A connection built by connect-on-send removes its own entry when the + /// far side closes. + #[tokio::test] + async fn nym_connect_teardown_removes_its_own_entry() { + let (mut dest, mut dest_rx, mut nym, target) = nym_via_mock_proxy().await; + + let frame = build_msg1_frame(); + nym.send_async(&target, &frame).await.unwrap(); + let received = tokio::time::timeout(Duration::from_secs(5), dest_rx.recv()) + .await + .expect("timeout waiting for packet") + .expect("channel closed"); + assert_eq!(received.data, frame); + assert_eq!(nym.pool.lock().await.len(), 1); + + dest.stop_async().await.unwrap(); + assert!( + wait_until( + || nym.pool.try_lock().map(|p| p.is_empty()).unwrap_or(false), + Duration::from_secs(5) + ) + .await, + "the receive loop should remove its own entry" + ); + assert_eq!( + nym.stats().snapshot().recv_errors, + 1, + "the empty pool must be the receive loop's teardown" + ); + + nym.stop_async().await.unwrap(); + } + + // ======================================================================== + // Deliberate close finishes the frames already queued + // ======================================================================== + + /// A frame queued immediately before a deliberate close must still reach + /// the peer through the proxy, and the close must still end the + /// connection at the far side. + #[tokio::test] + async fn nym_frame_queued_just_before_close_still_reaches_the_peer() { + let (mut dest, mut dest_rx, mut nym, target) = nym_via_mock_proxy().await; + + let frame = build_msg1_frame(); + nym.send_async(&target, &frame).await.unwrap(); + let first = tokio::time::timeout(Duration::from_secs(2), dest_rx.recv()) + .await + .expect("timeout waiting for the first frame") + .expect("channel closed"); + assert_eq!(first.data, frame); + + nym.send_async(&target, &frame).await.unwrap(); + nym.close_connection_async(&target).await; + + let second = tokio::time::timeout(Duration::from_secs(2), dest_rx.recv()) + .await + .expect("a frame queued just before close was never written") + .expect("channel closed"); + assert_eq!(second.data, frame); + assert!( + wait_until( + || dest.stats().snapshot().pool_inbound == 0, + Duration::from_secs(5) + ) + .await, + "the close must still end the connection once the queue is written" + ); + + nym.stop_async().await.unwrap(); + dest.stop_async().await.unwrap(); + } } diff --git a/src/transport/socks5/pool.rs b/src/transport/socks5/pool.rs index 1d3dab42..969f8268 100644 --- a/src/transport/socks5/pool.rs +++ b/src/transport/socks5/pool.rs @@ -21,6 +21,7 @@ use tracing::{debug, trace}; use tokio::io::AsyncWriteExt; use crate::transport::framing::read_fmp_packet; +use crate::transport::stream::{ConnId, PooledConn, remove_own}; use crate::transport::{ ConnectionState, PacketTx, ReceivedPacket, TransportAddr, TransportError, TransportId, }; @@ -46,6 +47,17 @@ pub(crate) struct ProxiedConnection { pub established_at: Instant, /// Per-transport metadata (tor: `Direction`; nym: `()`). pub meta: M, + /// Identity of this connection, shared with its writer and receive loop. + /// Either loop removes the entry at its address only when the entry + /// carries this id, so a loop that outlives its connection cannot remove + /// a newer connection at the same address. + pub id: ConnId, +} + +impl PooledConn for ProxiedConnection { + fn conn_id(&self) -> ConnId { + self.id + } } /// Shared connection pool: addr -> per-connection state. @@ -152,13 +164,17 @@ pub(crate) const SEND_QUEUE_DEPTH: usize = 64; /// Teardown mirrors [`proxied_receive_loop`]: the pool entry is removed and /// `on_remove` fires only when the removal returned `Some`, taking the /// metadata from the removed entry, so a concurrent `close`/`stop` of the same -/// address cannot double-count. +/// address cannot double-count. The entry is removed only when it carries +/// this connection's `id`. A writer can outlive its entry, and by the time its +/// write fails a newer connection may hold the address; that one is left +/// alone. #[allow(clippy::too_many_arguments)] pub(crate) async fn proxied_send_loop( mut writer: OwnedWriteHalf, mut frames: mpsc::Receiver>, transport_id: TransportId, remote_addr: TransportAddr, + id: ConnId, pool: ProxiedPool, stats: Arc, label: &'static str, @@ -187,7 +203,7 @@ pub(crate) async fn proxied_send_loop( ); let removed = { let mut guard = pool.lock().await; - guard.remove(&remote_addr) + remove_own(&mut guard, &remote_addr, id) }; if let Some(conn) = removed { conn.recv_task.abort(); @@ -233,11 +249,20 @@ pub(crate) async fn proxied_send_loop( /// must not run its cleanup before the accept loop has inserted the pool entry /// and bumped its counter, or the removal finds nothing, `on_remove` never /// fires, and the increment is stranded for the life of the process. +/// +/// `id` is the connection's identity. The cleanup removes the entry at +/// `remote_addr` only when it carries this id, so a loop whose entry has +/// already been replaced by a newer connection at the same address leaves +/// that connection alone. When it does remove its own entry it also stops the +/// entry's writer: the loop ended on EOF, a read error or a missed deadline, +/// and frames still queued for a connection in that state are not worth +/// writing. #[allow(clippy::too_many_arguments)] pub(crate) async fn proxied_receive_loop( mut reader: OwnedReadHalf, transport_id: TransportId, remote_addr: TransportAddr, + id: ConnId, packet_tx: PacketTx, pool: ProxiedPool, mtu: u16, @@ -334,8 +359,308 @@ pub(crate) async fn proxied_receive_loop( // concurrent close/stop teardown of the same address can never // double-count. let mut pool_guard = pool.lock().await; - if let Some(removed) = pool_guard.remove(&remote_addr) { + if let Some(removed) = remove_own(&mut pool_guard, &remote_addr, id) { drop(pool_guard); + removed.send_task.abort(); on_remove(&*stats, &removed.meta); } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::transport::packet_channel; + use crate::transport::stream::next_conn_id; + use portable_atomic::{AtomicU64, Ordering}; + use tokio::io::AsyncReadExt; + use tokio::net::TcpListener; + use tokio::time::timeout; + + /// Counters the shared loops write, plus how often `on_remove` fired. + #[derive(Default)] + struct CountingStats { + send_errors: AtomicU64, + recv_errors: AtomicU64, + removed: AtomicU64, + } + + impl ProxiedStats for CountingStats { + fn record_recv(&self, _bytes: usize) {} + fn record_recv_error(&self) { + self.recv_errors.fetch_add(1, Ordering::Relaxed); + } + fn record_send(&self, _bytes: usize) {} + fn record_send_error(&self) { + self.send_errors.fetch_add(1, Ordering::Relaxed); + } + } + + /// The `on_remove` hook the tests pass to both loops. + fn count_removal(stats: &CountingStats, _meta: &()) { + stats.removed.fetch_add(1, Ordering::Relaxed); + } + + /// A pool entry that stands for some other connection at the same + /// address, marked by its MTU. + fn successor() -> ProxiedConnection<()> { + ProxiedConnection { + send_tx: mpsc::channel(1).0, + send_task: tokio::spawn(async {}), + recv_task: tokio::spawn(async {}), + mtu: 1234, + established_at: Instant::now(), + meta: (), + id: next_conn_id(), + } + } + + /// Poll `f` every 10ms until it holds or `limit` elapses. + async fn wait_until bool>(mut f: F, limit: Duration) -> bool { + let deadline = Instant::now() + limit; + loop { + if f() { + return true; + } + if Instant::now() >= deadline { + return false; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + } + + /// A writer whose write fails must not remove a newer connection that has + /// taken its address in the pool, nor run `on_remove` for it. + #[tokio::test] + async fn proxied_writer_error_leaves_a_newer_connection_at_the_same_address() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listen = listener.local_addr().unwrap(); + let client = TcpStream::connect(listen).await.unwrap(); + let (server, _) = listener.accept().await.unwrap(); + socket2::SockRef::from(&server) + .set_linger(Some(Duration::ZERO)) + .unwrap(); + drop(server); + let (_read_half, write_half) = client.into_split(); + let remote = TransportAddr::from_string(&listen.to_string()); + + let pool: ProxiedPool<()> = Arc::new(Mutex::new(HashMap::new())); + let stats = Arc::new(CountingStats::default()); + pool.lock().await.insert(remote.clone(), successor()); + + let (send_tx, send_rx) = mpsc::channel(SEND_QUEUE_DEPTH); + let writer = tokio::spawn(proxied_send_loop( + write_half, + send_rx, + TransportId::new(1), + remote.clone(), + next_conn_id(), + pool.clone(), + stats.clone(), + "Test", + count_removal, + )); + + let frame = vec![0xAB; 114]; + let deadline = Instant::now() + Duration::from_secs(5); + while !writer.is_finished() && Instant::now() < deadline { + let _ = send_tx.try_send(frame.clone()); + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert!(writer.is_finished(), "the writer never hit a write error"); + assert_eq!( + stats.send_errors.load(Ordering::Relaxed), + 1, + "the writer's error path must have run" + ); + + assert_eq!( + pool.lock().await.get(&remote).map(|c| c.mtu), + Some(1234), + "a failed writer removed the newer connection at its address" + ); + assert_eq!(stats.removed.load(Ordering::Relaxed), 0); + } + + /// A receive loop that ends on EOF must stop its writer rather than leave + /// it writing to a peer that has gone. + /// + /// The writer is parked on a peer that does not read, with a full queue. + /// The peer then half-closes, which ends the receive loop, and only + /// afterwards reads. A writer left running delivers every frame it had + /// queued; a stopped one delivers fewer. + #[tokio::test] + async fn proxied_receive_teardown_stops_the_writer() { + let socket = tokio::net::TcpSocket::new_v4().unwrap(); + socket.set_recv_buffer_size(64 * 1024).unwrap(); + socket.bind("127.0.0.1:0".parse().unwrap()).unwrap(); + let listener = socket.listen(8).unwrap(); + let listen = listener.local_addr().unwrap(); + + let client = TcpStream::connect(listen).await.unwrap(); + socket2::SockRef::from(&client) + .set_send_buffer_size(64 * 1024) + .unwrap(); + let (mut peer, _) = listener.accept().await.unwrap(); + let remote = TransportAddr::from_string(&listen.to_string()); + let (read_half, write_half) = client.into_split(); + + let (packet_tx, _packet_rx) = packet_channel(10); + let pool: ProxiedPool<()> = Arc::new(Mutex::new(HashMap::new())); + let stats = Arc::new(CountingStats::default()); + let (send_tx, send_rx) = mpsc::channel(SEND_QUEUE_DEPTH); + let id = next_conn_id(); + let send_task = tokio::spawn(proxied_send_loop( + write_half, + send_rx, + TransportId::new(1), + remote.clone(), + id, + pool.clone(), + stats.clone(), + "Test", + count_removal, + )); + let recv_task = tokio::spawn({ + let pool = pool.clone(); + let stats = stats.clone(); + let remote = remote.clone(); + async move { + proxied_receive_loop( + read_half, + TransportId::new(1), + remote, + id, + packet_tx, + pool, + 1400, + stats, + "Test", + None, + None, + count_removal, + ) + .await; + } + }); + pool.lock().await.insert( + remote.clone(), + ProxiedConnection { + send_tx, + send_task, + recv_task, + mtu: 1400, + established_at: Instant::now(), + meta: (), + id, + }, + ); + + // Fill without stopping at the first refusal, yielding so the writer + // runs, and never keep a sender past the fill. + let frame = vec![0xAB; 1400]; + let mut queued = 0usize; + let mut refused = 0usize; + for _ in 0..8000 { + let sent = { + let guard = pool.lock().await; + guard + .get(&remote) + .map(|c| c.send_tx.try_send(frame.clone()).is_ok()) + }; + if sent == Some(true) { + queued += 1; + } else { + refused += 1; + } + tokio::task::yield_now().await; + } + tokio::time::sleep(Duration::from_millis(300)).await; + let capacity = pool.lock().await.get(&remote).map(|c| c.send_tx.capacity()); + assert_eq!( + capacity, + Some(0), + "setup did not park the writer: queued={queued} refused={refused}" + ); + assert!(refused > 0, "setup never filled the queue: queued={queued}"); + + peer.shutdown().await.unwrap(); + assert!( + wait_until( + || stats.removed.load(Ordering::Relaxed) == 1, + Duration::from_secs(5) + ) + .await, + "the receive loop should have torn the connection down on EOF" + ); + + let mut buf = vec![0u8; 64 * 1024]; + let read = timeout(Duration::from_secs(10), async { + let mut total = 0usize; + loop { + match peer.read(&mut buf).await { + Ok(0) | Err(_) => return total, + Ok(n) => total += n, + } + } + }) + .await + .expect("the connection was never closed toward the peer"); + assert!(read > 0, "the kernel buffers held written frames"); + assert!( + read < queued * frame.len(), + "the writer kept writing after its receive loop tore the connection down: \ + read={read} queued_bytes={}", + queued * frame.len() + ); + } + + /// A receive loop's teardown must leave alone a newer entry at its address, + /// and must not run `on_remove` for it. + #[tokio::test] + async fn proxied_receive_teardown_leaves_a_newer_connection_at_the_same_address() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listen = listener.local_addr().unwrap(); + let client = TcpStream::connect(listen).await.unwrap(); + let (server, peer_addr) = listener.accept().await.unwrap(); + let remote = TransportAddr::from_string(&peer_addr.to_string()); + let (read_half, _write_half) = server.into_split(); + + let (packet_tx, _packet_rx) = packet_channel(10); + let pool: ProxiedPool<()> = Arc::new(Mutex::new(HashMap::new())); + let stats = Arc::new(CountingStats::default()); + pool.lock().await.insert(remote.clone(), successor()); + + drop(client); + proxied_receive_loop( + read_half, + TransportId::new(1), + remote.clone(), + next_conn_id(), + packet_tx, + pool.clone(), + 1400, + stats.clone(), + "Test", + None, + None, + count_removal, + ) + .await; + assert_eq!( + stats.recv_errors.load(Ordering::Relaxed), + 1, + "the loop should have ended on EOF" + ); + + assert_eq!( + pool.lock().await.get(&remote).map(|c| c.mtu), + Some(1234), + "the teardown removed a newer connection at its address" + ); + assert_eq!( + stats.removed.load(Ordering::Relaxed), + 0, + "the teardown ran on_remove for a connection it did not remove" + ); + } +} diff --git a/src/transport/stream.rs b/src/transport/stream.rs new file mode 100644 index 00000000..103ed68d --- /dev/null +++ b/src/transport/stream.rs @@ -0,0 +1,119 @@ +//! Connection lifecycle rules shared by the stream transports. +//! +//! TCP and the SOCKS5-proxied Tor and Nym transports each give a pooled +//! connection its own writer task and receive loop, and either loop can +//! outlive the pool entry it was created with. The rules for when such a loop +//! may touch the pool are written once here. + +use std::collections::HashMap; +use std::time::Duration; + +use portable_atomic::{AtomicU64, Ordering}; +use tokio::task::JoinHandle; + +use crate::transport::TransportAddr; + +/// Identity of one pooled stream connection. +/// +/// The pool is keyed by address, and a newer connection can take an address +/// while an older connection's writer or receive loop is still running. The +/// id tells the two apart. +pub(crate) type ConnId = u64; + +/// Source of connection ids. Process-wide rather than per transport, because +/// the accept loops that build connections are free functions with no +/// transport instance to hold a counter. +static NEXT_CONN_ID: AtomicU64 = AtomicU64::new(1); + +/// Hand out an id no other connection in this process has had. +pub(crate) fn next_conn_id() -> ConnId { + NEXT_CONN_ID.fetch_add(1, Ordering::Relaxed) +} + +/// A pooled stream connection that knows its own [`ConnId`]. +pub(crate) trait PooledConn { + /// The id this connection's writer and receive loop were given. + fn conn_id(&self) -> ConnId; +} + +/// Remove the entry at `addr`, but only if it is connection `id`. +/// +/// This is the only way a connection's own writer or receive loop removes a +/// pool entry. An entry with another id belongs to a newer connection at the +/// same address, and is left alone. +pub(crate) fn remove_own( + pool: &mut HashMap, + addr: &TransportAddr, + id: ConnId, +) -> Option { + if pool.get(addr)?.conn_id() != id { + return None; + } + pool.remove(addr) +} + +/// How long a deliberately closed connection's writer may keep writing the +/// frames already queued before it is stopped. +/// +/// It bounds only how long the socket and the writer task outlive the close; +/// no caller waits on it. A peer that is still reading drains a full queue in +/// far less, and one that has not drained it by then has stopped reading. +pub(crate) const WRITER_DRAIN_TIMEOUT: Duration = Duration::from_secs(5); + +/// Let a closed connection's writer finish the frames already queued, and stop +/// it if it is still running after `bound`. +/// +/// The caller must already have dropped the connection's queue, so the writer +/// exits once it has written what was queued. The wait runs on its own task, +/// which ends as soon as the writer does; the returned handle is that task's. +pub(crate) fn drain_writer(send_task: JoinHandle<()>, bound: Duration) -> JoinHandle<()> { + tokio::spawn(async move { + let mut send_task = send_task; + if tokio::time::timeout(bound, &mut send_task).await.is_err() { + send_task.abort(); + } + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A writer still running when the bound expires is stopped, and the + /// timer ends with it. + /// + /// The writer holds a oneshot sender and never finishes, so the sender is + /// dropped only if the task is aborted. + #[tokio::test] + async fn drain_writer_aborts_a_writer_that_outlives_the_bound() { + let (guard_tx, guard_rx) = tokio::sync::oneshot::channel::<()>(); + let writer = tokio::spawn(async move { + let _guard = guard_tx; + std::future::pending::<()>().await + }); + + let timer = drain_writer(writer, Duration::from_millis(50)); + assert!( + matches!( + tokio::time::timeout(Duration::from_secs(1), guard_rx).await, + Ok(Err(_)) + ), + "a writer still running at the bound was left running" + ); + tokio::time::timeout(Duration::from_secs(1), timer) + .await + .expect("the drain timer outlived the writer it stopped") + .unwrap(); + } + + /// The wait ends when the writer does, not when the bound expires. + #[tokio::test] + async fn drain_writer_ends_as_soon_as_the_writer_does() { + let writer = tokio::spawn(async {}); + let timer = drain_writer(writer, Duration::from_secs(30)); + tokio::time::timeout(Duration::from_millis(100), timer) + .await + .expect("the drain timer kept running after the writer exited") + .unwrap(); + } +} diff --git a/src/transport/tcp/mod.rs b/src/transport/tcp/mod.rs index c14575d3..97b74e62 100644 --- a/src/transport/tcp/mod.rs +++ b/src/transport/tcp/mod.rs @@ -32,6 +32,9 @@ use super::{ }; use crate::config::TcpConfig; use crate::transport::framing::read_fmp_packet; +use crate::transport::stream::{ + ConnId, WRITER_DRAIN_TIMEOUT, drain_writer, next_conn_id, remove_own, +}; use pool::{ConnectingEntry, ConnectingPool, ConnectionPool, Direction, TcpConnection}; use stats::TcpStats; @@ -425,12 +428,14 @@ impl TcpTransport { let recv_stats = self.stats.clone(); let remote_addr = addr.clone(); let mtu = mss_mtu; + let id = next_conn_id(); let recv_task = tokio::spawn(async move { tcp_receive_loop( read_half, transport_id, remote_addr.clone(), + id, packet_tx, pool, mtu, @@ -450,6 +455,7 @@ impl TcpTransport { send_rx, transport_id, addr.clone(), + id, self.pool.clone(), self.stats.clone(), )); @@ -461,6 +467,7 @@ impl TcpTransport { mtu: mss_mtu, established_at: Instant::now(), direction: Direction::Outbound, + id, }; let mut pool = self.pool.lock().await; @@ -481,21 +488,35 @@ impl TcpTransport { /// Close a specific connection asynchronously. /// - /// Removes the connection from the pool, aborts its receive task, - /// and drops the write half (sends FIN to remote). + /// Removes the connection from the pool and aborts its receive task. The + /// writer is not aborted: dropping the queue lets it finish writing the + /// frames already queued, such as a Disconnect sent just before this close, + /// and then exit, which drops the write half and sends FIN. A detached + /// timer aborts it if it is still writing after [`WRITER_DRAIN_TIMEOUT`], + /// so this call never waits on the peer. Stopping the transport, and every + /// teardown after a connection has failed, abort the writer instead and + /// discard what it had queued. pub async fn close_connection_async(&self, addr: &TransportAddr) { let mut pool = self.pool.lock().await; if let Some(conn) = pool.remove(addr) { - conn.recv_task.abort(); - conn.send_task.abort(); - match conn.direction { + let TcpConnection { + send_tx, + send_task, + recv_task, + direction, + .. + } = conn; + drop(send_tx); + recv_task.abort(); + drain_writer(send_task, WRITER_DRAIN_TIMEOUT); + match direction { Direction::Inbound => self.stats.record_pool_inbound_removed(), Direction::Outbound => self.stats.record_pool_outbound_removed(), } debug!( transport_id = %self.transport_id, remote_addr = %addr, - direction = ?conn.direction, + direction = ?direction, "TCP connection closed (close_connection)" ); } @@ -689,12 +710,14 @@ impl TcpTransport { let pool = self.pool.clone(); let recv_stats = self.stats.clone(); let remote_addr = addr.clone(); + let id = next_conn_id(); let recv_task = tokio::spawn(async move { tcp_receive_loop( read_half, transport_id, remote_addr.clone(), + id, packet_tx, pool, mss_mtu, @@ -714,6 +737,7 @@ impl TcpTransport { send_rx, transport_id, addr.clone(), + id, self.pool.clone(), self.stats.clone(), )); @@ -725,6 +749,7 @@ impl TcpTransport { mtu: mss_mtu, established_at: Instant::now(), direction: Direction::Outbound, + id, }; // Use try_lock since we're in a sync context and the pool @@ -932,12 +957,14 @@ async fn accept_loop( // or it would remove nothing and leave an orphaned entry with // a permanently incremented inbound counter. let (ready_tx, ready_rx) = tokio::sync::oneshot::channel(); + let id = next_conn_id(); let recv_task = tokio::spawn(async move { tcp_receive_loop( read_half, transport_id, recv_addr, + id, recv_packet_tx, recv_pool, conn_mtu, @@ -956,6 +983,7 @@ async fn accept_loop( send_rx, transport_id, remote_addr.clone(), + id, pool.clone(), stats.clone(), )); @@ -967,6 +995,7 @@ async fn accept_loop( mtu: conn_mtu, established_at: Instant::now(), direction: Direction::Inbound, + id, }; let mut pool_guard = pool.lock().await; @@ -1018,6 +1047,10 @@ async fn accept_loop( /// receive task is aborted here rather than left to notice on its own, because /// a half-closed connection is not something either side should keep. /// +/// The entry is removed only when it carries this connection's `id`. A writer +/// can outlive its entry, and by the time its write fails a newer connection +/// may hold the address; that one is left alone. +/// /// Frames are written whole. A partial write followed by an error takes the /// connection down with it, so the peer never sees a frame it cannot /// resynchronise from. @@ -1026,6 +1059,7 @@ async fn tcp_send_loop( mut frames: mpsc::Receiver>, transport_id: TransportId, remote_addr: TransportAddr, + id: ConnId, pool: ConnectionPool, stats: Arc, ) { @@ -1050,11 +1084,12 @@ async fn tcp_send_loop( ); let removed = { let mut pool = pool.lock().await; - pool.remove(&remote_addr) + remove_own(&mut pool, &remote_addr, id) }; + // The removed entry's `send_task` is this task, which returns + // below, so only the receive task needs stopping. if let Some(conn) = removed { conn.recv_task.abort(); - conn.send_task.abort(); match conn.direction { Direction::Inbound => stats.record_pool_inbound_removed(), Direction::Outbound => stats.record_pool_outbound_removed(), @@ -1087,11 +1122,20 @@ async fn tcp_send_loop( /// slot from accept) and `None` for outbound ones. `ready_rx`, when /// present, is the accept loop's readiness barrier: the loop must not run /// its cleanup before the accept loop has inserted the pool entry. +/// +/// `id` is the connection's identity. The cleanup removes the entry at +/// `remote_addr` only when it carries this id, so a loop whose entry has +/// already been replaced by a newer connection at the same address leaves +/// that connection alone. When it does remove its own entry it also stops the +/// entry's writer: the loop ended on EOF, a read error or a missed deadline, +/// and frames still queued for a connection in that state are not worth +/// writing. #[allow(clippy::too_many_arguments)] async fn tcp_receive_loop( mut reader: tokio::net::tcp::OwnedReadHalf, transport_id: TransportId, remote_addr: TransportAddr, + id: ConnId, packet_tx: PacketTx, pool: ConnectionPool, mtu: u16, @@ -1182,9 +1226,10 @@ async fn tcp_receive_loop( // entry actually being removed so a double-cleanup never drives // the counter below zero. let mut pool_guard = pool.lock().await; - let removed = pool_guard.remove(&remote_addr).is_some(); + let removed = remove_own(&mut pool_guard, &remote_addr, id); drop(pool_guard); - if removed { + if let Some(conn) = removed { + conn.send_task.abort(); match direction { Direction::Inbound => stats.record_pool_inbound_removed(), Direction::Outbound => stats.record_pool_outbound_removed(), @@ -2227,6 +2272,7 @@ mod tests { let pool: ConnectionPool = Arc::new(Mutex::new(HashMap::new())); let stats = Arc::new(TcpStats::new()); + let id = next_conn_id(); pool.lock().await.insert( remote.clone(), TcpConnection { @@ -2236,6 +2282,7 @@ mod tests { mtu: 1400, established_at: Instant::now(), direction: Direction::Inbound, + id, }, ); stats.record_pool_inbound_added(); @@ -2248,6 +2295,7 @@ mod tests { read_half, TransportId::new(1), remote.clone(), + id, tx, pool.clone(), 1400, @@ -2321,4 +2369,566 @@ mod tests { drop(client); transport.stop_async().await.unwrap(); } + + // ======================================================================== + // Connection identity and failure teardown + // ======================================================================== + + /// Bind a listener whose accepted sockets get a small receive buffer. + /// + /// Setting `SO_RCVBUF` before `listen` locks the size on every accepted + /// socket, so kernel autotuning cannot grow it past what a test's fill + /// can overrun. + fn capped_deaf_listener() -> TcpListener { + let socket = tokio::net::TcpSocket::new_v4().unwrap(); + socket.set_recv_buffer_size(64 * 1024).unwrap(); + socket.bind("127.0.0.1:0".parse().unwrap()).unwrap(); + socket.listen(8).unwrap() + } + + /// Fill `remote`'s send queue behind a peer that does not read, until the + /// writer is parked in `write_all`, and return how many frames were + /// queued. + /// + /// Every send is attempted whatever the previous one returned, with a + /// yield between sends so the writer runs. A queue still full 300 ms + /// after the last send means the writer could not drain it, which is the + /// state the caller's teardown needs; anything else panics rather than + /// letting the caller pass without it. + async fn park_tcp_writer(t: &TcpTransport, remote: &TransportAddr, frame: &[u8]) -> usize { + let mut queued = 0usize; + let mut refused = 0usize; + for _ in 0..8000 { + match timeout(Duration::from_secs(2), t.send_async(remote, frame)).await { + Ok(Ok(_)) => queued += 1, + Ok(Err(_)) => refused += 1, + Err(_) => panic!("send blocked on a peer that stopped reading"), + } + tokio::task::yield_now().await; + } + tokio::time::sleep(Duration::from_millis(300)).await; + let capacity = t + .pool + .lock() + .await + .get(remote) + .map(|c| c.send_tx.capacity()); + assert_eq!( + capacity, + Some(0), + "setup did not park the writer: queued={queued} refused={refused}" + ); + assert!(refused > 0, "setup never filled the queue: queued={queued}"); + queued + } + + /// Read `stream` to EOF within `limit`, returning the byte count, or + /// `None` if EOF did not arrive in time. + async fn read_to_eof(stream: &mut TcpStream, limit: Duration) -> Option { + use tokio::io::AsyncReadExt; + let mut buf = vec![0u8; 64 * 1024]; + timeout(limit, async { + let mut total = 0usize; + loop { + match stream.read(&mut buf).await { + Ok(0) | Err(_) => return total, + Ok(n) => total += n, + } + } + }) + .await + .ok() + } + + /// A writer whose write fails must not remove a newer connection that has + /// taken its address in the pool. + /// + /// The successor is marked by its MTU. The peer resets the connection, and + /// frames are pushed until the writer's write fails, since the first write + /// after a reset can still succeed. + #[tokio::test] + async fn tcp_writer_error_leaves_a_newer_connection_at_the_same_address() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listen = listener.local_addr().unwrap(); + let client = TcpStream::connect(listen).await.unwrap(); + let (server, _) = listener.accept().await.unwrap(); + socket2::SockRef::from(&server) + .set_linger(Some(Duration::ZERO)) + .unwrap(); + drop(server); + let (_read_half, write_half) = client.into_split(); + let remote = TransportAddr::from_string(&listen.to_string()); + + let pool: ConnectionPool = Arc::new(Mutex::new(HashMap::new())); + let stats = Arc::new(TcpStats::new()); + pool.lock().await.insert( + remote.clone(), + TcpConnection { + send_tx: mpsc::channel(1).0, + send_task: tokio::spawn(async {}), + recv_task: tokio::spawn(async {}), + mtu: 1234, + established_at: Instant::now(), + direction: Direction::Outbound, + id: next_conn_id(), + }, + ); + + let (send_tx, send_rx) = mpsc::channel(pool::SEND_QUEUE_DEPTH); + let writer = tokio::spawn(tcp_send_loop( + write_half, + send_rx, + TransportId::new(1), + remote.clone(), + next_conn_id(), + pool.clone(), + stats.clone(), + )); + + let frame = build_msg1_frame(); + let deadline = Instant::now() + Duration::from_secs(5); + while !writer.is_finished() && Instant::now() < deadline { + let _ = send_tx.try_send(frame.clone()); + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert!(writer.is_finished(), "the writer never hit a write error"); + assert_eq!( + stats.snapshot().send_errors, + 1, + "the writer's error path must have run" + ); + + assert_eq!( + pool.lock().await.get(&remote).map(|c| c.mtu), + Some(1234), + "a failed writer removed the newer connection at its address" + ); + } + + /// A receive loop that ends on EOF must stop its writer rather than leave + /// it writing to a peer that has gone. + /// + /// The writer is parked on a peer that does not read, with a full queue. + /// The peer then half-closes, which ends the receive loop, and only + /// afterwards reads. A writer left running delivers every frame it had + /// queued; a stopped one delivers fewer, since the queue alone holds + /// more frames than the kernel buffers leave unread. + #[tokio::test] + async fn tcp_receive_teardown_stops_the_writer() { + let (tx1, _rx1) = packet_channel(100); + let mut t1 = TcpTransport::new(TransportId::new(1), None, make_outbound_config(), tx1); + t1.start_async().await.unwrap(); + let listener = capped_deaf_listener(); + let remote = TransportAddr::from_string(&listener.local_addr().unwrap().to_string()); + let frame = vec![0xAB; 1400]; + + let queued = park_tcp_writer(&t1, &remote, &frame).await; + let (mut peer, _) = listener.accept().await.unwrap(); + + peer.shutdown().await.unwrap(); + assert!( + wait_until( + || t1.stats().snapshot().pool_outbound == 0, + Duration::from_secs(5) + ) + .await, + "the receive loop should have torn the connection down on EOF" + ); + + let read = read_to_eof(&mut peer, Duration::from_secs(10)) + .await + .expect("the connection was never closed toward the peer"); + assert!(read > 0, "the kernel buffers held written frames"); + assert!( + read < queued * frame.len(), + "the writer kept writing after its receive loop tore the connection down: \ + read={read} queued_bytes={}", + queued * frame.len() + ); + + t1.stop_async().await.unwrap(); + } + + /// A connection displaced from the pool by a newer one at the same address + /// must not remove the newer one when its own receive loop ends. + /// + /// Both are built by `promote_connection`, and the MTU marks which entry + /// is pooled. The last step checks the newer connection still removes its + /// own entry. + #[tokio::test] + async fn tcp_displaced_connection_cannot_remove_its_successor() { + let (tx1, _rx1) = packet_channel(100); + let mut t1 = TcpTransport::new(TransportId::new(1), None, make_outbound_config(), tx1); + t1.start_async().await.unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listen = listener.local_addr().unwrap(); + let remote = TransportAddr::from_string(&listen.to_string()); + + let a = TcpStream::connect(listen).await.unwrap(); + let (sa, _) = listener.accept().await.unwrap(); + let b = TcpStream::connect(listen).await.unwrap(); + let (sb, _) = listener.accept().await.unwrap(); + + t1.promote_connection(&remote, a, 1400); + t1.promote_connection(&remote, b, 1300); + { + let pool = t1.pool.lock().await; + assert_eq!(pool.len(), 1); + assert_eq!(pool.get(&remote).map(|c| c.mtu), Some(1300)); + } + + drop(sa); + assert!( + wait_until( + || t1.stats().snapshot().recv_errors == 1, + Duration::from_secs(2) + ) + .await, + "the displaced connection's receive loop should have read EOF" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + assert_eq!( + t1.pool.lock().await.get(&remote).map(|c| c.mtu), + Some(1300), + "the displaced connection's teardown removed its successor" + ); + + drop(sb); + assert!( + wait_until( + || t1.stats().snapshot().recv_errors == 2, + Duration::from_secs(2) + ) + .await, + "the newer connection's receive loop should have read EOF" + ); + assert!( + wait_until( + || t1.pool.try_lock().map(|p| p.is_empty()).unwrap_or(false), + Duration::from_secs(2) + ) + .await, + "the newer connection's teardown should remove its own entry" + ); + + t1.stop_async().await.unwrap(); + } + + /// A receive loop's teardown must leave alone a newer entry at its address, + /// and must not decrement the counter for it. + /// + /// Calls the loop directly against a hand-built successor, so the check is + /// on the teardown alone and not on how a constructor wires it. + #[tokio::test] + async fn tcp_receive_teardown_leaves_a_newer_connection_at_the_same_address() { + let (tx, _rx) = packet_channel(10); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listen = listener.local_addr().unwrap(); + let client = TcpStream::connect(listen).await.unwrap(); + let (server, peer_addr) = listener.accept().await.unwrap(); + let remote = TransportAddr::from_string(&peer_addr.to_string()); + let (read_half, _write_half) = server.into_split(); + + let pool: ConnectionPool = Arc::new(Mutex::new(HashMap::new())); + let stats = Arc::new(TcpStats::new()); + pool.lock().await.insert( + remote.clone(), + TcpConnection { + send_tx: mpsc::channel(1).0, + send_task: tokio::spawn(async {}), + recv_task: tokio::spawn(async {}), + mtu: 1234, + established_at: Instant::now(), + direction: Direction::Outbound, + id: next_conn_id(), + }, + ); + stats.record_pool_outbound_added(); + assert_eq!(stats.snapshot().pool_outbound, 1); + + drop(client); + tcp_receive_loop( + read_half, + TransportId::new(1), + remote.clone(), + next_conn_id(), + tx, + pool.clone(), + 1400, + stats.clone(), + Direction::Outbound, + None, + None, + ) + .await; + assert_eq!( + stats.snapshot().recv_errors, + 1, + "the loop should have ended on EOF" + ); + + assert_eq!( + pool.lock().await.get(&remote).map(|c| c.mtu), + Some(1234), + "the teardown removed a newer connection at its address" + ); + assert_eq!( + stats.snapshot().pool_outbound, + 1, + "the teardown decremented for a connection it did not remove" + ); + } + + /// A connection built by connect-on-send removes its own entry when its + /// receive loop ends. + #[tokio::test] + async fn tcp_connect_teardown_removes_its_own_entry() { + use tokio::io::AsyncReadExt; + let (tx1, _rx1) = packet_channel(100); + let mut t1 = TcpTransport::new(TransportId::new(1), None, make_outbound_config(), tx1); + t1.start_async().await.unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let remote = TransportAddr::from_string(&listener.local_addr().unwrap().to_string()); + let frame = build_msg1_frame(); + + t1.send_async(&remote, &frame).await.unwrap(); + let (mut server, _) = listener.accept().await.unwrap(); + let mut buf = vec![0u8; frame.len()]; + timeout(Duration::from_secs(2), server.read_exact(&mut buf)) + .await + .expect("timeout waiting for the frame") + .unwrap(); + assert_eq!(buf, frame); + assert_eq!(t1.stats().snapshot().pool_outbound, 1); + + drop(server); + assert!( + wait_until( + || t1.stats().snapshot().pool_outbound == 0, + Duration::from_secs(2) + ) + .await, + "the receive loop should release the outbound slot" + ); + assert!( + t1.pool.lock().await.is_empty(), + "the receive loop should remove its own entry" + ); + + t1.stop_async().await.unwrap(); + } + + /// Two live inbound connections can share one remote address when they + /// reach a wildcard listener on different local addresses. Closing the + /// older one must not remove the newer one's pool entry. + /// + /// Known gap, not covered here: two live inbound connections that share a + /// remote address still share one pool key. The second accept replaces + /// the first entry without stopping its tasks and counts a second inbound + /// slot, so the inbound counter ends one above the pool once both + /// connections close. This test checks only that the older connection's + /// teardown no longer removes the newer connection's entry. + /// + /// Linux only: it needs `127.0.0.2` on the loopback interface and Linux + /// `SO_REUSEADDR` semantics to bind two client sockets to one port. + #[cfg(target_os = "linux")] + #[tokio::test] + async fn closing_the_older_of_two_inbound_connections_sharing_a_remote_address_keeps_the_newer_entry() + { + use socket2::{Domain, Socket, Type}; + use tokio::io::AsyncReadExt; + + let (tx, mut rx) = packet_channel(100); + let config = TcpConfig { + bind_addr: Some("0.0.0.0:0".to_string()), + mtu: Some(1400), + ..Default::default() + }; + let mut transport = TcpTransport::new(TransportId::new(1), None, config, tx); + transport.start_async().await.unwrap(); + let port = transport.local_addr().unwrap().port(); + + let client = |local: SocketAddr| { + let sock = Socket::new(Domain::IPV4, Type::STREAM, None).unwrap(); + sock.set_reuse_address(true).unwrap(); + sock.bind(&local.into()).unwrap(); + sock + }; + let into_tokio = |sock: Socket| { + let std_stream: std::net::TcpStream = sock.into(); + std_stream.set_nonblocking(true).unwrap(); + TcpStream::from_std(std_stream).unwrap() + }; + let sock_a = client("127.0.0.1:0".parse().unwrap()); + let source = sock_a.local_addr().unwrap().as_socket().unwrap(); + let sock_b = client(source); + let remote = TransportAddr::from_string(&source.to_string()); + let frame = build_msg1_frame(); + + // Admit A before B dials, so B's entry is the one left in the pool. + let target_a: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap(); + sock_a.connect(&target_a.into()).unwrap(); + let mut a = into_tokio(sock_a); + a.write_all(&frame).await.unwrap(); + let first = timeout(Duration::from_secs(2), rx.recv()) + .await + .expect("timeout waiting for A's frame") + .expect("packet channel closed"); + assert_eq!(first.remote_addr, remote); + assert_eq!(transport.stats().snapshot().connections_accepted, 1); + + let target_b: SocketAddr = format!("127.0.0.2:{port}").parse().unwrap(); + sock_b.connect(&target_b.into()).unwrap(); + let mut b = into_tokio(sock_b); + b.write_all(&frame).await.unwrap(); + let second = timeout(Duration::from_secs(2), rx.recv()) + .await + .expect("timeout waiting for B's frame") + .expect("packet channel closed"); + assert_eq!( + second.remote_addr, remote, + "B must arrive with the same remote address as A" + ); + assert_eq!(transport.stats().snapshot().connections_accepted, 2); + { + let pool = transport.pool.lock().await; + assert_eq!(pool.len(), 1); + assert!(pool.contains_key(&remote)); + } + + drop(a); + assert!( + wait_until( + || transport.stats().snapshot().recv_errors == 1, + Duration::from_secs(2) + ) + .await, + "A's receive loop should have read EOF" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + + let send_tx = transport + .pool + .lock() + .await + .get(&remote) + .map(|c| c.send_tx.clone()) + .expect("closing the older connection removed the newer connection's entry"); + send_tx.try_send(frame.clone()).unwrap(); + drop(send_tx); + let mut buf = vec![0u8; frame.len()]; + timeout(Duration::from_secs(2), b.read_exact(&mut buf)) + .await + .expect("the surviving entry is not B's live connection") + .unwrap(); + assert_eq!(buf, frame); + + drop(b); + assert!( + wait_until( + || transport.stats().snapshot().recv_errors == 2, + Duration::from_secs(2) + ) + .await, + "B's receive loop should have read EOF" + ); + assert!( + wait_until( + || transport + .pool + .try_lock() + .map(|p| p.is_empty()) + .unwrap_or(false), + Duration::from_secs(2) + ) + .await, + "B's teardown should remove its own entry" + ); + + transport.stop_async().await.unwrap(); + } + + // ======================================================================== + // Deliberate close finishes the frames already queued + // ======================================================================== + + /// A frame queued immediately before a deliberate close must still be + /// written. + /// + /// Sending only queues the frame for the connection's writer task. A close + /// that aborts that task before it has run discards the frame, which is + /// how a Disconnect sent just before a close, or a handshake message sent + /// just before the losing side of a crossed connection is closed, never + /// reaches the peer. The second half checks the close still closes: the + /// peer sees FIN and releases its inbound slot, so a writer that never + /// exits cannot pass. + #[tokio::test] + async fn a_frame_queued_just_before_close_still_reaches_the_peer() { + let (tx1, _rx1) = packet_channel(100); + let (tx2, mut rx2) = packet_channel(100); + let mut t1 = TcpTransport::new(TransportId::new(1), None, make_outbound_config(), tx1); + let mut t2 = TcpTransport::new(TransportId::new(2), None, make_config(), tx2); + t1.start_async().await.unwrap(); + t2.start_async().await.unwrap(); + let remote = TransportAddr::from_string(&t2.local_addr().unwrap().to_string()); + let frame = build_msg1_frame(); + + // Pool the connection and let its writer go idle. + t1.send_async(&remote, &frame).await.unwrap(); + let first = timeout(Duration::from_secs(2), rx2.recv()) + .await + .expect("timeout waiting for the first frame") + .expect("packet channel closed"); + assert_eq!(first.data, frame); + + // Queue, then close with nothing in between. + t1.send_async(&remote, &frame).await.unwrap(); + t1.close_connection_async(&remote).await; + + let second = timeout(Duration::from_secs(2), rx2.recv()) + .await + .expect("a frame queued just before close was never written") + .expect("packet channel closed"); + assert_eq!(second.data, frame); + + assert!( + wait_until( + || t2.stats().snapshot().pool_inbound == 0, + Duration::from_secs(5) + ) + .await, + "the close must still end the connection once the queue is written" + ); + + t1.stop_async().await.unwrap(); + t2.stop_async().await.unwrap(); + } + + /// A deliberate close must return at once even when the writer cannot + /// finish, because the peer has stopped reading. Draining happens after + /// the close returns, never inside it. + #[tokio::test] + async fn close_does_not_wait_for_a_writer_parked_on_a_deaf_peer() { + let (tx1, _rx1) = packet_channel(100); + let mut t1 = TcpTransport::new(TransportId::new(1), None, make_outbound_config(), tx1); + t1.start_async().await.unwrap(); + let listener = capped_deaf_listener(); + let remote = TransportAddr::from_string(&listener.local_addr().unwrap().to_string()); + let frame = vec![0xAB; 1400]; + + park_tcp_writer(&t1, &remote, &frame).await; + let (_peer, _) = listener.accept().await.unwrap(); + + assert!( + timeout( + Duration::from_millis(200), + t1.close_connection_async(&remote) + ) + .await + .is_ok(), + "close waited on a writer that cannot finish" + ); + + t1.stop_async().await.unwrap(); + } } diff --git a/src/transport/tcp/pool.rs b/src/transport/tcp/pool.rs index e8e678ca..ccacc4d7 100644 --- a/src/transport/tcp/pool.rs +++ b/src/transport/tcp/pool.rs @@ -10,6 +10,7 @@ use tokio::sync::{Mutex, mpsc}; use tokio::task::JoinHandle; use tokio::time::Instant; +use crate::transport::stream::{ConnId, PooledConn}; use crate::transport::{TransportAddr, TransportError}; /// Direction of a pooled connection, used to drive separate @@ -53,6 +54,17 @@ pub(crate) struct TcpConnection { pub(crate) established_at: Instant, /// Direction of the connection — drives pool-inbound/outbound accounting. pub(crate) direction: Direction, + /// Identity of this connection, shared with its writer and receive loop. + /// Either loop removes the entry at its address only when the entry + /// carries this id, so a loop that outlives its connection cannot remove + /// a newer connection at the same address. + pub(crate) id: ConnId, +} + +impl PooledConn for TcpConnection { + fn conn_id(&self) -> ConnId { + self.id + } } /// Shared connection pool. diff --git a/src/transport/tor/mod.rs b/src/transport/tor/mod.rs index ba586e25..fe7d0420 100644 --- a/src/transport/tor/mod.rs +++ b/src/transport/tor/mod.rs @@ -34,6 +34,7 @@ use crate::transport::socks5::{ Socks5Auth, Socks5Dialer, SocksTarget, poll_connecting, proxied_receive_loop, proxied_send_loop, }; +use crate::transport::stream::{ConnId, WRITER_DRAIN_TIMEOUT, drain_writer, next_conn_id}; use crate::transport::tcp::INBOUND_FIRST_FRAME_TIMEOUT; use control::{ControlAuth, TorControlClient, TorMonitoringInfo}; use stats::TorStats; @@ -756,12 +757,14 @@ impl TorTransport { let recv_stats = self.stats.clone(); let remote_addr = addr.clone(); let mtu = self.config.mtu(); + let id = next_conn_id(); let recv_task = tokio::spawn(async move { tor_receive_loop( read_half, transport_id, remote_addr.clone(), + id, packet_tx, pool, mtu, @@ -781,6 +784,7 @@ impl TorTransport { send_rx, transport_id, addr.clone(), + id, self.pool.clone(), self.stats.clone(), "Tor", @@ -797,6 +801,7 @@ impl TorTransport { mtu, established_at: Instant::now(), meta: Direction::Outbound, + id, }; let mut pool = self.pool.lock().await; @@ -940,12 +945,14 @@ impl TorTransport { let pool = self.pool.clone(); let recv_stats = self.stats.clone(); let remote_addr = addr.clone(); + let id = next_conn_id(); let recv_task = tokio::spawn(async move { tor_receive_loop( read_half, transport_id, remote_addr.clone(), + id, packet_tx, pool, mtu, @@ -965,6 +972,7 @@ impl TorTransport { send_rx, transport_id, addr.clone(), + id, self.pool.clone(), self.stats.clone(), "Tor", @@ -981,6 +989,7 @@ impl TorTransport { mtu, established_at: Instant::now(), meta: Direction::Outbound, + id, }; // Use try_lock since we're in a sync context and the pool @@ -1007,12 +1016,24 @@ impl TorTransport { } /// Close a specific connection asynchronously. + /// + /// Aborts the receive task and lets the writer finish the frames already + /// queued, within [`WRITER_DRAIN_TIMEOUT`], without waiting for it. This + /// mirrors `TcpTransport::close_connection_async`. pub async fn close_connection_async(&self, addr: &TransportAddr) { let mut pool = self.pool.lock().await; if let Some(conn) = pool.remove(addr) { - conn.recv_task.abort(); - conn.send_task.abort(); - match conn.meta { + let ProxiedConnection { + send_tx, + send_task, + recv_task, + meta, + .. + } = conn; + drop(send_tx); + recv_task.abort(); + drain_writer(send_task, WRITER_DRAIN_TIMEOUT); + match meta { Direction::Inbound => self.stats.record_pool_inbound_removed(), Direction::Outbound => self.stats.record_pool_outbound_removed(), } @@ -1096,6 +1117,7 @@ async fn tor_receive_loop( reader: tokio::net::tcp::OwnedReadHalf, transport_id: TransportId, remote_addr: TransportAddr, + id: ConnId, packet_tx: PacketTx, pool: ProxiedPool, mtu: u16, @@ -1108,6 +1130,7 @@ async fn tor_receive_loop( reader, transport_id, remote_addr.clone(), + id, packet_tx, pool, mtu, @@ -1236,12 +1259,14 @@ async fn tor_accept_loop( // nothing and leave an orphaned entry with a permanently incremented // inbound counter. let (ready_tx, ready_rx) = tokio::sync::oneshot::channel(); + let id = next_conn_id(); let recv_task = tokio::spawn(async move { tor_receive_loop( read_half, transport_id, recv_addr, + id, recv_tx, recv_pool, mtu, @@ -1259,6 +1284,7 @@ async fn tor_accept_loop( send_rx, transport_id, remote_addr.clone(), + id, pool.clone(), stats.clone(), "Tor", @@ -1275,6 +1301,7 @@ async fn tor_accept_loop( mtu, established_at: Instant::now(), meta: Direction::Inbound, + id, }; let evicted = { @@ -1289,6 +1316,7 @@ async fn tor_accept_loop( // just inserted and decrement for it, leaking one slot and // orphaning a live connection. old.recv_task.abort(); + old.send_task.abort(); match old.meta { Direction::Inbound => stats.record_pool_inbound_removed(), Direction::Outbound => stats.record_pool_outbound_removed(), @@ -2153,6 +2181,7 @@ mod tests { let pool: ProxiedPool = Arc::new(Mutex::new(HashMap::new())); let stats = Arc::new(TorStats::new()); + let id = next_conn_id(); pool.lock().await.insert( remote.clone(), ProxiedConnection { @@ -2162,6 +2191,7 @@ mod tests { mtu: 1400, established_at: Instant::now(), meta: Direction::Inbound, + id, }, ); stats.record_pool_inbound_added(); @@ -2174,6 +2204,7 @@ mod tests { read_half, TransportId::new(1), remote.clone(), + id, tx, pool.clone(), 1400, @@ -2223,11 +2254,13 @@ mod tests { let recv_pool = pool.clone(); let recv_stats = stats.clone(); let recv_addr = remote.clone(); + let id = next_conn_id(); let mut handle = tokio::spawn(async move { tor_receive_loop( read_half, TransportId::new(1), recv_addr, + id, tx, recv_pool, 1400, @@ -2256,6 +2289,7 @@ mod tests { mtu: 1400, established_at: Instant::now(), meta: Direction::Inbound, + id, }, ); stats.record_pool_inbound_added(); @@ -2313,6 +2347,7 @@ mod tests { mtu: 1400, established_at: Instant::now(), meta: Direction::Inbound, + id: next_conn_id(), }, ); stats.record_pool_inbound_added(); @@ -2349,4 +2384,248 @@ mod tests { accept.abort(); drop(sock); } + + // ======================================================================== + // Connection identity and failure teardown + // ======================================================================== + + /// A connection displaced from the pool by a newer one at the same address + /// must not remove the newer one when its own receive loop ends. + /// + /// Both are built by `promote_connection`, and the MTU marks which entry + /// is pooled. The last step checks the newer connection still removes its + /// own entry. + #[tokio::test] + async fn tor_displaced_connection_cannot_remove_its_successor() { + let (tx, _rx) = packet_channel(32); + let tor = TorTransport::new(TransportId::new(1), None, make_config(), tx); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listen = listener.local_addr().unwrap(); + let remote = TransportAddr::from_string(&listen.to_string()); + + let a = TcpStream::connect(listen).await.unwrap(); + let (sa, _) = listener.accept().await.unwrap(); + let b = TcpStream::connect(listen).await.unwrap(); + let (sb, _) = listener.accept().await.unwrap(); + + tor.promote_connection(&remote, a, 1400); + tor.promote_connection(&remote, b, 1300); + { + let pool = tor.pool.lock().await; + assert_eq!(pool.len(), 1); + assert_eq!(pool.get(&remote).map(|c| c.mtu), Some(1300)); + } + + drop(sa); + assert!( + wait_until( + || tor.stats().snapshot().recv_errors == 1, + Duration::from_secs(2) + ) + .await, + "the displaced connection's receive loop should have read EOF" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + assert_eq!( + tor.pool.lock().await.get(&remote).map(|c| c.mtu), + Some(1300), + "the displaced connection's teardown removed its successor" + ); + + drop(sb); + assert!( + wait_until( + || tor.stats().snapshot().recv_errors == 2, + Duration::from_secs(2) + ) + .await, + "the newer connection's receive loop should have read EOF" + ); + assert!( + wait_until( + || tor.pool.try_lock().map(|p| p.is_empty()).unwrap_or(false), + Duration::from_secs(2) + ) + .await, + "the newer connection's teardown should remove its own entry" + ); + } + + /// A connection admitted by the onion accept loop removes its own entry and + /// releases its inbound slot when its receive loop ends. + #[tokio::test] + async fn onion_accept_teardown_removes_its_own_entry() { + use socket2::{Domain, Socket, Type}; + + let (tx, mut rx) = packet_channel(10); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listen = listener.local_addr().unwrap(); + + let sock = Socket::new(Domain::IPV4, Type::STREAM, None).unwrap(); + sock.bind(&"127.0.0.1:0".parse::().unwrap().into()) + .unwrap(); + let client_addr = sock.local_addr().unwrap().as_socket().unwrap(); + let remote = TransportAddr::from_string(&client_addr.to_string()); + + let (pool, stats, accept) = spawn_onion_accept_loop(listener, tx, Duration::from_secs(5)); + + sock.connect(&listen.into()).unwrap(); + let std_stream: std::net::TcpStream = sock.into(); + std_stream.set_nonblocking(true).unwrap(); + let mut client = TcpStream::from_std(std_stream).unwrap(); + client.write_all(&build_msg1_frame()).await.unwrap(); + + let packet = tokio::time::timeout(Duration::from_secs(2), rx.recv()) + .await + .expect("timeout waiting for the frame") + .expect("packet channel closed"); + assert_eq!(packet.remote_addr, remote); + assert!(pool.lock().await.contains_key(&remote)); + assert_eq!(stats.pool_inbound_count(), 1); + + drop(client); + assert!( + wait_until(|| stats.pool_inbound_count() == 0, Duration::from_secs(2)).await, + "the receive loop should release the inbound slot" + ); + assert!( + !pool.lock().await.contains_key(&remote), + "the receive loop should remove its own entry" + ); + + accept.abort(); + } + + /// When the onion accept loop evicts an entry at a colliding address, it + /// must stop that entry's writer as well as its receive task. + /// + /// The stale writer holds a oneshot sender and never finishes, so the + /// sender is dropped only if the task is aborted: dropping its handle + /// alone leaves it running. + #[tokio::test] + async fn evicting_a_colliding_onion_entry_stops_its_writer() { + use socket2::{Domain, Socket, Type}; + + let (tx, _rx) = packet_channel(10); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let listen = listener.local_addr().unwrap(); + + let sock = Socket::new(Domain::IPV4, Type::STREAM, None).unwrap(); + sock.bind(&"127.0.0.1:0".parse::().unwrap().into()) + .unwrap(); + let client_addr = sock.local_addr().unwrap().as_socket().unwrap(); + let remote = TransportAddr::from_string(&client_addr.to_string()); + + let pool: ProxiedPool = Arc::new(Mutex::new(HashMap::new())); + let stats = Arc::new(TorStats::new()); + let (guard_tx, guard_rx) = tokio::sync::oneshot::channel::<()>(); + pool.lock().await.insert( + remote.clone(), + ProxiedConnection { + send_tx: tokio::sync::mpsc::channel(1).0, + send_task: tokio::spawn(async move { + let _guard = guard_tx; + std::future::pending::<()>().await + }), + recv_task: tokio::spawn(std::future::pending::<()>()), + mtu: 1400, + established_at: Instant::now(), + meta: Direction::Inbound, + id: next_conn_id(), + }, + ); + stats.record_pool_inbound_added(); + + let accept = tokio::spawn(tor_accept_loop( + listener, + TransportId::new(1), + tx, + pool.clone(), + 1400, + 64, + Duration::from_secs(5), + stats.clone(), + )); + + sock.connect(&listen.into()).unwrap(); + assert!( + wait_until( + || stats.snapshot().connections_accepted == 1, + Duration::from_secs(2) + ) + .await, + "the colliding connection should have been accepted" + ); + + assert!( + matches!( + tokio::time::timeout(Duration::from_secs(1), guard_rx).await, + Ok(Err(_)) + ), + "the evicted entry's writer was left running" + ); + + accept.abort(); + drop(sock); + } + + // ======================================================================== + // Deliberate close finishes the frames already queued + // ======================================================================== + + /// A frame queued immediately before a deliberate close must still reach + /// the peer through the proxy, and the close must still end the + /// connection at the far side. + #[tokio::test] + async fn tor_frame_queued_just_before_close_still_reaches_the_peer() { + let (dest_tx, mut dest_rx) = packet_channel(32); + let dest_config = TcpConfig { + bind_addr: Some("127.0.0.1:0".to_string()), + ..Default::default() + }; + let mut dest = TcpTransport::new(TransportId::new(100), None, dest_config, dest_tx); + dest.start_async().await.unwrap(); + let dest_addr = dest.local_addr().unwrap(); + + let mock = MockSocks5Server::new(dest_addr).await.unwrap(); + let proxy_addr = mock.addr(); + let _proxy_handle = mock.spawn(); + + let (tor_tx, _tor_rx) = packet_channel(32); + let tor_config = TorConfig { + socks5_addr: Some(proxy_addr.to_string()), + ..Default::default() + }; + let mut tor = TorTransport::new(TransportId::new(200), None, tor_config, tor_tx); + tor.start_async().await.unwrap(); + + let target = TransportAddr::from_string(&dest_addr.to_string()); + let frame = build_msg1_frame(); + tor.send_async(&target, &frame).await.unwrap(); + let first = tokio::time::timeout(Duration::from_secs(2), dest_rx.recv()) + .await + .expect("timeout waiting for the first frame") + .expect("channel closed"); + assert_eq!(first.data, frame); + + tor.send_async(&target, &frame).await.unwrap(); + tor.close_connection_async(&target).await; + + let second = tokio::time::timeout(Duration::from_secs(2), dest_rx.recv()) + .await + .expect("a frame queued just before close was never written") + .expect("channel closed"); + assert_eq!(second.data, frame); + assert!( + wait_until( + || dest.stats().snapshot().pool_inbound == 0, + Duration::from_secs(5) + ) + .await, + "the close must still end the connection once the queue is written" + ); + + tor.stop_async().await.unwrap(); + dest.stop_async().await.unwrap(); + } } diff --git a/src/transport/udp/io/mod.rs b/src/transport/udp/io/mod.rs index 8acbb630..64a25cd3 100644 --- a/src/transport/udp/io/mod.rs +++ b/src/transport/udp/io/mod.rs @@ -99,6 +99,50 @@ mod tests { ); } + /// The traversal path binds its socket plainly on port zero, so the + /// socket reaches `adopt` carrying neither reuse flag. Adoption has to + /// add them after the fact, or the per-peer connected socket's bind to + /// the same address is refused with `EADDRINUSE`. Either flag on the + /// holder admits that bind on Linux, so the successful open cannot tell + /// whether both were set; each flag is also read back. + #[cfg(target_os = "linux")] + #[test] + fn an_adopted_plain_socket_carries_both_reuse_flags_and_admits_a_connected_socket_on_its_port() + { + use std::os::fd::{AsRawFd, BorrowedFd}; + + let peer = std::net::UdpSocket::bind("127.0.0.1:0").expect("failed to bind the peer"); + let peer_addr = peer.local_addr().expect("peer local address"); + + let plain = std::net::UdpSocket::bind(("0.0.0.0", 0)).expect("failed to bind the holder"); + { + let probe = socket2::SockRef::from(&plain); + assert!( + !probe.reuse_address().expect("read SO_REUSEADDR"), + "precondition: a plain bind must arrive without SO_REUSEADDR", + ); + assert!( + !probe.reuse_port().expect("read SO_REUSEPORT"), + "precondition: a plain bind must arrive without SO_REUSEPORT", + ); + } + + let adopted = UdpRawSocket::adopt(plain, 65536, 65536).expect("failed to adopt the holder"); + + let joined = super::open_connected_fd(adopted.local_addr(), peer_addr, 65536, 65536); + // SAFETY: `adopted` owns this fd and outlives every use of the borrow. + let fd = unsafe { BorrowedFd::borrow_raw(adopted.as_raw_fd()) }; + let flags = socket2::SockRef::from(&fd); + let reuse_address = flags.reuse_address().expect("read SO_REUSEADDR"); + let reuse_port = flags.reuse_port().expect("read SO_REUSEPORT"); + + if let Err(err) = &joined { + panic!("a connected socket must be able to bind the adopted socket's port: {err}"); + } + assert!(reuse_address, "the adopted socket must carry SO_REUSEADDR"); + assert!(reuse_port, "the adopted socket must carry SO_REUSEPORT"); + } + #[tokio::test] async fn test_async_udp_socket_send_recv() { let sock1 = UdpRawSocket::open("127.0.0.1:0".parse().unwrap(), 65536, 65536) @@ -124,6 +168,131 @@ mod tests { assert_eq!(src, addr1); } + /// Measurement: duplicate local ports among many concurrently held + /// port-zero UDP binds, with reuse flags set before the bind and with + /// reuse flags set after it by `adopt`. + /// + /// A reuse flag set before a port-zero bind lets the kernel hand out a + /// port another flagged socket already holds; set after the bind it only + /// lets a later socket join. Three arms, each run over several trials: + /// + /// - `pre` flags each socket and then binds it, which is the ordering + /// that must not be used for a traversal socket. It is built here from + /// socket2 because no such helper exists in the tree. It must show + /// duplicates: if it shows none, this run could not have seen the + /// hazard, and the measurement fails rather than passing. + /// - `post` binds exactly as the traversal path does and then adopts, + /// so `adopt` flags each socket while other threads are still binding. + /// - `orphan` holds sockets flagged before their bind, as a connected + /// socket is, and counts later plain binds handed one of their ports. + /// + /// Coverage gap: nothing that gates runs this, because the port + /// allocator is probabilistic. It measures the kernel behaviour the + /// adoption path relies on, not the traversal binds themselves, so a + /// change that flagged those binds before binding would not turn it red. + /// + /// Run with: + /// cargo test --lib transport::udp::io::tests::measure_duplicate_ephemeral_ports -- --ignored --nocapture + #[cfg(target_os = "linux")] + #[test] + #[ignore = "probabilistic kernel port-allocator measurement; run explicitly with --ignored --nocapture"] + fn measure_duplicate_ephemeral_ports_for_reuse_flags_set_before_and_after_bind() { + use socket2::{Domain, Protocol, Socket, Type}; + use std::collections::HashSet; + use std::sync::Barrier; + + const N: usize = 500; + const TRIALS: usize = 5; + const THREADS: usize = 8; + const HELD: usize = 200; + const BUF: usize = 65536; + + fn plain_bind() -> std::net::UdpSocket { + std::net::UdpSocket::bind(("0.0.0.0", 0)).expect("plain port-zero bind") + } + + fn flagged_bind() -> std::net::UdpSocket { + let sock = Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP)).expect("socket"); + sock.set_reuse_port(true).expect("set SO_REUSEPORT"); + sock.set_reuse_address(true).expect("set SO_REUSEADDR"); + let any: SocketAddr = "0.0.0.0:0".parse().unwrap(); + sock.bind(&any.into()).expect("flagged port-zero bind"); + sock.into() + } + + /// Bind `n` sockets across `THREADS` threads released together, + /// adopting each one as soon as it is bound, and hold every socket + /// until all the threads have finished. + fn bind_concurrently(n: usize, bind: fn() -> std::net::UdpSocket) -> Vec { + let barrier = Barrier::new(THREADS); + std::thread::scope(|s| { + let handles: Vec<_> = (0..THREADS) + .map(|t| { + let share = n / THREADS + usize::from(t < n % THREADS); + let barrier = &barrier; + s.spawn(move || { + barrier.wait(); + (0..share) + .map(|_| UdpRawSocket::adopt(bind(), BUF, BUF).expect("adopt")) + .collect::>() + }) + }) + .collect(); + handles + .into_iter() + .flat_map(|h| h.join().expect("bind thread panicked")) + .collect() + }) + } + + fn duplicates(socks: &[UdpRawSocket]) -> usize { + let distinct: HashSet = socks.iter().map(|s| s.local_addr().port()).collect(); + socks.len() - distinct.len() + } + + let mut pre = Vec::with_capacity(TRIALS); + let mut post = Vec::with_capacity(TRIALS); + let mut orphan = Vec::with_capacity(TRIALS); + for _ in 0..TRIALS { + pre.push(duplicates(&bind_concurrently(N, flagged_bind))); + post.push(duplicates(&bind_concurrently(N, plain_bind))); + + let held: Vec = (0..HELD).map(|_| flagged_bind()).collect(); + let held_ports: HashSet = held + .iter() + .map(|s| s.local_addr().expect("held local address").port()) + .collect(); + let later = bind_concurrently(N, plain_bind); + orphan.push( + later + .iter() + .filter(|s| held_ports.contains(&s.local_addr().port())) + .count(), + ); + } + + eprintln!("duplicate ports per {N} binds, {THREADS} threads, {TRIALS} trials"); + eprintln!(" pre (flags before bind): {pre:?}"); + eprintln!(" post (flags after bind): {post:?}"); + eprintln!(" orphan ({HELD} held pre-flagged, later plain binds): {orphan:?}"); + + assert!( + pre.iter().sum::() > 0, + "the before-bind arm showed no duplicates, so this run could not have seen the \ + hazard at N = {N}; raise N rather than reading the other arms as clean", + ); + assert_eq!( + post.iter().sum::(), + 0, + "flags set after bind: {post:?}" + ); + assert_eq!( + orphan.iter().sum::(), + 0, + "orphan collisions: {orphan:?}" + ); + } + /// Microbench: compare per-packet `recv_from` (single recvmsg syscall + /// task wakeup per datagram — the macOS pre-recvmsg_x baseline) vs /// `recv_batch` (the new recvmsg_x path, up to 32 datagrams per syscall). diff --git a/src/transport/udp/io/unix.rs b/src/transport/udp/io/unix.rs index 60a4cb43..32341810 100644 --- a/src/transport/udp/io/unix.rs +++ b/src/transport/udp/io/unix.rs @@ -125,6 +125,8 @@ impl UdpRawSocket { /// Adopt an existing bound UDP socket. /// /// This preserves socket identity/NAT mapping created by bootstrap code. + /// The adopted socket is also made joinable by per-peer connected + /// sockets, which bind its local address. pub fn adopt( socket: std::net::UdpSocket, recv_buf_size: usize, @@ -135,6 +137,15 @@ impl UdpRawSocket { sock.set_nonblocking(true) .map_err(|e| TransportError::StartFailed(format!("set nonblocking failed: {}", e)))?; + // A per-peer connected socket later binds this socket's own address, + // and the kernel admits that joiner only when the holder carries a + // reuse flag too. The socket arrives already bound, so setting the + // flags here cannot change which port it was given, unlike flags set + // ahead of a port-zero bind; see the comment in `open`. Best-effort, + // as there: without them only the connected fast path is refused. + let _ = sock.set_reuse_port(true); + let _ = sock.set_reuse_address(true); + sock.set_recv_buffer_size(recv_buf_size) .map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?; sock.set_send_buffer_size(send_buf_size)