2026-08-14 09:46:45 +00:00
8 changed files with 1079 additions and 36 deletions
@@ -57,7 +57,7 @@ Tokio must remain optional for `irtt-client`; default builds must remain runtime
 ### Server architecture (`irtt-server`)
-`irtt-server` is a first-class reusable server crate. It contains the deterministic OPEN/ECHO/session core — packet admission, authentication policy, open negotiation, the bounded session table, open reply construction, echo processing with its per-session receive state and timestamps, per-session rate limiting, session lifetime with idle expiry, the maximum-duration server-initiated close, and client-initiated close — plus a reusable Tokio UDP `Server` that runs one sequential `ServerCore` per listener with caller-controlled shutdown and scheduled idle maintenance. Logical expiry remains exact on authenticated, structurally valid requests, while the runtime also reclaims expired sessions once per second when no traffic arrives. Each reply carries the raw traffic class it must be sent with, which the runtime applies to the listener before every send. A wildcard listener additionally recovers each request's local destination from packet metadata and sends that request's reply from it, on Linux, macOS and FreeBSD; a wildcard bind on a target without that path is refused at construction rather than served from a routing-table source address. Explicit-address listeners are unaffected everywhere. Echo replies fill their payload from the session's negotiated server fill; the policy and its deliberate divergences are in `crates/irtt-server/AGENTS.md`. Two optional negotiation policies restrict what a session may ask the server to provide — a timestamp allowance and DSCP permission — both off by default and both settled during open negotiation rather than in the runtime. Serving multiple listeners in one process is not implemented; that, and ordinary process-level polish and observability decisions, is what remains of the server architecture.
+`irtt-server` is a first-class reusable server crate. It contains the deterministic OPEN/ECHO/session core — packet admission, authentication policy, open negotiation, the bounded session table, open reply construction, echo processing with its per-session receive state and timestamps, per-session rate limiting, session lifetime with idle expiry, the maximum-duration server-initiated close, and client-initiated close — plus a reusable Tokio UDP `Server` that runs one sequential `ServerCore` per listener with caller-controlled shutdown and scheduled idle maintenance. Logical expiry remains exact on authenticated, structurally valid requests, while the runtime also reclaims expired sessions once per second when no traffic arrives. Each reply carries the raw traffic class it must be sent with, which the runtime applies to the listener before every send. A wildcard listener additionally recovers each request's local destination from packet metadata and sends that request's reply from it, on Linux, macOS and FreeBSD; a wildcard bind on a target without that path is refused at construction rather than served from a routing-table source address. Explicit-address listeners are unaffected everywhere. Echo replies fill their payload from the session's negotiated server fill; the policy and its deliberate divergences are in `crates/irtt-server/AGENTS.md`. Two optional negotiation policies restrict what a session may ask the server to provide — a timestamp allowance and DSCP permission — both off by default and both settled during open negotiation rather than in the runtime. Above that one-listener primitive sits `ServerSet`, the service-level owner of one or more independent `Server`s: it binds them all or none, runs each in its own Tokio task, fans one external shutdown out to them, joins them all, and fails the group if any listener fails or stops early. Ordinary process-level polish and observability decisions are what remains of the server architecture.
 - It shares wire semantics with the client through `irtt-proto` and follows the same clean-room boundary.
 - It must **not** depend on `irtt-client` merely to reuse protocol behavior.
@@ -65,7 +65,8 @@ Tokio must remain optional for `irtt-client`; default builds must remain runtime
 - There is currently no requirement for a blocking server API, alternate async runtimes, a runtime abstraction layer, or a runtime-free server feature. Do not build blocking/runtime variants merely for symmetry with the client, and do not invent a mirrored client-style API family (`BlockingServer`, `AsyncServer`, `ManagedServer`, `BlockingManagedServer`, or similar) ahead of agreement.
 - Protocol/session logic should be separated from socket/runtime orchestration where that materially improves deterministic testing, ownership clarity, state-machine reasoning, or maintainability. A `ServerCore`-like internal boundary is therefore desirable, but its exact name/API is not prescribed, it need not be public, it is not a promise of runtime independence, and simple code should not be tortured merely to make the core artificially pure.
 - Runtime/server orchestration belongs around that core and may freely use Tokio for UDP sockets, `recv_from`/`send_to`, IPv4/IPv6 listeners, timers, session expiry, max-duration deadlines, shutdown, bounded control/config channels where useful, and `select!`-style orchestration.
