diff --git a/crates/irtt-server/AGENTS.md b/crates/irtt-server/AGENTS.md index 1c64007..752ba20 100644 --- a/crates/irtt-server/AGENTS.md +++ b/crates/irtt-server/AGENTS.md @@ -58,6 +58,58 @@ material does not establish a semantic, state the ambiguity instead of guessing. no product requirement for any of them, and `ServerCore` being public is not a promise of runtime independence. +## One listener and several + +`Server` is the packet/runtime primitive and stays that: one socket, one core, +one sequential loop. `ServerSet` is the service around it and owns no protocol +or session state of its own — it binds listeners, supervises them and shuts them +down. Nothing it does is visible on the wire. + +- **A set of one is an ordinary set.** Singleton sets are normal and supported, + and the server CLI runs *every* invocation through one, single bind included. + There is no bind-count branch to reintroduce: one orchestration path is what + keeps startup, shutdown and failure semantics identical for one listener and + for five, and what keeps the singleton case exercised in normal use. +- **Nothing is shared between listeners but configuration.** Each gets its own + socket, `ServerCore`, session table, tokens, receive/rate/lifetime state, + clock origin and traffic-class socket state. Do not add an + `Arc>`, a process-global token allocator or session + registry, or a shared clock: a token from one listener is an unknown token at + the others, and that is the point rather than a limitation. +- **`ServerConfig` is cloned per listener, so every bound in it is per + listener.** `max_sessions` of 100 across two listeners admits 100 each. Say so + in the rustdoc, the CLI help and the README; do not add a process-wide session + cap here. +- **Construction is all-or-nothing and completes before anything runs.** Bind + every listener first, spawn nothing while iterating, and drop what was bound + if a later one fails. Startup diagnostics belong after the whole set exists, + so a set that never started cannot print that it is listening. +- **One task per listener, and the listener stays sequential inside it.** Do not + spawn per datagram or add a second sender: one sender per socket is what makes + the socket-wide traffic class race-free, and that invariant is unchanged by + there being several sockets. +- **One external shutdown, fanned out internally.** The library knows nothing + about Ctrl-C, signals or atomics. Every listener task is joined before `run` + returns — no detached tasks, no `abort_all()` for ordinary shutdown, and a + `JoinError` is never discarded. +- **A listener failure fails the set.** A listener that errors, or that returns + before shutdown was requested, shuts its siblings down; the group drains and + returns the first meaningful failure. A service configured for IPv4 and IPv6 + must not continue silently as IPv4 only. There is no error aggregation. +- **Same-port coexistence is settled before the bind, and only there.** + `IPV6_V6ONLY` can only be set before binding, so a genuine IPv6 listener + sharing an explicit port with an IPv4 sibling is created through a small + `socket2` helper and handed to `Server::from_socket`. That helper owns nothing + else: wildcard destination metadata, the core and traffic-class state stay + with `Server`. No `SO_REUSEPORT`, no bind-order dependence, no unsafe code, + and standalone `Server::bind` semantics are unchanged. +- **An IPv4-mapped address is an IPv4 listener here too**, exactly as it is for + wildcard handling and the traffic-class option. It is never given + `IPV6_V6ONLY` and never counted as the IPv6 half of a same-port pair. +- Listener sets are fixed at construction. Do not add dynamic add/remove, + hostname or interface expansion, or per-listener configuration without + agreement. + ## Reply source address A reply must leave from the address its request was sent to, or a client reading diff --git a/crates/irtt-server/src/lib.rs b/crates/irtt-server/src/lib.rs index 5d1a68f..1947b72 100644 --- a/crates/irtt-server/src/lib.rs +++ b/crates/irtt-server/src/lib.rs @@ -13,9 +13,24 @@ //! engine stays testable. Each reply it produces is an [`OutboundDatagram`]: //! the packet and the raw traffic class it must be sent with. [`Server`] owns //! one Tokio UDP listener and one core, and provides sequential receive, reply, -//! scheduled expiry maintenance and caller-controlled shutdown. The crate is -//! intentionally Tokio-native; there is no blocking or alternate-runtime -//! counterpart and no transport abstraction. +//! scheduled expiry maintenance and caller-controlled shutdown. [`ServerSet`] +//! owns one or more `Server`s as a single service: it binds them all or none, +//! runs each in its own task, fans one shutdown out to them and fails the group +//! if any listener does. The crate is intentionally Tokio-native; there is no +//! blocking or alternate-runtime counterpart and no transport abstraction. +//! +//! # One listener or several +//! +//! `Server` is the primitive and stays directly usable. `ServerSet` is the +//! service around it, and a set of one is an ordinary set rather than a special +//! case — the server CLI runs every invocation, single-bind included, through +//! one. +//! +//! Listeners in a set share nothing but the configuration they were cloned +//! from. Each has its own socket, core, session table and tokens, so a token +//! issued by one is unknown at the others, and every bound in +//! [`ServerConfig`] — `max_sessions` included — applies **per listener**, not +//! per process. //! //! # Rejection is silence //! @@ -60,8 +75,8 @@ //! per-session rate limiting, idle expiry, the maximum-duration //! server-initiated close, the negotiated per-session reply traffic class, //! server fill, the optional timestamp-allowance and DSCP capability -//! restrictions, wildcard reply-source selection and Tokio UDP orchestration. -//! Running several listeners in one process remains a separate slice. +//! restrictions, wildcard reply-source selection, Tokio UDP orchestration and +//! multi-listener supervision through [`ServerSet`]. //! //! # Capability restrictions //! @@ -99,6 +114,7 @@ mod fill; mod negotiate; mod runtime; mod session; +mod set; mod socket_io; mod socket_options; mod token; @@ -113,3 +129,4 @@ pub use config::{ pub use core::{OutboundDatagram, ServerCore}; pub use error::ServerError; pub use runtime::{Server, ServerRuntimeError}; +pub use set::{ServerSet, ServerSetError}; diff --git a/crates/irtt-server/src/set.rs b/crates/irtt-server/src/set.rs new file mode 100644 index 0000000..637a589 --- /dev/null +++ b/crates/irtt-server/src/set.rs @@ -0,0 +1,412 @@ +//! Service-level supervision of one or more independent [`Server`] listeners. +//! +//! [`Server`] is the one-listener primitive: one socket, one core, one session +//! namespace, one sequential receive/send loop. A host that must answer on more +//! than one address needs several of them, and something has to own their +//! construction, their shutdown and each other's failures. That owner is +//! [`ServerSet`], and it is deliberately not a protocol object: it holds no +//! session state, no tokens and no clock, and nothing it does is visible on the +//! wire. + +use std::{ + collections::HashMap, + future::Future, + io, + net::{IpAddr, SocketAddr}, +}; + +use socket2::{Domain, Protocol, SockAddr, Socket, Type}; +use thiserror::Error; +use tokio::{ + net::UdpSocket, + sync::watch, + task::{Id, JoinError, JoinSet}, +}; + +use crate::{Server, ServerConfig, ServerRuntimeError}; + +/// One or more independent UDP listeners run as a single service. +/// +/// Every listener is a complete [`Server`]: its own socket, its own +/// [`ServerCore`](crate::ServerCore), its own session table and its own tokens. +/// Nothing is shared between them but the configuration they were built from +/// and the supervision described below. A token issued by one listener is an +/// unknown token at any other, which is what makes the listeners genuinely +/// independent rather than one server wearing several addresses. +/// +/// # Configuration is cloned, so bounds are per listener +/// +/// The [`ServerConfig`] passed to [`ServerSet::bind`] is cloned into each +/// listener, so every resource bound in it applies **per listener** and not to +/// the process. A set of two listeners built from a configuration with +/// `max_sessions` of 100 admits up to 100 sessions on each, 200 in total. There +/// is no process-wide session cap. +/// +/// # A set of one is an ordinary set +/// +/// One listener is not a special case and gets no shortcut: it is bound, +/// supervised and shut down exactly as a member of a larger set is. The server +/// CLI runs every invocation through a `ServerSet` for that reason, so the +/// single-listener case exercises this path continuously. +/// +/// # Supervision +/// +/// [`run`](ServerSet::run) takes one shutdown future and fans it out to every +/// listener, then waits for all of them. A listener that fails, or that stops on +/// its own before shutdown was asked for, shuts its siblings down and fails the +/// whole set: a service configured for IPv4 and IPv6 must not quietly continue +/// as IPv4 only. Every listener task is joined before `run` returns, so no task +/// outlives it. +#[derive(Debug)] +pub struct ServerSet { + servers: Vec, + /// The addresses the listeners actually bound, in requested order and + /// positionally matching `servers`. + local_addrs: Vec, +} + +impl ServerSet { + /// Binds every requested listener, or none of them. + /// + /// Requested order is preserved, and each listener is bound before any is + /// served: a later failure drops the sockets already opened and starts + /// nothing. `config` is cloned into every listener, so its bounds are per + /// listener. + /// + /// A port of `0` is resolved per listener — two such requests select two + /// unrelated ephemeral ports, not one shared one. Ask + /// [`local_addrs`](ServerSet::local_addrs) for what was actually bound. + /// + /// # Same-port IPv4 and IPv6 + /// + /// Requesting a genuine IPv6 listener and an IPv4 one on the same explicit + /// port — `0.0.0.0:2112` with `[::]:2112`, say — is a normal configuration, + /// but on hosts whose IPv6 sockets are dual-stack by default the IPv6 bind + /// would claim the IPv4 port too and the second bind would fail. Such an + /// IPv6 listener is therefore created `IPV6_V6ONLY` before it is bound, so + /// the two coexist and each serves its own family. Bind order is not relied + /// on, and a listener requested on its own is bound exactly as + /// [`Server::bind`] would. + /// + /// An IPv4-mapped address such as `[::ffff:0.0.0.0]` is an IPv4 listener + /// here, not an IPv6 one, matching how the runtime already reads it. + /// + /// # Errors + /// + /// Returns [`ServerSetError::NoListeners`] if `addrs` is empty — a set + /// serving nothing has no meaning — and + /// [`ServerSetError::ListenerSetup`] naming the requested address if any + /// listener could not be bound or prepared. + pub async fn bind(addrs: I, config: ServerConfig) -> Result + where + I: IntoIterator, + { + let requested: Vec = addrs.into_iter().collect(); + if requested.is_empty() { + return Err(ServerSetError::NoListeners); + } + + let mut servers = Vec::with_capacity(requested.len()); + let mut local_addrs = Vec::with_capacity(requested.len()); + for &addr in &requested { + let server = bind_listener(addr, config.clone(), needs_ipv6_only(addr, &requested)) + .await + .map_err(|source| ServerSetError::ListenerSetup { addr, source })?; + let local_addr = server + .local_addr() + .map_err(|source| ServerSetError::ListenerSetup { addr, source })?; + servers.push(server); + local_addrs.push(local_addr); + } + + Ok(Self { + servers, + local_addrs, + }) + } + + /// The addresses the listeners actually bound, in requested order. + /// + /// These are the resolved endpoints, so a requested port of `0` appears + /// here as the port the host selected. + #[must_use] + pub fn local_addrs(&self) -> &[SocketAddr] { + &self.local_addrs + } + + /// Serves every listener until `shutdown` completes or one of them fails. + /// + /// Each listener runs in its own Tokio task and stays internally sequential + /// there, so the one-sender-per-socket ownership the reply traffic class + /// depends on is unchanged. The single `shutdown` future is fanned out + /// internally; nothing about how the caller produces it — a signal, a + /// channel, a timer — is this crate's business. + /// + /// The set consumes itself because each listener is moved into its task. A + /// caller that wants to bind and serve again binds a new set; the + /// lower-level [`Server`] keeps its reusable `&mut self` loop. + /// + /// # Errors + /// + /// A listener that fails, that returns before shutdown was requested, or + /// whose task did not complete shuts down its siblings; every remaining + /// listener is then drained and the first such failure is returned. An + /// external shutdown that completes first still returns a listener's error + /// if one appears while draining. `Ok(())` means every listener stopped + /// cleanly after being asked to. + pub async fn run(self, shutdown: F) -> Result<(), ServerSetError> + where + F: Future, + { + let (shutdown_tx, _) = watch::channel(false); + let mut listeners = JoinSet::new(); + // A panicking task cannot report which listener it was, so the mapping + // is kept here rather than recovered from the task's return value. + let mut listener_addrs: HashMap = HashMap::new(); + + for (mut server, addr) in self.servers.into_iter().zip(self.local_addrs) { + let mut shutdown_rx = shutdown_tx.subscribe(); + let handle = listeners.spawn(async move { + let outcome = server + .run(async move { + // A closed channel means the set is being torn down and + // is as good a reason to stop as an explicit request. + let _ = shutdown_rx.wait_for(|requested| *requested).await; + }) + .await; + (addr, outcome) + }); + listener_addrs.insert(handle.id(), addr); + } + + tokio::pin!(shutdown); + let mut shutting_down = false; + let mut failure: Option = None; + + loop { + tokio::select! { + () = &mut shutdown, if !shutting_down => { + shutting_down = true; + let _ = shutdown_tx.send(true); + } + joined = listeners.join_next() => { + let Some(joined) = joined else { break }; + let listener_failure = match joined { + // The only clean exit: it was asked to stop, and did. + Ok((_, Ok(()))) if shutting_down => None, + Ok((addr, Ok(()))) => Some(ServerSetError::ListenerStopped { addr }), + Ok((addr, Err(source))) => { + Some(ServerSetError::ListenerRun { addr, source }) + } + Err(source) => Some(ServerSetError::ListenerTask { + addr: listener_addrs.get(&source.id()).copied(), + source, + }), + }; + if let Some(listener_failure) = listener_failure { + failure.get_or_insert(listener_failure); + if !shutting_down { + shutting_down = true; + let _ = shutdown_tx.send(true); + } + } + } + } + } + + failure.map_or(Ok(()), Err) + } +} + +/// Binds one listener, pre-configuring the socket where the bind itself depends +/// on it. +/// +/// The ordinary path is [`Server::bind`] unchanged. Only an IPv6 listener that +/// has to leave an IPv4 sibling's port alone needs its own socket, because +/// `IPV6_V6ONLY` can only be set before the bind it affects. +async fn bind_listener( + addr: SocketAddr, + config: ServerConfig, + ipv6_only: bool, +) -> Result { + if !ipv6_only { + return Server::bind(addr, config).await; + } + + let socket = + bind_ipv6_only(addr).map_err(|source| ServerRuntimeError::Bind { addr, source })?; + Server::from_socket(socket, config) +} + +/// Creates an IPv6-only UDP socket and binds it. +/// +/// Everything past the bind — destination metadata for a wildcard listener, +/// the core, traffic-class state — stays with [`Server::from_socket`]. This +/// helper exists solely for the one option that must precede the bind. +fn bind_ipv6_only(addr: SocketAddr) -> io::Result { + let socket = Socket::new(Domain::IPV6, Type::DGRAM, Some(Protocol::UDP))?; + socket.set_only_v6(true)?; + socket.set_nonblocking(true)?; + socket.bind(&SockAddr::from(addr))?; + UdpSocket::from_std(socket.into()) +} + +/// Whether this listener must be bound IPv6-only to leave an IPv4 sibling's +/// port available. +/// +/// Only an explicit port can collide: two ephemeral requests are resolved +/// independently and never name the same port to begin with. +fn needs_ipv6_only(addr: SocketAddr, requested: &[SocketAddr]) -> bool { + is_genuine_ipv6(addr) + && addr.port() != 0 + && requested + .iter() + .any(|other| other.port() == addr.port() && serves_ipv4(*other)) +} + +/// Whether a bind address names an IPv6 endpoint in its own right. +/// +/// An IPv4-mapped address is not one: `[::ffff:0.0.0.0]` is an IPv4 wildcard +/// bind that the kernel reports back in IPv6 clothing, and the runtime already +/// reads it as IPv4 for wildcard handling and for the traffic-class option. It +/// must not be given `IPV6_V6ONLY` — that would refuse the very traffic it +/// exists to serve — nor counted as the IPv6 half of a same-port pair. +fn is_genuine_ipv6(addr: SocketAddr) -> bool { + addr.is_ipv6() && !serves_ipv4(addr) +} + +/// Whether a bind address will carry IPv4 traffic, in either spelling. +fn serves_ipv4(addr: SocketAddr) -> bool { + match addr.ip() { + IpAddr::V4(_) => true, + IpAddr::V6(address) => address.to_ipv4_mapped().is_some(), + } +} + +/// Failure to construct or supervise a [`ServerSet`]. +#[derive(Debug, Error)] +pub enum ServerSetError { + /// No listener address was requested. + #[error("a server set needs at least one listener address")] + NoListeners, + /// A requested listener could not be bound or prepared. Nothing was served: + /// any listener already bound for this set has been released. + #[error("could not set up listener {addr}: {source}")] + ListenerSetup { + addr: SocketAddr, + #[source] + source: ServerRuntimeError, + }, + /// A running listener failed. Its siblings were shut down. + #[error("listener {addr} failed: {source}")] + ListenerRun { + addr: SocketAddr, + #[source] + source: ServerRuntimeError, + }, + /// A listener returned before the set was asked to shut down. A set does + /// not continue with fewer listeners than it was configured for. + #[error("listener {addr} stopped before the server set was asked to shut down")] + ListenerStopped { addr: SocketAddr }, + /// A listener task did not complete — it panicked or was aborted. The + /// address is absent only where the task could not be identified. + #[error( + "listener task{} did not complete: {source}", + .addr.map(|addr| format!(" for {addr}")).unwrap_or_default() + )] + ListenerTask { + addr: Option, + #[source] + source: JoinError, + }, +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Which listener of a requested set has to be bound IPv6-only. + /// + /// The interesting rows are the mapped ones. `[::ffff:0.0.0.0]` is an IPv4 + /// wildcard bind wearing an IPv6 address, so it is neither a candidate for + /// `IPV6_V6ONLY` — which would refuse the traffic it exists to serve — nor + /// the IPv6 half of a same-port pair. A genuine IPv6 address paired with it + /// still is. + #[test] + fn only_a_genuine_ipv6_listener_sharing_an_explicit_port_is_bound_ipv6_only() { + for (addr, requested, ipv6_only, why) in [ + ( + "[::]:2112", + vec!["0.0.0.0:2112", "[::]:2112"], + true, + "the wildcard pair is the reason this exists", + ), + ( + "[::1]:2112", + vec!["127.0.0.1:2112", "[::1]:2112"], + true, + "and so is the explicit loopback pair", + ), + ( + "0.0.0.0:2112", + vec!["0.0.0.0:2112", "[::]:2112"], + false, + "the IPv4 half is bound exactly as it always was", + ), + ( + "[::]:2112", + vec!["[::]:2112"], + false, + "an IPv6 listener alone keeps ordinary bind semantics", + ), + ( + "[::]:2112", + vec!["0.0.0.0:2113", "[::]:2112"], + false, + "different ports cannot collide", + ), + ( + "[::]:0", + vec!["0.0.0.0:0", "[::]:0"], + false, + "two ephemeral requests resolve independently", + ), + ( + "[::ffff:0.0.0.0]:2112", + vec!["0.0.0.0:2112", "[::ffff:0.0.0.0]:2112"], + false, + "a mapped wildcard is an IPv4 listener, not an IPv6 one", + ), + ( + "[::]:2112", + vec!["[::ffff:0.0.0.0]:2112", "[::]:2112"], + true, + "and it is the IPv4 half a genuine IPv6 listener must avoid", + ), + ( + "[::ffff:127.0.0.1]:2112", + vec!["[::ffff:127.0.0.1]:2112", "[::1]:2112"], + false, + "a mapped host address is explicit IPv4 too", + ), + ] { + let requested: Vec = requested + .iter() + .map(|addr| addr.parse().expect("a test address must parse")) + .collect(); + assert_eq!( + needs_ipv6_only(addr.parse().unwrap(), &requested), + ipv6_only, + "{addr} among {requested:?}: {why}" + ); + } + } + + #[tokio::test(flavor = "current_thread")] + async fn a_set_serving_nothing_is_refused() { + assert!(matches!( + ServerSet::bind([], ServerConfig::default()).await, + Err(ServerSetError::NoListeners) + )); + } +} diff --git a/crates/irtt-server/tests/server_set.rs b/crates/irtt-server/tests/server_set.rs new file mode 100644 index 0000000..49aeab0 --- /dev/null +++ b/crates/irtt-server/tests/server_set.rs @@ -0,0 +1,467 @@ +//! One or more listeners run as a single service. +//! +//! Everything here is asserted from the outside, over real sockets, with the +//! production encoders: what the listeners answer, what they refuse, and what +//! survives one shared shutdown. The two properties worth the setup are that a +//! set of listeners is genuinely several independent servers — separate tokens, +//! separate session bounds — and that a set of *one* behaves exactly as a set of +//! several does, because the server CLI runs every invocation through this path. + +use std::{ + future::Future, + io, + net::{Ipv4Addr, Ipv6Addr, SocketAddr, UdpSocket as StdUdpSocket}, + time::Duration, +}; + +use irtt_proto::{ + decode_echo_reply, decode_open_reply, encode_request, Clock, EchoReply, OpenReply, Params, + ReceivedStats, RequestToEncode, StampAt, +}; +use irtt_server::{ServerConfig, ServerSet, ServerSetError}; +use tokio::{net::UdpSocket, sync::oneshot, task::JoinHandle, time::timeout}; + +/// How long a reply is waited for. +/// +/// Rejection is silence, so this doubles as the window a listener that must not +/// answer is listened to: it only has to outlast a reply that would have come, +/// and loopback replies take microseconds. +const REPLY_TIMEOUT: Duration = Duration::from_secs(1); +/// Every test is bounded, so a hang is a failure with a message rather than a +/// suite that never finishes. +const TEST_TIMEOUT: Duration = Duration::from_secs(10); + +#[tokio::test(flavor = "current_thread")] +async fn a_single_listener_set_binds_serves_and_stops() { + // The CLI's ordinary case, and the reason it has no second code path: one + // requested address, one resolved listener, one whole session over it. + bounded(async { + let set = ServerSet::bind([loopback(0)], unthrottled()).await.unwrap(); + assert_eq!(set.local_addrs().len(), 1); + let addr = set.local_addrs()[0]; + assert_eq!(addr.ip(), Ipv4Addr::LOCALHOST); + assert_ne!(addr.port(), 0, "a requested port of 0 is resolved"); + + let running = spawn(set); + let mut peer = Peer::at(addr).await; + let session = peer.open().await.expect("the listener answers an open"); + let reply = peer.echo(&session, 0).await.expect("and an echo"); + assert_eq!(reply.token, session.token); + assert_eq!(reply.sequence, 0); + peer.close(&session).await; + assert!( + peer.echo(&session, 1).await.is_none(), + "a closed session is an unknown token" + ); + + running.stop().await; + }) + .await; +} + +#[tokio::test(flavor = "current_thread")] +async fn two_listeners_keep_their_order_and_serve_interleaved_traffic() { + bounded(async { + let set = ServerSet::bind([loopback(0), loopback(0)], unthrottled()) + .await + .unwrap(); + let addrs = set.local_addrs().to_vec(); + assert_eq!(addrs.len(), 2); + assert!(addrs.iter().all(|addr| addr.port() != 0)); + assert_ne!( + addrs[0].port(), + addrs[1].port(), + "each ephemeral bind selects its own port" + ); + + let running = spawn(set); + let mut first = Peer::at(addrs[0]).await; + let mut second = Peer::at(addrs[1]).await; + + // Interleaved on purpose: both listener tasks have to be live at the + // same time under one service, not served one after the other. + let first_session = first.open().await.expect("the first listener opens"); + let second_session = second.open().await.expect("the second listener opens"); + for sequence in 0..2 { + assert_eq!( + first + .echo(&first_session, sequence) + .await + .expect("the first listener echoes") + .sequence, + sequence + ); + assert_eq!( + second + .echo(&second_session, sequence) + .await + .expect("the second listener echoes") + .sequence, + sequence + ); + } + + running.stop().await; + }) + .await; +} + +#[tokio::test(flavor = "current_thread")] +async fn a_token_means_nothing_at_another_listener() { + // Session namespaces are per listener because each listener owns its own + // core. This is the externally visible consequence, and it is deliberate: + // there is no process-wide token or session table to make a token portable. + bounded(async { + let set = ServerSet::bind([loopback(0), loopback(0)], unthrottled()) + .await + .unwrap(); + let addrs = set.local_addrs().to_vec(); + let running = spawn(set); + + let mut owner = Peer::at(addrs[0]).await; + let session = owner.open().await.expect("the first listener opens"); + + let mut stranger = Peer::at(addrs[1]).await; + assert!( + stranger.echo(&session, 0).await.is_none(), + "the second listener does not know a token it never issued" + ); + assert!( + owner.echo(&session, 0).await.is_some(), + "and the listener that issued it still does" + ); + + running.stop().await; + }) + .await; +} + +#[tokio::test(flavor = "current_thread")] +async fn the_session_bound_applies_to_each_listener_separately() { + // A cloned configuration is per-listener policy, not a process-wide budget: + // a maximum of one session means one session *each*. + bounded(async { + let set = ServerSet::bind( + [loopback(0), loopback(0)], + unthrottled().with_max_sessions(1), + ) + .await + .unwrap(); + let addrs = set.local_addrs().to_vec(); + let running = spawn(set); + + let mut first = Peer::at(addrs[0]).await; + let mut second = Peer::at(addrs[1]).await; + assert!(first.open().await.is_some()); + assert!( + second.open().await.is_some(), + "the second listener has its own session budget" + ); + + let mut crowd = Peer::at(addrs[0]).await; + assert!( + crowd.open().await.is_none(), + "a full listener refuses a second session silently" + ); + + running.stop().await; + }) + .await; +} + +#[tokio::test(flavor = "current_thread")] +async fn every_listener_serves_the_configured_policy() { + // One representative policy is enough; the negotiation itself is covered + // where it lives. What is under test is that the configuration reached every + // listener, not what it does. + bounded(async { + let config = unthrottled().with_timestamp_allowance(irtt_server::TimestampAllowance::None); + let set = ServerSet::bind([loopback(0), loopback(0)], config) + .await + .unwrap(); + let addrs = set.local_addrs().to_vec(); + let running = spawn(set); + + for addr in addrs { + let mut peer = Peer::at(addr).await; + let session = peer.open().await.expect("the listener opens"); + assert_eq!( + session.params.stamp_at, + StampAt::None, + "{addr} negotiated its own configuration" + ); + } + + running.stop().await; + }) + .await; +} + +#[tokio::test(flavor = "current_thread")] +async fn one_shutdown_stops_every_listener_and_releases_its_socket() { + bounded(async { + let set = ServerSet::bind([loopback(0), loopback(0)], unthrottled()) + .await + .unwrap(); + let addrs = set.local_addrs().to_vec(); + let running = spawn(set); + + // Prove both are actually serving before asking them to stop. + for &addr in &addrs { + assert!(Peer::at(addr).await.open().await.is_some()); + } + running.stop().await; + + for addr in addrs { + StdUdpSocket::bind(addr) + .unwrap_or_else(|error| panic!("{addr} was not released: {error}")); + } + }) + .await; +} + +#[tokio::test(flavor = "current_thread")] +async fn a_listener_that_cannot_bind_leaves_the_whole_set_unbound() { + // All or nothing: the set is built completely before anything is served, so + // a failure part-way through releases what it had and starts no task. + bounded(async { + let occupied = UdpSocket::bind(loopback(0)).await.unwrap(); + let occupied_addr = occupied.local_addr().unwrap(); + let wanted = free_loopback_port(); + let wanted_addr = SocketAddr::from((Ipv4Addr::LOCALHOST, wanted)); + + let error = ServerSet::bind([wanted_addr, occupied_addr], unthrottled()) + .await + .expect_err("the second listener cannot bind an occupied port"); + assert!( + matches!(error, ServerSetError::ListenerSetup { addr, .. } if addr == occupied_addr), + "the error names the listener that failed: {error}" + ); + + StdUdpSocket::bind(wanted_addr) + .expect("the listener bound before the failure was released again"); + }) + .await; +} + +#[tokio::test(flavor = "current_thread")] +async fn an_ipv4_and_a_genuine_ipv6_listener_share_one_port() { + // The configuration this exists for. On a host whose IPv6 sockets are + // dual-stack by default, the IPv6 bind would take the IPv4 port with it and + // the pair could not coexist; the set binds that listener IPv6-only first. + bounded(async { + if !ipv6_loopback_available() { + eprintln!("skipping the same-port test: IPv6 loopback is unavailable"); + return; + } + let port = free_loopback_port(); + let v4 = SocketAddr::from((Ipv4Addr::LOCALHOST, port)); + let v6 = SocketAddr::from((Ipv6Addr::LOCALHOST, port)); + + let set = ServerSet::bind([v4, v6], unthrottled()).await.unwrap(); + assert_eq!(set.local_addrs(), &[v4, v6][..]); + let running = spawn(set); + + for addr in [v4, v6] { + let mut peer = Peer::at(addr).await; + let session = peer.open().await.unwrap_or_else(|| panic!("{addr} opens")); + assert!(peer.echo(&session, 0).await.is_some(), "{addr} echoes"); + } + + running.stop().await; + }) + .await; +} + +/// The wildcard form of the same coexistence, where a wildcard listener is +/// served at all. Reply source selection itself is covered by its own tests; +/// this only asserts that both families bind on one port and answer. +#[cfg(any(target_os = "linux", target_os = "macos", target_os = "freebsd"))] +#[tokio::test(flavor = "current_thread")] +async fn wildcard_listeners_of_both_families_share_one_port() { + bounded(async { + if !ipv6_loopback_available() { + eprintln!("skipping the wildcard same-port test: IPv6 loopback is unavailable"); + return; + } + let port = free_loopback_port(); + let v4: SocketAddr = SocketAddr::from((Ipv4Addr::UNSPECIFIED, port)); + let v6: SocketAddr = SocketAddr::from((Ipv6Addr::UNSPECIFIED, port)); + + let set = ServerSet::bind([v4, v6], unthrottled()).await.unwrap(); + let running = spawn(set); + + for addr in [ + SocketAddr::from((Ipv4Addr::LOCALHOST, port)), + SocketAddr::from((Ipv6Addr::LOCALHOST, port)), + ] { + let mut peer = Peer::at(addr).await; + let session = peer.open().await.unwrap_or_else(|| panic!("{addr} opens")); + assert!(peer.echo(&session, 0).await.is_some(), "{addr} echoes"); + } + + running.stop().await; + }) + .await; +} + +/// A minimal compliant client for one listener. +/// +/// It uses the production encoders and decoders, and reads from an unconnected +/// socket so that a reply from the wrong endpoint would be visible as a failed +/// assertion rather than as silence. +struct Peer { + socket: UdpSocket, + server: SocketAddr, + buffer: Vec, +} + +impl Peer { + async fn at(server: SocketAddr) -> Self { + let bind = if server.is_ipv4() { + SocketAddr::from((Ipv4Addr::LOCALHOST, 0)) + } else { + SocketAddr::from((Ipv6Addr::LOCALHOST, 0)) + }; + Self { + socket: UdpSocket::bind(bind).await.unwrap(), + server, + buffer: vec![0; 4096], + } + } + + /// Opens a session, or reports the silence that means it was refused. + async fn open(&mut self) -> Option { + let request = encode_request( + RequestToEncode::Open { + params: &requested_params(), + no_test: false, + }, + None, + ) + .unwrap(); + let len = self.exchange(&request).await?; + Some(decode_open_reply(&self.buffer[..len], None).unwrap()) + } + + async fn echo(&mut self, session: &OpenReply, sequence: u32) -> Option { + let request = encode_request( + RequestToEncode::Echo { + token: session.token, + sequence, + params: &session.params, + payload: &[], + }, + None, + ) + .unwrap(); + let len = self.exchange(&request).await?; + Some(decode_echo_reply(&self.buffer[..len], &session.params, None).unwrap()) + } + + async fn close(&mut self, session: &OpenReply) { + let request = encode_request( + RequestToEncode::Close { + token: session.token, + }, + None, + ) + .unwrap(); + self.socket.send_to(&request, self.server).await.unwrap(); + } + + /// Sends one request and returns the length of its reply, or `None` if the + /// listener stayed silent. + async fn exchange(&mut self, request: &[u8]) -> Option { + self.socket.send_to(request, self.server).await.unwrap(); + match timeout(REPLY_TIMEOUT, self.socket.recv_from(&mut self.buffer)).await { + Ok(received) => { + let (len, from) = received.unwrap(); + assert_eq!(from, self.server, "a reply came from the wrong endpoint"); + Some(len) + } + Err(_) => None, + } + } +} + +/// The parameters every peer here asks for. Nothing in these tests depends on +/// the values beyond their being a session a server will run. +fn requested_params() -> Params { + Params { + protocol_version: 1, + duration_ns: 10_000_000_000, + interval_ns: 100_000_000, + received_stats: ReceivedStats::Both, + stamp_at: StampAt::Both, + clock: Clock::Both, + ..Params::default() + } +} + +fn unthrottled() -> ServerConfig { + ServerConfig::default().with_min_send_interval(Duration::ZERO) +} + +fn loopback(port: u16) -> SocketAddr { + SocketAddr::from((Ipv4Addr::LOCALHOST, port)) +} + +/// A loopback port nothing currently holds. +fn free_loopback_port() -> u16 { + StdUdpSocket::bind(loopback(0)) + .unwrap() + .local_addr() + .unwrap() + .port() +} + +fn ipv6_loopback_available() -> bool { + match StdUdpSocket::bind(SocketAddr::from((Ipv6Addr::LOCALHOST, 0))) { + Ok(_) => true, + Err(error) + if matches!( + error.kind(), + io::ErrorKind::AddrNotAvailable | io::ErrorKind::Unsupported + ) => + { + false + } + Err(error) => panic!("unexpected IPv6 loopback bind failure: {error}"), + } +} + +struct Running { + shutdown: oneshot::Sender<()>, + task: JoinHandle>, +} + +impl Running { + async fn stop(self) { + self.shutdown.send(()).unwrap(); + self.task + .await + .expect("the supervising task completed") + .expect("every listener stopped cleanly"); + } +} + +fn spawn(set: ServerSet) -> Running { + let (shutdown, shutdown_rx) = oneshot::channel(); + let task = tokio::spawn(async move { + set.run(async { + let _ = shutdown_rx.await; + }) + .await + }); + Running { shutdown, task } +} + +async fn bounded(test: F) +where + F: Future, +{ + timeout(TEST_TIMEOUT, test) + .await + .expect("the test exceeded its bounded runtime"); +}