From 4fb5ca6d1586904debcb838521bbd66f72711b70 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Thu, 13 Aug 2026 01:32:05 +0700 Subject: [PATCH 1/4] feat(server): add tokio udp runtime --- Cargo.lock | 13 + crates/irtt-server/AGENTS.md | 13 + crates/irtt-server/Cargo.toml | 1 + crates/irtt-server/src/core.rs | 33 ++- crates/irtt-server/src/lib.rs | 19 +- crates/irtt-server/src/runtime.rs | 278 ++++++++++++++++++++++ crates/irtt-server/src/tests/lifecycle.rs | 20 ++ 7 files changed, 356 insertions(+), 21 deletions(-) create mode 100644 crates/irtt-server/src/runtime.rs diff --git a/Cargo.lock b/Cargo.lock index 590c779..e916603 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -733,6 +733,7 @@ dependencies = [ "getrandom 0.4.3", "irtt-proto", "thiserror 2.0.20", + "tokio", ] [[package]] @@ -1737,9 +1738,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" diff --git a/crates/irtt-server/AGENTS.md b/crates/irtt-server/AGENTS.md index a3a584b..be4e8b7 100644 --- a/crates/irtt-server/AGENTS.md +++ b/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 diff --git a/crates/irtt-server/Cargo.toml b/crates/irtt-server/Cargo.toml index 3ad8b5a..6dfbc7a 100644 --- a/crates/irtt-server/Cargo.toml +++ b/crates/irtt-server/Cargo.toml @@ -15,3 +15,4 @@ 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"] } diff --git a/crates/irtt-server/src/core.rs b/crates/irtt-server/src/core.rs index eefdd96..e9e2de3 100644 --- a/crates/irtt-server/src/core.rs +++ b/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: diff --git a/crates/irtt-server/src/lib.rs b/crates/irtt-server/src/lib.rs index 629323d..bfeab27 100644 --- a/crates/irtt-server/src/lib.rs +++ b/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}; diff --git a/crates/irtt-server/src/runtime.rs b/crates/irtt-server/src/runtime.rs new file mode 100644 index 0000000..dd3a4e6 --- /dev/null +++ b/crates/irtt-server/src/runtime.rs @@ -0,0 +1,278 @@ +//! 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, +} + +impl Server { + /// Binds one UDP listener at `addr` and creates its independent session + /// namespace. + pub async fn bind(addr: SocketAddr, config: ServerConfig) -> Result { + 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 { + 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. 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(&mut self, shutdown: F) -> Result<(), ServerRuntimeError> + where + F: Future, + { + 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 sent = self.socket.send_to(&reply, peer).await; + 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) + } +} diff --git a/crates/irtt-server/src/tests/lifecycle.rs b/crates/irtt-server/src/tests/lifecycle.rs index 76b1463..ea67c95 100644 --- a/crates/irtt-server/src/tests/lifecycle.rs +++ b/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(¶ms_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( -- 2.54.0 From 85c432fcf87f22bdf0b0b25cc9038edb927b28e5 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Thu, 13 Aug 2026 01:35:34 +0700 Subject: [PATCH 2/4] test(server): exercise async client interop --- Cargo.lock | 1 + crates/irtt-server/Cargo.toml | 3 + crates/irtt-server/tests/async_client.rs | 90 ++++++++++++++++++++++++ 3 files changed, 94 insertions(+) create mode 100644 crates/irtt-server/tests/async_client.rs diff --git a/Cargo.lock b/Cargo.lock index e916603..b423b22 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -731,6 +731,7 @@ name = "irtt-server" version = "0.4.0" dependencies = [ "getrandom 0.4.3", + "irtt-client", "irtt-proto", "thiserror 2.0.20", "tokio", diff --git a/crates/irtt-server/Cargo.toml b/crates/irtt-server/Cargo.toml index 6dfbc7a..1689773 100644 --- a/crates/irtt-server/Cargo.toml +++ b/crates/irtt-server/Cargo.toml @@ -16,3 +16,6 @@ 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"] } diff --git a/crates/irtt-server/tests/async_client.rs b/crates/irtt-server/tests/async_client.rs new file mode 100644 index 0000000..4d411a3 --- /dev/null +++ b/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(); +} -- 2.54.0 From 41e39bd5b95da2296283fd6b5954ebd0469846eb Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Thu, 13 Aug 2026 01:50:25 +0700 Subject: [PATCH 3/4] fix(server): keep shutdown responsive during sends --- crates/irtt-server/AGENTS.md | 17 ++++++++--------- crates/irtt-server/src/runtime.rs | 17 +++++++++++++---- 2 files changed, 21 insertions(+), 13 deletions(-) diff --git a/crates/irtt-server/AGENTS.md b/crates/irtt-server/AGENTS.md index be4e8b7..59334f7 100644 --- a/crates/irtt-server/AGENTS.md +++ b/crates/irtt-server/AGENTS.md @@ -202,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 diff --git a/crates/irtt-server/src/runtime.rs b/crates/irtt-server/src/runtime.rs index dd3a4e6..158196a 100644 --- a/crates/irtt-server/src/runtime.rs +++ b/crates/irtt-server/src/runtime.rs @@ -62,9 +62,10 @@ impl Server { /// Serves datagrams until `shutdown` completes. /// /// Graceful shutdown sends no session-close packets and leaves no hidden - /// receive task behind. Per-packet send failures and short sends drop only - /// that reply; receive failures and internal core failures terminate the - /// loop with an error. + /// 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(&mut self, shutdown: F) -> Result<(), ServerRuntimeError> where F: Future, @@ -89,7 +90,15 @@ impl Server { } if let Some(reply) = self.core.handle_datagram(peer, &self.recv_buffer[..len])? { - let sent = self.socket.send_to(&reply, peer).await; + 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; } -- 2.54.0 From 8d240e352d996a51e5133a7b631c166e22dc4589 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Thu, 13 Aug 2026 02:06:41 +0700 Subject: [PATCH 4/4] docs(server): update runtime status --- AGENTS.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/AGENTS.md b/AGENTS.md index 1b536ce..4fd9bbe 100644 --- a/AGENTS.md +++ b/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. -- 2.54.0