- The CLI server applet exists behind the `irtt-cli` `server` feature, which is part of that crate's default build. It is thin orchestration/configuration over reusable `irtt-server` — one current-thread Tokio runtime, one explicitly bound listener, one `Server` — and must not grow a second server state machine. Multi-listener orchestration, hostname resolution, daemonization, and configuration files are deliberately absent.
+- Ownership is layered and the layers do not blur: `Server` is the one-listener primitive (one socket, one `ServerCore`, one sequential loop), and `ServerSet` supervises one or more of them without owning protocol or session state itself. A common `ServerConfig` is cloned into each listener, so its bounds — `max_sessions` included — are per listener and not per process. Listeners share nothing else: no shared `ServerCore`, no global session table or token namespace, no shared clock, and one sequential sender per socket, which is what keeps the socket-wide reply traffic class race-free.
 - The CLI server applet exists behind the `irtt-cli` `server` feature, which is part of that crate's default build. It is thin orchestration/configuration over reusable `irtt-server` — one current-thread Tokio runtime, one repeatable `--bind`, one `ServerSet` — and must not grow a second server state machine. **It always uses `ServerSet`, including for a single bind**: there is no direct CLI-to-`Server` path and no bind-count branch, so startup, shutdown and listener-failure semantics are identical whatever was configured, and the singleton case exercises the same orchestration in normal use. Hostname resolution, interface expansion, daemonization, and configuration files are deliberately absent.
 - Resource policy: the server should use bounded session/resource policy. Upstream's observed lack of useful default session/per-peer bounds, and its never expiring a session that has not carried an echo request (see `IRTT_SERVER_PROTOCOL_SPEC.md`), are not interoperability requirements and must not be reproduced merely for compatibility.
 - Testing: prefer `irtt-server` for normal compliant fake-peer behavior where practical, as far as its implemented behavior reaches. Keep narrow adversarial/raw peers for intentionally malformed, ambiguous, or non-compliant wire behavior.
@@ -168,7 +168,7 @@ Quit with `q` or `Ctrl-C`.
 ## Server
-The server applet serves one UDP listener and requires an explicit bind address:
+The server applet requires at least one explicit bind address:
 ```sh
 irtt-server --bind 127.0.0.1:2112
@@ -180,17 +180,34 @@ The same applet is reachable through the dispatcher:
 irtt-rs server --bind 192.0.2.10:2112
 ```
 Repeat `--bind` to serve several addresses from one process, in the order given:
 ```sh
 irtt-server \
    --bind 0.0.0.0:2112 \
    --bind [::]:2112
 ```
 Every invocation runs through the same multi-listener path, so one address is an
 ordinary set of one rather than a separate mode. The policy options below are
 applied to every listener, but the listeners are otherwise independent: each has
 its own sessions and tokens, so a session belongs to the address it was opened
 on and `--max-sessions` bounds each listener rather than the process. Binding is
 all or nothing — if any address cannot be bound, none are served — and a port of
 `0` selects an unused port per listener. If a listener fails while running, the
 others are shut down with it rather than leaving a service configured for two
 families answering on one.
 An explicit interface address works on every supported system. A wildcard bind
 such as `--bind 0.0.0.0:2112` reads each request's local destination from packet
 metadata and sends that request's reply from the same address, so a client on a
 multi-homed host is answered from the endpoint it contacted. That is implemented
 on Linux, macOS and FreeBSD; elsewhere a wildcard bind is refused rather than
