AGENTS.md

@@ -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 currently 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. DSCP application and the full server fill policy remain unimplemented, as does the Tokio runtime, so session expiry is evaluated when a datagram arrives rather than on a timer; see `crates/irtt-server/AGENTS.md`. The rest of this section describes the boundary the crate is built to, including the parts not yet written.
+`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. DSCP application and the full server fill policy remain unimplemented; see `crates/irtt-server/AGENTS.md`. The rest of this section describes the implemented boundary and the parts not yet written.
 
 - 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.

Cargo.lock

@@ -731,8 +731,10 @@ name = "irtt-server"
 version = "0.4.0"
 dependencies = [
  "getrandom 0.4.3",
+ "irtt-client",
  "irtt-proto",
  "thiserror 2.0.20",
+ "tokio",
 ]
 
 [[package]]
@@ -1737,9 +1739,21 @@ dependencies = [
  "mio",
  "pin-project-lite",
  "socket2",
+ "tokio-macros",
  "windows-sys",
 ]
 
+[[package]]
+name = "tokio-macros"
+version = "2.7.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.3",
+]
+
 [[package]]
 name = "typenum"
 version = "1.20.1"

crates/irtt-server/AGENTS.md

@@ -43,6 +43,19 @@ material does not establish a semantic, state the ambiguity instead of guessing.
   belongs around the core.
 - The server is intentionally **Tokio-native**. Tokio does not need to be
   optional for this crate.
+- `Server` owns exactly one Tokio `UdpSocket` and one `ServerCore`. Run protocol
+  processing sequentially in that one task: do not spawn per datagram, put the
+  core behind a lock, or start an internal detached task.
+- The runtime must remain a thin `recv_from` -> `handle_datagram` -> optional
+  `send_to` loop. Do not duplicate protocol parsing or encoding around the core.
+- A receive failure or core failure may terminate `Server::run`; a per-packet
+  send failure (including a short send) drops only that reply and must not stop
+  the listener. Core state is not rolled back after a transport send failure.
+- Runtime maintenance calls the core's expiry hook. It must not inspect session
+  internals or grow a second expiry rule.
+- Explicit-address binds are the source-address-safe supported path. A wildcard
+  bind on a multi-homed host remains subject to kernel source-address selection
+  until packet-info handling is implemented.
 - Do not add a blocking server, an alternate-runtime variant, a transport trait,
   a runtime abstraction layer, or a mirrored client-style API family. There is
   no product requirement for any of them, and `ServerCore` being public is not a
@@ -189,15 +202,14 @@ Do not rewrite the protocol evidence to match them.
   The only interoperability constraint is negative — expiry must never be
   signalled — and silence satisfies it. A zero idle timeout expires a session at
   the next evaluation; it is not "never expire".
-- **Expiry is packet-driven until the runtime exists.** Logical expiry is
-  timestamp-based, but reclamation happens in a global sweep run after
-  authentication and before dispatch, so an expired session may stay resident
-  until the next authenticated, structurally valid request. Nothing observable
-  turns on it: the table is already bounded, and the sweep runs before capacity
-  is judged, so a stale session cannot deny an open. Because the sweep is global
-  it also runs for a no-test open, where upstream was observed not to reclaim.
-  Do not add a public timer or tick abstraction for this; the Tokio slice may add
-  scheduled sweeps.
+- **Logical expiry is request-exact; physical reclamation also runs on the
+  runtime's fixed maintenance cadence.** Every authenticated, structurally valid
+  request runs the global sweep before dispatch, so a request at the deadline
+  finds the session expired and a stale session cannot deny an open. The Tokio
+  runtime invokes that same core sweep once per second so dead sessions do not
+  remain resident when there is no further traffic. This cadence is internal
+  housekeeping, not protocol policy or public configuration, and the runtime
+  must not inspect session internals or duplicate the expiry rule.
 - **The maximum-duration deadline is `first served echo + maximum + 2 s`.** The
   origin is measured behavior: neither the open, nor a rejected first echo, nor a
   rate-limited one starts it. The two-second grace matches the measured upstream

crates/irtt-server/Cargo.toml

@@ -15,3 +15,7 @@ categories = ["network-programming"]
 irtt-proto.workspace = true
 getrandom = "0.4.3"
 thiserror = "2.0.20"
+tokio = { version = "1.53.1", default-features = false, features = ["rt", "net", "time", "sync", "macros"] }
+
+[dev-dependencies]
+irtt-client = { workspace = true, features = ["tokio"] }

crates/irtt-server/src/core.rs

@@ -25,7 +25,7 @@ use crate::{
 /// reply's timestamps come from — both of which are private injected seams, so
 /// the whole engine stays deterministically testable.
 ///
-/// The Tokio runtime slice will wrap it directly:
+/// The Tokio runtime wraps it directly:
 ///
 /// ```text
 /// recv_from -> core.handle_datagram(peer, packet) -> Some(reply) -> send_to
