laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-10-06 03:28:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(transport): take the stream write off the caller's task

Browse Source 
TCP, Tor, Nym and BLE wrote to their links from whatever task called
`send`. `write_all` on a stream blocks once the peer's receive window and
this node's send buffer are both full, and it blocks for as long as that
lasts — there is no bound on it. A peer that has stopped reading, or one
whose path has just changed medium, produces exactly that state.

The callers are the rx loop's tick handlers. The heartbeat sweep is one
of them, which is how this surfaced: filtering connection-oriented peers
out of the medium-change fan-out took the write out of that handler, but
not out of the sweep ten seconds behind it — same peer, same state. And
while the rx loop sits in that write it is not serving anything else, so
one unresponsive peer stalls every other peer's liveness, the forwarding
path, and the control socket.

Each connection now owns a writer task holding the write half, and
`send` enqueues onto a bounded channel. The only code that can await the
wire is a task with nothing else to do, so the property holds by
construction rather than by a timeout that has to be tuned against a
healthy slow link — a Tor circuit legitimately stalls for seconds, and
any budget short enough to protect the rx loop is short enough to kill
one of those.

`try_send`, not `send`: awaiting a full queue would reinstate the same
block one level up. A full queue means the writer has not drained a
frame in the time it took to offer 64, which is a peer that is not
receiving, so the send fails and the caller's existing retry and
liveness handling takes over — the same shape as the connect gate above
it, which already refuses rather than waits.

Frames are written whole by a single task, so ordering is preserved and
the half-written-frame hazard cannot arise: a write error takes the
connection down with it, and the peer sees a closed connection rather
than a frame it cannot resynchronise from. Teardown mirrors the receive
loop's existing contract exactly — the pool entry is removed and the
direction counter decremented only when the removal returned `Some`, so
a concurrent close or stop of the same address cannot double-count.

BLE was the worst of the four. `send_async` awaited the L2CAP write
while holding the connection-pool mutex, so a peer that stopped draining
its link blocked not only its own sender but every other BLE operation
behind that guard: connects, evictions, and each receive loop's
teardown. The other three only blocked the caller. The fix is the same
shape, and the pooled stream was already an `Arc` with a `Send` future,
so the writer task holds a clone and no generic surgery was needed.
Queue depth is 16 there rather than 64: a BLE link carries a fraction of
the throughput, so the same depth would be seconds of backlog rather
than a burst.

This also settles the Android backend without changing it. Its
`BleStream::send` pushes onto the embedder's queue and waits for a slot
rather than dropping, which was an unbounded wait on whatever task
called it. That call now happens only in the writer task, where waiting
is the job, and the layer above it is the connection's own bounded
queue, which fills and refuses without waiting on anything.

Two costs, both deliberate. The byte count `send` returns is now what
was queued rather than what reached the wire, which is the prediction
the UDP fast path already reports when it dispatches to the encrypt
workers; bytes actually written are recorded by the writer task as they
go. And the frame is copied into the queue, because `write_all` borrows
and a queue must own — one memcpy of at most an MTU, against an
unbounded stall.

Tor and Nym share `socks5::pool`, so the writer loop is written once
there and once for TCP, matching each transport's existing receive loop.

Both regression tests drive a peer that accepts and then never reads,
push far more than any buffer holds, and assert every send returns
promptly. Both are mutation-checked: replacing `try_send` with an
awaiting `send` fails the TCP one on the timeout, and reinstating the
write under the pool guard fails the BLE one, which also asserts the
pool stays lockable throughout.

What the evidence does and does not cover. The BLE change is exercised
through `MockBleIo`, not over the radio, so the argument that a stalled
radio link behaves like a stalled mock channel is reasoning rather than
measurement. The Android backend is `target_os = "android"` and there is
no NDK on the author's host, so it has no local compile evidence; the
`android-check` leg is the only one. Integration suites were run one at
a time rather than as a single pass, since the full runner exhausts
memory there: `chaos-tcp-mesh`, `chaos-congestion-stress` and
`chaos-churn-mixed-10` all pass.

Folded in on landing, all documentation and no behaviour change. Two doc
comments were left attached to the wrong function when the writer loop was
inserted above the receive loop, so `tcp_receive_loop` and BLE's
`receive_loop` each lost their own documentation to the new function above
them; both are moved back, and the TCP section banner now names both loops.
The Android backend's safety comment claimed `BleStream::send` is reached
from the writer task and from nowhere else, which is not so — `pubkey_exchange`
calls it too, safely, because it wraps the call in a timeout — so the comment
now names both callers and the reason each is bounded. And the netmon
fan-out's filter documented itself by the hazard it avoided, an unbounded
`write_all` reached from the rx loop; that hazard is gone, so the rationale
and the matching changelog clause now record that the filter has outlived it
and that widening the fan-out is open work left out here so the two changes
stay separable.