-served from a routing-table source address a client would discard. One process
+served from a routing-table source address a client would discard. Wildcard
-serves exactly one address, so run a second process to serve a second address or
+IPv4 and IPv6 listeners may share one port, as above.
 family.
-The bound endpoint is printed on startup, which also resolves a port of `0`.
+Each bound endpoint is printed on startup once every listener is up, which also
-`Ctrl-C` stops the server gracefully.
+resolves a port of `0`. `Ctrl-C` stops the server gracefully.
 Session policy is set with options that map directly onto the server library's
 configuration; anything left unset keeps the library default:
@@ -326,10 +343,10 @@ The client, event stream, machine-readable output, multi-target execution, local
 The server library and the server applet are implemented: open negotiation,
 session state, echo processing, per-session rate limiting, idle expiry, the
 maximum-duration close, HMAC authentication, the negotiated per-session reply
-traffic class, which is applied on sockets that support it, server fill, and
+traffic class, which is applied on sockets that support it, server fill,
-wildcard reply-source selection on Linux, macOS and FreeBSD. Multiple listeners
+wildcard reply-source selection on Linux, macOS and FreeBSD, the optional
-in one process are not implemented, and optional timestamp and DSCP restriction
+timestamp and DSCP restriction controls, and serving several listeners from one
-controls remain separate concerns.
+process with independent sessions per listener.
 Server replies fill their payload with the negotiated ServerFill mode: `none`,
 which is zero-filled, `rand`, or `pattern:` with a repeating hexadecimal
@@ -10,16 +10,17 @@ use irtt_server::{ServerConfig, TimestampAllowance};
 #[command(
    name = "irtt-server",
    about = "Minimal IRTT-compatible UDP server",
-    after_help = "Policy options left unset keep the irtt-server library defaults.\nOne process serves exactly one listener; run a second process for a second address."
+    after_help = "Policy options left unset keep the irtt-server library defaults.\nRepeat --bind to serve multiple addresses in one process. All listeners use the same policy but keep independent session namespaces."
 )]
 pub struct ServerArgs {
-    /// Local UDP address to bind, as ADDR:PORT.
+    /// Local UDP address to bind, as ADDR:PORT. Repeatable.
    #[arg(
        long,
        required = true,
        value_name = "ADDR",
-        long_help = "Local UDP address to bind, as ADDR:PORT, for example 127.0.0.1:2112 or [::1]:2112. Host names are not resolved.\n\nA wildcard bind such as 0.0.0.0:2112 answers each request from the address it was sent to, using per-packet destination metadata. That is supported on Linux, macOS and FreeBSD; on other systems a wildcard bind is refused and an explicit interface address is required."
+        long_help = "Local UDP address to bind, as ADDR:PORT, for example 127.0.0.1:2112 or [::1]:2112. Host names are not resolved.\n\nRepeat the option to serve several addresses from one process, in the order given. Every listener applies the same policy options, but each keeps its own sessions and tokens, so a session belongs to the address it was opened on. Binding is all or nothing: if any address cannot be bound, none are served.\n\nA wildcard bind such as 0.0.0.0:2112 answers each request from the address it was sent to, using per-packet destination metadata. That is supported on Linux, macOS and FreeBSD; on other systems a wildcard bind is refused and an explicit interface address is required.\n\nA port of 0 selects an unused port per listener, so two such binds get two different ports."
    )]
-    pub bind: SocketAddr,
+    pub bind: Vec<SocketAddr>,
    /// HMAC key; every request must then carry a valid MAC.
    #[arg(
@@ -29,11 +30,11 @@ pub struct ServerArgs {
    )]
    pub hmac: Option<String>,
-    /// Maximum number of simultaneously live sessions.
+    /// Maximum simultaneously live sessions per listener.
    #[arg(
        long,
        value_name = "COUNT",
