Merge master into next, carrying the connectivity removal, the stream writer drain and the adopted UDP socket reuse flags

Carries show_peers stale reporting, derived from how long a peer has been silent. Carries the removal of the stored peer connectivity, which on this line also drops the always-true health conjunct from the msg3 cross-connection and rekey-responder gates and from the rekey scan, and drops the send filter from next-hop candidate selection. Carries per-connection ids and draining closes on TCP, Tor and Nym, so a crossed TCP handshake's msg3 is written before the losing connection closes. Carries the reuse flags on an adopted traversal socket, whose node test now drives this line's three-message handshake directly instead of one receive-loop pass per node. Carries the netmon test and the probe's measured syscall cost.

The rekey msg2 rollback from maint is not carried here: its read rollback, handler arm, node test and changelog entry are written for IK, so this merge keeps this line's XX rekey msg2 path as it was, and the XX change follows in its own commit.

Two tests reached the msg2 resend arm through a health lever production never sets. The node test now reaches it through a rekey declaration naming keys the responder no longer holds, after replacing the responder's keys the way a cross-connection swap does, and the core test declares a mismatched rekey on the peer's own link. A new core test covers a matching declaration with no session, so the rekey-responder gate's session check keeps a test that can fail.

The establish characterisation test file and the IK duplicate-msg1 decision test stay out, as before. The conflicts were next's XX establish classification and its rewritten establish tests against master's edits to the IK versions; every region resolves to next's text with only the health and send-state reads removed and their doc comments reworded.
This commit is contained in:
Johnathan Corgan
2026-09-16 19:58:05 +00:00
36 changed files with 2366 additions and 336 deletions
+42 -1
View File
@@ -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
+9 -4
View File
@@ -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.
+1 -1
View File
@@ -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
+8 -5
View File
@@ -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:
+2 -1
View File
@@ -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<Value> = 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(),
-2
View File
@@ -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,
+3 -5
View File
@@ -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);
+3 -3
View File
@@ -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<PeerSnapshot> {
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(),
+2 -8
View File
@@ -2725,7 +2725,7 @@ impl Node {
let peer_addrs: Vec<NodeAddr> = 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(
+43 -16
View File
@@ -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<snap::PeerRow> = 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<Item = &ActivePeer> {
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()
}
+24 -15
View File
@@ -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,
}
+37
View File
@@ -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
+129
View File
@@ -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();
}
}
+108
View File
@@ -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);
}
+85 -35
View File
@@ -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;
+1 -1
View File
@@ -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;
+1 -44
View File
@@ -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),
+40
View File
@@ -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;
}
+6 -6
View File
@@ -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 ===
+14 -110
View File
@@ -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<ParentDeclaration>,
@@ -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();
+1 -7
View File
@@ -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]));
+8 -14
View File
@@ -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<ConnSnapshot>;
/// 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<PeerSnapshot>;
@@ -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 {
+19 -4
View File
@@ -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),
+2 -3
View File
@@ -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<u8>) ->
}
}
/// 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,
+8 -10
View File
@@ -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;
}
+15 -18
View File
@@ -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 {
+1 -5
View File
@@ -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<NodeAddr>,
pub(super) can_send: bool,
pub(super) link_cost: f64,
pub(super) coords: Option<TreeCoordinate>,
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
}
+1
View File
@@ -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;
+212 -2
View File
@@ -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<SocksTarget, TransportError
/// exhaust, and the mixnet's Sphinx routing makes a first frame legitimately
/// slow, so a TCP-scale deadline here would drop good connections to defend
/// a cap that does not exist.
#[allow(clippy::too_many_arguments)]
async fn nym_receive_loop(
reader: tokio::net::tcp::OwnedReadHalf,
transport_id: TransportId,
remote_addr: TransportAddr,
id: ConnId,
packet_tx: PacketTx,
pool: ProxiedPool<()>,
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<F: FnMut() -> 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();
}
}
+328 -3
View File
@@ -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<M> {
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<M> PooledConn for ProxiedConnection<M> {
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<S: ProxiedStats, M>(
mut writer: OwnedWriteHalf,
mut frames: mpsc::Receiver<Vec<u8>>,
transport_id: TransportId,
remote_addr: TransportAddr,
id: ConnId,
pool: ProxiedPool<M>,
stats: Arc<S>,
label: &'static str,
@@ -187,7 +203,7 @@ pub(crate) async fn proxied_send_loop<S: ProxiedStats, M>(
);
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<S: ProxiedStats, M>(
/// 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<S: ProxiedStats, M>(
mut reader: OwnedReadHalf,
transport_id: TransportId,
remote_addr: TransportAddr,
id: ConnId,
packet_tx: PacketTx,
pool: ProxiedPool<M>,
mtu: u16,
@@ -334,8 +359,308 @@ pub(crate) async fn proxied_receive_loop<S: ProxiedStats, M>(
// 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<F: FnMut() -> 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"
);
}
}
+119
View File
@@ -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<C: PooledConn>(
pool: &mut HashMap<TransportAddr, C>,
addr: &TransportAddr,
id: ConnId,
) -> Option<C> {
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();
}
}
+620 -10
View File
@@ -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<Vec<u8>>,
transport_id: TransportId,
remote_addr: TransportAddr,
id: ConnId,
pool: ConnectionPool,
stats: Arc<TcpStats>,
) {
@@ -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<usize> {
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();
}
}
+12
View File
@@ -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.
+282 -3
View File
@@ -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<Direction>,
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<Direction> = 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::<SocketAddr>().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::<SocketAddr>().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<Direction> = 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();
}
}
+169
View File
@@ -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<UdpRawSocket> {
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::<Vec<_>>()
})
})
.collect();
handles
.into_iter()
.flat_map(|h| h.join().expect("bind thread panicked"))
.collect()
})
}
fn duplicates(socks: &[UdpRawSocket]) -> usize {
let distinct: HashSet<u16> = 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<std::net::UdpSocket> = (0..HELD).map(|_| flagged_bind()).collect();
let held_ports: HashSet<u16> = 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::<usize>() > 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::<usize>(),
0,
"flags set after bind: {post:?}"
);
assert_eq!(
orphan.iter().sum::<usize>(),
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).
+11
View File
@@ -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)