Merge pull request #81 from Lochnair/feat/server-tokio-runtime

feat(server): add Tokio UDP runtime

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
+- **Logical expiry is request-exact; physical reclamation also runs on the
-  timestamp-based, but reclamation happens in a global sweep run after
+  runtime's fixed maintenance cadence.** Every authenticated, structurally valid
-  authentication and before dispatch, so an expired session may stay resident
+  request runs the global sweep before dispatch, so a request at the deadline
-  until the next authenticated, structurally valid request. Nothing observable
+  finds the session expired and a stale session cannot deny an open. The Tokio
-  turns on it: the table is already bounded, and the sweep runs before capacity
+  runtime invokes that same core sweep once per second so dead sessions do not
-  is judged, so a stale session cannot deny an open. Because the sweep is global
+  remain resident when there is no further traffic. This cadence is internal
-  it also runs for a no-test open, where upstream was observed not to reclaim.
+  housekeeping, not protocol policy or public configuration, and the runtime
-  Do not add a public timer or tick abstraction for this; the Tokio slice may add
+  must not inspect session internals or duplicate the expiry rule.
-  scheduled sweeps.
 - **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
+/// full server fill policy are separate slices, and echo payloads are
-/// expiry is evaluated when a datagram arrives rather than on a timer, and
+/// zero-filled. Incoming authenticated requests enforce logical expiry exactly;
-/// echo payloads are zero-filled.
+/// 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
+    /// passes. Authenticated, structurally valid requests run this sweep before
-    /// no runtime and so no timer. An expired session may therefore stay
+    /// dispatch, while the runtime's maintenance timer also invokes it for
-    /// resident until the next authenticated, structurally valid request causes
+    /// physical reclamation when no traffic arrives. The table remains bounded,
-    /// this sweep. Nothing observable turns on the difference: total state is
+    /// and a stale session cannot deny capacity to a live one.
-    /// 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.
     ///
     /// 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,
+//! engine stays testable. [`Server`] owns one Tokio UDP listener and one core,
-//! address handling, timers, scheduled sweeps and shutdown — will live around
+//! and provides sequential receive, reply, scheduled expiry maintenance and
-//! it and is intentionally Tokio-native. There is no blocking or
+//! caller-controlled shutdown. The crate is intentionally Tokio-native; there
-//! alternate-runtime counterpart and no transport abstraction.
+//! 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
+//! per-session rate limiting, idle expiry, the maximum-duration
-//! server-initiated close. DSCP application, the full server fill policy and
+//! server-initiated close and Tokio UDP orchestration. DSCP application, the
-//! the Tokio runtime are separate slices: echo payloads are zero-filled, and
+//! full server fill policy, wildcard-bind destination-address metadata and the
-//! expiry is evaluated when a datagram arrives rather than on a timer.
+//! 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();
+}