-        long_help = "Maximum number of simultaneously live sessions. Once the table is full, a session-creating open is dropped silently; nothing is evicted. Zero refuses every session-creating open."
+        long_help = "Maximum number of simultaneously live sessions per listener. Once a listener's table is full, a session-creating open to it is dropped silently; nothing is evicted. Zero refuses every session-creating open.\n\nThis is a per-listener bound, not a process-wide one: with two --bind options, each listener admits up to this many sessions."
    )]
    pub max_sessions: Option<usize>,
@@ -226,16 +227,67 @@ mod tests {
    #[test]
    fn bind_is_required_and_parses_both_families() {
-        assert!(parse(&[]).is_err());
+        assert!(
            parse(&[]).is_err(),
            "a server with no address serves nothing"
        );
        let ipv4 = parse(&["--bind", "127.0.0.1:2112"]).unwrap();
-        assert_eq!(ipv4.bind, SocketAddr::from((Ipv4Addr::LOCALHOST, 2112)));
+        assert_eq!(ipv4.bind, [SocketAddr::from((Ipv4Addr::LOCALHOST, 2112))]);
        let ipv6 = parse(&["--bind", "[::1]:2112"]).unwrap();
-        assert_eq!(ipv6.bind, SocketAddr::from((Ipv6Addr::LOCALHOST, 2112)));
+        assert_eq!(ipv6.bind, [SocketAddr::from((Ipv6Addr::LOCALHOST, 2112))]);
        assert!(parse(&["--bind", "127.0.0.1"]).is_err());
        assert!(parse(&["--bind", "localhost:2112"]).is_err());
        // Still one address per option: comma-separated syntax is not a thing
        // here, and accepting it silently would be worse than rejecting it.
        assert!(parse(&["--bind", "127.0.0.1:2112,[::1]:2112"]).is_err());
    }
    #[test]
    fn bind_is_repeatable_and_keeps_the_requested_order() {
        let args = parse(&[
            "--bind",
            "[::1]:2112",
            "--bind",
            "127.0.0.1:2112",
            "--bind",
            "127.0.0.1:2113",
        ])
        .unwrap();
        assert_eq!(
            args.bind,
            [
                SocketAddr::from((Ipv6Addr::LOCALHOST, 2112)),
                SocketAddr::from((Ipv4Addr::LOCALHOST, 2112)),
                SocketAddr::from((Ipv4Addr::LOCALHOST, 2113)),
            ],
            "listeners are served in the order they were requested"
        );
    }
    #[test]
    fn the_configuration_is_one_policy_however_many_listeners_there_are() {
        // Addresses never enter the server configuration: every listener is
        // built from a clone of this one value.
        let one = parse(&["--bind", "127.0.0.1:2112", "--max-sessions", "4"])
            .unwrap()
            .server_config();
        let several = parse(&[
            "--bind",
            "127.0.0.1:2112",
            "--bind",
            "[::1]:2112",
            "--max-sessions",
            "4",
        ])
        .unwrap()
        .server_config();
        assert_eq!(one, several);
        assert_eq!(several.max_sessions(), 4, "and it is a per-listener bound");
    }
    #[test]
@@ -3,21 +3,24 @@ use std::{
    time::Duration,
 };
-use irtt_server::Server;
+use irtt_server::ServerSet;
 use super::args::ServerArgs;
 /// How often the shutdown future samples the process-wide signal flag.
 ///
 /// The flag a signal handler sets has no async wake mechanism, so the future
-/// polls it on a timer rather than spinning. `Server::run` owns everything after
+/// polls it on a timer rather than spinning. `ServerSet::run` owns everything
-/// that future becomes ready.
+/// after that future becomes ready, including fanning it out to the listeners.
 const SHUTDOWN_POLL_INTERVAL: Duration = Duration::from_millis(50);
-/// Runs one listener until the process is asked to shut down.
+/// Runs the configured listeners until the process is asked to shut down.
 ///
-/// The applet owns exactly one current-thread Tokio runtime, one bound socket
+/// The applet owns exactly one current-thread Tokio runtime and one
-/// and one reusable [`Server`]; the server's own loop is the event loop.
+/// [`ServerSet`], whatever number of addresses was requested. A single `--bind`
 /// is an ordinary set of one and takes the same path as a set of several: one
 /// construction, one shutdown future, one supervision rule. Listener
 /// concurrency is async I/O concurrency and needs no worker threads.
 pub fn run_server(
    args: ServerArgs,
    shutdown_requested: &AtomicBool,
@@ -32,9 +35,14 @@ async fn serve(
    args: ServerArgs,
    shutdown_requested: &AtomicBool,
 ) -> Result<(), Box<dyn std::error::Error>> {
-    let mut server = Server::bind(args.bind, args.server_config()).await?;
+    let config = args.server_config();
-    eprintln!("irtt-server: listening on {}", server.local_addr()?);
+    let set = ServerSet::bind(args.bind, config).await?;
-    server.run(shutdown_signal(shutdown_requested)).await?;
+    // Only once every listener is up. Announcing one address and then failing
    // to bind the next would report a service that never started.
    for addr in set.local_addrs() {
        eprintln!("irtt-server: listening on {addr}");
    }
    set.run(shutdown_signal(shutdown_requested)).await?;
    Ok(())
 }
@@ -50,11 +58,28 @@ mod tests {
    use super::*;
    /// The ordinary invocation, and the one that keeps the single-listener case
    /// on the same orchestration path as every other.
    #[test]
-    fn an_already_requested_shutdown_binds_and_returns() {
+    fn an_already_requested_shutdown_binds_one_listener_and_returns() {
        let args = ServerArgs::try_parse_from(["irtt-server", "--bind", "127.0.0.1:0"]).unwrap();
        let shutdown_requested = AtomicBool::new(true);
        run_server(args, &shutdown_requested).unwrap();
    }
    #[test]
    fn an_already_requested_shutdown_binds_every_listener_and_returns() {
        let args = ServerArgs::try_parse_from([
            "irtt-server",
            "--bind",
            "127.0.0.1:0",
            "--bind",
            "127.0.0.1:0",
        ])
        .unwrap();
        let shutdown_requested = AtomicBool::new(true);
        run_server(args, &shutdown_requested).unwrap();
    }
 }
@@ -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<Mutex<ServerCore>>`, 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
@@ -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
+//! scheduled expiry maintenance and caller-controlled shutdown. [`ServerSet`]
-//! intentionally Tokio-native; there is no blocking or alternate-runtime
+//! owns one or more `Server`s as a single service: it binds them all or none,
-//! counterpart and no transport abstraction.
+//! 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.
+//! restrictions, wildcard reply-source selection, Tokio UDP orchestration and
-//! Running several listeners in one process remains a separate slice.
+//! 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};
@@ -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<Server>,
    /// The addresses the listeners actually bound, in requested order and
    /// positionally matching `servers`.
    local_addrs: Vec<SocketAddr>,
 }
 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<I>(addrs: I, config: ServerConfig) -> Result<Self, ServerSetError>
    where
        I: IntoIterator<Item = SocketAddr>,
    {
        let requested: Vec<SocketAddr> = 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<F>(self, shutdown: F) -> Result<(), ServerSetError>
    where
        F: Future<Output = ()>,
    {
        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<Id, SocketAddr> = 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<ServerSetError> = 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<Server, ServerRuntimeError> {
    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<UdpSocket> {
    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<SocketAddr>,
        #[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<SocketAddr> = 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)
        ));
    }
 }
@@ -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<u8>,
 }
 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<OpenReply> {
        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<EchoReply> {
        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<usize> {
        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<Result<(), ServerSetError>>,
 }
 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<F>(test: F)
 where
    F: Future<Output = ()>,
 {
    timeout(TEST_TIMEOUT, test)
        .await
        .expect("the test exceeded its bounded runtime");
 }