@@ -39,12 +39,13 @@ use crate::{
 ///
 /// # Scope
 ///
-/// This slice implements open handling, session creation and negotiation,
+/// The core implements open handling, session creation and negotiation,
 /// normal echo processing, per-session rate limiting, idle expiry,
 /// maximum-duration close and client-initiated close. DSCP application and the
-/// full server fill policy are separate slices, as is the Tokio runtime — so
-/// expiry is evaluated when a datagram arrives rather than on a timer, and
-/// echo payloads are zero-filled.
+/// full server fill policy are separate slices, and echo payloads are
+/// zero-filled. Incoming authenticated requests enforce logical expiry exactly;
+/// the runtime also calls a private maintenance hook to reclaim idle sessions
+/// when no traffic arrives.
 #[derive(Debug)]
 pub struct ServerCore {
     config: ServerConfig,
@@ -198,17 +199,23 @@ impl ServerCore {
         }
     }
 
+    /// Reclaims sessions whose idle deadline has passed.
+    ///
+    /// The Tokio runtime calls this for physical background reclamation. Every
+    /// authenticated structurally valid datagram still performs the same sweep
+    /// before dispatch, so the timer cadence never weakens logical expiry.
+    pub(crate) fn maintain(&mut self) {
+        let now_ns = self.clock.sample().mono_ns;
+        self.release_expired_sessions(now_ns);
+    }
+
     /// Drops every session whose idle deadline has passed, as of `now_ns`.
     ///
     /// Expiry is *logical* — a session is expired the instant its deadline
-    /// passes — but reclamation here is *packet-driven*, because this slice has
-    /// no runtime and so no timer. An expired session may therefore stay
-    /// resident until the next authenticated, structurally valid request causes
-    /// this sweep. Nothing observable turns on the difference: total state is
-    /// already bounded by [`max_sessions`](ServerConfig::max_sessions), the
-    /// sweep runs before a new open is admitted so a stale session cannot deny
-    /// capacity to a live one, and while no traffic arrives at all there is
-    /// no peer to observe anything. The Tokio slice may add scheduled sweeps.
+    /// passes. Authenticated, structurally valid requests run this sweep before
+    /// dispatch, while the runtime's maintenance timer also invokes it for
+    /// physical reclamation when no traffic arrives. The table remains bounded,
+    /// and a stale session cannot deny capacity to a live one.
     ///
     /// The sweep is global and runs before the request is dispatched, so a
     /// request touching one session can release another. That is deliberate:

crates/irtt-server/src/lib.rs

@@ -10,10 +10,11 @@
 //! receive state, rate and lifetime policy, resource decisions and reply
 //! construction. It performs no I/O; the clock its timestamps and deadlines
 //! come from is a private injected seam, not a runtime abstraction, so the
-//! engine stays testable. Socket and runtime orchestration — the UDP listener,
-//! address handling, timers, scheduled sweeps and shutdown — will live around
-//! it and is intentionally Tokio-native. There is no blocking or
-//! alternate-runtime counterpart and no transport abstraction.
+//! engine stays testable. [`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.
 //!
 //! # Rejection is silence
 //!
@@ -40,10 +41,10 @@
 //! # Current scope
 //!
 //! Open handling and negotiation, session creation, normal echo processing,
-//! per-session rate limiting, idle expiry and the maximum-duration
-//! server-initiated close. DSCP application, the full server fill policy and
-//! the Tokio runtime are separate slices: echo payloads are zero-filled, and
-//! expiry is evaluated when a datagram arrives rather than on a timer.
+//! per-session rate limiting, idle expiry, the maximum-duration
+//! server-initiated close and Tokio UDP orchestration. DSCP application, the
+//! full server fill policy, wildcard-bind destination-address metadata and the
+//! CLI server applet remain separate slices. Echo payloads are zero-filled.
 #![forbid(unsafe_code)]
 
 mod clock;
@@ -51,6 +52,7 @@ mod config;
 mod core;
 mod error;
 mod negotiate;
+mod runtime;
 mod session;
 mod token;
 
@@ -63,3 +65,4 @@ pub use config::{
 };
 pub use core::ServerCore;
 pub use error::ServerError;
+pub use runtime::{Server, ServerRuntimeError};

crates/irtt-server/src/runtime.rs

@@ -0,0 +1,287 @@
+//! Tokio UDP orchestration around [`ServerCore`](crate::ServerCore).
+
+use std::{future::Future, io, net::SocketAddr, time::Duration};
+
+use thiserror::Error;
+use tokio::{net::UdpSocket, time::MissedTickBehavior};
+
+use crate::{ServerConfig, ServerCore, ServerError};
+
+/// Fixed transport receive capacity.
+///
+/// Standard UDP payloads fit in this buffer. A datagram reported as filling it
+/// exactly is dropped conservatively instead of passing potentially truncated
+/// bytes to the protocol core. IPv6 jumbograms are not supported.
+const RECEIVE_BUFFER_LEN: usize = 65_536;
+const MAINTENANCE_INTERVAL: Duration = Duration::from_secs(1);
+
+/// A Tokio UDP listener backed by one deterministic server core.
+///
+/// Processing is sequential: one task receives a datagram, lets the core
+/// process it, and sends the optional reply from the same socket. No
+/// per-datagram tasks or locks are involved.
+///
+/// Bind an explicit local address when reply source-address identity matters.
+/// Wildcard binds rely on the kernel's source-address choice; destination
+/// packet metadata and per-packet source selection are not yet implemented for
+/// multi-homed hosts.
+#[derive(Debug)]
+pub struct Server {
+    socket: UdpSocket,
+    core: ServerCore,
+    recv_buffer: Vec<u8>,
+}
+
+impl Server {
+    /// Binds one UDP listener at `addr` and creates its independent session
+    /// namespace.
+    pub async fn bind(addr: SocketAddr, config: ServerConfig) -> Result<Self, ServerRuntimeError> {
+        let socket = UdpSocket::bind(addr)
+            .await
+            .map_err(|source| ServerRuntimeError::Bind { addr, source })?;
+        Ok(Self::from_socket(socket, config))
+    }
+
+    /// Wraps an already prepared Tokio UDP socket in a server.
+    #[must_use]
+    pub fn from_socket(socket: UdpSocket, config: ServerConfig) -> Self {
+        Self {
+            socket,
+            core: ServerCore::new(config),
+            recv_buffer: vec![0; RECEIVE_BUFFER_LEN],
+        }
+    }
+
+    /// Returns the local endpoint selected for the listener.
+    pub fn local_addr(&self) -> Result<SocketAddr, ServerRuntimeError> {
+        self.socket
+            .local_addr()
+            .map_err(|source| ServerRuntimeError::LocalAddr { source })
+    }
+
+    /// Serves datagrams until `shutdown` completes.
+    ///
+    /// Graceful shutdown sends no session-close packets and leaves no hidden
+    /// receive task behind, even while a reply send is waiting for socket
+    /// writability. Per-packet send failures and short sends drop only that
+    /// reply; receive failures and internal core failures terminate the loop
+    /// with an error.
+    pub async fn run<F>(&mut self, shutdown: F) -> Result<(), ServerRuntimeError>
+    where
+        F: Future<Output = ()>,
+    {
+        tokio::pin!(shutdown);
+        let mut maintenance = tokio::time::interval(MAINTENANCE_INTERVAL);
+        maintenance.set_missed_tick_behavior(MissedTickBehavior::Skip);
+        maintenance.tick().await;
+
+        loop {
+            tokio::select! {
+                _ = &mut shutdown => return Ok(()),
+                _ = maintenance.tick() => self.core.maintain(),
+                received = self.socket.recv_from(&mut self.recv_buffer) => {
+                    let (len, peer) = match received {
+                        Ok(received) => received,
+                        Err(source) if source.kind() == io::ErrorKind::Interrupted => continue,
+                        Err(source) => return Err(ServerRuntimeError::Receive { source }),
+                    };
+                    if len == RECEIVE_BUFFER_LEN {
+                        continue;
+                    }
+
+                    if let Some(reply) = self.core.handle_datagram(peer, &self.recv_buffer[..len])? {
+                        let send = self.socket.send_to(&reply, peer);
+                        tokio::pin!(send);
+                        let sent = loop {
+                            tokio::select! {
+                                _ = &mut shutdown => return Ok(()),
+                                _ = maintenance.tick() => self.core.maintain(),
+                                sent = &mut send => break sent,
+                            }
+                        };
+                        if !matches!(sent, Ok(len) if len == reply.len()) {
+                            continue;
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
+
+/// Failure to create or run a Tokio UDP server.
+#[derive(Debug, Error)]
+pub enum ServerRuntimeError {
+    /// Binding the requested listener failed.
+    #[error("could not bind UDP listener at {addr}: {source}")]
+    Bind {
+        addr: SocketAddr,
+        #[source]
+        source: io::Error,
+    },
+    /// Querying the listener's selected local endpoint failed.
+    #[error("could not query UDP listener address: {source}")]
+    LocalAddr {
+        #[source]
+        source: io::Error,
+    },
+    /// Receiving from the listener failed irrecoverably.
+    #[error("UDP receive failed: {source}")]
+    Receive {
+        #[source]
+        source: io::Error,
+    },
+    /// The deterministic protocol/session core failed internally.
+    #[error(transparent)]
+    Core(#[from] ServerError),
+}
+
+#[cfg(test)]
+mod tests {
+    use std::{net::Ipv4Addr, time::Duration};
+
+    use irtt_proto::{
+        decode_echo_reply, decode_open_reply, encode_request, Clock, Params, ReceivedStats,
+        RequestToEncode, StampAt,
+    };
+    use tokio::{net::UdpSocket, sync::oneshot, time::timeout};
+
+    use super::*;
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn bind_reports_local_address_and_shutdown_stops_run() {
+        let requested = SocketAddr::from((Ipv4Addr::LOCALHOST, 0));
+        let mut server = Server::bind(requested, ServerConfig::default())
+            .await
+            .unwrap();
+        let bound = server.local_addr().unwrap();
+        assert_eq!(bound.ip(), requested.ip());
+        assert_ne!(bound.port(), 0);
+
+        let (shutdown_tx, shutdown_rx) = oneshot::channel();
+        shutdown_tx.send(()).unwrap();
+        server
+            .run(async {
+                let _ = shutdown_rx.await;
+            })
+            .await
+            .unwrap();
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn ipv4_open_echo_close_crosses_a_real_udp_socket() {
+        exercise_udp_path(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))).await;
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn ipv6_open_echo_close_when_loopback_is_available() {
+        let addr = "[::1]:0".parse().unwrap();
+        let server = match Server::bind(addr, unthrottled()).await {
+            Ok(server) => server,
+            Err(ServerRuntimeError::Bind { source, .. })
+                if matches!(
+                    source.kind(),
+                    io::ErrorKind::AddrNotAvailable | io::ErrorKind::Unsupported
+                ) =>
+            {
+                return;
+            }
+            Err(error) => panic!("unexpected IPv6 loopback bind failure: {error}"),
+        };
+        exercise_bound_server(server).await;
+    }
+
+    async fn exercise_udp_path(addr: SocketAddr) {
+        let server = Server::bind(addr, unthrottled()).await.unwrap();
+        exercise_bound_server(server).await;
+    }
+
+    async fn exercise_bound_server(mut server: Server) {
+        let server_addr = server.local_addr().unwrap();
+        let client_bind = match server_addr {
+            SocketAddr::V4(_) => "127.0.0.1:0",
+            SocketAddr::V6(_) => "[::1]:0",
+        };
+        let client = UdpSocket::bind(client_bind).await.unwrap();
+        let (shutdown_tx, shutdown_rx) = oneshot::channel();
+        let server_task = tokio::spawn(async move {
+            server
+                .run(async {
+                    let _ = shutdown_rx.await;
+                })
+                .await
+        });
+
+        let requested = Params {
+            protocol_version: 1,
+            duration_ns: 1_000_000_000,
+            interval_ns: 100_000_000,
+            received_stats: ReceivedStats::Both,
+            stamp_at: StampAt::Both,
+            clock: Clock::Both,
+            ..Params::default()
+        };
+        let open = encode_request(
+            RequestToEncode::Open {
+                params: &requested,
+                no_test: false,
+            },
+            None,
+        )
+        .unwrap();
+        client.send_to(&open, server_addr).await.unwrap();
+
+        let mut buffer = vec![0; RECEIVE_BUFFER_LEN];
+        let (len, reply_source) = timeout(Duration::from_secs(1), client.recv_from(&mut buffer))
+            .await
+            .expect("OPEN reply timeout")
+            .unwrap();
+        assert_eq!(reply_source, server_addr);
+        let open_reply = decode_open_reply(&buffer[..len], None).unwrap();
+        assert_ne!(open_reply.token, 0);
+
+        let echo = encode_request(
+            RequestToEncode::Echo {
+                token: open_reply.token,
+                sequence: 0,
+                params: &open_reply.params,
+                payload: &[],
+            },
+            None,
+        )
+        .unwrap();
+        client.send_to(&echo, server_addr).await.unwrap();
+        let (len, reply_source) = timeout(Duration::from_secs(1), client.recv_from(&mut buffer))
+            .await
+            .expect("ECHO reply timeout")
+            .unwrap();
+        assert_eq!(reply_source, server_addr);
+        let echo_reply = decode_echo_reply(&buffer[..len], &open_reply.params, None).unwrap();
+        assert_eq!(echo_reply.token, open_reply.token);
+        assert_eq!(echo_reply.sequence, 0);
+        assert_eq!(echo_reply.recv_count, Some(1));
+        assert_eq!(echo_reply.recv_window, Some(1));
+
+        let close = encode_request(
+            RequestToEncode::Close {
+                token: open_reply.token,
+            },
+            None,
+        )
+        .unwrap();
+        client.send_to(&close, server_addr).await.unwrap();
+        client.send_to(&echo, server_addr).await.unwrap();
+        assert!(
+            timeout(Duration::from_millis(100), client.recv_from(&mut buffer))
+                .await
+                .is_err()
+        );
+
+        shutdown_tx.send(()).unwrap();
+        server_task.await.unwrap().unwrap();
+    }
+
+    fn unthrottled() -> ServerConfig {
+        ServerConfig::default().with_min_send_interval(Duration::ZERO)
+    }
+}

crates/irtt-server/src/tests/lifecycle.rs

@@ -110,6 +110,26 @@ fn a_session_that_never_carried_an_echo_still_expires() {
     );
 }
 
+#[test]
+fn maintenance_reclaims_an_idle_session_without_traffic() {
+    let clock = ManualClock::at(0);
+    let mut core = core_with_sources(
+        idle_after(Duration::from_secs(2)),
+        ScriptedTokens::new([TOKEN_A]),
+        clock.clone(),
+    );
+
+    assert!(core
+        .handle_datagram(peer(), &open_request(&params_at(SECOND), None))
+        .unwrap()
+        .is_some());
+    assert_eq!(core.session_count(), 1);
+
+    clock.set(2 * SECOND);
+    core.maintain();
+    assert_eq!(core.session_count(), 0);
+}
+
 #[test]
 fn a_served_echo_moves_the_idle_deadline_along() {
     let (mut core, clock, token, negotiated) = manual_core(

crates/irtt-server/tests/async_client.rs

@@ -0,0 +1,90 @@
+use std::time::Duration;
+
+use irtt_client::{AsyncClient, ClientConfig, ClientEvent, OpenOutcome, SocketConfig};
+use irtt_proto::{Clock, ReceivedStats, StampAt};
+use irtt_server::{Server, ServerConfig};
+use tokio::{sync::oneshot, time::timeout};
+
+#[tokio::test(flavor = "current_thread")]
+async fn async_client_opens_echoes_receives_and_closes() {
+    timeout(Duration::from_secs(2), exercise_client_server())
+        .await
+        .expect("client/server integration exceeded its bounded runtime");
+}
+
+async fn exercise_client_server() {
+    let mut server = Server::bind("127.0.0.1:0".parse().unwrap(), ServerConfig::default())
+        .await
+        .unwrap();
+    let server_addr = server.local_addr().unwrap();
+    let (shutdown_tx, shutdown_rx) = oneshot::channel();
+    let server_task = tokio::spawn(async move {
+        server
+            .run(async {
+                let _ = shutdown_rx.await;
+            })
+            .await
+    });
+
+    let config = ClientConfig {
+        server_addr: server_addr.to_string(),
+        duration: Some(Duration::from_secs(1)),
+        interval: Duration::from_millis(100),
+        length: 64,
+        received_stats: ReceivedStats::Both,
+        stamp_at: StampAt::Both,
+        clock: Clock::Both,
+        open_timeouts: vec![Duration::from_millis(200)],
+        socket_config: SocketConfig {
+            recv_timeout: Some(Duration::from_millis(200)),
+            ..SocketConfig::default()
+        },
+        probe_timeout: Duration::from_millis(200),
+        ..ClientConfig::default()
+    };
+    let mut client = AsyncClient::connect(config).await.unwrap();
+
+    let opened = client.open().await.unwrap();
+    match &opened {
+        OpenOutcome::Started {
+            remote,
+            token,
+            negotiated,
+            event,
+        } => {
+            assert_eq!(*remote, server_addr);
+            assert_ne!(*token, 0);
+            assert_eq!(negotiated.params.received_stats, ReceivedStats::Both);
+            assert_eq!(negotiated.params.stamp_at, StampAt::Both);
+            assert_eq!(negotiated.params.clock, Clock::Both);
+            assert!(matches!(event, ClientEvent::SessionStarted { .. }));
+        }
+        OpenOutcome::NoTestCompleted { .. } => panic!("normal client unexpectedly ran no-test"),
+    }
+
+    assert!(matches!(
+        client.send_probe().await.unwrap().as_slice(),
+        [ClientEvent::EchoSent {
+            seq: 0,
+            remote,
+            ..
+        }] if *remote == server_addr
+    ));
+    assert!(matches!(
+        client.recv().await.unwrap().as_slice(),
+        [ClientEvent::EchoReply {
+            seq: 0,
+            remote,
+            server_timing: Some(_),
+            received_stats: Some(_),
+            ..
+        }] if *remote == server_addr
+    ));
+    assert!(matches!(
+        client.close().await.unwrap().as_slice(),
+        [ClientEvent::SessionClosed { remote, .. }] if *remote == server_addr
+    ));
+
+    shutdown_tx.send(()).unwrap();
+    server_task.await.unwrap().unwrap();
+}