diff --git a/CHANGELOG.md b/CHANGELOG.md index 7cd14768..2e30a11d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -722,6 +722,34 @@ with v0.5.x or earlier peers. - An Ethernet peer's address is always shown as a colon-separated MAC address in `fipsctl show links` and in log lines. An address whose six bytes happened to be valid UTF-8 was printed as text instead. +- The node no longer dials from its receive loop. When a peer's TCP, Tor or + Nym connection closed before the node answered or resent a handshake or + rekey message on it, the send dialed the peer's address and the receive + loop waited out the whole connect timeout (5 s by default on TCP) before + handling anything else, one send after another. For an inbound connection + that address is the peer's source port, where nothing listens. Meanwhile + UDP receive queues filled, other peers' links timed out and dropped, and the + log showed a `transport timeout` warning on `Failed to send msg2` about + every 5 s (#176). These sends now use only a connection the transport + already holds, and fail at once without one. For a peer this node dialed, a + connect is started in the background and the next send uses it; an inbound + peer's address is never dialed. +- `show_transports` (`fipsctl show transports`) reports transport counters as + they stand when asked. They were a copy refreshed by the node's periodic + tick, so when the receive loop was too busy for the tick to run they stopped + moving with nothing to show it: in #176, `connect_timeouts` read 0 while the + log recorded hundreds of timeouts. The output is unchanged. +- The UDP transport's `packets_recv` and `bytes_recv` now count datagrams + received on per-peer connected sockets, and `packets_sent`, `bytes_sent` and + `send_errors` count the data frames the encrypt workers send. Both were + missed, so on a node with connected UDP on (the Linux and macOS default) + the counters were far below the traffic the transport carried. +- A failed connect made in the background by TCP, Tor or Nym is now counted, + as a failed inline connect is. TCP counted none, Tor missed timeouts, and + Nym missed timeouts and SOCKS5 errors. TCP's `connect_refused` counts every + connect that fails before its timeout, not only a refused one: an + unreachable host or network, a reset or any other connect error lands there + too, with the error in the debug log. #### Gateway diff --git a/src/control/queries.rs b/src/control/queries.rs index cdab2b36..52d2c2ee 100644 --- a/src/control/queries.rs +++ b/src/control/queries.rs @@ -1443,9 +1443,11 @@ pub fn show_transports(node: &Node) -> Value { /// Off-loop variant of [`show_transports`]: renders from the tick-published /// [`EntitySnapshot`](super::snapshot::EntitySnapshot) transport table. The -/// `stats` and `tor_monitoring` blocks are already-projected `serde_json::Value` -/// data captured at publish time. Output is byte-identical to -/// [`show_transports`]. +/// `tor_monitoring` block is already-projected `serde_json::Value` data +/// captured at publish time. The `stats` block is read now from the +/// transport's shared counters, as `show_status` reads `forwarding`, so it is +/// current even when the tick has not run since the counters moved. Output +/// is byte-identical to [`show_transports`]. pub(crate) fn show_transports_from_handle(handle: &super::read_handle::ControlReadHandle) -> Value { let entities = handle.entities(); let transports: Vec = entities @@ -1488,7 +1490,7 @@ pub(crate) fn show_transports_from_handle(handle: &super::read_handle::ControlRe }); } - t_json["stats"] = t.stats.clone(); + t_json["stats"] = t.stats.to_json(); t_json }) diff --git a/src/control/read_handle.rs b/src/control/read_handle.rs index ad4d1f80..4d46ebe2 100644 --- a/src/control/read_handle.rs +++ b/src/control/read_handle.rs @@ -27,7 +27,10 @@ //! //! A projection is a point-in-time copy, not a live view. The entity tables in //! particular are mutated on the packet path between ticks, so a reader sees -//! the state as of the last publish. +//! the state as of the last publish. The exceptions are counters that live in +//! shared atomics, which are read at request time: the `MetricsRegistry` +//! families, and the transport counters `show_transports` reports, whose rows +//! hold the transports' shared stats rather than a copy of them. //! //! [`snapshot_dispatch`] is the seam: it serves the commands in its match arms //! directly from the handle and returns `None` for everything else, so the diff --git a/src/control/snapshot.rs b/src/control/snapshot.rs index 0ce2ca8a..fe3bccb2 100644 --- a/src/control/snapshot.rs +++ b/src/control/snapshot.rs @@ -765,10 +765,18 @@ pub(crate) struct ConnectionRow { pub expected_peer: Option, } -/// One transport instance in `show_transports`. The `stats` and -/// `tor_monitoring` fields are stored as already-projected `serde_json::Value` -/// (data, produced by the transport handle), not as rendered `Response` -/// envelopes. +/// One transport instance in `show_transports`. The `tor_monitoring` field is +/// stored as an already-projected `serde_json::Value` (data, produced by the +/// transport handle), not as a rendered `Response` envelope. +/// +/// `stats` is the exception to the snapshot being a point-in-time copy: it +/// holds the transport's shared counters, which the query reads at request +/// time. They are atomics the transport updates from its own tasks, so they +/// stay current when the tick that publishes this row is late or held, which +/// is when an operator most needs them. The rest of the row is copied at +/// publish time like every other row. Equality compares the counters by +/// identity, so a row whose other fields are unchanged is reused across +/// publishes however much traffic the transport carried. #[derive(Clone, PartialEq)] pub(crate) struct TransportRow { pub transport_id: u32, @@ -780,7 +788,7 @@ pub(crate) struct TransportRow { pub tor_mode: Option, pub onion_address: Option, pub tor_monitoring: Option, - pub stats: serde_json::Value, + pub stats: crate::transport::LiveStats, /// Interface presence for interface-bound transports; `None` for the rest. pub interface: Option, } diff --git a/src/node/dataplane/connected_udp.rs b/src/node/dataplane/connected_udp.rs index 090dd315..7cf88f14 100644 --- a/src/node/dataplane/connected_udp.rs +++ b/src/node/dataplane/connected_udp.rs @@ -119,7 +119,7 @@ impl Node { // the UDP transport's DNS cache. This may await on a DNS // lookup the very first time we see a hostname; subsequent // calls hit the cache. - let (peer_socket_addr, local_addr, recv_buf, send_buf, packet_tx) = { + let (peer_socket_addr, local_addr, recv_buf, send_buf, packet_tx, stats) = { let Some(transport) = self.transports.get(&transport_id) else { return Ok(()); }; @@ -137,7 +137,7 @@ impl Node { let recv_buf = udp.recv_buf_size(); let send_buf = udp.send_buf_size(); let tx = udp.clone_packet_tx(); - (peer_sa, local, recv_buf, send_buf, tx) + (peer_sa, local, recv_buf, send_buf, tx, udp.stats().clone()) }; // Open the connected socket on the kernel side, then adopt the @@ -155,13 +155,15 @@ impl Node { local_addr, )); - // Spawn the drain thread. It feeds `packet_tx` exactly like - // the wildcard listen socket — rx_loop dispatches identically. + // Spawn the drain thread. It feeds `packet_tx` and counts into + // the transport's stats exactly like the wildcard listen socket, + // so rx_loop dispatches identically. let drain = crate::transport::udp::PeerRecvDrain::spawn( socket.clone(), transport_id, peer_socket_addr, packet_tx, + stats, ) .map_err(|e| format!("PeerRecvDrain::spawn: {e}"))?; diff --git a/src/node/dataplane/peer_actions.rs b/src/node/dataplane/peer_actions.rs index c2e4730f..ced50c2e 100644 --- a/src/node/dataplane/peer_actions.rs +++ b/src/node/dataplane/peer_actions.rs @@ -3,8 +3,9 @@ //! The per-peer FSM in [`crate::peer::machine`] is a sans-IO reducer: it decides //! *what* must happen and returns a `Vec`; this module is the *doing* //! half — the thin driver that maps each action onto the exact shell call it -//! stands for (`build_msg2` + `transport.send`, `promote_connection`, -//! `remove_active_peer`, `index_allocator.free`, `note_link_dead`, …). +//! stands for (`build_msg2` + `send_existing`, `send_stored_msg1`, +//! `promote_connection`, `remove_active_peer`, `index_allocator.free`, +//! `note_link_dead`, …). //! //! ## Progressive cutover //! @@ -189,11 +190,15 @@ impl Node { // (`Node::msg2_failed`: a transient one keeps the leg, // a terminal one disposes it) and, either way, ABORTS // the remaining queue so the queued `PromoteToActive` - // never runs. + // never runs. The send never dials: if the msg1's + // connection has closed, a dial to its address (for an + // inbound connection, the initiator's ephemeral port) + // would hold the rx loop for up to the connect timeout. let send_err = match self.transports.get(&ambient.transport_id) { - Some(transport) => { - transport.send(&ambient.remote_addr, &frame).await.err() - } + Some(transport) => transport + .send_existing(&ambient.remote_addr, &frame) + .await + .err(), None => None, }; if let Some(e) = send_err { diff --git a/src/node/encrypt_worker.rs b/src/node/encrypt_worker.rs index c68019d3..e8f20451 100644 --- a/src/node/encrypt_worker.rs +++ b/src/node/encrypt_worker.rs @@ -55,6 +55,7 @@ use crate::node::worker_set::{TestWorker, test_spawner}; use crate::node::worker_set::{WorkerLiveness, WorkerSet, worth_logging}; use crate::proto::fmp::wire::ESTABLISHED_HEADER_SIZE; use crate::proto::fsp::wire::FSP_HEADER_SIZE; +use crate::transport::udp::UdpStats; use crate::transport::udp::io::AsyncUdpSocket; #[cfg(not(target_os = "macos"))] use crossbeam_channel::{Receiver, SendError, Sender, TrySendError, bounded}; @@ -140,6 +141,12 @@ pub(crate) struct FmpSendJob { /// strong ref remains. #[cfg(any(target_os = "linux", target_os = "macos"))] pub connected_socket: Option>, + /// The sending UDP transport's counters. The worker's sends bypass + /// `UdpTransport::send_async`, so the worker counts each datagram it + /// hands the kernel, and each it gives up on, here; the transport's + /// send counters then cover every datagram it sends, on the wildcard + /// socket and on per-peer connected sockets alike. + pub stats: Arc, /// Bulk endpoint data may be dropped when the kernel reports UDP /// send-queue exhaustion. Control/rekey frames keep retrying so /// congestion cannot strand the session. @@ -719,6 +726,7 @@ impl MacSequencedSendFlows { job.socket.clone(), job.connected_socket.clone(), job.dest_addr, + job.stats.clone(), now_ms, ); flows.insert(key, Arc::clone(&flow)); @@ -842,6 +850,9 @@ struct MacSequencedSendFlow { socket: AsyncUdpSocket, connected_socket: Option>, dest_addr: SocketAddr, + /// The sending transport's counters, which this flow's sender thread + /// counts each datagram into. + stats: Arc, next_seq: portable_atomic::AtomicU64, last_used_ms: portable_atomic::AtomicU64, state: Mutex, @@ -950,6 +961,7 @@ impl MacSequencedSendFlow { socket: AsyncUdpSocket, connected_socket: Option>, dest_addr: SocketAddr, + stats: Arc, now_ms: u64, ) -> Arc { let flow = Arc::new(Self { @@ -957,6 +969,7 @@ impl MacSequencedSendFlow { socket, connected_socket, dest_addr, + stats, next_seq: portable_atomic::AtomicU64::new(0), last_used_ms: portable_atomic::AtomicU64::new(now_ms), state: Mutex::new(MacSendFlowState::default()), @@ -1106,6 +1119,7 @@ impl MacSequencedSendFlow { &packet, &mut backpressure, drop_on_backpressure, + &self.stats, ) { debug!( socket_fd = self.key.socket_fd, @@ -1329,6 +1343,7 @@ fn flush_batch_sync( #[cfg(any(target_os = "linux", target_os = "macos"))] connected_socket: Option>, dest_addr: SocketAddr, + stats: Arc, wire_packets: Vec>, drop_on_backpressure: bool, } @@ -1352,6 +1367,7 @@ fn flush_batch_sync( dest_addr, #[cfg(any(target_os = "linux", target_os = "macos"))] connected_socket, + stats, drop_on_backpressure, queued_at, } = job; @@ -1394,7 +1410,7 @@ fn flush_batch_sync( if g.dest_addr != dest_addr { return false; } - if g.socket.as_raw_fd() != socket_fd { + if g.socket.as_raw_fd() != socket_fd || !Arc::ptr_eq(&g.stats, &stats) { return false; } #[cfg(any(target_os = "linux", target_os = "macos"))] @@ -1414,6 +1430,7 @@ fn flush_batch_sync( #[cfg(any(target_os = "linux", target_os = "macos"))] connected_socket, dest_addr, + stats, wire_packets: vec![wire_buf], drop_on_backpressure, }); @@ -1424,7 +1441,7 @@ fn flush_batch_sync( // Windows: encrypt worker pool isn't spawned (see // lifecycle.rs); this function is unreachable. Drop // values explicitly so the compiler sees them as used. - let _ = (socket, dest_addr, wire_buf); + let _ = (socket, dest_addr, stats, wire_buf); } } @@ -1458,13 +1475,27 @@ fn flush_batch_sync( // full so this is the cold path. let _t2 = crate::perf_profile::Timer::start(crate::perf_profile::Stage::UdpSend); + // Every datagram is counted once in its transport's stats: as sent + // when the kernel takes it, as a send error when it is given up on. + // A hard error ends the flush, so the groups after the failing one + // are never tried; `abandon` counts their datagrams as send errors. + #[cfg(unix)] + let mut groups = groups.into_iter(); + #[cfg(unix)] + let abandon = |rest: std::vec::IntoIter| { + for group in rest { + group.stats.record_unsent(group.wire_packets.len() as u64); + } + }; + #[cfg(target_os = "linux")] - for group in groups { + while let Some(group) = groups.next() { let mut backpressure = SendBackpressurePacer::default(); let EncryptedGroup { socket, connected_socket, dest_addr, + stats, wire_packets, drop_on_backpressure: _, } = group; @@ -1481,6 +1512,7 @@ fn flush_batch_sync( match send_batch_gso(fd, &wire_packets, dest_addr, connected) { Ok(()) => { record_udp_send_path(connected, wire_packets.len() as u64); + stats.record_sends(wire_packets.len() as u64, wire_bytes(&wire_packets)); continue; } Err(err) @@ -1500,6 +1532,8 @@ fn flush_batch_sync( // sendmmsg retry loop. No GSO_DISABLED toggle. } Err(err) => { + stats.record_unsent(wire_packets.len() as u64); + abandon(groups); return Err(format!("sendmsg+UDP_GSO failed: {err}").into()); } } @@ -1514,19 +1548,23 @@ fn flush_batch_sync( continue; } Err(err) => { + stats.record_unsent((wire_packets.len() - sent) as u64); + abandon(groups); return Err(format!("sendmmsg(2) failed: {err}").into()); } }; if n == 0 { + stats.record_unsent((wire_packets.len() - sent) as u64); break; } + stats.record_sends(n as u64, wire_bytes(&wire_packets[sent..sent + n])); sent += n; backpressure.record_success(); record_udp_send_path(connected, n as u64); } } #[cfg(all(unix, not(target_os = "linux")))] - for group in groups { + while let Some(group) = groups.next() { let mut backpressure = SendBackpressurePacer::default(); #[cfg(target_os = "macos")] let (fd, connected) = match group.connected_socket.as_ref() { @@ -1535,7 +1573,7 @@ fn flush_batch_sync( }; #[cfg(not(target_os = "macos"))] let (fd, connected) = (group.socket.as_raw_fd(), false); - for data in &group.wire_packets { + for (i, data) in group.wire_packets.iter().enumerate() { if let Err(err) = send_one_with_backpressure( fd, connected, @@ -1543,10 +1581,14 @@ fn flush_batch_sync( data, &mut backpressure, group.drop_on_backpressure, + &group.stats, ) { if group.drop_on_backpressure && is_send_backpressure(&err) { continue; } + let rest = group.wire_packets.len() - i - 1; + group.stats.record_unsent(rest as u64); + abandon(groups); return Err(format!("sendto failed: {err}").into()); } } @@ -1575,6 +1617,13 @@ fn record_udp_send_path(connected: bool, count: u64) { crate::perf_profile::record_event_count(event, count); } +/// Total bytes in `packets`: the UDP payload bytes one batched send +/// hands the kernel when it takes every packet in the slice. +#[cfg(target_os = "linux")] +fn wire_bytes(packets: &[Vec]) -> u64 { + packets.iter().map(|p| p.len() as u64).sum() +} + fn is_send_backpressure(err: &std::io::Error) -> bool { err.kind() == std::io::ErrorKind::WouldBlock || err.raw_os_error().is_some_and(raw_send_backpressure_code) @@ -1822,6 +1871,12 @@ impl MacSendRatePacer { #[cfg(target_os = "linux")] static GSO_DISABLED: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false); +/// The most packets one UDP GSO send carries, the kernel's segment limit +/// on older kernels. `send_batch_gso` reports a whole group as sent, so a +/// larger group must not reach it. +#[cfg(target_os = "linux")] +const GSO_SEGMENTS: usize = 64; + /// Size-only GSO eligibility check. Callers MUST ensure all packets /// share one destination + send target — `flush_batch_sync` does this /// by grouping. A batch is GSO-eligible iff every packet is the same @@ -1835,6 +1890,10 @@ fn gso_eligible_sizes(packets: &[Vec]) -> bool { // saving) and just add cmsg overhead. return false; } + if packets.len() > GSO_SEGMENTS { + // More than one GSO send can carry; sendmmsg's loop sends them all. + return false; + } let seg = packets[0].len(); if seg == 0 { return false; @@ -1864,8 +1923,8 @@ fn send_batch_gso( connected: bool, ) -> std::io::Result<()> { debug_assert!(!packets.is_empty()); - const MAX_BATCH: usize = 64; - let n = packets.len().min(MAX_BATCH); + debug_assert!(packets.len() <= GSO_SEGMENTS); + let n = packets.len().min(GSO_SEGMENTS); if n == 0 { return Ok(()); } @@ -1874,7 +1933,7 @@ fn send_batch_gso( let sa: socket2::SockAddr = dest.into(); // Stack-allocated arrays sized for the worst case in this batch. - let mut iovs: [libc::iovec; MAX_BATCH] = unsafe { std::mem::zeroed() }; + let mut iovs: [libc::iovec; GSO_SEGMENTS] = unsafe { std::mem::zeroed() }; for (i, data) in packets[..n].iter().enumerate() { iovs[i].iov_base = data.as_ptr() as *mut libc::c_void; iovs[i].iov_len = data.len(); @@ -2082,6 +2141,7 @@ mod unix_tests { dest_addr: recv_addr, #[cfg(any(target_os = "linux", target_os = "macos"))] connected_socket: None, + stats: Arc::new(UdpStats::new()), drop_on_backpressure: true, queued_at: None, }]; @@ -2215,6 +2275,7 @@ mod unix_tests { dest_addr: recv_addr, #[cfg(any(target_os = "linux", target_os = "macos"))] connected_socket: None, + stats: Arc::new(UdpStats::new()), drop_on_backpressure: true, queued_at: None, }]; @@ -2345,6 +2406,7 @@ mod unix_tests { dest_addr: "127.0.0.1:9".parse().unwrap(), #[cfg(any(target_os = "linux", target_os = "macos"))] connected_socket: None, + stats: Arc::new(UdpStats::new()), drop_on_backpressure: true, queued_at: None, }; @@ -2390,6 +2452,7 @@ mod unix_tests { dest_addr: "127.0.0.1:9".parse().unwrap(), #[cfg(any(target_os = "linux", target_os = "macos"))] connected_socket: None, + stats: Arc::new(UdpStats::new()), drop_on_backpressure: false, queued_at: None, }; @@ -2427,6 +2490,254 @@ mod unix_tests { Err(SealError::Layout) )); } + + /// A job carrying `plaintext_len` bytes of plaintext for `dest`, sent + /// on `socket` and counted in `stats`. Its wire packet is + /// `ESTABLISHED_HEADER_SIZE + plaintext_len + TAG_SIZE` bytes. + fn counted_job( + socket: &AsyncUdpSocket, + dest: SocketAddr, + plaintext_len: usize, + counter: u64, + stats: &Arc, + ) -> FmpSendJob { + let mut wire_buf = + Vec::with_capacity(ESTABLISHED_HEADER_SIZE + plaintext_len + crate::noise::TAG_SIZE); + wire_buf.extend_from_slice(&[0xA5; ESTABLISHED_HEADER_SIZE]); + wire_buf.resize(ESTABLISHED_HEADER_SIZE + plaintext_len, 0); + FmpSendJob { + cipher: test_cipher(3), + counter, + wire_buf, + fsp_seal: None, + socket: socket.clone(), + dest_addr: dest, + #[cfg(any(target_os = "linux", target_os = "macos"))] + connected_socket: None, + stats: stats.clone(), + drop_on_backpressure: false, + queued_at: None, + } + } + + /// The wire length of a `counted_job` carrying `plaintext_len` bytes. + fn wire_len(plaintext_len: usize) -> u64 { + (ESTABLISHED_HEADER_SIZE + plaintext_len + crate::noise::TAG_SIZE) as u64 + } + + /// Receive datagrams on `sock` until it goes quiet, returning how many + /// arrived. + fn count_received(sock: &UdpSocket) -> u64 { + sock.set_read_timeout(Some(std::time::Duration::from_millis(200))) + .expect("set_read_timeout"); + let mut buf = [0u8; 2048]; + let mut n = 0; + while sock.recv_from(&mut buf).is_ok() { + n += 1; + } + n + } + + /// A nonblocking loopback send socket registered with `rt`'s reactor. + fn open_async(rt: &tokio::runtime::Runtime) -> AsyncUdpSocket { + let _enter = rt.enter(); + UdpRawSocket::open("127.0.0.1:0".parse().unwrap(), 1 << 20, 1 << 20) + .expect("open send socket") + .into_async() + .expect("into_async") + } + + /// The worker's sends bypass `UdpTransport::send_async`, so the worker + /// must count them itself: each datagram it hands the kernel counts + /// once, in the stats of the transport whose job it was. The batch + /// mixes a same-size run (the UDP GSO group on Linux), a lone packet + /// to a second destination (plain `sendmmsg`), and a job from a second + /// transport, whose datagram must land in that transport's stats only. + #[test] + fn each_datagram_a_flush_sends_counts_once_in_its_own_transports_stats() { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_io() + .build() + .expect("tokio rt"); + let recv_a = UdpSocket::bind("127.0.0.1:0").expect("bind recv_a"); + let recv_b = UdpSocket::bind("127.0.0.1:0").expect("bind recv_b"); + let addr_a = recv_a.local_addr().unwrap(); + let addr_b = recv_b.local_addr().unwrap(); + let first = open_async(&rt); + let second = open_async(&rt); + let stats = Arc::new(UdpStats::new()); + let other = Arc::new(UdpStats::new()); + + const RUN: u64 = 6; + const RUN_LEN: usize = 100; + const LONE_LEN: usize = 40; + const OTHER_LEN: usize = 60; + let mut batch: Vec = (0..RUN) + .map(|i| counted_job(&first, addr_a, RUN_LEN, i, &stats)) + .collect(); + batch.push(counted_job(&first, addr_b, LONE_LEN, RUN, &stats)); + batch.push(counted_job(&second, addr_b, OTHER_LEN, RUN + 1, &other)); + flush_direct_batch_sync(&mut batch).expect("flush ok"); + + assert_eq!(count_received(&recv_a), RUN); + assert_eq!(count_received(&recv_b), 2); + let counted = stats.snapshot(); + assert_eq!( + counted.packets_sent, + RUN + 1, + "each datagram the worker sent must count once" + ); + assert_eq!( + counted.bytes_sent, + RUN * wire_len(RUN_LEN) + wire_len(LONE_LEN) + ); + assert_eq!(counted.send_errors, 0); + let counted = other.snapshot(); + assert_eq!( + counted.packets_sent, 1, + "a datagram must count in the stats of the transport whose job it was" + ); + assert_eq!(counted.bytes_sent, wire_len(OTHER_LEN)); + } + + /// Datagrams sent on a per-peer connected socket count in the stats + /// of the transport the job came from, as wildcard sends do. + #[cfg(any(target_os = "linux", target_os = "macos"))] + #[test] + fn datagrams_sent_on_a_connected_socket_count_once_in_the_transports_stats() { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_io() + .build() + .expect("tokio rt"); + let recv = UdpSocket::bind("127.0.0.1:0").expect("bind recv"); + let peer = recv.local_addr().unwrap(); + let local: SocketAddr = "127.0.0.1:0".parse().unwrap(); + let owned = crate::transport::udp::open_connected_fd(local, peer, 1 << 16, 1 << 16) + .expect("open a connected UDP socket"); + let connected = Arc::new(crate::transport::udp::ConnectedPeerSocket::from_fd( + owned, peer, local, + )); + let wildcard = open_async(&rt); + let stats = Arc::new(UdpStats::new()); + + const N: u64 = 4; + const LEN: usize = 80; + let mut batch: Vec = (0..N) + .map(|i| { + let mut job = counted_job(&wildcard, peer, LEN, i, &stats); + job.connected_socket = Some(connected.clone()); + job + }) + .collect(); + flush_direct_batch_sync(&mut batch).expect("flush ok"); + + assert_eq!(count_received(&recv), N); + let counted = stats.snapshot(); + assert_eq!(counted.packets_sent, N); + assert_eq!(counted.bytes_sent, N * wire_len(LEN)); + assert_eq!(counted.send_errors, 0); + } + + /// A hard send error ends the flush. The failing datagram and every + /// datagram the flush then never tries count as send errors, none as + /// sent. The failing destination is an IPv6 address on an IPv4 + /// socket, which the kernel refuses outright; it is a single packet so + /// the Linux GSO path, whose refusal would switch GSO off for the + /// whole process, is not taken. + #[test] + fn datagrams_a_flush_abandons_on_a_send_error_count_as_send_errors() { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_io() + .build() + .expect("tokio rt"); + let recv = UdpSocket::bind("127.0.0.1:0").expect("bind recv"); + let good = recv.local_addr().unwrap(); + let bad: SocketAddr = "[::1]:9".parse().unwrap(); + let socket = open_async(&rt); + let stats = Arc::new(UdpStats::new()); + + let mut batch = vec![ + counted_job(&socket, bad, 40, 0, &stats), + counted_job(&socket, good, 50, 1, &stats), + counted_job(&socket, good, 70, 2, &stats), + ]; + assert!( + flush_direct_batch_sync(&mut batch).is_err(), + "precondition: the kernel refuses the IPv6 destination" + ); + + assert_eq!(count_received(&recv), 0); + let counted = stats.snapshot(); + assert_eq!(counted.packets_sent, 0); + assert_eq!(counted.bytes_sent, 0); + assert_eq!( + counted.send_errors, 3, + "the refused datagram and the two never tried must each count once" + ); + } + + /// A hard error on the Linux UDP GSO path ends the flush as one on + /// `sendmmsg` does: every datagram of the failing group and of the + /// groups after it counts as a send error, none as sent. The error is + /// the ECONNREFUSED a connected socket reports after an ICMP port + /// unreachable; unlike EINVAL, it does not switch GSO off for the + /// whole process. On a kernel without UDP GSO the group takes + /// `sendmmsg` and the counts must hold the same. + #[cfg(target_os = "linux")] + #[test] + fn datagrams_a_gso_send_error_abandons_count_as_send_errors() { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_io() + .build() + .expect("tokio rt"); + let closed = UdpSocket::bind("127.0.0.1:0").expect("bind closed"); + let peer = closed.local_addr().unwrap(); + drop(closed); + let recv = UdpSocket::bind("127.0.0.1:0").expect("bind recv"); + let good = recv.local_addr().unwrap(); + let local: SocketAddr = "127.0.0.1:0".parse().unwrap(); + let owned = crate::transport::udp::open_connected_fd(local, peer, 1 << 16, 1 << 16) + .expect("open a connected UDP socket"); + let connected = Arc::new(crate::transport::udp::ConnectedPeerSocket::from_fd( + owned, peer, local, + )); + let wildcard = open_async(&rt); + let to_peer = |len: usize, counter: u64, stats: &Arc| { + let mut job = counted_job(&wildcard, peer, len, counter, stats); + job.connected_socket = Some(connected.clone()); + job + }; + + // One datagram to the closed port draws the ICMP port unreachable + // that the socket's next send reports as ECONNREFUSED. + let mut prime = vec![to_peer(40, 0, &Arc::new(UdpStats::new()))]; + flush_direct_batch_sync(&mut prime).expect("priming send ok"); + std::thread::sleep(std::time::Duration::from_millis(50)); + + const RUN: u64 = 4; + const LEN: usize = 100; + let stats = Arc::new(UdpStats::new()); + let mut batch: Vec = (1..=RUN).map(|i| to_peer(LEN, i, &stats)).collect(); + batch.push(counted_job(&wildcard, good, 40, RUN + 1, &stats)); + let err = flush_direct_batch_sync(&mut batch) + .expect_err("precondition: the connected socket reports the refused port"); + if !GSO_DISABLED.load(std::sync::atomic::Ordering::Relaxed) { + assert!( + err.to_string().contains("UDP_GSO"), + "precondition: the same-size group must fail on the GSO path, got: {err}" + ); + } + + assert_eq!(count_received(&recv), 0); + let counted = stats.snapshot(); + assert_eq!(counted.packets_sent, 0); + assert_eq!(counted.bytes_sent, 0); + assert_eq!( + counted.send_errors, + RUN + 1, + "the failing group's datagrams and the one never tried must each count once" + ); + } } /// Standalone tests for the GSO-eligibility predicate. The full @@ -2460,6 +2771,14 @@ mod tests { assert!(gso_eligible_sizes(&batch)); } + #[test] + fn gso_eligible_rejects_a_group_larger_than_one_gso_send_carries() { + let at_cap: Vec<_> = (0..GSO_SEGMENTS).map(|_| pkt(1500)).collect(); + assert!(gso_eligible_sizes(&at_cap)); + let over: Vec<_> = (0..=GSO_SEGMENTS).map(|_| pkt(1500)).collect(); + assert!(!gso_eligible_sizes(&over)); + } + #[test] fn gso_eligible_rejects_mixed_sizes() { let mut batch: Vec<_> = (0..18).map(|_| pkt(1500)).collect(); @@ -2665,6 +2984,7 @@ mod tests { dest_addr: dest, #[cfg(any(target_os = "linux", target_os = "macos"))] connected_socket: None, + stats: Arc::new(UdpStats::new()), drop_on_backpressure: true, queued_at: None, } @@ -2732,6 +3052,9 @@ fn send_connected_raw(fd: std::os::unix::io::RawFd, data: &[u8]) -> std::io::Res } } +/// Send one datagram, retrying while the kernel reports backpressure, and +/// count it once in `stats`: as sent, or as a send error when it is +/// dropped under backpressure or the send fails. #[cfg(all(unix, not(target_os = "linux")))] fn send_one_with_backpressure( fd: std::os::unix::io::RawFd, @@ -2740,6 +3063,7 @@ fn send_one_with_backpressure( data: &[u8], backpressure: &mut SendBackpressurePacer, drop_on_backpressure: bool, + stats: &UdpStats, ) -> std::io::Result<()> { loop { let result = if connected { @@ -2748,18 +3072,23 @@ fn send_one_with_backpressure( send_one_raw(fd, data, dest) }; match result { - Ok(_) => { + Ok(bytes) => { backpressure.record_success(); record_udp_send_path(connected, 1); + stats.record_send(bytes); return Ok(()); } Err(err) if is_send_backpressure(&err) => { if backpressure.pause(&err) && drop_on_backpressure { record_udp_send_backpressure_drop(&err); + stats.record_send_error(); return Err(err); } } - Err(err) => return Err(err), + Err(err) => { + stats.record_send_error(); + return Err(err); + } } } } @@ -2976,6 +3305,7 @@ mod mac_ordered_tests { socket: socket.clone(), dest_addr: dest, connected_socket: None, + stats: Arc::new(UdpStats::new()), drop_on_backpressure: false, queued_at: None, } @@ -3232,6 +3562,7 @@ mod pool_tests { dest_addr: dest, #[cfg(any(target_os = "linux", target_os = "macos"))] connected_socket: None, + stats: Arc::new(UdpStats::new()), drop_on_backpressure: false, queued_at: None, } diff --git a/src/node/handlers/netmon.rs b/src/node/handlers/netmon.rs index fb77735d..8e658c10 100644 --- a/src/node/handlers/netmon.rs +++ b/src/node/handlers/netmon.rs @@ -167,11 +167,12 @@ impl Node { /// /// So a peer on TCP, Tor, Nym or BLE keeps the periodic heartbeat it had /// before this detector existed, and `link_dead_timeout_secs` remains the - /// backstop. Note that it does *not* recover by redialling: `send_async` - /// only dials when the pool holds no connection for the address, and a - /// connection stranded by a medium change is still in the pool. It is - /// evicted after a write to it fails, so the redial happens on the send - /// after the failure, not on the first one. Doing better for them means + /// backstop. Note that it does *not* recover by redialling. The link send + /// never dials, and a connection stranded by a medium change is still in + /// the pool until its writer's write to it fails. The send after that + /// finds no connection and fails, starting a background connect only if + /// this node dialed the peer at that address, and a later send uses it; + /// a peer that dialed in is never redialled. Doing better for them means /// dropping the stale connection /// rather than writing into it, which is a different change with a real /// cost behind it — a Tor peer pays a fresh circuit — and is not this one. diff --git a/src/node/handlers/rekey.rs b/src/node/handlers/rekey.rs index 92c31478..b8cba908 100644 --- a/src/node/handlers/rekey.rs +++ b/src/node/handlers/rekey.rs @@ -415,9 +415,15 @@ impl Node { let wire_msg1 = build_msg1(our_index, &noise_msg1); - // Send msg1 on the existing link (same transport + address) + // Send msg1 on the existing link (same transport + address). This + // runs on the tick, so it never dials: with the connection gone it + // fails at once, the index is freed, and the next tick's trigger + // tries again, by then over any connection the failed send started. if let Some(transport) = self.transports.get(&transport_id) { - match transport.send(&remote_addr, &wire_msg1).await { + match self + .send_nowait(transport, link_id, &remote_addr, &wire_msg1) + .await + { Ok(_) => { debug!( peer = %self.peer_display_name(node_addr), @@ -532,16 +538,21 @@ impl Node { bytes, next_resend_at_ms, } => { - let (transport_id, remote_addr) = match self.peers.get(&node_addr) { + let (link_id, transport_id, remote_addr) = match self.peers.get(&node_addr) { Some(p) => match (p.transport_id(), p.current_addr()) { - (Some(tid), Some(addr)) => (tid, addr.clone()), + (Some(tid), Some(addr)) => (p.link_id(), tid, addr.clone()), _ => continue, }, None => continue, }; + // A failed send records no resend, so the msg1 stays due + // and is retried next tick, over any connection the + // failed send started. let sent = if let Some(transport) = self.transports.get(&transport_id) { - transport.send(&remote_addr, &bytes).await.is_ok() + self.send_nowait(transport, link_id, &remote_addr, &bytes) + .await + .is_ok() } else { false }; diff --git a/src/node/handlers/session.rs b/src/node/handlers/session.rs index f20372a0..cff3e0e4 100644 --- a/src/node/handlers/session.rs +++ b/src/node/handlers/session.rs @@ -2959,6 +2959,7 @@ impl Node { let Some(socket) = udp.async_socket() else { return Ok(false); }; + let stats = udp.stats().clone(); // FSP cipher + counter — separate session from next-hop FMP session. let (fsp_counter, fsp_cipher) = { @@ -3099,6 +3100,7 @@ impl Node { dest_addr: socket_addr, #[cfg(any(target_os = "linux", target_os = "macos"))] connected_socket, + stats, // Bulk endpoint data: drop on UDP backpressure so the // worker queue keeps moving instead of stranding under // sustained congestion. @@ -3138,7 +3140,7 @@ impl Node { debug!(next_hop = %next_hop_addr, "Transport gone before inline send of session data"); return; }; - if let Err(error) = transport.send(remote_addr, &wire).await { + if let Err(error) = transport.send_existing(remote_addr, &wire).await { debug!(next_hop = %next_hop_addr, %error, "Inline send of session data failed"); } } diff --git a/src/node/handlers/timeout.rs b/src/node/handlers/timeout.rs index ba3237cf..96807cd2 100644 --- a/src/node/handlers/timeout.rs +++ b/src/node/handlers/timeout.rs @@ -373,7 +373,10 @@ impl Node { }; let sent = if let Some(transport) = self.transports.get(&transport_id) { - match transport.send(&remote_addr, &bytes).await { + match self + .send_nowait(transport, link, &remote_addr, &bytes) + .await + { Ok(_) => true, Err(e) => { debug!( diff --git a/src/node/lifecycle/mod.rs b/src/node/lifecycle/mod.rs index f3fb2af3..ab248ecd 100644 --- a/src/node/lifecycle/mod.rs +++ b/src/node/lifecycle/mod.rs @@ -943,9 +943,15 @@ impl Node { .filter(|machine| machine.leg().is_some()) .and_then(|machine| machine.our_index()); - // Send the wire format handshake message + // Send the wire format handshake message. It never dials: if the + // connection the dial resolved to has gone, the send fails at once + // into the failure path below, after starting a background connect + // to the dial address that the next attempt picks up. if let Some(transport) = self.transports.get(&transport_id) { - match transport.send(remote_addr, &wire_msg1).await { + match self + .send_nowait(transport, link_id, remote_addr, &wire_msg1) + .await + { Ok(bytes) => { if let Some(idx) = our_index { debug!( diff --git a/src/node/mod.rs b/src/node/mod.rs index 4a29c997..1d277561 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -85,7 +85,7 @@ use crate::transport::tcp::TcpTransport; use crate::transport::tor::TorTransport; use crate::transport::udp::UdpTransport; use crate::transport::{ - ConnectionState, Link, LinkId, PacketRx, PacketTx, TransportAddr, TransportError, + Link, LinkDirection, LinkId, PacketRx, PacketTx, TransportAddr, TransportError, TransportHandle, TransportId, }; use crate::upper::hosts::HostMap; @@ -2591,6 +2591,8 @@ impl Node { // Ascending id, matching `show_transports`; see the note there. The // off-loop renderer reads this table verbatim, so the two paths would // otherwise disagree about ordering as well as being arbitrary. + // `stats` shares the transport's counters rather than copying them, so + // the query reads them live even if the next tick is held. let mut transport_ids: Vec<_> = self.transport_ids().copied().collect(); transport_ids.sort_by_key(|id| id.as_u32()); let transport_rows: Vec = transport_ids @@ -2609,7 +2611,7 @@ impl Node { tor_monitoring: handle .tor_monitoring() .map(|m| serde_json::to_value(&m).unwrap_or_default()), - stats: handle.transport_stats(), + stats: handle.live_stats(), interface: handle.interface_presence().map(|p| snap::InterfaceRow { name: handle.interface_name().unwrap_or_default().to_string(), presence: p.presence, @@ -3965,6 +3967,7 @@ impl Node { node_addr: *node_addr, reason: "no current_addr".into(), })?; + let link_id = peer.link_id(); // Prepend 4-byte session-relative timestamp (inner header) let timestamp_ms = peer.session_elapsed_ms(); @@ -4056,6 +4059,7 @@ impl Node { dest_addr: dest_socket_addr, #[cfg(any(target_os = "linux", target_os = "macos"))] connected_socket, + stats: udp.stats().clone(), drop_on_backpressure, queued_at: None, }); @@ -4070,7 +4074,7 @@ impl Node { reason: format!("encryption failed: {}", e), })?; transport - .send(&remote_addr, &wire) + .send_existing(&remote_addr, &wire) .await .map_err(|e| link_send_error(*node_addr, e))? } @@ -4113,29 +4117,15 @@ impl Node { .get(&transport_id) .ok_or(NodeError::TransportNotFound(transport_id))?; - // Gate: don't drive connect-on-send from the tick path. If the - // transport's connection isn't ready, kick off a non-blocking - // background connect (no-op if already in flight or pooled) and - // fail this send fast. A subsequent tick will retry once the - // pool entry exists. The historical connect-on-send wedged the - // rx_loop tick body for up to `connect_timeout_ms` (5 s default) - // per unreachable peer, which under convergence-phase mesh - // pressure cascaded into multi-tick stalls and control-RPC HOL. - match transport.connection_state(&remote_addr) { - ConnectionState::Connected => {} - other => { - if matches!(other, ConnectionState::None) { - let _ = transport.connect(&remote_addr).await; - } - return Err(NodeError::SendFailed { - node_addr: *node_addr, - reason: format!("transport connection not ready: {:?}", other), - }); - } - } - - let bytes_sent = transport - .send(&remote_addr, &wire_packet) + // Never connect-on-send here: this runs on the tick and forwarding + // paths, and a dial held the rx loop for up to `connect_timeout_ms` + // (5 s default) per unreachable peer, which under convergence-phase + // mesh pressure cascaded into multi-tick stalls and control-RPC + // head-of-line blocking. With no connection the send fails fast, + // after starting a background connect if the address is one this + // node dialed; a later send uses the connection once it is up. + let bytes_sent = self + .send_nowait(transport, link_id, &remote_addr, &wire_packet) .await .map_err(|e| link_send_error(*node_addr, e))?; @@ -4208,6 +4198,59 @@ impl Node { }, }) } + + /// Send `data` to `addr` over a connection the transport already holds, + /// without ever waiting on a dial. + /// + /// When there is no connection the send fails at once with + /// [`TransportError::NotConnected`]. If `addr` is the address this node + /// dialed for `link_id` (see [`Self::may_dial`]), a background connect + /// is started first; it returns at once, and is a no-op while one is in + /// flight. A later send takes the finished connection into the pool and + /// uses it. + /// + /// This is the send for the tick, dial and forwarding paths, which the + /// rx loop awaits. Replies to a msg1 use + /// [`TransportHandle::send_existing`] directly: their address is the + /// sender's, and nothing should be dialed there. + pub(in crate::node) async fn send_nowait( + &self, + transport: &TransportHandle, + link_id: LinkId, + addr: &TransportAddr, + data: &[u8], + ) -> Result { + let sent = transport.send_existing(addr, data).await; + if matches!(sent, Err(TransportError::NotConnected)) + && self.may_dial(link_id, transport.transport_id(), addr) + && let Err(e) = transport.connect(addr).await + { + tracing::debug!( + link_id = %link_id, + remote_addr = %addr, + error = %e, + "Background connect not started" + ); + } + sent + } + + /// Whether a send for `link_id` may start a connect to `addr` on + /// `transport_id`: only when the link is outbound and `addr` is the + /// address it was dialed at. + /// + /// Direction alone is not enough. An inbound link's address is the + /// remote's source port, where nothing listens, and an outbound link's + /// current address follows the peer's authenticated frames, so it can + /// move to such a port too. The dial address recorded on the link is + /// the one address known to have a listener. + fn may_dial(&self, link_id: LinkId, transport_id: TransportId, addr: &TransportAddr) -> bool { + self.links.get(&link_id).is_some_and(|link| { + link.direction() == LinkDirection::Outbound + && link.transport_id() == transport_id + && link.remote_addr() == addr + }) + } } /// Map a transport's refusal of an encrypted link frame to `node_addr` onto diff --git a/src/node/tests/connected_udp.rs b/src/node/tests/connected_udp.rs index 6919775a..fd2e3107 100644 --- a/src/node/tests/connected_udp.rs +++ b/src/node/tests/connected_udp.rs @@ -14,6 +14,10 @@ //! `bool` carries no `#[must_use]`, so discarding the return here is //! silent under `-D warnings`; the assertions below are what makes the //! difference between binding it and dropping it observable. +//! +//! The tests at the end check that datagrams the node receives through a +//! connected socket's drain, and sends through the encrypt workers on +//! either kind of socket, are counted once in the UDP transport's stats. use super::*; use crate::noise::NoiseSession; @@ -201,3 +205,236 @@ async fn a_frame_from_the_address_the_peer_is_already_on_keeps_the_connected_soc "a frame from the address already in use must leave the socket alone" ); } + +/// Datagrams a peer sends after its connected socket is installed reach +/// the node through that socket's drain thread, not the wildcard listen +/// socket, and must still be counted in the UDP transport's own stats: +/// `packets_recv` means datagrams received on this transport. +/// +/// The runtime thread is blocked while the count is read, so the +/// wildcard socket's receive task (a task on this runtime) cannot be the +/// one counting; only the drain thread can. +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[tokio::test] +async fn datagrams_on_a_peers_connected_socket_are_counted_in_the_udp_transport_stats() { + use crate::config::UdpConfig; + use crate::transport::udp::UdpTransport; + + const SENT: u64 = 5; + let transport_id = TransportId::new(1); + let (mut node, node_addr, _, _) = promoted_peer_with_the_far_side_session(transport_id); + let (tx, mut rx) = packet_channel(64); + let udp_cfg = UdpConfig { + bind_addr: Some("127.0.0.1:0".to_string()), + ..Default::default() + }; + let mut udp = UdpTransport::new(transport_id, None, udp_cfg, tx); + udp.start_async().await.unwrap(); + let local = udp.local_addr().unwrap(); + let stats = udp.stats().clone(); + node.transports + .insert(transport_id, TransportHandle::Udp(udp)); + + let remote = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + node.get_peer_mut(&node_addr).unwrap().set_current_addr( + transport_id, + TransportAddr::from_string(&remote.local_addr().unwrap().to_string()), + ); + node.activate_connected_udp_sessions().await; + assert!( + node.get_peer(&node_addr).unwrap().connected_udp().is_some(), + "precondition: the tick activation installed a connected socket" + ); + + for i in 0..SENT { + remote.send_to(&[i as u8; 8], local).unwrap(); + } + // Block the runtime thread: the wildcard receive task cannot run. + let deadline = std::time::Instant::now() + Duration::from_secs(2); + while stats.snapshot().packets_recv < SENT && std::time::Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + let counted = stats.snapshot(); + assert_eq!( + counted.packets_recv, SENT, + "datagrams read by the connected socket's drain were not counted" + ); + assert_eq!(counted.bytes_recv, SENT * 8); + + for _ in 0..SENT { + let packet = tokio::time::timeout(Duration::from_secs(1), rx.recv()) + .await + .expect("a counted datagram was not delivered") + .expect("packet channel closed"); + assert_eq!(packet.transport_id, transport_id); + } + node.clear_connected_udp_for_peer(&node_addr); + for (_, t) in node.transports.iter_mut() { + t.stop().await.ok(); + } +} + +/// A promoted peer reached over a real UDP transport on loopback, with an +/// encrypt worker pool, so the node's sends to it take the worker path +/// rather than `UdpTransport::send_async`. Returns the node, the peer's +/// address, the UDP transport's stats, and the plain socket standing in +/// for the peer. +#[cfg(unix)] +async fn peer_behind_the_encrypt_workers() -> ( + Node, + NodeAddr, + std::sync::Arc, + std::net::UdpSocket, +) { + use crate::config::UdpConfig; + use crate::transport::udp::UdpTransport; + + let transport_id = TransportId::new(1); + let (mut node, node_addr, _, _) = promoted_peer_with_the_far_side_session(transport_id); + let (tx, _rx) = packet_channel(64); + let udp_cfg = UdpConfig { + bind_addr: Some("127.0.0.1:0".to_string()), + ..Default::default() + }; + let mut udp = UdpTransport::new(transport_id, None, udp_cfg, tx); + udp.start_async().await.unwrap(); + let stats = udp.stats().clone(); + node.transports + .insert(transport_id, TransportHandle::Udp(udp)); + node.supervisor.encrypt_workers = + Some(crate::node::encrypt_worker::EncryptWorkerPool::spawn(1)); + + let remote = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + remote + .set_read_timeout(Some(Duration::from_secs(2))) + .unwrap(); + node.get_peer_mut(&node_addr).unwrap().set_current_addr( + transport_id, + TransportAddr::from_string(&remote.local_addr().unwrap().to_string()), + ); + (node, node_addr, stats, remote) +} + +/// Wait for the encrypt worker to count `sent` datagrams, then a little +/// longer so a second count of any of them would show. The worker counts +/// after its send returns, so a datagram can reach the peer just before +/// its count lands. +#[cfg(unix)] +fn settle_sent(stats: &crate::transport::udp::UdpStats, sent: u64) { + let deadline = std::time::Instant::now() + Duration::from_secs(2); + while stats.snapshot().packets_sent < sent && std::time::Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(5)); + } + std::thread::sleep(Duration::from_millis(50)); +} + +/// A link message the encrypt worker sends counts once in the UDP +/// transport's own stats, whether it leaves on the wildcard socket or on +/// the peer's connected socket: `packets_sent` means datagrams sent on +/// this transport, and the worker's sends never pass through +/// `UdpTransport::send_async`, which counts the rest. +#[cfg(any(target_os = "linux", target_os = "macos"))] +#[tokio::test] +async fn a_link_message_the_encrypt_worker_sends_counts_once_in_the_udp_transport_stats() { + let (mut node, node_addr, stats, remote) = peer_behind_the_encrypt_workers().await; + let mut buf = [0u8; 2048]; + // One tick publishes the transport rows `show transports` serves off + // the rx loop; no tick runs after this. + node.record_stats_history(); + let handle = node.control_read_handle(); + + node.send_encrypted_link_message(&node_addr, &[0x51]) + .await + .expect("send on the wildcard socket"); + let (wildcard_len, _) = remote.recv_from(&mut buf).expect("the peer receives it"); + settle_sent(&stats, 1); + let counted = stats.snapshot(); + assert_eq!( + counted.packets_sent, 1, + "a datagram the worker sent on the wildcard socket must count once" + ); + assert_eq!(counted.bytes_sent, wildcard_len as u64); + + node.activate_connected_udp_sessions().await; + assert!( + node.get_peer(&node_addr).unwrap().connected_udp().is_some(), + "precondition: the tick activation installed a connected socket" + ); + node.send_encrypted_link_message(&node_addr, &[0x51]) + .await + .expect("send on the connected socket"); + let (connected_len, _) = remote.recv_from(&mut buf).expect("the peer receives it"); + settle_sent(&stats, 2); + let counted = stats.snapshot(); + assert_eq!( + counted.packets_sent, 2, + "a datagram the worker sent on the connected socket must count once" + ); + assert_eq!(counted.bytes_sent, (wildcard_len + connected_len) as u64); + assert_eq!(counted.send_errors, 0); + + // `show transports`, on the rx loop and off it, reports the same counts. + let off_loop = crate::control::queries::show_transports_from_handle(&handle); + assert_eq!(off_loop, crate::control::queries::show_transports(&node)); + let row = off_loop["transports"] + .as_array() + .unwrap() + .iter() + .find(|t| t["transport_id"] == 1) + .unwrap_or_else(|| panic!("no row for the UDP transport: {off_loop}")) + .clone(); + assert_eq!( + row["stats"]["packets_sent"], 2, + "show transports must report the datagrams the worker sent" + ); + assert_eq!( + row["stats"]["bytes_sent"], + (wildcard_len + connected_len) as u64 + ); + + node.clear_connected_udp_for_peer(&node_addr); + for (_, t) in node.transports.iter_mut() { + t.stop().await.ok(); + } +} + +/// Session data takes the pipelined path, where the worker seals both +/// layers and sends; its datagram counts once in the UDP transport's +/// stats, as a link message's does. +#[cfg(unix)] +#[tokio::test] +async fn session_data_the_encrypt_worker_sends_counts_once_in_the_udp_transport_stats() { + use crate::node::session::{EndToEndState, SessionEntry}; + + let (mut node, node_addr, stats, remote) = peer_behind_the_encrypt_workers().await; + let far_side = Identity::generate(); + let session = super::session::make_noise_session(node.identity(), &far_side); + node.sessions.insert( + node_addr, + SessionEntry::new( + node_addr, + far_side.pubkey_full(), + EndToEndState::Established(session), + 1_000, + true, + ), + ); + + node.send_session_data(&node_addr, 0, 0, b"counted once") + .await + .expect("send session data"); + let mut buf = [0u8; 2048]; + let (len, _) = remote.recv_from(&mut buf).expect("the peer receives it"); + settle_sent(&stats, 1); + let counted = stats.snapshot(); + assert_eq!( + counted.packets_sent, 1, + "a session datagram the worker sent must count once" + ); + assert_eq!(counted.bytes_sent, len as u64); + assert_eq!(counted.send_errors, 0); + + for (_, t) in node.transports.iter_mut() { + t.stop().await.ok(); + } +} diff --git a/src/node/tests/control.rs b/src/node/tests/control.rs index dbb08b99..2cc1042f 100644 --- a/src/node/tests/control.rs +++ b/src/node/tests/control.rs @@ -529,3 +529,101 @@ fn peer_connectivity_turns_stale_one_millisecond_past_the_heartbeat_interval() { assert_eq!(at(&node, seen + 1_000), ConnectivityState::Connected); assert_eq!(at(&node, seen + 1_001), ConnectivityState::Stale); } + +/// `show_transports` served off the rx loop reports the transport counters +/// as they are now, not as the last tick published them: a counter that +/// moves after the publish shows up without another tick, in the same +/// fields and shape the on-loop render gives. +#[tokio::test] +async fn show_transports_off_loop_reports_counters_that_moved_since_the_last_tick() { + use crate::config::{TcpConfig, UdpConfig}; + use crate::control::queries::{show_transports, show_transports_from_handle}; + use crate::transport::tcp::TcpTransport; + use crate::transport::udp::UdpTransport; + + let mut node = make_node(); + let (tx, _rx) = packet_channel(16); + let udp_cfg = UdpConfig { + bind_addr: Some("127.0.0.1:0".to_string()), + ..Default::default() + }; + let mut udp = UdpTransport::new(TransportId::new(1), None, udp_cfg, tx.clone()); + udp.start_async().await.unwrap(); + let udp_stats = udp.stats().clone(); + let tcp_cfg = TcpConfig { + bind_addr: Some("127.0.0.1:0".to_string()), + ..Default::default() + }; + let mut tcp = TcpTransport::new(TransportId::new(2), None, tcp_cfg, tx); + tcp.start_async().await.unwrap(); + let tcp_stats = tcp.stats().clone(); + node.transports + .insert(TransportId::new(1), TransportHandle::Udp(udp)); + node.transports + .insert(TransportId::new(2), TransportHandle::Tcp(tcp)); + + // One tick publishes the transport rows. + node.record_stats_history(); + let handle = node.control_read_handle(); + let published = std::sync::Arc::clone(&*handle.entities()); + let row = |v: &serde_json::Value, id: u32| { + v["transports"] + .as_array() + .unwrap() + .iter() + .find(|t| t["transport_id"] == id) + .unwrap_or_else(|| panic!("no row for transport {id}: {v}")) + .clone() + }; + let before = show_transports_from_handle(&handle); + assert_eq!(row(&before, 1)["stats"]["packets_recv"], 0); + assert_eq!(row(&before, 2)["stats"]["connect_timeouts"], 0); + + // The counters move; no tick runs. + for _ in 0..3 { + udp_stats.record_recv(100); + } + tcp_stats.record_connect_timeout(); + tcp_stats.record_connect_refused(); + + let after = show_transports_from_handle(&handle); + assert!( + std::sync::Arc::ptr_eq(&published, &handle.entities()), + "precondition: nothing republished the snapshot" + ); + let udp_row = row(&after, 1); + let tcp_row = row(&after, 2); + assert_eq!( + udp_row["stats"]["packets_recv"], 3, + "off-loop show_transports did not see the UDP counter move" + ); + assert_eq!(udp_row["stats"]["bytes_recv"], 300); + assert_eq!( + tcp_row["stats"]["connect_timeouts"], 1, + "off-loop show_transports did not see the TCP counter move" + ); + assert_eq!(tcp_row["stats"]["connect_refused"], 1); + + // Same fields and shape as before: each block is its stats snapshot + // serialized whole, and the whole render matches the on-loop one. + assert_eq!( + udp_row["stats"], + serde_json::to_value(udp_stats.snapshot()).unwrap() + ); + assert_eq!( + tcp_row["stats"], + serde_json::to_value(tcp_stats.snapshot()).unwrap() + ); + assert_eq!(after, show_transports(&node)); + let keys = |v: &serde_json::Value| { + let mut k: Vec = v.as_object().unwrap().keys().cloned().collect(); + k.sort(); + k + }; + assert_eq!(keys(&udp_row), keys(&row(&before, 1))); + assert_eq!(keys(&tcp_row["stats"]), keys(&row(&before, 2)["stats"])); + + for (_, t) in node.transports.iter_mut() { + t.stop().await.ok(); + } +} diff --git a/src/node/tests/mod.rs b/src/node/tests/mod.rs index 06e18165..eacb0894 100644 --- a/src/node/tests/mod.rs +++ b/src/node/tests/mod.rs @@ -85,6 +85,7 @@ pub(super) fn install_connected_udp( transport_id, peer_sa, packet_tx, + std::sync::Arc::new(crate::transport::udp::UdpStats::new()), ) .expect("spawn the peer recv drain"); diff --git a/src/node/tests/session.rs b/src/node/tests/session.rs index ddb95e3e..1c636c8d 100644 --- a/src/node/tests/session.rs +++ b/src/node/tests/session.rs @@ -2244,7 +2244,7 @@ async fn test_tun_outbound_pending_queue_flush() { // ============================================================================ /// Helper: complete a Noise XX handshake and return the initiator's NoiseSession. -fn make_noise_session( +pub(super) fn make_noise_session( our_identity: &Identity, remote_identity: &Identity, ) -> crate::noise::NoiseSession { diff --git a/src/testutil.rs b/src/testutil.rs index f55e42e8..442890d7 100644 --- a/src/testutil.rs +++ b/src/testutil.rs @@ -78,3 +78,100 @@ pub(crate) fn capture_logs_scoped() -> (LogCapture, tracing::subscriber::Default let guard = tracing::subscriber::set_default(subscriber); (capture, guard) } + +/// Poll `f` every 10ms until it holds or `limit` elapses. +/// +/// Uses tokio's clock, so a test running with paused time advances through +/// the waits instead of sleeping. +pub(crate) async fn wait_until bool>(mut f: F, limit: std::time::Duration) -> bool { + let deadline = tokio::time::Instant::now() + limit; + loop { + if f() { + return true; + } + if tokio::time::Instant::now() >= deadline { + return false; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } +} + +/// A local TCP address whose SYNs go unanswered once filled. +/// +/// A listener with a backlog of one whose accept queue is filled: Linux, +/// macOS and the BSDs drop further SYNs to a listener whose accept queue is +/// full, so a connect to it times out rather than completing or being +/// refused. How many connects the queue takes before it is full differs by +/// kernel (one on Linux, more on macOS), so filling stops at the first +/// connect that times out. The listener and the fillers must be kept alive +/// for as long as that is relied on. +pub(crate) struct Blackhole { + pub(crate) listener: socket2::Socket, + fillers: Vec, + pub(crate) addr: std::net::SocketAddr, +} + +impl Blackhole { + /// A listener with backlog 1. Not 0: macOS reads a backlog of 0 as the + /// system default (about 128), so its queue would not fill. + /// + /// When `fill` is false the accept queue is left empty, so at least one + /// connect completes; [`Blackhole::fill`] fills it later. + pub(crate) fn open(fill: bool) -> Self { + use socket2::{Domain, Socket, Type}; + let listener = Socket::new(Domain::IPV4, Type::STREAM, None).unwrap(); + let bind: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap(); + listener.bind(&bind.into()).unwrap(); + listener.listen(1).unwrap(); + let addr = listener.local_addr().unwrap().as_socket().unwrap(); + let mut bh = Self { + listener, + fillers: Vec::new(), + addr, + }; + if fill { + bh.fill(); + } + bh + } + + /// A listener whose SYNs already go unanswered. + pub(crate) fn silent() -> Self { + Self::open(true) + } + + /// Fill the listener's accept queue, stopping at the first connect that + /// times out, which shows a further connect now times out instead of + /// completing or being refused. + pub(crate) fn fill(&mut self) { + const MAX_FILLERS: usize = 64; + for _ in 0..=MAX_FILLERS { + let probe = std::net::TcpStream::connect_timeout( + &self.addr, + std::time::Duration::from_millis(200), + ); + match probe { + Ok(filler) => self.fillers.push(filler), + Err(e) if e.kind() == std::io::ErrorKind::TimedOut => return, + Err(e) => panic!("blackhole is not silent: probe connect returned {e:?}"), + } + } + panic!("blackhole is not silent: {MAX_FILLERS} connects completed"); + } + + /// Accept every filler still queued on the listener, so the address + /// answers again: the next connect to it, or the next retransmitted SYN + /// of one already waiting, completes. Returns the accepted far ends, + /// which the caller keeps alive while it relies on that. + pub(crate) fn drain(&mut self) -> Vec { + self.fillers + .iter() + .map(|_| std::net::TcpStream::from(self.listener.accept().unwrap().0)) + .collect() + } + + /// The address in the transport form. + pub(crate) fn transport_addr(&self) -> crate::transport::TransportAddr { + crate::transport::TransportAddr::from_string(&self.addr.to_string()) + } +} diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 4efd1d8f..a0227300 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -240,6 +240,12 @@ pub enum TransportError { #[error("connection refused")] ConnectionRefused, + /// No connection to the address exists, and the send did not open one. + /// The connection a reply was meant for has gone; the remote's next + /// attempt arrives on a new one. + #[error("not connected")] + NotConnected, + #[error("transport not supported: {0}")] NotSupported(String), @@ -284,6 +290,10 @@ impl TransportError { | Self::MtuExceeded { .. } | Self::Timeout | Self::ConnectionRefused + // The connection the frame was meant for has gone, and the send + // did not reopen it; the remote's next attempt arrives on a new + // connection. + | Self::NotConnected | Self::NotSupported(_) | Self::Io(_) => false, } @@ -796,6 +806,46 @@ pub struct TransportCongestion { pub recv_drops: Option, } +// ============================================================================ +// Background Connects +// ============================================================================ + +/// What a background connect task yields: the connected stream and the MTU +/// to use on it. +pub(crate) type ConnectOutcome = Result<(tokio::net::TcpStream, u16), TransportError>; + +/// Take the background connect for `addr` out of `connecting` if its task +/// has finished, and return what it produced. +/// +/// Returns `None`, leaving the map untouched, when there is no attempt for +/// the address or it is still running. +pub(crate) fn take_finished_connect( + connecting: &mut std::collections::HashMap, + addr: &TransportAddr, +) -> Option +where + E: AsMut>, +{ + use futures::FutureExt; + + let task = connecting.get_mut(addr)?.as_mut(); + if !task.is_finished() { + return None; + } + // Polling a JoinHandle spends the caller's cooperative budget, and with + // none left it reads pending even though the task has finished. Poll it + // unconstrained, and remove the entry only once its output is in hand, + // so a connected stream is never dropped unread. + let joined = tokio::task::unconstrained(task).now_or_never()?; + connecting.remove(addr); + match joined { + Ok(outcome) => Some(outcome), + Err(e) => Some(Err(TransportError::LinkFailed(format!( + "connect task failed: {e}" + )))), + } +} + // ============================================================================ // Transport Handle // ============================================================================ @@ -873,6 +923,36 @@ impl TransportHandle { } } + /// Send data only over a connection that already exists; never dial. + /// + /// On TCP, Tor and Nym, a background connect that has finished is taken + /// into the pool and used; otherwise the call fails at once with + /// [`TransportError::NotConnected`] and opens nothing. Connectionless + /// transports have no connection to look up and send as + /// [`send`](Self::send) does. BLE's send never waits on a connect and is + /// used as is. + /// + /// Use this from anything the rx loop awaits: a dial there holds every + /// other frame for up to the transport's connect timeout. + pub async fn send_existing( + &self, + addr: &TransportAddr, + data: &[u8], + ) -> Result { + match self { + TransportHandle::Udp(t) => t.send_async(addr, data).await, + #[cfg(any(target_os = "linux", target_os = "macos"))] + TransportHandle::Ethernet(t) => t.send_async(addr, data).await, + TransportHandle::Tcp(t) => t.send_existing(addr, data).await, + TransportHandle::Tor(t) => t.send_existing(addr, data).await, + TransportHandle::Nym(t) => t.send_existing(addr, data).await, + #[cfg(ble_available)] + TransportHandle::Ble(t) => t.send_async(addr, data).await, + #[cfg(test)] + TransportHandle::Loopback(t) => t.send_async(addr, data).await, + } + } + /// Get the transport ID. pub fn transport_id(&self) -> TransportId { match self { @@ -1232,29 +1312,92 @@ impl TransportHandle { /// /// Returns a snapshot of counters for the specific transport type. pub fn transport_stats(&self) -> serde_json::Value { + self.live_stats().to_json() + } + + /// The transport's shared counters, for reading live off the rx loop. + /// + /// The counters are atomics the transport updates from its own tasks and + /// threads, so a holder sees them move without anything republishing + /// them. + pub(crate) fn live_stats(&self) -> LiveStats { match self { - TransportHandle::Udp(t) => { - serde_json::to_value(t.stats().snapshot()).unwrap_or_default() - } + TransportHandle::Udp(t) => LiveStats::Udp(t.stats().clone()), #[cfg(any(target_os = "linux", target_os = "macos"))] - TransportHandle::Ethernet(t) => { - serde_json::to_value(t.stats().snapshot()).unwrap_or_default() - } - TransportHandle::Tcp(t) => { - serde_json::to_value(t.stats().snapshot()).unwrap_or_default() - } - TransportHandle::Tor(t) => { - serde_json::to_value(t.stats().snapshot()).unwrap_or_default() - } - TransportHandle::Nym(t) => { - serde_json::to_value(t.stats().snapshot()).unwrap_or_default() - } + TransportHandle::Ethernet(t) => LiveStats::Ethernet(t.stats().clone()), + TransportHandle::Tcp(t) => LiveStats::Tcp(t.stats().clone()), + TransportHandle::Tor(t) => LiveStats::Tor(t.stats().clone()), + TransportHandle::Nym(t) => LiveStats::Nym(t.stats().clone()), #[cfg(ble_available)] - TransportHandle::Ble(t) => { - serde_json::to_value(t.stats().snapshot()).unwrap_or_default() - } + TransportHandle::Ble(t) => LiveStats::Ble(t.stats().clone()), #[cfg(test)] - TransportHandle::Loopback(_) => serde_json::json!({}), + TransportHandle::Loopback(_) => LiveStats::Empty, + } + } +} + +/// A transport's shared counters, held by reference rather than copied. +/// +/// Cloning shares the counters. `show_transports` holds one per transport in +/// its published row and reads it at request time, so the counters it shows +/// are current even when nothing has republished the row. +#[derive(Clone)] +pub(crate) enum LiveStats { + /// UDP transport counters. + Udp(std::sync::Arc), + /// Ethernet transport counters. + #[cfg(any(target_os = "linux", target_os = "macos"))] + Ethernet(std::sync::Arc), + /// TCP transport counters. + Tcp(std::sync::Arc), + /// Tor transport counters. + Tor(std::sync::Arc), + /// Nym transport counters. + Nym(std::sync::Arc), + /// BLE transport counters. + #[cfg(ble_available)] + Ble(std::sync::Arc), + /// No counters (the test loopback transport). + #[cfg(test)] + Empty, +} + +impl LiveStats { + /// Read the counters now, as the JSON `show_transports` reports under + /// `stats`. + pub(crate) fn to_json(&self) -> serde_json::Value { + match self { + LiveStats::Udp(s) => serde_json::to_value(s.snapshot()).unwrap_or_default(), + #[cfg(any(target_os = "linux", target_os = "macos"))] + LiveStats::Ethernet(s) => serde_json::to_value(s.snapshot()).unwrap_or_default(), + LiveStats::Tcp(s) => serde_json::to_value(s.snapshot()).unwrap_or_default(), + LiveStats::Tor(s) => serde_json::to_value(s.snapshot()).unwrap_or_default(), + LiveStats::Nym(s) => serde_json::to_value(s.snapshot()).unwrap_or_default(), + #[cfg(ble_available)] + LiveStats::Ble(s) => serde_json::to_value(s.snapshot()).unwrap_or_default(), + #[cfg(test)] + LiveStats::Empty => serde_json::json!({}), + } + } +} + +/// Two values are equal when they share the same counters, not when the +/// counters happen to read the same. +impl PartialEq for LiveStats { + fn eq(&self, other: &Self) -> bool { + use std::sync::Arc; + match (self, other) { + (LiveStats::Udp(a), LiveStats::Udp(b)) => Arc::ptr_eq(a, b), + #[cfg(any(target_os = "linux", target_os = "macos"))] + (LiveStats::Ethernet(a), LiveStats::Ethernet(b)) => Arc::ptr_eq(a, b), + (LiveStats::Tcp(a), LiveStats::Tcp(b)) => Arc::ptr_eq(a, b), + (LiveStats::Tor(a), LiveStats::Tor(b)) => Arc::ptr_eq(a, b), + (LiveStats::Nym(a), LiveStats::Nym(b)) => Arc::ptr_eq(a, b), + #[cfg(ble_available)] + (LiveStats::Ble(a), LiveStats::Ble(b)) => Arc::ptr_eq(a, b), + #[cfg(test)] + (LiveStats::Empty, LiveStats::Empty) => true, + _ => false, } } } @@ -1299,6 +1442,47 @@ pub(crate) async fn resolve_socket_addrs( mod tests { use super::*; + /// A connecting-pool entry holding only its connect task. + struct TaskEntry(tokio::task::JoinHandle); + + impl AsMut> for TaskEntry { + /// The background connect task. + fn as_mut(&mut self) -> &mut tokio::task::JoinHandle { + &mut self.0 + } + } + + /// A finished connect is handed back even when the calling task has + /// spent its cooperative budget, which makes a plain poll of the task + /// read pending. Losing it there would close a connected stream unseen. + #[tokio::test] + async fn take_finished_connect_returns_the_outcome_when_the_caller_budget_is_spent() { + let addr = TransportAddr::from_socket_addr("192.0.2.1:2121".parse().unwrap()); + let task = tokio::spawn(async { Err(TransportError::ConnectionRefused) }); + while !task.is_finished() { + tokio::task::yield_now().await; + } + let mut connecting = std::collections::HashMap::from([(addr.clone(), TaskEntry(task))]); + + for _ in 0..10_000 { + if !tokio::task::coop::has_budget_remaining() { + break; + } + tokio::task::consume_budget().await; + } + assert!( + !tokio::task::coop::has_budget_remaining(), + "the test runtime must budget this task, or the case is not exercised" + ); + + let outcome = take_finished_connect(&mut connecting, &addr); + assert!( + matches!(outcome, Some(Err(TransportError::ConnectionRefused))), + "finished connect not returned: {outcome:?}" + ); + assert!(connecting.is_empty()); + } + #[test] fn test_transport_id() { let id = TransportId::new(42); @@ -1812,6 +1996,8 @@ mod tests { // answer, not this node's inability to transmit. TransportError::Timeout, TransportError::ConnectionRefused, + // The connection is gone and nothing will bring it back. + TransportError::NotConnected, ] { assert!( !terminal.is_transient(), diff --git a/src/transport/nym/mod.rs b/src/transport/nym/mod.rs index e850bad8..b83adf31 100644 --- a/src/transport/nym/mod.rs +++ b/src/transport/nym/mod.rs @@ -21,7 +21,7 @@ use super::{ use crate::config::NymConfig; use crate::transport::socks5::{ ConnectingEntry, ConnectingPool, DialError, ProxiedConnection, ProxiedPool, SEND_QUEUE_DEPTH, - Socks5Auth, Socks5Dialer, SocksTarget, poll_connecting, proxied_receive_loop, + Socks5Auth, Socks5Dialer, SocksTarget, existing_sender, poll_connecting, proxied_receive_loop, proxied_send_loop, }; use crate::transport::stream::{ConnId, WRITER_DRAIN_TIMEOUT, drain_writer, next_conn_id}; @@ -246,16 +246,7 @@ impl NymTransport { if !self.state.is_operational() { return Err(TransportError::NotStarted); } - - // Pre-send MTU check - let mtu = self.config.mtu() as usize; - if data.len() > mtu { - self.stats.record_mtu_exceeded(); - return Err(TransportError::MtuExceeded { - packet_size: data.len(), - mtu: self.config.mtu(), - }); - } + self.check_mtu(data)?; // Get or create the connection's send queue. Never the write half: // this function must not be able to await the wire (see @@ -273,6 +264,54 @@ impl NymTransport { } }; + self.enqueue(addr, &send_tx, data) + } + + /// Send a packet only over a connection that already exists. + /// + /// Uses the pooled connection for `addr`, or one a finished background + /// connect has produced, which it moves into the pool. Never dials: with + /// neither, it fails at once with [`TransportError::NotConnected`]. Like + /// `send_async`, it only queues the frame for the connection's writer + /// task and never awaits the wire. + pub async fn send_existing( + &self, + addr: &TransportAddr, + data: &[u8], + ) -> Result { + if !self.state.is_operational() { + return Err(TransportError::NotStarted); + } + self.check_mtu(data)?; + let send_tx = existing_sender(&self.pool, &self.connecting, addr, |stream, mtu| { + let conn = self.outbound_connection(addr, stream, mtu); + self.record_promoted(addr); + conn + }) + .await + .ok_or(TransportError::NotConnected)?; + self.enqueue(addr, &send_tx, data) + } + + /// Reject a packet larger than the transport MTU before queueing it. + fn check_mtu(&self, data: &[u8]) -> Result<(), TransportError> { + if data.len() > self.config.mtu() as usize { + self.stats.record_mtu_exceeded(); + return Err(TransportError::MtuExceeded { + packet_size: data.len(), + mtu: self.config.mtu(), + }); + } + Ok(()) + } + + /// Queue one packet for the writer task of the connection to `addr`. + fn enqueue( + &self, + addr: &TransportAddr, + send_tx: &tokio::sync::mpsc::Sender>, + data: &[u8], + ) -> Result { // 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 @@ -441,6 +480,7 @@ impl NymTransport { let transport_id = self.transport_id; let remote_addr = addr.clone(); let config = self.config.clone(); + let stats = self.stats.clone(); debug!( transport_id = %transport_id, @@ -475,7 +515,9 @@ impl NymTransport { ); stream } + // Counted as connect() counts them. Err(DialError::Socks(e)) => { + stats.record_socks5_error(); warn!( transport_id = %transport_id, remote_addr = %remote_addr, @@ -486,6 +528,7 @@ impl NymTransport { return Err(TransportError::ConnectionRefused); } Err(DialError::Timeout) => { + stats.record_connect_timeout(); warn!( transport_id = %transport_id, remote_addr = %remote_addr, @@ -519,6 +562,30 @@ impl NymTransport { /// Promote a completed background connection to the established pool. fn promote_connection(&self, addr: &TransportAddr, stream: TcpStream, mtu: u16) { + let conn = self.outbound_connection(addr, stream, mtu); + + if let Ok(mut pool) = self.pool.try_lock() { + pool.insert(addr.clone(), conn); + self.record_promoted(addr); + } else { + conn.recv_task.abort(); + conn.send_task.abort(); + warn!( + transport_id = %self.transport_id, + remote_addr = %addr, + "Failed to promote Nym connection (pool locked)" + ); + } + } + + /// Build the pool entry for a finished background connect: split the + /// stream and spawn its receive loop and writer task. + fn outbound_connection( + &self, + addr: &TransportAddr, + stream: TcpStream, + mtu: u16, + ) -> ProxiedConnection<()> { let (read_half, write_half) = stream.into_split(); let transport_id = self.transport_id; @@ -555,7 +622,7 @@ impl NymTransport { |_stats: &NymStats, _meta: &()| {}, )); - let conn = ProxiedConnection { + ProxiedConnection { send_tx, send_task, recv_task, @@ -563,27 +630,19 @@ impl NymTransport { established_at: Instant::now(), meta: (), id, - }; - - if let Ok(mut pool) = self.pool.try_lock() { - pool.insert(addr.clone(), conn); - self.stats.record_connection_established(); - debug!( - transport_id = %self.transport_id, - remote_addr = %addr, - "Nym connection established (background connect)" - ); - } else { - conn.recv_task.abort(); - conn.send_task.abort(); - warn!( - transport_id = %self.transport_id, - remote_addr = %addr, - "Failed to promote Nym connection (pool locked)" - ); } } + /// Count and log a background connection that has entered the pool. + fn record_promoted(&self, addr: &TransportAddr) { + self.stats.record_connection_established(); + debug!( + transport_id = %self.transport_id, + remote_addr = %addr, + "Nym connection established (background connect)" + ); + } + /// Close a specific connection asynchronously. /// /// Aborts the receive task and lets the writer finish the frames already @@ -763,6 +822,7 @@ fn validate_host_port(addr: &str, field: &str) -> Result<(), TransportError> { #[cfg(test)] mod tests { use super::*; + use crate::testutil::wait_until; use crate::transport::packet_channel; /// Test config: a syntactically valid loopback proxy address, with the @@ -1057,20 +1117,6 @@ mod tests { // Connection identity and failure teardown // ======================================================================== - /// Poll `f` every 10ms until it holds or `limit` elapses. - async fn wait_until bool>(mut f: F, limit: Duration) -> bool { - let deadline = Instant::now() + limit; - loop { - if f() { - return true; - } - if Instant::now() >= deadline { - return false; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - } - /// A destination TCP transport behind a mock SOCKS5 proxy, and a started /// Nym transport dialing through it. async fn nym_via_mock_proxy() -> ( @@ -1239,4 +1285,154 @@ mod tests { nym.stop_async().await.unwrap(); dest.stop_async().await.unwrap(); } + + /// With no pooled connection and no connect under way, `send_existing` + /// fails with `NotConnected` and opens nothing. + #[tokio::test] + async fn send_existing_without_connection_fails_fast_and_dials_nothing() { + let (mut dest, _dest_rx, mut t, target) = nym_via_mock_proxy().await; + + let result = t.send_existing(&target, &build_msg1_frame()).await; + + assert!( + matches!(result, Err(TransportError::NotConnected)), + "expected NotConnected, got {result:?}" + ); + assert!( + t.connecting.lock().await.is_empty(), + "a connect was started" + ); + assert_eq!(t.stats().snapshot().connections_established, 0); + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!( + dest.stats().snapshot().connections_accepted, + 0, + "the destination saw a connection" + ); + + t.stop_async().await.unwrap(); + dest.stop_async().await.unwrap(); + } + + /// A background connect that has finished is moved into the pool and + /// carries the send, with no second connection opened. + #[tokio::test] + async fn send_existing_promotes_a_finished_background_connect_and_sends_on_it() { + let (mut dest, mut dest_rx, mut t, target) = nym_via_mock_proxy().await; + + t.connect_async(&target).await.unwrap(); + let mut waited = 0; + while !t + .connecting + .try_lock() + .is_ok_and(|c| c.get(&target).is_some_and(|e| e.task.is_finished())) + { + assert!(waited < 150, "background connect never finished"); + tokio::time::sleep(Duration::from_millis(20)).await; + waited += 1; + } + let frame = build_msg1_frame(); + t.send_existing(&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!( + t.connecting.lock().await.is_empty(), + "the finished connect was left in the connecting map" + ); + assert_eq!(t.stats().snapshot().connections_established, 1); + assert_eq!( + dest.stats().snapshot().connections_accepted, + 1, + "the send used the background connection, not a new one" + ); + + t.stop_async().await.unwrap(); + dest.stop_async().await.unwrap(); + } + + /// Wait until the background connect to `target` has finished, leaving + /// it in the connecting map. + async fn wait_background_finished(t: &NymTransport, target: &TransportAddr) { + let finished = wait_until( + || { + t.connecting + .try_lock() + .is_ok_and(|c| c.get(target).is_some_and(|e| e.task.is_finished())) + }, + Duration::from_secs(3), + ) + .await; + assert!(finished, "background connect to {target} never finished"); + } + + /// A background connect through a proxy that accepts the TCP connection + /// but never answers the SOCKS5 greeting times out, and is counted in + /// `connect_timeouts` as an inline one is. + #[tokio::test] + async fn background_connect_timeout_is_counted() { + // Bound and listening, never accepted: the kernel completes the TCP + // handshake and nothing ever replies. + let silent = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let (tx, _rx) = packet_channel(32); + let config = NymConfig { + socks5_addr: Some(silent.local_addr().unwrap().to_string()), + startup_timeout_secs: Some(5), + connect_timeout_ms: Some(200), + ..Default::default() + }; + let mut t = NymTransport::new(TransportId::new(200), None, config, tx); + t.start_async().await.unwrap(); + let target = TransportAddr::from_string("127.0.0.1:9"); + + t.connect_async(&target).await.unwrap(); + wait_background_finished(&t, &target).await; + + let stats = t.stats().snapshot(); + assert_eq!(stats.connect_timeouts, 1, "the timeout was not counted"); + assert_eq!(stats.socks5_errors, 0); + let result = t.send_existing(&target, &build_msg1_frame()).await; + assert!(matches!(result, Err(TransportError::NotConnected))); + assert_eq!( + t.stats().snapshot().connect_timeouts, + 1, + "taking the result counted the timeout again" + ); + + t.stop_async().await.unwrap(); + } + + /// A background connect the proxy answers with a SOCKS5 failure is + /// counted in `socks5_errors` as an inline one is. + #[tokio::test] + async fn background_connect_socks5_error_is_counted() { + let dummy_target: std::net::SocketAddr = "127.0.0.1:1".parse().unwrap(); + let mock = MockSocks5Server::with_reply_code(dummy_target, 0x01) + .await + .unwrap(); + let proxy_addr = mock.addr(); + let _proxy_handle = mock.spawn(); + let (tx, _rx) = packet_channel(32); + let config = NymConfig { + socks5_addr: Some(proxy_addr.to_string()), + startup_timeout_secs: Some(5), + connect_timeout_ms: Some(2000), + ..Default::default() + }; + let mut t = NymTransport::new(TransportId::new(200), None, config, tx); + t.start_async().await.unwrap(); + let target = TransportAddr::from_string("127.0.0.1:9"); + + t.connect_async(&target).await.unwrap(); + wait_background_finished(&t, &target).await; + + let stats = t.stats().snapshot(); + assert_eq!(stats.socks5_errors, 1, "the SOCKS5 error was not counted"); + assert_eq!(stats.connect_timeouts, 0); + + t.stop_async().await.unwrap(); + } } diff --git a/src/transport/socks5/mod.rs b/src/transport/socks5/mod.rs index 4a0911ab..0c7d6d1e 100644 --- a/src/transport/socks5/mod.rs +++ b/src/transport/socks5/mod.rs @@ -12,7 +12,7 @@ mod stats; pub use dialer::{DialError, Socks5Auth, Socks5Dialer, SocksTarget}; pub(crate) use pool::{ ConnectingEntry, ConnectingPool, ProxiedConnection, ProxiedPool, ProxiedStats, - SEND_QUEUE_DEPTH, poll_connecting, proxied_receive_loop, proxied_send_loop, + SEND_QUEUE_DEPTH, existing_sender, poll_connecting, proxied_receive_loop, proxied_send_loop, }; pub(crate) use stats::ProxiedStatsBase; diff --git a/src/transport/socks5/pool.rs b/src/transport/socks5/pool.rs index fbaf9f28..6163337d 100644 --- a/src/transport/socks5/pool.rs +++ b/src/transport/socks5/pool.rs @@ -24,6 +24,7 @@ use crate::transport::stream::{ConnId, PooledConn, remove_own}; use crate::transport::tcp::InboundDeadline; use crate::transport::{ ConnectionState, PacketTx, ReceivedPacket, TransportAddr, TransportError, TransportId, + take_finished_connect, }; /// State for a single pooled connection to a peer. @@ -72,6 +73,13 @@ pub(crate) struct ConnectingEntry { pub task: JoinHandle>, } +impl AsMut>> for ConnectingEntry { + /// The background connect task. + fn as_mut(&mut self) -> &mut JoinHandle> { + &mut self.task + } +} + /// Map of addresses with background connection attempts in progress. pub(crate) type ConnectingPool = Arc>>; @@ -132,6 +140,42 @@ pub(crate) fn poll_connecting( } } +/// The writer queue for an established connection to `addr`, promoting a +/// background connect that has finished since it was started. +/// +/// `promote` builds the pool entry for a finished connect and does the +/// transport's own accounting; it runs with the pool lock held, so the +/// connection cannot be inserted twice. A finished connect that failed is +/// dropped, so the next `connect_async` starts a new attempt. Never dials. +pub(crate) async fn existing_sender( + pool: &ProxiedPool, + connecting: &ConnectingPool, + addr: &TransportAddr, + promote: impl FnOnce(TcpStream, u16) -> ProxiedConnection, +) -> Option>> { + let mut pool = pool.lock().await; + if let Some(conn) = pool.get(addr) { + return Some(conn.send_tx.clone()); + } + let finished = take_finished_connect(&mut *connecting.lock().await, addr)?; + match finished { + Ok((stream, mtu)) => { + let conn = promote(stream, mtu); + let send_tx = conn.send_tx.clone(); + pool.insert(addr.clone(), conn); + Some(send_tx) + } + Err(e) => { + debug!( + remote_addr = %addr, + error = %e, + "Background SOCKS5 connect failed, nothing to send on" + ); + None + } + } +} + /// Minimal stats surface the shared receive loop needs. /// /// The per-transport stats structs implement this by delegating to their @@ -373,6 +417,7 @@ pub(crate) async fn proxied_receive_loop( #[cfg(test)] mod tests { use super::*; + use crate::testutil::wait_until; use crate::transport::packet_channel; use crate::transport::stream::{next_conn_id, park_writer}; use portable_atomic::{AtomicU64, Ordering}; @@ -419,20 +464,6 @@ mod tests { } } - /// Poll `f` every 10ms until it holds or `limit` elapses. - async fn wait_until bool>(mut f: F, limit: Duration) -> bool { - let deadline = Instant::now() + limit; - loop { - if f() { - return true; - } - if Instant::now() >= deadline { - return false; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - } - /// A writer whose write fails must not remove a newer connection that has /// taken its address in the pool, nor run `on_remove` for it. #[tokio::test] diff --git a/src/transport/tcp/mod.rs b/src/transport/tcp/mod.rs index 21103567..9f63e6da 100644 --- a/src/transport/tcp/mod.rs +++ b/src/transport/tcp/mod.rs @@ -25,11 +25,11 @@ mod pool; pub mod stats; -use super::resolve_socket_addrs; use super::{ ConnectionState, DiscoveredPeer, PacketTx, ReceivedPacket, Transport, TransportAddr, TransportError, TransportId, TransportState, TransportType, }; +use super::{resolve_socket_addrs, take_finished_connect}; use crate::config::TcpConfig; use crate::transport::framing::read_fmp_packet; use crate::transport::stream::{ @@ -338,19 +338,7 @@ impl TcpTransport { if !self.state.is_operational() { return Err(TransportError::NotStarted); } - - // Pre-send MTU check: reject oversize packets before writing them - // to the TCP stream. Without this, the receiver's FMP stream reader - // would see payload_len > max and close the connection, causing a - // disruptive reset-reconnect cycle. - let mtu = self.config.mtu() as usize; - if data.len() > mtu { - self.stats.record_mtu_exceeded(); - return Err(TransportError::MtuExceeded { - packet_size: data.len(), - mtu: self.config.mtu(), - }); - } + self.check_mtu(data)?; // Get or create connection. What comes back is the queue into the // connection's writer task, never the write half itself: this function @@ -368,6 +356,87 @@ impl TcpTransport { } }; + self.enqueue(addr, &send_tx, data) + } + + /// Send a packet only over a connection that already exists. + /// + /// Uses the pooled connection for `addr`, or one a finished background + /// connect has produced, which it moves into the pool. Never dials: with + /// neither, it fails at once with [`TransportError::NotConnected`]. Like + /// `send_async`, it only queues the frame for the connection's writer + /// task and never awaits the wire. + pub async fn send_existing( + &self, + addr: &TransportAddr, + data: &[u8], + ) -> Result { + if !self.state.is_operational() { + return Err(TransportError::NotStarted); + } + self.check_mtu(data)?; + let send_tx = self + .existing_sender(addr) + .await + .ok_or(TransportError::NotConnected)?; + self.enqueue(addr, &send_tx, data) + } + + /// The writer queue for an established connection to `addr`, promoting + /// a background connect that has finished since it was started. + /// + /// Holds the pool lock across the promotion, so the connection cannot be + /// inserted twice. A finished connect that failed is dropped, so the next + /// `connect_async` starts a new attempt. + async fn existing_sender(&self, addr: &TransportAddr) -> Option>> { + let mut pool = self.pool.lock().await; + if let Some(conn) = key_for_remote(&pool, addr).and_then(|key| pool.get(&key)) { + return Some(conn.send_tx.clone()); + } + let finished = take_finished_connect(&mut *self.connecting.lock().await, addr)?; + match finished { + Ok((stream, mss_mtu)) => { + let conn = self.outbound_connection(addr, stream, mss_mtu); + let send_tx = conn.send_tx.clone(); + pool.insert(PoolKey::outbound(addr.clone()), conn); + self.record_promoted(addr, mss_mtu); + Some(send_tx) + } + Err(e) => { + debug!( + transport_id = %self.transport_id, + remote_addr = %addr, + error = %e, + "Background TCP connect failed, nothing to send on" + ); + None + } + } + } + + /// Reject a packet larger than the transport MTU before queueing it. + /// + /// Without this, the receiver's FMP stream reader would see + /// payload_len > max and close the connection, causing a disruptive + /// reset-reconnect cycle. + fn check_mtu(&self, data: &[u8]) -> Result<(), TransportError> { + if data.len() > self.config.mtu() as usize { + self.stats.record_mtu_exceeded(); + return Err(TransportError::MtuExceeded { + packet_size: data.len(), + mtu: self.config.mtu(), + }); + } + Ok(()) + } + + /// Queue one packet for the writer task of the connection to `addr`. + fn enqueue( + &self, + addr: &TransportAddr, + send_tx: &mpsc::Sender>, + data: &[u8], + ) -> Result { // 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 @@ -583,6 +652,7 @@ impl TcpTransport { let config = self.config.clone(); let transport_id = self.transport_id; let remote_addr = addr.clone(); + let stats = self.stats.clone(); debug!( transport_id = %transport_id, @@ -596,11 +666,18 @@ impl TcpTransport { let socket_addrs: Vec<_> = resolve_socket_addrs(&remote_addr).await?.collect(); // A refusal is logged with its OS error by connect_to_any_addr. + // Failures are counted here, where they happen, as connect() + // counts them; whoever later takes the result does not. let connected = connect_to_any_addr(transport_id, &remote_addr, &socket_addrs, timeout_ms).await; let stream = match connected { Ok(stream) => stream, + Err(error @ TransportError::ConnectionRefused) => { + stats.record_connect_refused(); + return Err(error); + } Err(error @ TransportError::Timeout) => { + stats.record_connect_timeout(); debug!( transport_id = %transport_id, remote_addr = %remote_addr, @@ -698,6 +775,33 @@ impl TcpTransport { /// Splits the stream, spawns a receive loop, and inserts into the pool. /// Called from `connection_state_sync()` when a background task completes. fn promote_connection(&self, addr: &TransportAddr, stream: TcpStream, mss_mtu: u16) { + let conn = self.outbound_connection(addr, stream, mss_mtu); + + // Use try_lock since we're in a sync context and the pool + // should be available (connection_state_sync already checked it) + if let Ok(mut pool) = self.pool.try_lock() { + pool.insert(PoolKey::outbound(addr.clone()), conn); + self.record_promoted(addr, mss_mtu); + } 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, + "Failed to promote connection (pool locked)" + ); + } + } + + /// Build the pool entry for a finished background connect: split the + /// stream and spawn its receive loop and writer task. + fn outbound_connection( + &self, + addr: &TransportAddr, + stream: TcpStream, + mss_mtu: u16, + ) -> TcpConnection { let (read_half, write_half) = stream.into_split(); let transport_id = self.transport_id; @@ -706,7 +810,7 @@ impl TcpTransport { let recv_stats = self.stats.clone(); let key = PoolKey::outbound(addr.clone()); let recv_key = key.clone(); - let send_key = key.clone(); + let send_key = key; let id = next_conn_id(); let recv_task = tokio::spawn(async move { @@ -739,7 +843,7 @@ impl TcpTransport { self.stats.clone(), )); - let conn = TcpConnection { + TcpConnection { send_tx, send_task, recv_task, @@ -747,31 +851,20 @@ impl TcpTransport { established_at: Instant::now(), direction: Direction::Outbound, id, - }; - - // Use try_lock since we're in a sync context and the pool - // should be available (connection_state_sync already checked it) - if let Ok(mut pool) = self.pool.try_lock() { - pool.insert(key, conn); - self.stats.record_connection_established(); - self.stats.record_pool_outbound_added(); - debug!( - transport_id = %self.transport_id, - remote_addr = %addr, - mtu = mss_mtu, - "TCP connection established (background connect)" - ); - } 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, - "Failed to promote connection (pool locked)" - ); } } + + /// Count and log a background connection that has entered the pool. + fn record_promoted(&self, addr: &TransportAddr, mss_mtu: u16) { + self.stats.record_connection_established(); + self.stats.record_pool_outbound_added(); + debug!( + transport_id = %self.transport_id, + remote_addr = %addr, + mtu = mss_mtu, + "TCP connection established (background connect)" + ); + } } impl Transport for TcpTransport { @@ -1456,6 +1549,7 @@ fn read_mss_mtu(stream: &std::net::TcpStream, default_mtu: u16) -> u16 { mod tests { use super::pool::PoolMap; use super::*; + use crate::testutil::{Blackhole, wait_until}; use crate::transport::framing::{build_established_frame, build_msg1_frame}; use crate::transport::packet_channel; use crate::transport::stream::park_writer; @@ -1470,20 +1564,6 @@ mod tests { pool.get(&key) } - /// Poll `f` every 10ms until it holds or `limit` elapses. - async fn wait_until bool>(mut f: F, limit: Duration) -> bool { - let deadline = Instant::now() + limit; - loop { - if f() { - return true; - } - if Instant::now() >= deadline { - return false; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - } - fn capped_config(max_inbound: usize) -> TcpConfig { TcpConfig { bind_addr: Some("127.0.0.1:0".to_string()), @@ -3423,4 +3503,244 @@ mod tests { drop(peer); transport.stop_async().await.unwrap(); } + + /// A minimal msg1-phase frame the receive loop accepts. + fn msg1_frame() -> Vec { + build_msg1_frame() + } + + /// A minimal XX msg2-phase frame the receive loop accepts: 118 bytes, + /// version 1. + fn msg2_frame() -> Vec { + let mut frame = vec![0xBB; 118]; + frame[0] = 0x12; + frame[1] = 0x00; + frame[2..4].copy_from_slice(&114u16.to_le_bytes()); + frame + } + + /// Wait until the background connect to `remote` has finished, leaving + /// it unpromoted in the connecting map. + async fn wait_connect_finished(t: &TcpTransport, remote: &TransportAddr) { + let finished = wait_until( + || { + t.connecting + .try_lock() + .is_ok_and(|c| c.get(remote).is_some_and(|e| e.task.is_finished())) + }, + Duration::from_secs(3), + ) + .await; + assert!(finished, "background connect to {remote} never finished"); + } + + /// With no pooled connection and no connect under way, `send_existing` + /// fails with `NotConnected` and opens nothing. + #[tokio::test] + async fn send_existing_without_connection_fails_fast_and_dials_nothing() { + let (tx1, _rx1) = packet_channel(100); + let (tx2, _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(); + // Reachable, so a send that dialed would succeed. + let remote = TransportAddr::from_string(&t2.local_addr().unwrap().to_string()); + + let result = t1.send_existing(&remote, &msg1_frame()).await; + + assert!( + matches!(result, Err(TransportError::NotConnected)), + "expected NotConnected, got {result:?}" + ); + assert!( + t1.connecting.lock().await.is_empty(), + "a connect was started" + ); + assert_eq!(t1.connection_state_sync(&remote), ConnectionState::None); + assert_eq!(t1.stats().snapshot().connections_established, 0); + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!( + t2.stats().snapshot().connections_accepted, + 0, + "the remote saw a connection" + ); + + t1.stop_async().await.unwrap(); + t2.stop_async().await.unwrap(); + } + + /// A background connect that has finished is moved into the pool and + /// carries the send, with no second connection opened. + #[tokio::test] + async fn send_existing_promotes_a_finished_background_connect_and_sends_on_it() { + 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()); + + t1.connect_async(&remote).await.unwrap(); + wait_connect_finished(&t1, &remote).await; + let frame = msg1_frame(); + let sent = t1.send_existing(&remote, &frame).await.unwrap(); + assert_eq!(sent, frame.len()); + + let packet = timeout(Duration::from_secs(2), rx2.recv()) + .await + .expect("timeout") + .expect("channel closed"); + assert_eq!(packet.data, frame); + assert!( + t1.connecting.lock().await.is_empty(), + "the finished connect was left in the connecting map" + ); + let stats = t1.stats().snapshot(); + assert_eq!(stats.connections_established, 1); + assert_eq!(stats.pool_outbound, 1); + assert_eq!( + t2.stats().snapshot().connections_accepted, + 1, + "the send used the background connection, not a new one" + ); + + t1.stop_async().await.unwrap(); + t2.stop_async().await.unwrap(); + } + + /// A background connect that failed is dropped from the connecting map, + /// the send fails with `NotConnected`, and `connect_async` can start again. + #[tokio::test] + async fn send_existing_after_a_failed_background_connect_drops_it_so_a_new_one_can_start() { + let (tx, _rx) = packet_channel(100); + let mut t = TcpTransport::new(TransportId::new(1), None, make_outbound_config(), tx); + t.start_async().await.unwrap(); + // A port nothing listens on: the connect is refused. + let closed = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let remote = TransportAddr::from_string(&closed.local_addr().unwrap().to_string()); + drop(closed); + + t.connect_async(&remote).await.unwrap(); + wait_connect_finished(&t, &remote).await; + let result = t.send_existing(&remote, &msg1_frame()).await; + + assert!( + matches!(result, Err(TransportError::NotConnected)), + "expected NotConnected, got {result:?}" + ); + assert!( + t.connecting.lock().await.get(&remote).is_none(), + "the failed connect was left in the connecting map" + ); + t.connect_async(&remote).await.unwrap(); + assert!( + t.connecting.lock().await.get(&remote).is_some(), + "connect_async did not start a new attempt" + ); + + t.stop_async().await.unwrap(); + } + + /// A reply to a peer that connected in goes back on that inbound + /// connection. + #[tokio::test] + async fn send_existing_replies_on_a_live_inbound_connection() { + let (tx1, mut 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()); + + t1.send_async(&remote, &msg1_frame()).await.unwrap(); + let inbound = timeout(Duration::from_secs(2), rx2.recv()) + .await + .expect("timeout") + .expect("channel closed"); + // The frame is only forwarded once the accept loop has pooled the + // connection, so a reply to its address finds it. + let reply = msg2_frame(); + t2.send_existing(&inbound.remote_addr, &reply) + .await + .unwrap(); + + let packet = timeout(Duration::from_secs(2), rx1.recv()) + .await + .expect("timeout") + .expect("channel closed"); + assert_eq!(packet.data, reply); + assert_eq!( + t2.stats().snapshot().connections_established, + 0, + "the reply dialed" + ); + + t1.stop_async().await.unwrap(); + t2.stop_async().await.unwrap(); + } + + /// A background connect that times out is counted in `connect_timeouts`, + /// as an inline one is. + #[tokio::test] + async fn background_connect_timeout_is_counted() { + let bh = Blackhole::silent(); + let (tx, _rx) = packet_channel(100); + let config = TcpConfig { + connect_timeout_ms: Some(200), + ..make_outbound_config() + }; + let mut t = TcpTransport::new(TransportId::new(1), None, config, tx); + t.start_async().await.unwrap(); + let remote = bh.transport_addr(); + + t.connect_async(&remote).await.unwrap(); + wait_connect_finished(&t, &remote).await; + + let stats = t.stats().snapshot(); + assert_eq!(stats.connect_timeouts, 1, "the timeout was not counted"); + assert_eq!(stats.connect_refused, 0); + assert_eq!( + t.connection_state_sync(&remote), + ConnectionState::Failed("transport timeout".into()) + ); + assert_eq!( + t.stats().snapshot().connect_timeouts, + 1, + "taking the result counted the timeout again" + ); + + t.stop_async().await.unwrap(); + } + + /// A background connect that is refused is counted in `connect_refused`, + /// as an inline one is. + #[tokio::test] + async fn background_connect_refusal_is_counted() { + let (tx, _rx) = packet_channel(100); + let mut t = TcpTransport::new(TransportId::new(1), None, make_outbound_config(), tx); + t.start_async().await.unwrap(); + // A port nothing listens on: the connect is refused. + let closed = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let remote = TransportAddr::from_string(&closed.local_addr().unwrap().to_string()); + drop(closed); + + t.connect_async(&remote).await.unwrap(); + wait_connect_finished(&t, &remote).await; + + let stats = t.stats().snapshot(); + assert_eq!(stats.connect_refused, 1, "the refusal was not counted"); + assert_eq!(stats.connect_timeouts, 0); + let result = t.send_existing(&remote, &msg1_frame()).await; + assert!(matches!(result, Err(TransportError::NotConnected))); + assert_eq!( + t.stats().snapshot().connect_refused, + 1, + "taking the result counted the refusal again" + ); + + t.stop_async().await.unwrap(); + } } diff --git a/src/transport/tcp/pool.rs b/src/transport/tcp/pool.rs index 26323151..ba85db49 100644 --- a/src/transport/tcp/pool.rs +++ b/src/transport/tcp/pool.rs @@ -139,5 +139,12 @@ pub(crate) struct ConnectingEntry { pub(crate) task: JoinHandle>, } +impl AsMut>> for ConnectingEntry { + /// The background connect task. + fn as_mut(&mut self) -> &mut JoinHandle> { + &mut self.task + } +} + /// Map of addresses with background connection attempts in progress. pub(crate) type ConnectingPool = Arc>>; diff --git a/src/transport/tcp/stats.rs b/src/transport/tcp/stats.rs index 62340e4f..be8e73d7 100644 --- a/src/transport/tcp/stats.rs +++ b/src/transport/tcp/stats.rs @@ -21,7 +21,14 @@ pub struct TcpStats { pub connections_established: AtomicU64, pub connections_accepted: AtomicU64, pub connections_rejected: AtomicU64, + /// Outbound connects, inline or in the background, that did not complete + /// within `connect_timeout_ms`. pub connect_timeouts: AtomicU64, + /// Outbound connects, inline or in the background, that failed before + /// the timeout. Despite the name this counts every such failure, not + /// only `ECONNREFUSED`: an unreachable host or network, a reset or any + /// other connect error lands here too. The debug log "TCP connect + /// failed" carries the OS error for each one. pub connect_refused: AtomicU64, /// Inbound/outbound connection-pool occupancy. Inbound drives the /// `max_inbound_connections` admission check. @@ -95,7 +102,8 @@ impl TcpStats { self.connect_timeouts.fetch_add(1, Ordering::Relaxed); } - /// Record a connection refused. + /// Record an outbound connect that failed before its timeout, for any + /// reason (see `connect_refused`). pub fn record_connect_refused(&self) { self.connect_refused.fetch_add(1, Ordering::Relaxed); } diff --git a/src/transport/tor/mod.rs b/src/transport/tor/mod.rs index 7b06578e..bd8d051f 100644 --- a/src/transport/tor/mod.rs +++ b/src/transport/tor/mod.rs @@ -31,7 +31,7 @@ use super::{ use crate::config::TorConfig; use crate::transport::socks5::{ ConnectingEntry, ConnectingPool, DialError, ProxiedConnection, ProxiedPool, SEND_QUEUE_DEPTH, - Socks5Auth, Socks5Dialer, SocksTarget, poll_connecting, proxied_receive_loop, + Socks5Auth, Socks5Dialer, SocksTarget, existing_sender, poll_connecting, proxied_receive_loop, proxied_send_loop, }; use crate::transport::stream::{ConnId, WRITER_DRAIN_TIMEOUT, drain_writer, next_conn_id}; @@ -654,16 +654,7 @@ impl TorTransport { if !self.state.is_operational() { return Err(TransportError::NotStarted); } - - // Pre-send MTU check - let mtu = self.config.mtu() as usize; - if data.len() > mtu { - self.stats.record_mtu_exceeded(); - return Err(TransportError::MtuExceeded { - packet_size: data.len(), - mtu: self.config.mtu(), - }); - } + self.check_mtu(data)?; // Get or create connection let send_tx = { @@ -679,6 +670,54 @@ impl TorTransport { } }; + self.enqueue(addr, &send_tx, data) + } + + /// Send a packet only over a connection that already exists. + /// + /// Uses the pooled connection for `addr`, or one a finished background + /// connect has produced, which it moves into the pool. Never dials: with + /// neither, it fails at once with [`TransportError::NotConnected`]. Like + /// `send_async`, it only queues the frame for the connection's writer + /// task and never awaits the wire. + pub async fn send_existing( + &self, + addr: &TransportAddr, + data: &[u8], + ) -> Result { + if !self.state.is_operational() { + return Err(TransportError::NotStarted); + } + self.check_mtu(data)?; + let send_tx = existing_sender(&self.pool, &self.connecting, addr, |stream, mtu| { + let conn = self.outbound_connection(addr, stream, mtu); + self.record_promoted(addr); + conn + }) + .await + .ok_or(TransportError::NotConnected)?; + self.enqueue(addr, &send_tx, data) + } + + /// Reject a packet larger than the transport MTU before queueing it. + fn check_mtu(&self, data: &[u8]) -> Result<(), TransportError> { + if data.len() > self.config.mtu() as usize { + self.stats.record_mtu_exceeded(); + return Err(TransportError::MtuExceeded { + packet_size: data.len(), + mtu: self.config.mtu(), + }); + } + Ok(()) + } + + /// Queue one packet for the writer task of the connection to `addr`. + fn enqueue( + &self, + addr: &TransportAddr, + send_tx: &tokio::sync::mpsc::Sender>, + data: &[u8], + ) -> Result { // 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 @@ -924,6 +963,7 @@ impl TorTransport { return Err(TransportError::ConnectionRefused); } Err(DialError::Timeout) => { + stats.record_connect_timeout(); debug!( transport_id = %transport_id, remote_addr = %remote_addr, @@ -964,6 +1004,33 @@ impl TorTransport { /// Splits the stream, spawns a receive loop, and inserts into the pool. /// Called from `connection_state_sync()` when a background task completes. fn promote_connection(&self, addr: &TransportAddr, stream: TcpStream, mtu: u16) { + let conn = self.outbound_connection(addr, stream, mtu); + + // Use try_lock since we're in a sync context and the pool + // should be available (connection_state_sync already checked it) + if let Ok(mut pool) = self.pool.try_lock() { + pool.insert(addr.clone(), conn); + self.record_promoted(addr); + } 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, + "Failed to promote Tor connection (pool locked)" + ); + } + } + + /// Build the pool entry for a finished background connect: split the + /// stream and spawn its receive loop and writer task. + fn outbound_connection( + &self, + addr: &TransportAddr, + stream: TcpStream, + mtu: u16, + ) -> ProxiedConnection { let (read_half, write_half) = stream.into_split(); let transport_id = self.transport_id; @@ -1008,7 +1075,7 @@ impl TorTransport { }, )); - let conn = ProxiedConnection { + ProxiedConnection { send_tx, send_task, recv_task, @@ -1016,31 +1083,20 @@ impl TorTransport { established_at: Instant::now(), meta: Direction::Outbound, id, - }; - - // Use try_lock since we're in a sync context and the pool - // should be available (connection_state_sync already checked it) - if let Ok(mut pool) = self.pool.try_lock() { - pool.insert(addr.clone(), conn); - self.stats.record_connection_established(); - self.stats.record_pool_outbound_added(); - debug!( - transport_id = %self.transport_id, - remote_addr = %addr, - "Tor connection established (background connect)" - ); - } 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, - "Failed to promote Tor connection (pool locked)" - ); } } + /// Count and log a background connection that has entered the pool. + fn record_promoted(&self, addr: &TransportAddr) { + self.stats.record_connection_established(); + self.stats.record_pool_outbound_added(); + debug!( + transport_id = %self.transport_id, + remote_addr = %addr, + "Tor connection established (background connect)" + ); + } + /// Close a specific connection asynchronously. /// /// Aborts the receive task and lets the writer finish the frames already @@ -1391,6 +1447,7 @@ fn validate_host_port(addr: &str, field_name: &str) -> Result<(), TransportError #[cfg(test)] mod tests { use super::*; + use crate::testutil::wait_until; use crate::transport::packet_channel; use tokio::io::AsyncWriteExt; @@ -2050,20 +2107,6 @@ mod tests { // Inbound first-frame deadline (onion listener) // ======================================================================== - /// Poll `f` every 10ms until it holds or `limit` elapses. - async fn wait_until bool>(mut f: F, limit: Duration) -> bool { - let deadline = Instant::now() + limit; - loop { - if f() { - return true; - } - if Instant::now() >= deadline { - return false; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - } - /// Drives `tor_accept_loop` directly: the only production path to it is /// `start_directory_mode`, which needs a Tor-managed hostname file and a /// running daemon, so it is not reachable from a unit test. @@ -2757,4 +2800,204 @@ mod tests { tor.stop_async().await.unwrap(); dest.stop_async().await.unwrap(); } + + /// A destination TCP transport behind a mock SOCKS5 proxy, and a + /// started Tor transport pointed at the proxy. + async fn tor_behind_mock_proxy() -> ( + TorTransport, + TcpTransport, + crate::transport::PacketRx, + 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 (tx, _rx) = packet_channel(32); + let config = TorConfig { + socks5_addr: Some(proxy_addr.to_string()), + ..Default::default() + }; + let mut t = TorTransport::new(TransportId::new(200), None, config, tx); + t.start_async().await.unwrap(); + let target = TransportAddr::from_string(&dest_addr.to_string()); + (t, dest, dest_rx, target) + } + + /// With no pooled connection and no connect under way, `send_existing` + /// fails with `NotConnected` and opens nothing. + #[tokio::test] + async fn send_existing_without_connection_fails_fast_and_dials_nothing() { + let (mut t, mut dest, _dest_rx, target) = tor_behind_mock_proxy().await; + + let result = t.send_existing(&target, &build_msg1_frame()).await; + + assert!( + matches!(result, Err(TransportError::NotConnected)), + "expected NotConnected, got {result:?}" + ); + assert!( + t.connecting.lock().await.is_empty(), + "a connect was started" + ); + assert_eq!(t.stats().snapshot().connections_established, 0); + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!( + dest.stats().snapshot().connections_accepted, + 0, + "the destination saw a connection" + ); + + t.stop_async().await.unwrap(); + dest.stop_async().await.unwrap(); + } + + /// A background connect that has finished is moved into the pool and + /// carries the send, with no second connection opened. + #[tokio::test] + async fn send_existing_promotes_a_finished_background_connect_and_sends_on_it() { + let (mut t, mut dest, mut dest_rx, target) = tor_behind_mock_proxy().await; + + t.connect_async(&target).await.unwrap(); + let mut waited = 0; + while !t + .connecting + .try_lock() + .is_ok_and(|c| c.get(&target).is_some_and(|e| e.task.is_finished())) + { + assert!(waited < 150, "background connect never finished"); + tokio::time::sleep(Duration::from_millis(20)).await; + waited += 1; + } + let frame = build_msg1_frame(); + t.send_existing(&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!( + t.connecting.lock().await.is_empty(), + "the finished connect was left in the connecting map" + ); + assert_eq!(t.stats().snapshot().connections_established, 1); + assert_eq!( + dest.stats().snapshot().connections_accepted, + 1, + "the send used the background connection, not a new one" + ); + + t.stop_async().await.unwrap(); + dest.stop_async().await.unwrap(); + } + + /// When the background connect fails because the SOCKS5 proxy is + /// unreachable, `send_existing` fails with `NotConnected`, drops the + /// failed connect, and `connect_async` can start a new attempt. + #[tokio::test] + async fn send_existing_after_a_failed_background_connect_drops_it_so_a_new_one_can_start() { + // A proxy port nothing listens on: every connect through it fails. + let closed = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let proxy_addr = closed.local_addr().unwrap(); + drop(closed); + let (tx, _rx) = packet_channel(32); + let config = TorConfig { + socks5_addr: Some(proxy_addr.to_string()), + ..Default::default() + }; + let mut t = TorTransport::new(TransportId::new(200), None, config, tx); + t.start_async().await.unwrap(); + let target = TransportAddr::from_string("127.0.0.1:9"); + + t.connect_async(&target).await.unwrap(); + let mut waited = 0; + while !t + .connecting + .try_lock() + .is_ok_and(|c| c.get(&target).is_some_and(|e| e.task.is_finished())) + { + assert!(waited < 150, "background connect never finished"); + tokio::time::sleep(Duration::from_millis(20)).await; + waited += 1; + } + let result = t.send_existing(&target, &build_msg1_frame()).await; + + assert!( + matches!(result, Err(TransportError::NotConnected)), + "expected NotConnected, got {result:?}" + ); + assert!( + t.connecting.lock().await.get(&target).is_none(), + "the failed connect was left in the connecting map" + ); + assert!(t.pool.lock().await.is_empty()); + t.connect_async(&target).await.unwrap(); + assert!( + t.connecting.lock().await.get(&target).is_some(), + "connect_async did not start a new attempt" + ); + + t.stop_async().await.unwrap(); + } + + /// Wait until the background connect to `target` has finished, leaving + /// it in the connecting map. + async fn wait_background_finished(t: &TorTransport, target: &TransportAddr) { + let finished = wait_until( + || { + t.connecting + .try_lock() + .is_ok_and(|c| c.get(target).is_some_and(|e| e.task.is_finished())) + }, + Duration::from_secs(3), + ) + .await; + assert!(finished, "background connect to {target} never finished"); + } + + /// A background connect through a proxy that accepts the TCP connection + /// but never answers the SOCKS5 greeting times out, and is counted in + /// `connect_timeouts` as an inline one is. + #[tokio::test] + async fn background_connect_timeout_is_counted() { + // Bound and listening, never accepted: the kernel completes the TCP + // handshake and nothing ever replies. + let silent = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let (tx, _rx) = packet_channel(32); + let config = TorConfig { + socks5_addr: Some(silent.local_addr().unwrap().to_string()), + connect_timeout_ms: Some(200), + ..Default::default() + }; + let mut t = TorTransport::new(TransportId::new(200), None, config, tx); + t.start_async().await.unwrap(); + let target = TransportAddr::from_string("127.0.0.1:9"); + + t.connect_async(&target).await.unwrap(); + wait_background_finished(&t, &target).await; + + let stats = t.stats().snapshot(); + assert_eq!(stats.connect_timeouts, 1, "the timeout was not counted"); + assert_eq!(stats.connect_refused, 0); + assert_eq!(stats.socks5_errors, 0); + let result = t.send_existing(&target, &build_msg1_frame()).await; + assert!(matches!(result, Err(TransportError::NotConnected))); + assert_eq!( + t.stats().snapshot().connect_timeouts, + 1, + "taking the result counted the timeout again" + ); + + t.stop_async().await.unwrap(); + } } diff --git a/src/transport/udp/io/connected/drain.rs b/src/transport/udp/io/connected/drain.rs index de276509..04bbfdf8 100644 --- a/src/transport/udp/io/connected/drain.rs +++ b/src/transport/udp/io/connected/drain.rs @@ -12,7 +12,9 @@ //! `recvmsg_x(2)` on Darwin), push each packet into //! the existing `packet_tx` (the same channel that the wildcard listen //! socket feeds), and exit cleanly when the parent signals shutdown -//! via a self-pipe. +//! via a self-pipe. Each datagram is counted in the transport's +//! `UdpStats`, as the wildcard socket counts its own, so the transport's +//! receive counters cover every datagram it receives. //! //! Future: when the full data-plane shard lands, this per-peer thread //! becomes a `epoll_wait` arm inside the shard's event loop instead @@ -20,6 +22,7 @@ //! useful in either shape; only the wakeup mechanism differs. use super::socket::ConnectedPeerSocket; +use crate::transport::udp::UdpStats; use crate::transport::{PacketTx, ReceivedPacket, TransportAddr, TransportId}; use std::io; use std::net::SocketAddr; @@ -53,12 +56,14 @@ impl PeerRecvDrain { /// kernel fd alive while it's running. When this handle drops, /// the stop pipe fires; the thread exits; its `Arc` releases. /// If the parent also releases its `Arc`, the socket's `Drop` - /// closes the kernel fd. + /// closes the kernel fd. `stats` is the owning UDP transport's, so + /// datagrams read here are counted with the rest of its traffic. pub fn spawn( socket: Arc, transport_id: TransportId, peer_addr: SocketAddr, packet_tx: PacketTx, + stats: Arc, ) -> io::Result { // Self-pipe for shutdown signaling. The drain thread polls // (socket_fd | pipe_rx) so a write to pipe_tx wakes it. @@ -76,6 +81,7 @@ impl PeerRecvDrain { transport_id, peer_addr, packet_tx, + stats, pipe_rx, stop_clone, ); @@ -137,6 +143,7 @@ fn drain_loop( transport_id: TransportId, peer_addr: SocketAddr, packet_tx: PacketTx, + stats: Arc, stop_pipe_rx: RawFd, stop: Arc, ) { @@ -231,6 +238,7 @@ fn drain_loop( if len == 0 { continue; } + stats.record_recv(len); // Move the filled buffer out, refill the slot with a // fresh one. Same zero-copy pattern the wildcard listen // socket uses (see `transport/udp/mod.rs::run_receive_loop`). @@ -452,7 +460,8 @@ mod tests { /// End-to-end: open a ConnectedPeerSocket, spawn a drain thread /// on it, send packets at it from a remote, verify they land in - /// the packet_tx mpsc with the correct transport_id + peer_addr. + /// the packet_tx mpsc with the correct transport_id + peer_addr, + /// and that each is counted in the stats the drain was given. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn drain_delivers_packets_to_packet_tx() { // Peer (remote) — sends packets at our connected socket. @@ -499,8 +508,10 @@ mod tests { }; // Spawn the drain. - let _drain = PeerRecvDrain::spawn(socket.clone(), transport_id, peer_addr, tx) - .expect("PeerRecvDrain::spawn"); + let stats = Arc::new(UdpStats::new()); + let _drain = + PeerRecvDrain::spawn(socket.clone(), transport_id, peer_addr, tx, stats.clone()) + .expect("PeerRecvDrain::spawn"); // Send a couple of packets from the peer to our socket. for i in 0u8..5 { @@ -518,6 +529,12 @@ mod tests { assert_eq!(pkt.data.len(), 4); assert_eq!(pkt.data[0], i, "packet {i} payload mismatch"); } + let counted = stats.snapshot(); + assert_eq!( + counted.packets_recv, 5, + "the drain did not count what it delivered" + ); + assert_eq!(counted.bytes_recv, 20); // Drop the drain handle — should stop the thread within one // poll iteration. } diff --git a/src/transport/udp/mod.rs b/src/transport/udp/mod.rs index 9ea1a3f8..b5131ce7 100644 --- a/src/transport/udp/mod.rs +++ b/src/transport/udp/mod.rs @@ -14,7 +14,7 @@ use super::resolve_socket_addr; use crate::config::UdpConfig; use crate::nostr::is_punch_packet; use io::{AsyncUdpSocket, UdpRawSocket}; -use stats::UdpStats; +pub(crate) use stats::UdpStats; use std::collections::HashMap; use std::net::SocketAddr; use std::sync::{Arc, Mutex as StdMutex}; diff --git a/src/transport/udp/stats.rs b/src/transport/udp/stats.rs index a0bb79c6..3c2afdc1 100644 --- a/src/transport/udp/stats.rs +++ b/src/transport/udp/stats.rs @@ -40,6 +40,21 @@ impl UdpStats { self.bytes_sent.fetch_add(bytes as u64, Ordering::Relaxed); } + /// Record `packets` datagrams totalling `bytes` handed to the kernel in + /// one batched send (`sendmmsg(2)` or UDP GSO), counted as that many + /// separate sends. + pub fn record_sends(&self, packets: u64, bytes: u64) { + self.packets_sent.fetch_add(packets, Ordering::Relaxed); + self.bytes_sent.fetch_add(bytes, Ordering::Relaxed); + } + + /// Record `packets` datagrams a batched send did not deliver to the + /// kernel, whether after an error or because the kernel took none of + /// them, counted as that many send errors. + pub fn record_unsent(&self, packets: u64) { + self.send_errors.fetch_add(packets, Ordering::Relaxed); + } + /// Record a successful receive. pub fn record_recv(&self, bytes: usize) { self.packets_recv.fetch_add(1, Ordering::Relaxed); @@ -61,11 +76,14 @@ impl UdpStats { self.mtu_exceeded.fetch_add(1, Ordering::Relaxed); } - /// Update kernel drop count from SO_MEMINFO. + /// Update the kernel drop count. /// - /// Not yet wired up — requires `getsockopt(SO_MEMINFO)` on the raw fd - /// (via socket2 or libc) to read `SK_MEMINFO_DROPS`. Linux-only. - /// Until implemented, this counter will always be zero. + /// The value is the cumulative `SO_RXQ_OVFL` count the kernel attaches + /// to datagrams read from the wildcard listen socket, so it covers that + /// socket only, not the per-peer connected sockets. It is updated only + /// as that socket is read, so it stops moving whenever the listen socket + /// is not being read. It is wired only on Linux; elsewhere (Darwin, + /// Windows, other Unix) it stays zero. pub fn set_kernel_drops(&self, drops: u64) { self.kernel_drops.store(drops, Ordering::Relaxed); }