From 4fb5ca6d1586904debcb838521bbd66f72711b70 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Thu, 13 Aug 2026 01:32:05 +0700 Subject: [PATCH] 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(