Co-authored-by: Johnathan Corgan <johnathan@corganlabs.com>
This commit is contained in:
fr34aky
2026-09-08 17:07:23 +00:00
committed by Johnathan Corgan
co-authored by Johnathan Corgan
parent eaca285ad2
commit 813c50bb14
13 changed files with 762 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+23 -3
View File
@@ -11,6 +11,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
#### Data plane
- A peer that stops reading can no longer stall the node. TCP, Tor, Nym and
BLE wrote to their links directly from the caller's task, and a write
blocks once the peer's receive window and this node's send buffer are both
full — which is what a peer that has gone away, or whose path has just
changed medium, produces. The callers are the rx loop's tick handlers, so a
single unresponsive peer held every other arm of the event loop behind it:
other peers' liveness, forwarding, and control RPCs. Each connection now has
its own writer task owning the write half, and sending is an enqueue onto a
bounded queue, so the only code that can await the wire is a task with
nothing else to do. A peer that stops draining fills its queue and its sends
then fail immediately, which is the signal the caller's retry and liveness
handling already expects. Frames are written whole, and a write error takes
the connection down with it, so a peer never sees a partial frame it cannot
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.
- 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
send fast path; `open_connected_fd` binds the wildcard and then calls
@@ -34,9 +52,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
peer on a connectionless transport is also heartbeated at once, so the far
side re-pins to the new source address rather than waiting out its own
heartbeat interval. A peer on a connection-oriented transport keeps the
periodic heartbeat instead: that send awaits an unbounded `write_all` on a
stream the medium change has very likely just stranded, and this reaction
runs on the rx loop. Measured on a live
periodic heartbeat instead. That was because such a send awaited an unbounded
`write_all` on a stream the medium change had very likely just stranded, and
this reaction runs on the rx loop; the writer-task change below removes that
hazard, so widening the fan-out to those transports is now open work rather
than something the design forbids. Measured on a live
node, a WLAN/LAN switch in either direction now costs no reconnection at all —
the Noise session, tree position and routes survive it. Linux and macOS (the
platforms with the connected-socket fast path); elsewhere the heartbeat alone
src/node/handlers/netmon.rs
+21 -18
View File
@@ -41,11 +41,11 @@
//! side re-pins on receipt instead of waiting out its own
//! `heartbeat_interval_secs`. Without it the forward direction is fixed but
//! the reverse still points at the old address until the node next happens
//! to send. This runs on the rx loop, so it covers the connectionless
//! to send. This runs on the rx loop, and it covers the connectionless
//! transports only — see
//! [`Node::heartbeat_all_peers_after_net_change`] for why awaiting a
//! connection-oriented write here would hold the loop, and what a peer on
//! one of those gets instead.
//! [`Node::heartbeat_all_peers_after_net_change`] for what a peer on a
//! connection-oriented transport gets instead, and for why that filter has
//! outlived the reason it was written for.
//!
//! Nothing here tears a peering down. On a live node both WLAN→LAN and
//! LAN→WLAN now cost no reconnection at all — the Noise session, the tree
@@ -112,21 +112,24 @@ impl Node {
/// learns the node's new source address in one RTT rather than at the next
/// due interval. Returns how many went out.
///
/// The filter is not an optimisation. A connectionless transport's send
/// completes without ever awaiting the wire: the UDP fast path hands the
/// frame to the encrypt workers and returns, and a raw datagram write does
/// not wait for a peer. A connection-oriented one awaits `write_all` on a
/// stream, unbounded — the connect above it is wrapped in a timeout, the
/// write is not — and a medium change is precisely the condition that
/// leaves a send window full against a path that has just gone away. This
/// runs on the rx loop, so that write would hold every other arm of the
/// select for as long as the stranded socket takes to fail.
/// The filter was written for a hazard that no longer exists, and it is
/// kept deliberately rather than by oversight. It was this: a
/// connectionless transport's send completes without ever awaiting the
/// wire, because the UDP fast path hands the frame to the encrypt workers
/// and returns and a raw datagram write does not wait for a peer, while a
/// connection-oriented one awaited `write_all` on a stream, unbounded. A
/// medium change is precisely the condition that leaves a send window full
/// against a path that has just gone away, and this runs on the rx loop, so
/// that write held every other arm of the select for as long as the
/// stranded socket took to fail.
///
/// Bounding it with a timeout is not the fix either: dropping a partial
/// `write_all` would leave a half-written frame on the stream, which the
/// peer cannot resynchronise from. Nor can the fan-out simply be spawned,
/// because the send needs `&mut self` for the session counter and the MMP
/// sender record.
/// Every connection-oriented send now enqueues onto its connection's
/// bounded queue and returns, so none of them can await the wire from here.
/// Widening the fan-out to those transports is therefore open work rather
/// than something the send path forbids; it is left out of the change that
/// removed the hazard so the two stay separable. Note that the fan-out
/// still cannot simply be spawned, because the send needs `&mut self` for
/// the session counter and the MMP sender record.
///
/// So a peer on TCP, Tor, Nym or BLE keeps the periodic heartbeat it had
/// before this detector existed. It is not stranded by the omission: those
src/transport/ble/io_android.rs
+11
View File
@@ -700,6 +700,17 @@ impl BleStream for AndroidStream {
// this out via `next_send`. The queue is shallow (`SEND_QUEUE_CAP`) and
// this waits for a slot rather than dropping, so backpressure reaches
// the layers above instead of the link bufferbloating.
//
// The wait is only safe because of who calls it, and there are exactly
// two callers. The connection's writer task is one: an embedder that
// stops draining parks that task alone, and the layer above it is the
// connection's own bounded queue, which fills and starts refusing
// sends without waiting on anything. `pubkey_exchange` is the other,
// and it is safe for a different reason: it wraps this call in
// `tokio::time::timeout`, so its wait is bounded whatever the embedder
// does. Adding a third caller on a caller's task would put an
// unbounded wait back on the rx loop, which is the defect the writer
// task exists to remove.
let mut payload = data.to_vec();
loop {
if self.closed.load(Ordering::Relaxed) {
src/transport/ble/mod.rs
+201 -18
View File
@@ -373,12 +373,19 @@ impl<I: BleIo> BleTransport<I> {
addr: &TransportAddr,
data: &[u8],
) -> Result<usize, TransportError> {
let pool = self.pool.lock().await;
let conn = match pool.get(addr) {
Some(c) => c,
// Take the MTU and the connection's send queue, then release the pool
// lock. Everything after this point must be lock-free: the previous
// shape awaited the L2CAP write while still holding this guard, so a
// peer that stopped draining blocked not just its own sender but every
// other BLE operation — connect, eviction, the receive loops' teardown.
let found = {
let pool = self.pool.lock().await;
pool.get(addr)
.map(|c| (c.effective_mtu() as usize, c.send_tx.clone()))
};
let (mtu, send_tx) = match found {
Some(pair) => pair,
None => {
// Drop pool lock before triggering background connect
drop(pool);
// Fire-and-forget: connect_async spawns a background task
let _ = self.connect_async(addr).await;
return Err(TransportError::SendFailed("not connected".into()));
@@ -386,7 +393,6 @@ impl<I: BleIo> BleTransport<I> {
};
// MTU check
let mtu = conn.effective_mtu() as usize;
if data.len() > mtu {
self.stats.record_mtu_exceeded();
return Err(TransportError::MtuExceeded {
@@ -395,19 +401,27 @@ impl<I: BleIo> BleTransport<I> {
});
}
match conn.stream.send(data).await {
Ok(()) => {
self.stats.record_send(data.len());
Ok(data.len())
}
Err(e) => {
// Queue the frame for the connection's writer task. `try_send`, not
// `send`: waiting for a slot is the same stall in a different shape,
// which is what the Android backend's own queue does one layer down.
// The byte count is what was queued; bytes on the link are recorded by
// the writer task.
match send_tx.try_send(data.to_vec()) {
Ok(()) => Ok(data.len()),
Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
self.stats.record_send_error();
// Drop pool lock before removing to avoid deadlock
drop(pool);
let mut pool = self.pool.lock().await;
pool.remove(addr);
warn!(addr = %addr, error = %e, "BLE send failed, connection removed");
Err(e)
debug!(
addr = %addr,
depth = pool::SEND_QUEUE_DEPTH,
"BLE outbound queue full; peer is not draining"
);
Err(TransportError::SendFailed(
"outbound queue full: peer not draining".into(),
))
}
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
self.stats.record_send_error();
Err(TransportError::SendFailed("connection writer gone".into()))
}
}
}
@@ -509,8 +523,19 @@ impl<I: BleIo> BleTransport<I> {
recv_mtu,
));
let (send_tx, send_rx) = tokio::sync::mpsc::channel(pool::SEND_QUEUE_DEPTH);
let send_task = tokio::spawn(send_loop(
Arc::clone(&stream),
send_rx,
addr.clone(),
Arc::clone(&self.pool),
Arc::clone(&self.stats),
));
let conn = BleConnection {
stream,
send_tx,
send_task: Some(send_task),
recv_task: Some(recv_task),
send_mtu,
recv_mtu,
@@ -630,8 +655,19 @@ impl<I: BleIo> BleTransport<I> {
recv_mtu,
));
let (send_tx, send_rx) = tokio::sync::mpsc::channel(pool::SEND_QUEUE_DEPTH);
let send_task = tokio::spawn(send_loop(
Arc::clone(&stream),
send_rx,
addr_clone.clone(),
Arc::clone(&pool),
Arc::clone(&stats),
));
let conn = BleConnection {
stream,
send_tx,
send_task: Some(send_task),
recv_task: Some(recv_task),
send_mtu,
recv_mtu,
@@ -1093,8 +1129,19 @@ async fn admit_inbound<S>(
recv_mtu,
));
let (send_tx, send_rx) = tokio::sync::mpsc::channel(pool::SEND_QUEUE_DEPTH);
let send_task = tokio::spawn(send_loop(
Arc::clone(&stream),
send_rx,
ta.clone(),
Arc::clone(&pool),
Arc::clone(&stats),
));
let conn = BleConnection {
stream,
send_tx,
send_task: Some(send_task),
recv_task: Some(recv_task),
send_mtu,
recv_mtu,
@@ -1125,6 +1172,39 @@ async fn admit_inbound<S>(
stats.record_connection_accepted();
}
/// Per-connection writer task: the only place a BLE write is ever awaited.
///
/// The BLE case was the worst of the connection-oriented transports. The write
/// was awaited by the caller *while holding the pool mutex*, so a peer that
/// stopped draining its L2CAP link blocked every other BLE operation as well
/// as the caller's task — connects, evictions and each receive loop's
/// teardown all queue behind that one guard. Moving the write here removes
/// both halves of that: the caller enqueues and returns, and the pool lock is
/// never held across the link.
///
/// On a write error the connection is removed from the pool, which is where
/// the old inline path put it too. Dropping the pool entry aborts this task
/// and the receive task through `BleConnection`'s `Drop`.
async fn send_loop<S: BleStream + 'static>(
stream: Arc<S>,
mut frames: tokio::sync::mpsc::Receiver<Vec<u8>>,
addr: TransportAddr,
pool: Arc<Mutex<ConnectionPool<Arc<S>>>>,
stats: Arc<BleStats>,
) {
while let Some(frame) = frames.recv().await {
match stream.send(&frame).await {
Ok(()) => stats.record_send(frame.len()),
Err(e) => {
stats.record_send_error();
warn!(addr = %addr, error = %e, "BLE send failed, connection removed");
pool.lock().await.remove(&addr);
return;
}
}
}
}
/// Receive loop: reads packets from a BLE stream and delivers to node.
///
/// Takes the connection's `BleStreamRead` — already positioned past the
@@ -1560,8 +1640,19 @@ async fn scan_probe_loop<I: io::BleIo>(
recv_mtu,
));
let (send_tx, send_rx) = tokio::sync::mpsc::channel(pool::SEND_QUEUE_DEPTH);
let send_task = tokio::spawn(send_loop(
Arc::clone(&stream),
send_rx,
ta.clone(),
Arc::clone(&pool),
Arc::clone(&stats),
));
let conn = BleConnection {
stream,
send_tx,
send_task: Some(send_task),
recv_task: Some(recv_task),
send_mtu,
recv_mtu,
@@ -1628,6 +1719,7 @@ mod tests {
use crate::transport::framing::build_established_frame;
use io::{MockBleIo, MockBleStream};
use secp256k1::{Secp256k1, SecretKey};
use std::time::Duration;
// ------------------------------------------------------------------
// PendingProbes — the retry/backoff policy for discovered addresses
@@ -1909,6 +2001,93 @@ mod tests {
(transport, rx)
}
/// **The property the writer task exists to guarantee, on the transport
/// where it mattered most.**
///
/// BLE was the worst of the connection-oriented transports: `send_async`
/// awaited the L2CAP write *while holding the pool mutex*, so a peer that
/// stopped draining blocked not only its own sender but every other BLE
/// operation — connects, evictions, and each receive loop's teardown all
/// queue behind that guard.
///
/// The mock stream's send half is a bounded channel, so a peer that never
/// reads is a peer whose link has stopped draining. This inserts such a
/// connection with a real writer task behind it, pushes far more than
/// either queue holds, and asserts that every call returns promptly and
/// that the pool stays lockable throughout.
#[tokio::test]
async fn a_ble_peer_that_stops_reading_cannot_block_the_sender_or_the_pool() {
let io = MockBleIo::new("hci0", test_addr(1));
let (transport, _rx) = make_transport(io);
// `_deaf` is the far end. Holding it without ever calling `recv` is
// what makes this a stalled link rather than a closed one.
let (near, _deaf) = MockBleStream::pair(test_addr(1), test_addr(2), 2048);
let stream = Arc::new(near);
let ta = TransportAddr::from_string("AA:BB:CC:DD:EE:02");
let (send_tx, send_rx) = tokio::sync::mpsc::channel(pool::SEND_QUEUE_DEPTH);
let send_task = tokio::spawn(send_loop(
Arc::clone(&stream),
send_rx,
ta.clone(),
Arc::clone(&transport.pool),
Arc::clone(&transport.stats),
));
transport
.pool
.lock()
.await
.insert(
ta.clone(),
BleConnection {
stream,
send_tx,
send_task: Some(send_task),
recv_task: None,
send_mtu: 2048,
recv_mtu: 2048,
established_at: tokio::time::Instant::now(),
is_static: false,
addr: test_addr(2),
node_addr: None,
},
)
.unwrap();
let frame = vec![0xAB; 512];
let mut queued = 0usize;
let mut refused = 0usize;
for _ in 0..512 {
match tokio::time::timeout(Duration::from_secs(2), transport.send_async(&ta, &frame))
.await
{
Ok(Ok(_)) => queued += 1,
Ok(Err(_)) => refused += 1,
Err(_) => panic!(
"send blocked on a BLE peer that stopped reading; the write is back \
on the caller's task"
),
}
// The pool must stay available the whole time. Before the writer
// task this guard was held across the L2CAP write, so a stalled
// link froze every other BLE operation too.
let guard = tokio::time::timeout(Duration::from_millis(100), transport.pool.lock())
.await
.expect("the pool lock must never be held across a BLE write");
drop(guard);
}
assert!(queued > 0, "the first sends must be accepted");
assert!(
refused > 0,
"a BLE peer that never drains must eventually have sends refused rather \
than queued without bound: queued={queued}"
);
}
#[test]
fn test_transport_type() {
let io = MockBleIo::new("hci0", test_addr(1));
@@ -2137,6 +2316,8 @@ mod tests {
ta.clone(),
BleConnection {
stream: Arc::new(parked),
send_tx: tokio::sync::mpsc::channel(1).0,
send_task: None,
recv_task: None,
send_mtu: 2048,
recv_mtu: 2048,
@@ -2808,6 +2989,8 @@ mod tests {
ta.clone(),
BleConnection {
stream: Arc::new(parked),
send_tx: tokio::sync::mpsc::channel(1).0,
send_task: None,
recv_task: None,
send_mtu: 64,
recv_mtu: 64,
src/transport/ble/pool.rs
+18
View File
@@ -13,10 +13,23 @@ use crate::transport::{TransportAddr, TransportError};
use super::addr::BleAddr;
/// How many frames may be queued for one BLE connection before sends to it
/// fail. Shallower than the IP transports': a BLE link carries a fraction of
/// their throughput, so a queue of the same depth would represent seconds of
/// backlog rather than a burst.
pub const SEND_QUEUE_DEPTH: usize = 16;
/// A single BLE connection in the pool.
pub struct BleConnection<S> {
/// The L2CAP stream for this connection.
pub stream: S,
/// Frames queued for the writer task. Sending is an enqueue, never a
/// write: the write is awaited only by `send_task`, so no caller can be
/// held by a peer that has stopped draining — and, on this transport in
/// particular, no caller holds the pool lock while it happens.
pub send_tx: tokio::sync::mpsc::Sender<Vec<u8>>,
/// Writer task for this connection.
pub send_task: Option<JoinHandle<()>>,
/// Background receive task handle.
pub recv_task: Option<JoinHandle<()>>,
/// Negotiated L2CAP send MTU.
@@ -53,6 +66,9 @@ impl<S> Drop for BleConnection<S> {
if let Some(task) = self.recv_task.take() {
task.abort();
}
if let Some(task) = self.send_task.take() {
task.abort();
}
}
}
@@ -239,6 +255,8 @@ mod tests {
fn test_conn(n: u8, is_static: bool) -> BleConnection<()> {
BleConnection {
stream: (),
send_tx: tokio::sync::mpsc::channel(1).0,
send_task: None,
recv_task: None,
send_mtu: 2048,
recv_mtu: 2048,
src/transport/nym/mod.rs
+62 -33
View File
@@ -20,8 +20,9 @@ use super::{
};
use crate::config::NymConfig;
use crate::transport::socks5::{
ConnectingEntry, ConnectingPool, DialError, ProxiedConnection, ProxiedPool, Socks5Auth,
Socks5Dialer, SocksTarget, poll_connecting, proxied_receive_loop,
ConnectingEntry, ConnectingPool, DialError, ProxiedConnection, ProxiedPool, SEND_QUEUE_DEPTH,
Socks5Auth, Socks5Dialer, SocksTarget, poll_connecting, proxied_receive_loop,
proxied_send_loop,
};
use stats::NymStats;
@@ -29,12 +30,10 @@ use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpStream;
use tokio::net::tcp::OwnedWriteHalf;
use tokio::sync::Mutex;
use tokio::time::Instant;
use tracing::{debug, info, trace, warn};
use tracing::{debug, info, warn};
// ============================================================================
// Nym Transport
@@ -214,6 +213,7 @@ impl NymTransport {
let mut pool = self.pool.lock().await;
for (addr, conn) in pool.drain() {
conn.recv_task.abort();
conn.send_task.abort();
let _ = conn.recv_task.await;
debug!(
transport_id = %self.transport_id,
@@ -256,42 +256,45 @@ impl NymTransport {
});
}
// Get or create connection
let writer = {
// Get or create the connection's send queue. Never the write half:
// this function must not be able to await the wire (see
// `proxied_send_loop`).
let send_tx = {
let pool = self.pool.lock().await;
pool.get(addr).map(|c| c.writer.clone())
pool.get(addr).map(|c| c.send_tx.clone())
};
let writer = match writer {
Some(w) => w,
let send_tx = match send_tx {
Some(tx) => tx,
None => {
// Connect-on-send
self.connect(addr).await?
}
};
// Write packet
let mut w = writer.lock().await;
match w.write_all(data).await {
Ok(()) => {
self.stats.record_send(data.len());
trace!(
// Queue the frame. `try_send`, not `send`: awaiting a full queue would
// reinstate one level up exactly the block this removes. The byte
// count is what was queued; bytes on the wire are recorded by the
// writer task.
match send_tx.try_send(data.to_vec()) {
Ok(()) => Ok(data.len()),
Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
self.stats.record_send_error();
debug!(
transport_id = %self.transport_id,
remote_addr = %addr,
bytes = data.len(),
"Nym packet sent"
depth = SEND_QUEUE_DEPTH,
"Nym outbound queue full; peer is not draining"
);
Ok(data.len())
Err(TransportError::SendFailed(
"outbound queue full: peer not draining".to_string(),
))
}
Err(e) => {
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
self.stats.record_send_error();
drop(w);
// Remove failed connection from pool
let mut pool = self.pool.lock().await;
if let Some(conn) = pool.remove(addr) {
conn.recv_task.abort();
}
Err(TransportError::SendFailed(format!("{}", e)))
Err(TransportError::SendFailed(
"connection writer gone".to_string(),
))
}
}
}
@@ -300,7 +303,7 @@ impl NymTransport {
async fn connect(
&self,
addr: &TransportAddr,
) -> Result<Arc<Mutex<OwnedWriteHalf>>, TransportError> {
) -> Result<tokio::sync::mpsc::Sender<Vec<u8>>, TransportError> {
let target_addr = parse_target_addr(addr)?;
let proxy_addr = self.config.socks5_addr();
let timeout_ms = self.config.connect_timeout_ms();
@@ -348,7 +351,6 @@ impl NymTransport {
// Split and spawn receive task
let (read_half, write_half) = stream.into_split();
let writer = Arc::new(Mutex::new(write_half));
let transport_id = self.transport_id;
let packet_tx = self.packet_tx.clone();
@@ -370,8 +372,21 @@ impl NymTransport {
.await;
});
let (send_tx, send_rx) = tokio::sync::mpsc::channel(SEND_QUEUE_DEPTH);
let send_task = tokio::spawn(proxied_send_loop(
write_half,
send_rx,
transport_id,
addr.clone(),
self.pool.clone(),
self.stats.clone(),
"Nym",
|_stats: &NymStats, _meta: &()| {},
));
let conn = ProxiedConnection {
writer: writer.clone(),
send_tx: send_tx.clone(),
send_task,
recv_task,
mtu,
established_at: Instant::now(),
@@ -390,7 +405,7 @@ impl NymTransport {
"Nym mixnet connection established via SOCKS5"
);
Ok(writer)
Ok(send_tx)
}
/// Initiate a non-blocking connection to a remote address.
@@ -500,7 +515,6 @@ impl NymTransport {
/// Promote a completed background connection to the established pool.
fn promote_connection(&self, addr: &TransportAddr, stream: TcpStream, mtu: u16) {
let (read_half, write_half) = stream.into_split();
let writer = Arc::new(Mutex::new(write_half));
let transport_id = self.transport_id;
let packet_tx = self.packet_tx.clone();
@@ -521,8 +535,21 @@ impl NymTransport {
.await;
});
let (send_tx, send_rx) = tokio::sync::mpsc::channel(SEND_QUEUE_DEPTH);
let send_task = tokio::spawn(proxied_send_loop(
write_half,
send_rx,
transport_id,
addr.clone(),
self.pool.clone(),
self.stats.clone(),
"Nym",
|_stats: &NymStats, _meta: &()| {},
));
let conn = ProxiedConnection {
writer,
send_tx,
send_task,
recv_task,
mtu,
established_at: Instant::now(),
@@ -539,6 +566,7 @@ impl NymTransport {
);
} else {
conn.recv_task.abort();
conn.send_task.abort();
warn!(
transport_id = %self.transport_id,
remote_addr = %addr,
@@ -552,6 +580,7 @@ impl NymTransport {
let mut pool = self.pool.lock().await;
if let Some(conn) = pool.remove(addr) {
conn.recv_task.abort();
conn.send_task.abort();
debug!(
transport_id = %self.transport_id,
remote_addr = %addr,
src/transport/nym/stats.rs
+8
View File
@@ -94,6 +94,14 @@ impl ProxiedStats for NymStats {
fn record_recv_error(&self) {
self.base.record_recv_error();
}
fn record_send(&self, bytes: usize) {
NymStats::record_send(self, bytes);
}
fn record_send_error(&self) {
NymStats::record_send_error(self);
}
}
/// Point-in-time snapshot of Nym stats (non-atomic, copyable).
src/transport/socks5/mod.rs
+2 -2
View File
@@ -11,8 +11,8 @@ mod stats;
pub use dialer::{DialError, Socks5Auth, Socks5Dialer, SocksTarget};
pub(crate) use pool::{
ConnectingEntry, ConnectingPool, ProxiedConnection, ProxiedPool, ProxiedStats, poll_connecting,
proxied_receive_loop,
ConnectingEntry, ConnectingPool, ProxiedConnection, ProxiedPool, ProxiedStats,
SEND_QUEUE_DEPTH, poll_connecting, proxied_receive_loop, proxied_send_loop,
};
pub(crate) use stats::ProxiedStatsBase;
src/transport/socks5/pool.rs
+82 -3
View File
@@ -13,11 +13,13 @@ use std::time::Duration;
use futures::FutureExt;
use tokio::net::TcpStream;
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::sync::Mutex;
use tokio::sync::{Mutex, mpsc};
use tokio::task::JoinHandle;
use tokio::time::Instant;
use tracing::{debug, trace};
use tokio::io::AsyncWriteExt;
use crate::transport::framing::read_fmp_packet;
use crate::transport::{
ConnectionState, PacketTx, ReceivedPacket, TransportAddr, TransportError, TransportId,
@@ -28,8 +30,12 @@ use crate::transport::{
/// `M` is per-transport metadata: `Direction` for tor (drives
/// inbound/outbound pool accounting), `()` for nym.
pub(crate) struct ProxiedConnection<M> {
/// Write half of the split stream.
pub writer: Arc<Mutex<OwnedWriteHalf>>,
/// Frames queued for the writer task. Sending is an enqueue, never a
/// write: the write half belongs to `send_task`, so no caller can block on
/// the wire. A full queue is a peer that has stopped draining.
pub send_tx: mpsc::Sender<Vec<u8>>,
/// Writer task for this connection.
pub send_task: JoinHandle<()>,
/// Receive task for this connection.
pub recv_task: JoinHandle<()>,
/// MTU for this connection.
@@ -124,6 +130,79 @@ pub(crate) trait ProxiedStats: Send + Sync + 'static {
fn record_recv(&self, bytes: usize);
/// Record a receive error.
fn record_recv_error(&self);
/// Record `bytes` actually written to the wire.
fn record_send(&self, bytes: usize);
/// Record a send error.
fn record_send_error(&self);
}
/// How many frames may be queued for one connection before sends to it fail.
/// See `crate::transport::tcp::pool::SEND_QUEUE_DEPTH`, which this mirrors.
pub(crate) const SEND_QUEUE_DEPTH: usize = 64;
/// Per-connection writer task: the only place a write to a proxied stream is
/// ever awaited.
///
/// The reasoning is the TCP transport's, and the shape is deliberately the
/// same. `write_all` blocks once the local socket to the proxy stops draining,
/// and the callers are the rx loop's tick handlers, where that holds every
/// other arm of the select. The loop owns the write half, so nothing else can
/// block on it.
///
/// 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.
#[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,
pool: ProxiedPool<M>,
stats: Arc<S>,
label: &'static str,
on_remove: impl Fn(&S, &M) + Send + 'static,
) {
while let Some(frame) = frames.recv().await {
match writer.write_all(&frame).await {
Ok(()) => {
stats.record_send(frame.len());
trace!(
transport_id = %transport_id,
remote_addr = %remote_addr,
bytes = frame.len(),
"{} packet sent",
label
);
}
Err(e) => {
stats.record_send_error();
debug!(
transport_id = %transport_id,
remote_addr = %remote_addr,
error = %e,
"{} write failed; dropping connection",
label
);
let removed = {
let mut guard = pool.lock().await;
guard.remove(&remote_addr)
};
if let Some(conn) = removed {
conn.recv_task.abort();
on_remove(&stats, &conn.meta);
}
return;
}
}
}
trace!(
transport_id = %transport_id,
remote_addr = %remote_addr,
"{} writer task exiting",
label
);
}
/// Shared per-connection receive loop for the proxied transports.
src/transport/tcp/mod.rs
+216 -42
View File
@@ -42,9 +42,8 @@ use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use tokio::io::AsyncWriteExt;
use tokio::net::tcp::OwnedWriteHalf;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::Mutex;
use tokio::sync::{Mutex, mpsc};
use tokio::task::JoinHandle;
use tokio::time::Instant;
use tracing::{debug, info, trace, warn};
@@ -271,6 +270,7 @@ impl TcpTransport {
let mut pool = self.pool.lock().await;
for (addr, conn) in pool.drain() {
conn.recv_task.abort();
conn.send_task.abort();
let _ = conn.recv_task.await;
match conn.direction {
Direction::Inbound => self.stats.record_pool_inbound_removed(),
@@ -323,46 +323,56 @@ impl TcpTransport {
});
}
// Get or create connection
let writer = {
// Get or create connection. What comes back is the queue into the
// connection's writer task, never the write half itself: this function
// must not be able to await the wire (see `tcp_send_loop`).
let send_tx = {
let pool = self.pool.lock().await;
pool.get(addr).map(|c| c.writer.clone())
pool.get(addr).map(|c| c.send_tx.clone())
};
let writer = match writer {
Some(w) => w,
let send_tx = match send_tx {
Some(tx) => tx,
None => {
// Connect-on-send
self.connect(addr).await?
}
};
// Write packet directly (no framing transformation needed)
let mut w = writer.lock().await;
match w.write_all(data).await {
Ok(()) => {
self.stats.record_send(data.len());
trace!(
// Hand the frame to the writer task. The copy buys the caller its
// freedom from the wire: `write_all` borrows, a queue must own. One
// memcpy of at most an MTU is a good trade for not stalling the rx
// loop on a peer that has stopped reading.
//
// `try_send` rather than `send`: awaiting a full queue would reinstate
// exactly the block this removes, one level up. A full queue means the
// writer task has not drained a frame in the time it took to fill 64 of
// them, which is a peer that is not receiving, so the send fails and
// the caller's own retry policy takes over.
//
// The byte count is what was queued, not what reached the wire — the
// same prediction the UDP fast path reports when it dispatches to the
// encrypt workers. Bytes actually written are recorded by the writer
// task as they go.
match send_tx.try_send(data.to_vec()) {
Ok(()) => Ok(data.len()),
Err(mpsc::error::TrySendError::Full(_)) => {
self.stats.record_send_error();
debug!(
transport_id = %self.transport_id,
remote_addr = %addr,
bytes = data.len(),
"TCP packet sent"
depth = crate::transport::tcp::pool::SEND_QUEUE_DEPTH,
"TCP outbound queue full; peer is not draining"
);
Ok(data.len())
Err(TransportError::SendFailed(
"outbound queue full: peer not draining".to_string(),
))
}
Err(e) => {
Err(mpsc::error::TrySendError::Closed(_)) => {
self.stats.record_send_error();
drop(w);
// Remove failed connection from pool
let mut pool = self.pool.lock().await;
if let Some(conn) = pool.remove(addr) {
conn.recv_task.abort();
match conn.direction {
Direction::Inbound => self.stats.record_pool_inbound_removed(),
Direction::Outbound => self.stats.record_pool_outbound_removed(),
}
}
Err(TransportError::SendFailed(format!("{}", e)))
Err(TransportError::SendFailed(
"connection writer gone".to_string(),
))
}
}
}
@@ -371,10 +381,7 @@ impl TcpTransport {
///
/// Configures socket options, reads TCP_MAXSEG for MTU, splits the
/// stream, spawns a receive task, and stores in the pool.
async fn connect(
&self,
addr: &TransportAddr,
) -> Result<Arc<Mutex<OwnedWriteHalf>>, TransportError> {
async fn connect(&self, addr: &TransportAddr) -> Result<mpsc::Sender<Vec<u8>>, TransportError> {
let socket_addr = resolve_socket_addr(addr).await?;
let timeout_ms = self.config.connect_timeout_ms();
@@ -411,7 +418,6 @@ impl TcpTransport {
// Split and spawn receive task
let (read_half, write_half) = stream.into_split();
let writer = Arc::new(Mutex::new(write_half));
let transport_id = self.transport_id;
let packet_tx = self.packet_tx.clone();
@@ -438,8 +444,19 @@ impl TcpTransport {
.await;
});
let (send_tx, send_rx) = mpsc::channel(crate::transport::tcp::pool::SEND_QUEUE_DEPTH);
let send_task = tokio::spawn(tcp_send_loop(
write_half,
send_rx,
transport_id,
addr.clone(),
self.pool.clone(),
self.stats.clone(),
));
let conn = TcpConnection {
writer: writer.clone(),
send_tx: send_tx.clone(),
send_task,
recv_task,
mtu: mss_mtu,
established_at: Instant::now(),
@@ -459,7 +476,7 @@ impl TcpTransport {
"TCP connection established (connect-on-send)"
);
Ok(writer)
Ok(send_tx)
}
/// Close a specific connection asynchronously.
@@ -470,6 +487,7 @@ impl TcpTransport {
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 {
Direction::Inbound => self.stats.record_pool_inbound_removed(),
Direction::Outbound => self.stats.record_pool_outbound_removed(),
@@ -665,7 +683,6 @@ impl TcpTransport {
/// Called from `connection_state_sync()` when a background task completes.
fn promote_connection(&self, addr: &TransportAddr, stream: TcpStream, mss_mtu: u16) {
let (read_half, write_half) = stream.into_split();
let writer = Arc::new(Mutex::new(write_half));
let transport_id = self.transport_id;
let packet_tx = self.packet_tx.clone();
@@ -691,8 +708,19 @@ impl TcpTransport {
.await;
});
let (send_tx, send_rx) = mpsc::channel(crate::transport::tcp::pool::SEND_QUEUE_DEPTH);
let send_task = tokio::spawn(tcp_send_loop(
write_half,
send_rx,
transport_id,
addr.clone(),
self.pool.clone(),
self.stats.clone(),
));
let conn = TcpConnection {
writer,
send_tx,
send_task,
recv_task,
mtu: mss_mtu,
established_at: Instant::now(),
@@ -714,6 +742,7 @@ impl TcpTransport {
} else {
// Pool locked — abort the recv task, connection will be retried
conn.recv_task.abort();
conn.send_task.abort();
warn!(
transport_id = %self.transport_id,
remote_addr = %addr,
@@ -892,7 +921,6 @@ async fn accept_loop(
// Split and spawn receive task
let (read_half, write_half) = stream.into_split();
let writer = Arc::new(Mutex::new(write_half));
let recv_pool = pool.clone();
let recv_packet_tx = packet_tx.clone();
@@ -921,8 +949,20 @@ async fn accept_loop(
.await;
});
let (send_tx, send_rx) =
mpsc::channel(crate::transport::tcp::pool::SEND_QUEUE_DEPTH);
let send_task = tokio::spawn(tcp_send_loop(
write_half,
send_rx,
transport_id,
remote_addr.clone(),
pool.clone(),
stats.clone(),
));
let conn = TcpConnection {
writer,
send_tx,
send_task,
recv_task,
mtu: conn_mtu,
established_at: Instant::now(),
@@ -959,9 +999,80 @@ async fn accept_loop(
}
// ============================================================================
// Receive Loop (per-connection)
// Per-connection Loops (writer and receiver)
// ============================================================================
/// Per-connection writer task: the only place a TCP write is ever awaited.
///
/// This exists so the caller does not await the wire. `write_all` on a stream
/// blocks once the kernel send buffer fills, which is precisely what a peer
/// that has stopped draining causes — and the callers are the rx loop's tick
/// handlers, where blocking holds every other arm of the select behind it. The
/// loop owns the write half outright, so no other task can hold it and no
/// other task can be held by it.
///
/// On a write error the connection is removed from the pool, mirroring
/// `tcp_receive_loop`'s teardown contract: the pool entry is removed and the
/// direction counter decremented only when the removal returned `Some`, so a
/// concurrent `close`/`stop` of the same address cannot double-count. The
/// 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.
///
/// 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.
async fn tcp_send_loop(
mut writer: tokio::net::tcp::OwnedWriteHalf,
mut frames: mpsc::Receiver<Vec<u8>>,
transport_id: TransportId,
remote_addr: TransportAddr,
pool: ConnectionPool,
stats: Arc<TcpStats>,
) {
while let Some(frame) = frames.recv().await {
match writer.write_all(&frame).await {
Ok(()) => {
stats.record_send(frame.len());
trace!(
transport_id = %transport_id,
remote_addr = %remote_addr,
bytes = frame.len(),
"TCP packet sent"
);
}
Err(e) => {
stats.record_send_error();
debug!(
transport_id = %transport_id,
remote_addr = %remote_addr,
error = %e,
"TCP write failed; dropping connection"
);
let removed = {
let mut pool = pool.lock().await;
pool.remove(&remote_addr)
};
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(),
}
}
return;
}
}
}
// The sender side is gone: the pool entry was dropped, so the connection
// is already being torn down and there is nothing to clean up here.
trace!(
transport_id = %transport_id,
remote_addr = %remote_addr,
"TCP writer task exiting"
);
}
/// Per-connection TCP receive loop.
///
/// Reads complete FMP packets using the stream reader, delivers them to
@@ -1549,6 +1660,68 @@ mod tests {
assert!(!transport.accept_connections());
}
/// **The property this transport's send path exists to guarantee.**
///
/// A peer that stops reading fills its receive window, then this node's
/// kernel send buffer, and from that moment `write_all` blocks until the
/// peer drains or the connection dies. The callers are the rx loop's tick
/// handlers — the heartbeat sweep among them — so a blocking send holds
/// every other arm of the select behind it: control RPCs, forwarding,
/// every other peer's liveness. A medium change is precisely the condition
/// that produces such a peer, which is how this was found.
///
/// The writer task owns the write half, so `send_async` can only ever
/// enqueue. This drives a peer that accepts the connection and then never
/// reads, pushes far more than any socket buffer will hold, and asserts
/// every call returns promptly — failing once the queue fills, rather than
/// blocking on a peer that is not listening.
#[tokio::test]
async fn a_peer_that_stops_reading_cannot_block_the_sender() {
let (tx1, _rx1) = packet_channel(100);
let mut t1 = TcpTransport::new(TransportId::new(1), None, make_outbound_config(), tx1);
t1.start_async().await.unwrap();
// A listener that accepts and then never reads a byte. Holding the
// stream is the point: dropping it would close the connection and turn
// the writes into fast errors, which is not the case under test.
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let listen_addr = listener.local_addr().unwrap();
let deaf = tokio::spawn(async move {
let (stream, _) = listener.accept().await.unwrap();
tokio::time::sleep(Duration::from_secs(30)).await;
drop(stream);
});
let remote = TransportAddr::from_string(&listen_addr.to_string());
// Enough to overrun any plausible socket buffer plus the queue behind
// it, so the blocking case cannot be missed by sending too little.
let frame = vec![0xAB; 1400];
let mut queued = 0usize;
let mut refused = 0usize;
for _ in 0..8000 {
// The budget is per call and generous: a healthy enqueue is
// microseconds, while the old unbounded write parked here until
// the peer drained, which it never does.
match timeout(Duration::from_secs(2), t1.send_async(&remote, &frame)).await {
Ok(Ok(_)) => queued += 1,
Ok(Err(_)) => refused += 1,
Err(_) => panic!(
"send blocked on a peer that stopped reading; \
the write is back on the caller's task"
),
}
}
assert!(queued > 0, "the first sends must be accepted");
assert!(
refused > 0,
"a peer that never drains must eventually have sends refused rather than \
queued without bound: queued={queued}"
);
deaf.abort();
}
#[tokio::test]
async fn test_connection_drop_and_reconnect() {
let (tx1, _rx1) = packet_channel(100);
@@ -2050,14 +2223,15 @@ mod tests {
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 (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 {
writer: Arc::new(Mutex::new(write_half)),
send_tx: mpsc::channel(1).0,
send_task: tokio::spawn(async {}),
recv_task: tokio::spawn(async {}),
mtu: 1400,
established_at: Instant::now(),
src/transport/tcp/pool.rs
+19 -4
View File
@@ -6,8 +6,7 @@
use std::collections::HashMap;
use std::sync::Arc;
use tokio::net::TcpStream;
use tokio::net::tcp::OwnedWriteHalf;
use tokio::sync::Mutex;
use tokio::sync::{Mutex, mpsc};
use tokio::task::JoinHandle;
use tokio::time::Instant;
@@ -24,10 +23,26 @@ pub(crate) enum Direction {
Outbound,
}
/// How many frames may be queued for one connection before sends to it start
/// failing.
///
/// The queue exists so the caller never awaits the wire; the bound exists so a
/// peer that has stopped draining cannot turn that into unbounded memory. Deep
/// enough to absorb a burst — a heartbeat sweep plus the forwarding this node
/// does for one peer — and shallow enough that a stranded peer is recognised
/// within a tick or two rather than after megabytes have piled up behind it.
pub(crate) const SEND_QUEUE_DEPTH: usize = 64;
/// State for a single TCP connection to a peer.
pub(crate) struct TcpConnection {
/// Write half of the split stream.
pub(crate) writer: Arc<Mutex<OwnedWriteHalf>>,
/// Frames queued for the writer task. Sending is an enqueue, never a
/// write: the write half belongs to `send_task` and nothing else can
/// block on it. A full queue is a peer that has stopped draining, and the
/// send fails rather than waiting.
pub(crate) send_tx: mpsc::Sender<Vec<u8>>,
/// Writer task for this connection. Owns the write half of the split
/// stream, so the only code that can ever await `write_all` is this task.
pub(crate) send_task: JoinHandle<()>,
/// Receive task for this connection.
pub(crate) recv_task: JoinHandle<()>,
/// MSS-derived MTU for this connection (used for dynamic MTU re-reading).
src/transport/tor/mod.rs
+91 -53
View File
@@ -30,8 +30,9 @@ use super::{
};
use crate::config::TorConfig;
use crate::transport::socks5::{
ConnectingEntry, ConnectingPool, DialError, ProxiedConnection, ProxiedPool, Socks5Auth,
Socks5Dialer, SocksTarget, poll_connecting, proxied_receive_loop,
ConnectingEntry, ConnectingPool, DialError, ProxiedConnection, ProxiedPool, SEND_QUEUE_DEPTH,
Socks5Auth, Socks5Dialer, SocksTarget, poll_connecting, proxied_receive_loop,
proxied_send_loop,
};
use crate::transport::tcp::INBOUND_FIRST_FRAME_TIMEOUT;
use control::{ControlAuth, TorControlClient, TorMonitoringInfo};
@@ -42,13 +43,11 @@ use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use tokio::io::AsyncWriteExt;
use tokio::net::tcp::OwnedWriteHalf;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::Mutex;
use tokio::task::JoinHandle;
use tokio::time::Instant;
use tracing::{debug, info, trace, warn};
use tracing::{debug, info, warn};
// ============================================================================
// Address Parsing
@@ -497,6 +496,7 @@ impl TorTransport {
let mut pool = self.pool.lock().await;
for (addr, conn) in pool.drain() {
conn.recv_task.abort();
conn.send_task.abort();
let _ = conn.recv_task.await;
match conn.meta {
Direction::Inbound => self.stats.record_pool_inbound_removed(),
@@ -639,45 +639,42 @@ impl TorTransport {
}
// Get or create connection
let writer = {
let send_tx = {
let pool = self.pool.lock().await;
pool.get(addr).map(|c| c.writer.clone())
pool.get(addr).map(|c| c.send_tx.clone())
};
let writer = match writer {
Some(w) => w,
let send_tx = match send_tx {
Some(tx) => tx,
None => {
// Connect-on-send
self.connect(addr).await?
}
};
// Write packet directly (no framing transformation needed)
let mut w = writer.lock().await;
match w.write_all(data).await {
Ok(()) => {
self.stats.record_send(data.len());
trace!(
// Queue the frame for the connection's writer task. `try_send`, not
// `send`: awaiting a full queue would reinstate the block this removes
// one level up. The byte count is what was queued; bytes on the wire
// are recorded by the writer task.
match send_tx.try_send(data.to_vec()) {
Ok(()) => Ok(data.len()),
Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
self.stats.record_send_error();
debug!(
transport_id = %self.transport_id,
remote_addr = %addr,
bytes = data.len(),
"Tor packet sent"
depth = SEND_QUEUE_DEPTH,
"Tor outbound queue full; peer is not draining"
);
Ok(data.len())
Err(TransportError::SendFailed(
"outbound queue full: peer not draining".to_string(),
))
}
Err(e) => {
Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
self.stats.record_send_error();
drop(w);
// Remove failed connection from pool
let mut pool = self.pool.lock().await;
if let Some(conn) = pool.remove(addr) {
conn.recv_task.abort();
match conn.meta {
Direction::Inbound => self.stats.record_pool_inbound_removed(),
Direction::Outbound => self.stats.record_pool_outbound_removed(),
}
}
Err(TransportError::SendFailed(format!("{}", e)))
Err(TransportError::SendFailed(
"connection writer gone".to_string(),
))
}
}
}
@@ -690,7 +687,7 @@ impl TorTransport {
async fn connect(
&self,
addr: &TransportAddr,
) -> Result<Arc<Mutex<OwnedWriteHalf>>, TransportError> {
) -> Result<tokio::sync::mpsc::Sender<Vec<u8>>, TransportError> {
let tor_addr = parse_tor_addr(addr)?;
let proxy_addr = self.config.socks5_addr();
let timeout_ms = self.config.connect_timeout_ms();
@@ -752,7 +749,6 @@ impl TorTransport {
// Split and spawn receive task
let (read_half, write_half) = stream.into_split();
let writer = Arc::new(Mutex::new(write_half));
let transport_id = self.transport_id;
let packet_tx = self.packet_tx.clone();
@@ -779,8 +775,24 @@ impl TorTransport {
.await;
});
let (send_tx, send_rx) = tokio::sync::mpsc::channel(SEND_QUEUE_DEPTH);
let send_task = tokio::spawn(proxied_send_loop(
write_half,
send_rx,
transport_id,
addr.clone(),
self.pool.clone(),
self.stats.clone(),
"Tor",
|stats: &TorStats, meta: &Direction| match meta {
Direction::Inbound => stats.record_pool_inbound_removed(),
Direction::Outbound => stats.record_pool_outbound_removed(),
},
));
let conn = ProxiedConnection {
writer: writer.clone(),
send_tx: send_tx.clone(),
send_task,
recv_task,
mtu,
established_at: Instant::now(),
@@ -800,7 +812,7 @@ impl TorTransport {
"Tor circuit established via SOCKS5"
);
Ok(writer)
Ok(send_tx)
}
/// Initiate a non-blocking connection to a remote address.
@@ -922,7 +934,6 @@ impl TorTransport {
/// Called from `connection_state_sync()` when a background task completes.
fn promote_connection(&self, addr: &TransportAddr, stream: TcpStream, mtu: u16) {
let (read_half, write_half) = stream.into_split();
let writer = Arc::new(Mutex::new(write_half));
let transport_id = self.transport_id;
let packet_tx = self.packet_tx.clone();
@@ -948,8 +959,24 @@ impl TorTransport {
.await;
});
let (send_tx, send_rx) = tokio::sync::mpsc::channel(SEND_QUEUE_DEPTH);
let send_task = tokio::spawn(proxied_send_loop(
write_half,
send_rx,
transport_id,
addr.clone(),
self.pool.clone(),
self.stats.clone(),
"Tor",
|stats: &TorStats, meta: &Direction| match meta {
Direction::Inbound => stats.record_pool_inbound_removed(),
Direction::Outbound => stats.record_pool_outbound_removed(),
},
));
let conn = ProxiedConnection {
writer,
send_tx,
send_task,
recv_task,
mtu,
established_at: Instant::now(),
@@ -970,6 +997,7 @@ impl TorTransport {
} else {
// Pool locked — abort the recv task, connection will be retried
conn.recv_task.abort();
conn.send_task.abort();
warn!(
transport_id = %self.transport_id,
remote_addr = %addr,
@@ -983,6 +1011,7 @@ impl TorTransport {
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 {
Direction::Inbound => self.stats.record_pool_inbound_removed(),
Direction::Outbound => self.stats.record_pool_outbound_removed(),
@@ -1196,7 +1225,6 @@ async fn tor_accept_loop(
// Split stream and spawn receive task
let (read_half, write_half) = stream.into_split();
let writer = Arc::new(Mutex::new(write_half));
let recv_pool = pool.clone();
let recv_stats = stats.clone();
@@ -1225,8 +1253,24 @@ async fn tor_accept_loop(
.await;
});
let (send_tx, send_rx) = tokio::sync::mpsc::channel(SEND_QUEUE_DEPTH);
let send_task = tokio::spawn(proxied_send_loop(
write_half,
send_rx,
transport_id,
remote_addr.clone(),
pool.clone(),
stats.clone(),
"Tor",
|stats: &TorStats, meta: &Direction| match meta {
Direction::Inbound => stats.record_pool_inbound_removed(),
Direction::Outbound => stats.record_pool_outbound_removed(),
},
));
let conn = ProxiedConnection {
writer,
send_tx,
send_task,
recv_task,
mtu,
established_at: Instant::now(),
@@ -1294,6 +1338,7 @@ fn validate_host_port(addr: &str, field_name: &str) -> Result<(), TransportError
mod tests {
use super::*;
use crate::transport::packet_channel;
use tokio::io::AsyncWriteExt;
fn make_config() -> TorConfig {
TorConfig {
@@ -2090,16 +2135,6 @@ mod tests {
// Accept-loop readiness barrier
// ========================================================================
/// Build a throwaway `OwnedWriteHalf` for a hand-planted pool entry.
async fn spare_write_half() -> OwnedWriteHalf {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let client = TcpStream::connect(addr).await.unwrap();
let (_server, _) = listener.accept().await.unwrap();
let (_read, write) = client.into_split();
write
}
/// The accept loop can be torn down between the pool insert and the
/// `ready_tx.send()`: the sender is dropped, so `ready_rx.await` returns
/// `Err`. The receive loop must still fall through to its cleanup, or the
@@ -2114,14 +2149,15 @@ mod tests {
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 (read_half, _write_half) = server.into_split();
let pool: ProxiedPool<Direction> = Arc::new(Mutex::new(HashMap::new()));
let stats = Arc::new(TorStats::new());
pool.lock().await.insert(
remote.clone(),
ProxiedConnection {
writer: Arc::new(Mutex::new(write_half)),
send_tx: tokio::sync::mpsc::channel(1).0,
send_task: tokio::spawn(async {}),
recv_task: tokio::spawn(async {}),
mtu: 1400,
established_at: Instant::now(),
@@ -2178,7 +2214,7 @@ mod tests {
let (server, peer_addr) = listener.accept().await.unwrap();
let remote = TransportAddr::from_string(&peer_addr.to_string());
drop(client);
let (read_half, write_half) = server.into_split();
let (read_half, _write_half) = server.into_split();
let pool: ProxiedPool<Direction> = Arc::new(Mutex::new(HashMap::new()));
let stats = Arc::new(TorStats::new());
@@ -2214,7 +2250,8 @@ mod tests {
pool.lock().await.insert(
remote.clone(),
ProxiedConnection {
writer: Arc::new(Mutex::new(write_half)),
send_tx: tokio::sync::mpsc::channel(1).0,
send_task: tokio::spawn(async {}),
recv_task: tokio::spawn(async {}),
mtu: 1400,
established_at: Instant::now(),
@@ -2270,7 +2307,8 @@ mod tests {
pool.lock().await.insert(
remote.clone(),
ProxiedConnection {
writer: Arc::new(Mutex::new(spare_write_half().await)),
send_tx: tokio::sync::mpsc::channel(1).0,
send_task: tokio::spawn(async {}),
recv_task: tokio::spawn(std::future::pending::<()>()),
mtu: 1400,
established_at: Instant::now(),
src/transport/tor/stats.rs
+8
View File
@@ -159,6 +159,14 @@ impl ProxiedStats for TorStats {
fn record_recv_error(&self) {
self.base.record_recv_error();
}
fn record_send(&self, bytes: usize) {
TorStats::record_send(self, bytes);
}
fn record_send_error(&self) {
TorStats::record_send_error(self);
}
}
/// Point-in-time snapshot of Tor stats (non-atomic, copyable).