diff --git a/AGENTS.md b/AGENTS.md index fa671e9..1b536ce 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, and client-initiated close. Rate limiting, session lifetime, server-initiated close, DSCP application and the server fill policy remain unimplemented, as does the Tokio runtime; 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 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. - 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. @@ -66,7 +66,7 @@ Tokio must remain optional for `irtt-client`; default builds must remain runtime - Protocol/session logic should be separated from socket/runtime orchestration where that materially improves deterministic testing, ownership clarity, state-machine reasoning, or maintainability. A `ServerCore`-like internal boundary is therefore desirable, but its exact name/API is not prescribed, it need not be public, it is not a promise of runtime independence, and simple code should not be tortured merely to make the core artificially pure. - Runtime/server orchestration belongs around that core and may freely use Tokio for UDP sockets, `recv_from`/`send_to`, IPv4/IPv6 listeners, timers, session expiry, max-duration deadlines, shutdown, bounded control/config channels where useful, and `select!`-style orchestration. - The CLI server applet, when added, should remain thin orchestration/configuration over reusable `irtt-server`, not contain a second server state machine. -- Resource policy: the server should use bounded session/resource policy. Upstream's observed lack of useful default session/per-peer bounds (see `IRTT_SERVER_PROTOCOL_SPEC.md`) is not an interoperability requirement and must not be reproduced merely for compatibility. +- Resource policy: the server should use bounded session/resource policy. Upstream's observed lack of useful default session/per-peer bounds, and its never expiring a session that has not carried an echo request (see `IRTT_SERVER_PROTOCOL_SPEC.md`), are not interoperability requirements and must not be reproduced merely for compatibility. - Testing: prefer `irtt-server` for normal compliant fake-peer behavior where practical, as far as its implemented behavior reaches. Keep narrow adversarial/raw peers for intentionally malformed, ambiguous, or non-compliant wire behavior. ## Engineering rules diff --git a/crates/irtt-server/AGENTS.md b/crates/irtt-server/AGENTS.md index 333311a..a3a584b 100644 --- a/crates/irtt-server/AGENTS.md +++ b/crates/irtt-server/AGENTS.md @@ -144,9 +144,78 @@ effective parameters must be safe for the server to run. - Reply payload bytes are zero, which is the initial safe fill policy: payload carries no protocol meaning, and a server must never emit residue from another request or client. Request payload bytes never reach a reply. -- Rate limiting, idle expiry, maximum duration, server-initiated close, DSCP - socket application and the full `ServerFill` policy are later slices. Do not - add placeholder session fields for them. +- DSCP socket application and the full `ServerFill` policy are later slices. Do + not add placeholder session fields for them. + +## Rate, lifetime and server-initiated close + +Defaults: minimum send interval 10 ms, burst allowance 5, idle timeout 60 s, no +maximum test duration. All four are per-server configuration, and the burst +allowance is per session. + +Negotiation applies the interval floor first and the (idle timeout ÷ 4) cap +second, so a configured minimum above a quarter of the timeout ends with the cap +winning and the returned interval below that minimum. A configured maximum test +duration reduces a longer requested Duration and also replaces a *continuous* +request (an absent Duration), because restricting continuous mode to a finite +test is what configuring a maximum means. A zero configured maximum is stored as +no maximum: Duration zero on the wire means continuous, so a finite maximum of +zero could only be expressed by making a client's finite test endless. + +Several choices below deliberately differ from the reference server's observed +behavior. Each is **`irtt-rs` project policy**, taken where the clean evidence +records upstream behavior as policy rather than an interoperability requirement. +Do not rewrite the protocol evidence to match them. + +- **The refill interval is the shorter of the configured minimum and the + negotiated interval.** The specification records upstream replenishing on the + configured minimum regardless, so its own idle cap can hand a client a 2 s + interval while the limiter enforces 5 s and rate-limits a fully conforming + client. A server must not enforce a cadence it did not advertise. Ordinary + configurations are unaffected: a 10 ms minimum against a 1 s negotiated + interval still refills every 10 ms. +- **Zero means two different things and neither is "unlimited".** A zero minimum + send interval is no time-based throttling; a zero burst allowance is no + allowance at all, so every echo is rate-limited. +- **A rate-limited echo refreshes the idle deadline and advances no statistic.** + This one *is* observed behavior — the only tested drop class that refreshes — + and is kept deliberately. +- **The idle deadline runs from the open**, not from the first echo request. A + session that never carries one still ages out; upstream's never-expiring + unused session is a resource leak the specification recommends against. +- **Expiry is immediate and silent at the configured deadline.** No five-second + additive grace, no final lazy-release reply to the first echo that finds an + expired session, and no request-class-dependent "who consumed it" behavior. + 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. +- **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 + margin and is a fixed internal constant, not a configuration knob. +- **Rate allowance is judged before the maximum-duration close.** A + deadline-crossing echo with no allowance is dropped silently and the close is + carried by the next echo that is served. Other rejected classes likewise do not + trigger or defer it — the specification records that case as untested, so this + is the deterministic reading rather than an observed one. +- **A server close is an ordinary echo reply with `FLAG_CLOSE` added**, carrying + the triggering request's sequence number, statistics, timestamps and + authentication. There is no standalone close packet in protocol version 1. The + session is released once that reply has encoded, so every later request is an + unknown token. + +Every state transition — receive, rate and lifetime — is computed purely and +committed only after the reply has encoded, so an encoding failure leaves the +session exactly as it was and releases nothing. ## Resource policy @@ -174,7 +243,18 @@ half-created session. - Token generation and clock sampling are the only nondeterministic parts; tests inject a scripted source for each, so identity, collisions, allocation failure and timestamp values are all assertable. Keep both seams private, and keep - timestamp tests free of sleeps, tolerances and real wall-clock assertions. + timestamp, rate and lifetime tests free of sleeps, tolerances and real + wall-clock assertions. +- Two clock fakes, for two different questions. A scripted clock returns a fixed + list of samples and is for tests that assert individual timestamp *readings*; a + hand-moved clock stands still until the test moves it, and is for rate and + lifetime tests, which care when a datagram arrived and not how many times the + core read the clock on the way. Prefer the latter for anything about deadlines. +- Assert rate and lifetime behavior from replies: which datagrams came back, the + count and window they reported, and `session_count`. The token bucket is a + black-box inference about the reference server, not a protocol requirement, so + do not add accessors for allowance, activity instants or the maximum-duration + origin. - As the server gains behavior, prefer it over fake peers for normal compliant behavior in other crates' tests. Keep small raw/adversarial peers for intentionally malformed or non-compliant wire behavior. diff --git a/crates/irtt-server/src/core.rs b/crates/irtt-server/src/core.rs index 3f5977f..eefdd96 100644 --- a/crates/irtt-server/src/core.rs +++ b/crates/irtt-server/src/core.rs @@ -7,11 +7,11 @@ use irtt_proto::{ }; use crate::{ - clock::{ClockSample, ClockSource, SystemClock}, + clock::{saturating_ns, ClockSample, ClockSource, SystemClock}, config::ServerConfig, error::ServerError, negotiate::negotiate_params, - session::Session, + session::{RateDecision, Session}, token::{OsTokenSource, TokenSource, TOKEN_ATTEMPTS}, }; @@ -39,11 +39,12 @@ use crate::{ /// /// # Scope /// -/// This slice implements open handling, session creation, normal echo -/// processing and client-initiated close. Rate limiting, session lifetime and -/// expiry, server-initiated close, DSCP application and the server fill policy -/// are separate slices, so every otherwise admissible echo is answered and no -/// session ever ages out. +/// This slice 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. #[derive(Debug)] pub struct ServerCore { config: ServerConfig, @@ -122,7 +123,14 @@ impl ServerCore { /// /// `Ok(None)` is not only a rejection, though. A close request that does /// release its session is answered with nothing as well, because protocol - /// version 1 defines no acknowledgement for one. + /// version 1 defines no acknowledgement for one — and so is an echo that + /// found its session's rate allowance spent, which advances no statistic + /// but does keep the session alive. + /// + /// Sessions whose idle deadline has passed are released before the request + /// is dispatched, so a request may find the session it names already gone, + /// and one request may release an unrelated session. Expiry is never + /// signalled; a client sees only silence. /// /// Authentication is checked before open parameters are decoded and before /// any session is looked up. That ordering is deliberate: a packet that @@ -165,8 +173,23 @@ impl ServerCore { } } + // One monotonic instant per authenticated, structurally valid request, + // which is the request's whole lifecycle time: what expiry is judged + // against, what a new session's deadlines start from, and what the rate + // limiter and maximum-duration deadline read. An echo already has one — + // the instant its reply will report as the receive time — and taking a + // second would let a reply's timestamps disagree with the policy + // decisions made about the very same datagram. + let now_ns = match kind { + ClassifiedKind::Echo(ref echo) => echo.received_at.mono_ns, + _ => self.clock.sample().mono_ns, + }; + self.release_expired_sessions(now_ns); + match kind { - ClassifiedKind::Open { no_test, params } => self.handle_open(peer, no_test, params), + ClassifiedKind::Open { no_test, params } => { + self.handle_open(peer, no_test, params, now_ns) + } ClassifiedKind::Close { token } => { self.handle_close(peer, token); Ok(None) @@ -175,6 +198,32 @@ impl ServerCore { } } + /// 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. + /// + /// The sweep is global and runs before the request is dispatched, so a + /// request touching one session can release another. That is deliberate: + /// expiry is a resource decision about the table, not a side effect of what + /// the current datagram happens to name. It also means a *no-test* open — + /// which creates nothing of its own — reclaims expired sessions like any + /// other open, where the reference server was observed not to. Both are + /// upstream policy the clean specification declines to require, and a + /// client cannot tell the difference: expiry is never signalled. + fn release_expired_sessions(&mut self, now_ns: i64) { + let idle_timeout_ns = saturating_ns(self.config.idle_timeout()); + self.sessions + .retain(|_, session| !session.is_idle_expired(now_ns, idle_timeout_ns)); + } + /// Pairs a structurally decoded request with the clock sample an echo /// needs, leaving every other kind untouched. fn classify<'a>( @@ -206,9 +255,10 @@ impl ServerCore { /// discard: the datagram must not be longer than /// [`max_packet_length`](ServerConfig::max_packet_length), its token must /// name a live session, and it must have arrived from that session's own - /// endpoint. Authentication was already settled by the caller. A request - /// failing any of these leaves the receive state — of this session and of - /// every other — exactly where it was. + /// endpoint. Authentication was already settled by the caller, as was idle + /// expiry — a request reaching here found its session still alive. A + /// request failing any of these leaves the receive state — of this session + /// and of every other — exactly where it was. /// /// Only the token and the sequence number mean anything. The request's own /// length is judged against the configured maximum and then forgotten: it @@ -216,9 +266,24 @@ impl ServerCore { /// reply, which is laid out and sized from the session's negotiated /// parameters alone. /// - /// The next receive state is computed but not committed until the reply has + /// **Rate allowance is judged after admission and before the + /// maximum-duration deadline.** That order is the observed one and it + /// matters: a post-deadline echo with no allowance is simply dropped, and + /// the close it would have carried is not lost but handed to the next echo + /// that is served. A rate-limited request refreshes the session's idle + /// deadline — the one drop class the clean evidence records doing so — and + /// changes nothing else. + /// + /// **The maximum-duration close is an ordinary reply with the Close flag + /// added**, never a standalone packet: protocol version 1 has no other + /// in-band mechanism. It reports the triggering request in its statistics + /// exactly as any reply would, and the session is released once it has + /// encoded, so every later request is dropped as an unknown token. + /// + /// Every state transition is computed but not committed until the reply has /// been encoded, so a reply this server failed to build cannot leave a - /// session claiming to have answered a request it never did. + /// session claiming to have answered a request it never did — or, for a + /// close, leave the session released with no reply carrying the flag. fn handle_echo( &mut self, peer: SocketAddr, @@ -239,7 +304,29 @@ impl ServerCore { return Ok(None); } + let now_ns = request.received_at.mono_ns; + let next_rate = match session.rate_state().charge(now_ns) { + RateDecision::Limited { next } => { + // Silence, and no statistic moves — but the session lives on, + // and its idle deadline moves with it. + let next_lifetime = session.lifetime_state().refreshed(now_ns); + session.commit_rate_state(next); + session.commit_lifetime_state(next_lifetime); + return Ok(None); + } + RateDecision::Allowed { next } => next, + }; + + // Only a session that has already served an echo has a deadline at all, + // so this can never fire on the request that starts the test. + let closing = self.config.max_test_duration().is_some_and(|maximum| { + session + .lifetime_state() + .max_duration_reached(now_ns, saturating_ns(maximum)) + }); + let next = session.receive_state().accepted(request.sequence); + let next_lifetime = session.lifetime_state().served(now_ns); // The departure side, sampled as late as the core can: after admission // and the transition, immediately before the reply is built. There is // no socket here, so this is the core's departure instant rather than a @@ -248,11 +335,12 @@ impl ServerCore { let sent_at = self.clock.sample(); let params = session.params(); let reply = EchoReply { - // Reply only. Open must be clear or upstream clients abort, and - // Close belongs to the server-initiated close slice. The - // authentication flag and field are the encoder's, decided by the - // key it is given. - flags: FLAG_REPLY, + // Open must be clear or upstream clients abort. Close is set only + // for the maximum-duration limit, which is the sole trigger this + // server has. The authentication flag and field are the encoder's, + // decided by the key it is given — a close-flagged reply is + // authenticated by exactly the same path as any other. + flags: FLAG_REPLY | if closing { FLAG_CLOSE } else { 0 }, // Both are copied through unchanged: no token is allocated for an // echo, and no sequence number is normalized or range-checked. token: request.token, @@ -270,16 +358,34 @@ impl ServerCore { let packet = encode_echo_reply(&reply, params, self.config.hmac_key()) .map_err(|source| ServerError::ReplyEncoding { source })?; - // The last fallible step is behind us, so the transition can be made. - session.commit_receive_state(next); + // The last fallible step is behind us, so the transitions can be made. + if closing { + // The reply already carries the state this request produced, so + // there is nothing worth committing into a session about to go. A + // failure above would have left it live and untouched instead. + self.sessions.remove(&request.token); + } else { + session.commit_rate_state(next_rate); + session.commit_receive_state(next); + session.commit_lifetime_state(next_lifetime); + } Ok(Some(packet)) } + /// Negotiates an open request and, unless it is a no-test one, creates the + /// session it asks for. + /// + /// `now_ns` is the request's lifecycle instant, which becomes the new + /// session's idle origin: a session starts aging the moment it exists, not + /// when it first carries an echo, so one whose client never returns is + /// still reclaimed. Expired sessions were already swept, so the capacity + /// bound below is judged against sessions that are genuinely live. fn handle_open( &mut self, peer: SocketAddr, no_test: bool, encoded_params: &[u8], + now_ns: i64, ) -> Result>, ServerError> { // The parameter decoder already rejects a truncated tag or value, a // varint overflow, an out-of-range stats/stamp-at/clock enum, and a @@ -325,8 +431,10 @@ impl ServerCore { // The reply is encoded before its params move into the session, so // nothing is cloned to keep both. - self.sessions - .insert(token, Session::new(peer, reply.params)); + self.sessions.insert( + token, + Session::new(peer, reply.params, &self.config, now_ns), + ); Ok(Some(packet)) } diff --git a/crates/irtt-server/src/lib.rs b/crates/irtt-server/src/lib.rs index a93d657..629323d 100644 --- a/crates/irtt-server/src/lib.rs +++ b/crates/irtt-server/src/lib.rs @@ -6,14 +6,14 @@ //! # Structure //! //! [`ServerCore`] is the deterministic protocol and session engine: packet -//! admission, authentication policy, open negotiation, the session table, -//! echo receive state, resource decisions and reply construction. It performs -//! no I/O; the clock its echo timestamps 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, expiry -//! and shutdown — will live around it and is intentionally Tokio-native. There -//! is no blocking or alternate-runtime counterpart and no transport -//! abstraction. +//! admission, authentication policy, open negotiation, the session table, echo +//! 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. //! //! # Rejection is silence //! @@ -27,19 +27,23 @@ //! //! Total session state is bounded (see [`DEFAULT_MAX_SESSIONS`]), and so is the //! echo datagram a single session can negotiate (see -//! [`DEFAULT_MAX_PACKET_LENGTH`]). A single unauthenticated datagram creates a -//! session and opens are never deduplicated, so neither an unbounded table nor a -//! remotely chosen buffer size is a compatibility feature. Upstream's observed -//! lack of session bounds, and its unlimited default packet length, are -//! explicitly not compatibility targets. +//! [`DEFAULT_MAX_PACKET_LENGTH`]), the rate one session is answered at (see +//! [`DEFAULT_MIN_SEND_INTERVAL`] and [`DEFAULT_BURST_ALLOWANCE`]) and how long +//! one lives without traffic (see [`DEFAULT_IDLE_TIMEOUT`]). A single +//! unauthenticated datagram creates a session and opens are never deduplicated, +//! so neither an unbounded table nor a remotely chosen buffer size is a +//! compatibility feature. Upstream's observed lack of session bounds, its +//! unlimited default packet length, and its never expiring a session that has +//! not carried an echo request are explicitly not compatibility targets: an +//! `irtt-rs` session ages from the moment it is opened. //! //! # Current scope //! -//! Open handling, session creation, normal echo processing and -//! client-initiated close. Rate limiting, session lifetime and expiry, -//! server-initiated close, DSCP application, the server fill policy and the -//! Tokio runtime are separate slices: every otherwise admissible echo is -//! answered, and a session lives until its client closes it. +//! 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. #![forbid(unsafe_code)] mod clock; diff --git a/crates/irtt-server/src/session.rs b/crates/irtt-server/src/session.rs index 69b970c..c96c992 100644 --- a/crates/irtt-server/src/session.rs +++ b/crates/irtt-server/src/session.rs @@ -1,28 +1,62 @@ -use std::net::SocketAddr; +use std::{net::SocketAddr, time::Duration}; use irtt_proto::Params; +use crate::{clock::saturating_ns, config::ServerConfig}; + +/// How long past its negotiated maximum test duration a session is served +/// before the next echo it answers carries the Close flag. +/// +/// This is `irtt-rs` policy, not a protocol constant, chosen to match the +/// additive two-second margin the clean evidence measured of the reference +/// server at maxima of 500 ms, 1 s and 3 s. It is deliberately not +/// configurable: a conforming client stops at the negotiated duration and never +/// reaches the deadline at all, so the margin exists only to be forgiving of +/// clients that do not, and an operator has the maximum itself to tune. +const MAX_DURATION_CLOSE_GRACE: Duration = Duration::from_secs(2); + /// A live session created by an accepted open request. /// -/// A session records what identifies it, what was negotiated for it, and what -/// its accepted echo requests have added up to. Lifetime state (activity, -/// deadlines, expiry) and rate state belong to the slices that implement those -/// behaviors, and are deliberately absent rather than present as placeholders. +/// A session records what identifies it, what was negotiated for it, what its +/// accepted echo requests have added up to, how much reply allowance it has +/// left, and how long it has to live. /// -/// Deliberately crate-private. The remaining slices are all but certain to add -/// and restructure fields here, and nothing outside the crate needs a session; +/// The three mutable groups are kept apart because they move on different +/// events: [`ReceiveState`] advances only for an echo that is answered, +/// [`RateState`] for every echo that is judged, and [`LifetimeState`] for both — +/// a rate-limited echo refreshes the idle deadline while advancing no statistic. +/// +/// Deliberately crate-private. The remaining slices may still add and +/// restructure fields here, and nothing outside the crate needs a session; /// publishing internal session state would buy nothing and commit us to a shape -/// we are about to change. +/// we may yet change. #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct Session { peer: SocketAddr, params: Params, receive: ReceiveState, + rate: RateState, + lifetime: LifetimeState, } impl Session { - pub(crate) fn new(peer: SocketAddr, params: Params) -> Self { + /// Creates a session for parameters that have already been negotiated and + /// found executable, as of monotonic instant `now_ns`. + /// + /// Rate policy is derived from the configuration and the *negotiated* + /// parameters once, here, rather than recomputed per echo — see + /// [`RateState::new`] for why the two together decide it. The idle deadline + /// starts running immediately: a session that never carries an echo request + /// ages out like any other. + pub(crate) fn new( + peer: SocketAddr, + params: Params, + config: &ServerConfig, + now_ns: i64, + ) -> Self { Self { + rate: RateState::new(config, ¶ms, now_ns), + lifetime: LifetimeState::opened_at(now_ns), peer, params, receive: ReceiveState::default(), @@ -69,6 +103,36 @@ impl Session { pub(crate) fn commit_receive_state(&mut self, next: ReceiveState) { self.receive = next; } + + /// This session's reply allowance as it stands. + pub(crate) fn rate_state(&self) -> RateState { + self.rate + } + + /// Adopts the allowance computed for a judged echo request. + /// + /// Unlike the receive state, this is committed for a *rate-limited* request + /// too: declining to answer is itself bookkeeping the limiter has to keep, + /// or the replenishment it just accounted for would be recomputed from a + /// stale instant on the next request. + pub(crate) fn commit_rate_state(&mut self, next: RateState) { + self.rate = next; + } + + /// How long this session has to live, and when its test duration started. + pub(crate) fn lifetime_state(&self) -> LifetimeState { + self.lifetime + } + + /// Adopts the lifetime state computed for a request. + pub(crate) fn commit_lifetime_state(&mut self, next: LifetimeState) { + self.lifetime = next; + } + + /// Whether this session has gone `idle_timeout_ns` without an echo request. + pub(crate) fn is_idle_expired(&self, now_ns: i64, idle_timeout_ns: i64) -> bool { + self.lifetime.is_idle_expired(now_ns, idle_timeout_ns) + } } /// What a session's accepted echo requests have added up to. @@ -138,3 +202,231 @@ impl ReceiveState { } } } + +/// One session's reply allowance, and the policy that replenishes it. +/// +/// The externally visible behavior is what the clean specification's Section 9.3 +/// records: a session starts with its full burst allowance, spends one unit per +/// echo it is answered, replenishes at one unit per refill interval and never +/// exceeds the burst, and an echo arriving with no allowance gets no reply and +/// advances no reception statistic. The specification is explicit that a +/// token bucket is a *black-box inference* about the reference server, so this +/// representation is an implementation choice, not a protocol requirement — any +/// structure producing those outcomes would do. +/// +/// The policy is copied in per session rather than read from the configuration +/// per echo, because the refill interval depends on what that session +/// negotiated. Sibling sessions therefore limit independently, as they do +/// upstream. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct RateState { + /// The most allowance this session may hold. + burst: u32, + /// How long one unit of allowance takes to accrue. Zero means no + /// time-based throttling at all. + interval_ns: i64, + /// Allowance available now, never above `burst`. + allowance: u32, + /// The monotonic instant `allowance` was last accounted at. + refilled_at_ns: i64, +} + +impl RateState { + /// The full allowance a new session starts with, and the policy it will + /// replenish under. + /// + /// **The refill interval is the shorter of the configured minimum send + /// interval and the interval this session actually negotiated.** Normally + /// they are the configured minimum: a 10 ms minimum against a 1 s + /// negotiated interval refills every 10 ms, which is the ordinary reference + /// behavior. They differ only where negotiation's idle cap pulled the + /// returned interval *below* the configured minimum — a 5 s minimum against + /// an 8 s idle timeout returns 2 s — and there the negotiated value wins. + /// + /// That is a deliberate `irtt-rs` divergence. The clean specification + /// records the reference server replenishing on the configured minimum + /// regardless, so a fully conforming client sending at the 2 s interval it + /// was handed is rate-limited every time by a 5 s allowance. A server must + /// not enforce a cadence it did not advertise, so this one refills at + /// whichever cadence it told the client it would accept. + /// + /// Zero on either side is not throttling. A zero configured minimum + /// disables the time-based floor outright, and a session with no negotiated + /// interval — an absent Interval, which only survives negotiation when the + /// floor is zero as well — has no advertised cadence to honor, so the + /// configured minimum stands alone. A zero refill interval means every + /// otherwise admissible echo is allowed; a zero *burst* still means none is. + fn new(config: &ServerConfig, params: &Params, now_ns: i64) -> Self { + let configured = saturating_ns(config.min_send_interval()); + let interval_ns = if configured <= 0 || params.interval_ns <= 0 { + configured + } else { + configured.min(params.interval_ns) + }; + Self { + burst: config.burst_allowance(), + interval_ns, + allowance: config.burst_allowance(), + refilled_at_ns: now_ns, + } + } + + /// Judges one echo request arriving at monotonic instant `now_ns`. + /// + /// Pure, like [`ReceiveState::accepted`]: it returns the state the caller + /// should commit rather than mutating, so an allowed request can have its + /// reply built and encoded before any of the session moves. Both outcomes + /// carry a next state, because replenishment is accounted for whether or not + /// the request is served. + pub(crate) fn charge(mut self, now_ns: i64) -> RateDecision { + if self.burst == 0 { + // No allowance exists to replenish into, so nothing accrues and + // nothing is served. This is not a synonym for "unlimited". + return RateDecision::Limited { next: self }; + } + if self.interval_ns == 0 { + // No time-based throttling: allowance never has to be spent, + // because it never has to be waited for. + return RateDecision::Allowed { next: self }; + } + self.replenish(now_ns); + if self.allowance == 0 { + return RateDecision::Limited { next: self }; + } + self.allowance -= 1; + RateDecision::Allowed { next: self } + } + + /// Credits whatever whole intervals have elapsed, up to the burst. + fn replenish(&mut self, now_ns: i64) { + if self.allowance >= self.burst { + // Already full. Idle time is not bankable, so the next unit is + // measured from now rather than from a stretch the session spent + // sending nothing. + self.allowance = self.burst; + self.refilled_at_ns = now_ns; + return; + } + // Saturating, so a monotonic reading that somehow ran backwards yields a + // negative elapsed time and credits nothing rather than panicking. + let elapsed = now_ns.saturating_sub(self.refilled_at_ns); + if elapsed < self.interval_ns { + return; + } + let gained = elapsed / self.interval_ns; + let allowance = self + .allowance + .saturating_add(u32::try_from(gained).unwrap_or(u32::MAX)) + .min(self.burst); + self.refilled_at_ns = if allowance == self.burst { + now_ns + } else { + // Carry the unspent remainder of the interval, so a session sending + // at exactly its cadence does not lose a fraction of an interval on + // every request and drift into being limited. + self.refilled_at_ns + .saturating_add(gained.saturating_mul(self.interval_ns)) + }; + self.allowance = allowance; + } +} + +/// What the rate limiter decided about one echo request. +/// +/// Both variants carry the state to commit; they differ in whether a reply is +/// owed. `Limited` is a silent drop — the protocol has no rejection response — +/// and it must leave the reception statistics exactly where they were. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum RateDecision { + Allowed { next: RateState }, + Limited { next: RateState }, +} + +/// How long a session has left, and when its test duration started running. +/// +/// Both instants are monotonic. Lifetime decisions never consult the wall +/// clock: a session must not outlive or predecease its deadline because an +/// administrator or NTP stepped the host clock, and the wall readings a reply +/// carries are a separate concern with the opposite requirement — that they be +/// reported honestly, corrections and all. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct LifetimeState { + /// When this session last did something that keeps it alive: it was opened, + /// it had an echo answered, or it had one rate-limited. + last_activity_ns: i64, + /// When this session's first echo request was actually *served*, if one has + /// been. The maximum test duration is measured from here. + first_served_echo_ns: Option, +} + +impl LifetimeState { + /// The state of a session at the moment it is created. + /// + /// The idle deadline runs from the open, which is a deliberate `irtt-rs` + /// divergence: the clean evidence records the reference server starting it + /// at the first echo request instead, so a session that never carries one + /// never expires. A session is created by a single unauthenticated + /// datagram, so that is an unbounded-state hazard rather than an + /// interoperability feature, and the specification recommends against + /// reproducing it. + fn opened_at(now_ns: i64) -> Self { + Self { + last_activity_ns: now_ns, + first_served_echo_ns: None, + } + } + + /// Whether this session has gone `idle_timeout_ns` without an echo request. + /// + /// The deadline is exact and the comparison inclusive: at + /// `last_activity + idle_timeout` the session is already gone. `irtt-rs` + /// deliberately does not reproduce the reference server's additional + /// five-second grace, nor its lazy final reply to the first echo that finds + /// an expired session. The specification records both as observed upstream + /// policy and states explicitly that the one interoperability constraint is + /// negative — a server must never *signal* expiry — which silence satisfies. + /// + /// A zero timeout expires a session at the first evaluation. It is not a + /// synonym for "never expire". + fn is_idle_expired(&self, now_ns: i64, idle_timeout_ns: i64) -> bool { + now_ns.saturating_sub(self.last_activity_ns) >= idle_timeout_ns + } + + /// Whether an echo arriving now is the one that should carry Close. + /// + /// The deadline is `first served echo + maximum + ` + /// [`MAX_DURATION_CLOSE_GRACE`], so it does not exist at all until an echo + /// has actually been served: neither the open, nor an echo that was + /// rejected, nor one that was rate-limited starts it. That origin is + /// externally measured behavior, not an inference. + pub(crate) fn max_duration_reached(&self, now_ns: i64, max_duration_ns: i64) -> bool { + self.first_served_echo_ns.is_some_and(|first| { + let deadline = first + .saturating_add(max_duration_ns) + .saturating_add(saturating_ns(MAX_DURATION_CLOSE_GRACE)); + now_ns >= deadline + }) + } + + /// The state after an echo that kept the session alive without being + /// served — which is to say, one that was rate-limited. + /// + /// This is the one drop class the clean evidence records as refreshing the + /// idle deadline, and it was established against every other tested class, + /// which does not. It does **not** move the maximum-duration origin: a + /// request that was never answered cannot start the test. + pub(crate) fn refreshed(mut self, now_ns: i64) -> Self { + self.last_activity_ns = now_ns; + self + } + + /// The state after an echo that is about to be answered. + /// + /// Sets the maximum-duration origin if this is the session's first served + /// echo, and refreshes the idle deadline either way. + pub(crate) fn served(mut self, now_ns: i64) -> Self { + self.last_activity_ns = now_ns; + self.first_served_echo_ns.get_or_insert(now_ns); + self + } +} diff --git a/crates/irtt-server/src/tests/echo.rs b/crates/irtt-server/src/tests/echo.rs index 26e2faa..a3d8966 100644 --- a/crates/irtt-server/src/tests/echo.rs +++ b/crates/irtt-server/src/tests/echo.rs @@ -20,13 +20,24 @@ use irtt_proto::{ use super::support::{ core_for, core_with_sources, core_with_tokens, echo_params, echo_request, expect_echo_reply, - open_negotiated, other_peer, peer, sample, ScriptedClock, ScriptedTokens, KEY, OTHER_KEY, + open_negotiated, other_peer, peer, sample, unthrottled, ScriptedClock, ScriptedTokens, KEY, + OTHER_KEY, }; -use crate::ServerConfig; +use crate::{clock::ClockSample, ServerConfig}; const TOKEN_A: u64 = 0x0102_0304_0506_0708; const UNKNOWN_TOKEN: u64 = 0x2122_2324_2526_2728; +/// The lifecycle instant an open consumes before any echo is sent. +/// +/// The core reads the clock once per authenticated request, to judge idle expiry +/// and to start the new session's deadlines. It is earlier than every echo +/// sample below, as a real monotonic reading would be, and no reply ever carries +/// it: an open reply has no timestamp fields. +fn open_sample() -> ClockSample { + sample(900, 9_000) +} + /// Params selecting no optional field at all, so the echo layout is the /// mandatory field block and nothing else. fn bare_params() -> Params { @@ -48,10 +59,13 @@ fn counting_params(length: i64) -> Params { ) } -/// Opens one session on a default core and returns the core, the token and the -/// negotiated params. +/// Opens one session and returns the core, the token and the negotiated params. +/// +/// Rate limiting is disabled: these tests drive a whole vector of echoes at once +/// against the production clock, and what they are about is the reply, not the +/// allowance. `rate` covers the limiter. fn session_with(params: &Params) -> (crate::ServerCore, u64, Params) { - let mut core = core_with_tokens(ServerConfig::default(), ScriptedTokens::new([TOKEN_A])); + let mut core = core_with_tokens(unthrottled(), ScriptedTokens::new([TOKEN_A])); let (token, negotiated) = open_negotiated(&mut core, peer(), params, None); (core, token, negotiated) } @@ -558,7 +572,7 @@ fn only_the_negotiated_timestamps_are_emitted() { }, ), ] { - let clock_source = ScriptedClock::new([received, sent]); + let clock_source = ScriptedClock::new([open_sample(), received, sent]); let mut core = core_with_sources( ServerConfig::default(), ScriptedTokens::new([TOKEN_A]), @@ -569,7 +583,7 @@ fn only_the_negotiated_timestamps_are_emitted() { assert_eq!( clock_source.remaining(), 2, - "an open must not consume a clock sample" + "an open consumes exactly its own lifecycle sample" ); let packet = core @@ -603,7 +617,7 @@ fn a_backward_wall_step_never_reports_receive_after_send() { let mut core = core_with_sources( ServerConfig::default(), ScriptedTokens::new([TOKEN_A]), - ScriptedClock::new([sample(1_000, 10_000), sample(990, 10_300)]), + ScriptedClock::new([open_sample(), sample(1_000, 10_000), sample(990, 10_300)]), ); let params = echo_params(ReceivedStats::None, StampAt::Both, Clock::Both, 0); let (token, negotiated) = open_negotiated(&mut core, peer(), ¶ms, None); @@ -633,7 +647,7 @@ fn a_backward_wall_step_never_reports_receive_after_send() { let mut core = core_with_sources( ServerConfig::default(), ScriptedTokens::new([TOKEN_A]), - ScriptedClock::new([sample(1_000, 10_000), sample(990, 10_300)]), + ScriptedClock::new([open_sample(), sample(1_000, 10_000), sample(990, 10_300)]), ); let params = echo_params(ReceivedStats::None, StampAt::Midpoint, Clock::Wall, 0); let (token, negotiated) = open_negotiated(&mut core, peer(), ¶ms, None); @@ -678,7 +692,7 @@ fn a_single_clock_midpoint_emits_exactly_one_field() { let mut core = core_with_sources( ServerConfig::default(), ScriptedTokens::new([TOKEN_A]), - ScriptedClock::new([sample(1_000, 10_000), sample(1_200, 10_300)]), + ScriptedClock::new([open_sample(), sample(1_000, 10_000), sample(1_200, 10_300)]), ); let params = echo_params(ReceivedStats::None, StampAt::Midpoint, clock, 0); let (token, negotiated) = open_negotiated(&mut core, peer(), ¶ms, None); diff --git a/crates/irtt-server/src/tests/lifecycle.rs b/crates/irtt-server/src/tests/lifecycle.rs new file mode 100644 index 0000000..76b1463 --- /dev/null +++ b/crates/irtt-server/src/tests/lifecycle.rs @@ -0,0 +1,580 @@ +//! Session lifetime: idle expiry and the maximum-duration close. +//! +//! Time is a hand-moved monotonic clock, so every deadline here is exact and no +//! test sleeps, tolerates or races. +//! +//! Two of the behaviors pinned here are deliberate `irtt-rs` policy rather than +//! observed reference behavior, and each says so where it is tested: expiry is +//! immediate at the configured deadline with no reply of any kind, and the +//! deadline covers a session that has never carried an echo request. The clean +//! specification records the reference server's five-second grace, its final +//! lazy-release reply and its never-expiring unused sessions as upstream policy, +//! states that expiry is entirely a server's own business, and recommends +//! against reproducing the resource leak. The one interoperability constraint is +//! negative — a server must never *signal* expiry — and silence satisfies it. + +use std::time::Duration; + +use irtt_proto::{Clock, Params, ReceivedStats, StampAt, FLAG_CLOSE}; + +use super::support::{ + close_request, core_with_sources, echo_at, echo_params, echo_request, + expect_closing_echo_reply, expect_echo_reply, expect_no_test_reply, manual_core, + no_test_request, open_request, other_peer, peer, ManualClock, ScriptedTokens, KEY, +}; +use crate::ServerConfig; + +const TOKEN_A: u64 = 0x0102_0304_0506_0708; +const TOKEN_B: u64 = 0x1112_1314_1516_1718; + +const MS: i64 = 1_000_000; +const SECOND: i64 = 1_000_000_000; + +/// Params reporting both statistics and requesting `interval_ns`. +fn params_at(interval_ns: i64) -> Params { + Params { + interval_ns, + ..echo_params( + ReceivedStats::Both, + StampAt::None, + Clock::Unspecified, + /* length */ 0, + ) + } +} + +/// A configuration that expires after `idle` and rate-limits nothing. +fn idle_after(idle: Duration) -> ServerConfig { + ServerConfig::default() + .with_idle_timeout(idle) + .with_min_send_interval(Duration::ZERO) +} + +fn count_of(packet: Option>, params: &Params) -> Option { + let packet = packet?; + expect_echo_reply(&packet, params, None).recv_count +} + +#[test] +fn a_session_that_never_carried_an_echo_still_expires() { + // `irtt-rs` policy. The reference server starts the idle clock at the first + // echo request, so a session left unused never expires at all; a session is + // created by one unauthenticated datagram and opens are never deduplicated, + // which makes that an unbounded-state hazard rather than a compatibility + // feature. Here the deadline runs from the open. + // + // A one-session table makes the reclamation externally visible: the second + // open can only succeed if the first session was released before capacity + // was judged. + let config = idle_after(Duration::from_secs(2)).with_max_sessions(1); + let clock = ManualClock::at(0); + let mut core = core_with_sources( + config, + ScriptedTokens::new([TOKEN_A, TOKEN_B]), + clock.clone(), + ); + let requested = params_at(SECOND); + + clock.set(0); + assert!(core + .handle_datagram(peer(), &open_request(&requested, None)) + .unwrap() + .is_some()); + assert_eq!(core.session_count(), 1); + + // Still inside the deadline: the table is full and a second open is refused. + clock.set(2 * SECOND - 1); + assert_eq!( + core.handle_datagram(other_peer(), &open_request(&requested, None)) + .unwrap(), + None, + "a live session must still occupy the table" + ); + assert_eq!(core.session_count(), 1); + + // At the deadline the unused session is reclaimed, before capacity is + // judged, so the new open fits. + clock.set(2 * SECOND); + assert!( + core.handle_datagram(other_peer(), &open_request(&requested, None)) + .unwrap() + .is_some(), + "an expired session must not deny capacity to a new one" + ); + assert_eq!(core.session_count(), 1, "one replaced the other"); + + // And the reclaimed token names nothing. + assert!( + echo_at(&mut core, &clock, 2 * SECOND, TOKEN_A, 0, &requested, None).is_none(), + "the released token must be unusable" + ); +} + +#[test] +fn a_served_echo_moves_the_idle_deadline_along() { + let (mut core, clock, token, negotiated) = manual_core( + idle_after(Duration::from_secs(2)), + ¶ms_at(SECOND), + ScriptedTokens::new([TOKEN_A]), + None, + ); + + // Inside the deadline the open set, and past the one it would have had. + assert_eq!( + count_of( + echo_at(&mut core, &clock, SECOND, token, 0, &negotiated, None), + &negotiated + ), + Some(1) + ); + assert_eq!( + count_of( + echo_at(&mut core, &clock, 2_900 * MS, token, 1, &negotiated, None), + &negotiated + ), + Some(2), + "the echo at 1 s moved the deadline from 2 s to 3 s" + ); + + // Exactly at the deadline the last served echo set: gone, and silently. + assert!( + echo_at(&mut core, &clock, 4_900 * MS, token, 2, &negotiated, None).is_none(), + "the deadline is inclusive" + ); + assert_eq!(core.session_count(), 0); +} + +#[test] +fn an_expired_session_answers_nothing_at_all() { + // `irtt-rs` policy, and the reason this is its own test. The reference + // server was observed answering the first otherwise-serviceable echo to + // reach an expired session — a final lazy-release reply — and only dropping + // the one after it, with the release boundary a further five seconds past + // the configured timeout. `irtt-rs` reproduces neither: at the deadline the + // session is simply gone, and the client observes silence, which is the only + // thing the specification actually requires of expiry. + let (mut core, clock, token, negotiated) = manual_core( + idle_after(Duration::from_secs(2)), + ¶ms_at(SECOND), + ScriptedTokens::new([TOKEN_A]), + None, + ); + + assert_eq!( + count_of( + echo_at( + &mut core, + &clock, + 2 * SECOND - 1, + token, + 0, + &negotiated, + None + ), + &negotiated + ), + Some(1), + "one nanosecond short of the deadline is still served" + ); + + // Two seconds after that echo, to the nanosecond. + assert!( + echo_at( + &mut core, + &clock, + 4 * SECOND - 1, + token, + 1, + &negotiated, + None + ) + .is_none(), + "no final reply: an expired session is simply gone" + ); + assert_eq!(core.session_count(), 0); + assert!( + echo_at(&mut core, &clock, 4 * SECOND, token, 2, &negotiated, None).is_none(), + "and stays gone" + ); +} + +#[test] +fn a_rate_limited_echo_keeps_its_session_alive_without_counting() { + // The one drop class the clean evidence records as refreshing the idle + // deadline. It was established against every other tested class, which does + // not, so `irtt-rs` keeps it. + // + // A 2 s minimum interval against an 8 s idle timeout negotiates 2 s and + // refills every 2 s; a burst of one then makes the echo at 8.5 s — 1.5 s + // after the previous one — rate-limited. + let config = ServerConfig::default() + .with_min_send_interval(Duration::from_secs(2)) + .with_idle_timeout(Duration::from_secs(8)) + .with_burst_allowance(1); + let (mut core, clock, token, negotiated) = manual_core( + config, + ¶ms_at(2 * SECOND), + ScriptedTokens::new([TOKEN_A]), + None, + ); + assert_eq!(negotiated.interval_ns, 2 * SECOND); + + // Served, spending the burst. The idle deadline is now 15 s. + assert_eq!( + count_of( + echo_at(&mut core, &clock, 7 * SECOND, token, 0, &negotiated, None), + &negotiated + ), + Some(1) + ); + + // Rate-limited, 1.5 s later: no reply, and the deadline moves to 16.5 s. + assert!( + echo_at(&mut core, &clock, 8_500 * MS, token, 1, &negotiated, None).is_none(), + "no allowance, so no reply" + ); + + // Past the deadline the served echo alone would have set, and short of the + // one the rate-limited echo set. + assert_eq!( + count_of( + echo_at(&mut core, &clock, 16 * SECOND, token, 2, &negotiated, None), + &negotiated + ), + Some(2), + "the rate-limited echo kept the session alive, and was not counted" + ); +} + +#[test] +fn a_rejected_echo_does_not_move_the_idle_deadline() { + // Two representative classes; the admission rules themselves are pinned by + // the echo and hmac tests, and are not repeated here. + let config = idle_after(Duration::from_secs(2)).with_max_packet_length(64); + let (mut core, clock, token, negotiated) = manual_core( + config, + ¶ms_at(SECOND), + ScriptedTokens::new([TOKEN_A]), + None, + ); + + let oversized = echo_request( + token, + 0, + &Params { + length: 65, + ..negotiated.clone() + }, + &[], + None, + ); + assert_eq!(oversized.len(), 65); + + clock.set(2 * SECOND - 1); + assert_eq!(core.handle_datagram(peer(), &oversized).unwrap(), None); + assert_eq!( + core.handle_datagram( + other_peer(), + &echo_request(token, 0, &negotiated, &[], None) + ) + .unwrap(), + None, + "a foreign endpoint is refused whatever the session's state" + ); + assert_eq!(core.session_count(), 1, "neither released it early either"); + + // The deadline the open set, unmoved. + assert!( + echo_at(&mut core, &clock, 2 * SECOND, token, 0, &negotiated, None).is_none(), + "a rejected echo must not have extended the session" + ); + assert_eq!(core.session_count(), 0); +} + +#[test] +fn any_authenticated_request_reclaims_expired_sessions_including_a_no_test_open() { + // `irtt-rs` policy. Expiry is a decision about the table, so it runs before + // a request is dispatched and does not depend on what that request is for. + // The reference server was observed releasing an expired session on a burst + // of ordinary opens but not on no-test ones; a client cannot tell, since + // expiry is never signalled, and the distinction is not worth reproducing. + // + // A no-test open still creates nothing of its own, which the token source + // and the session count both check here. + let tokens = ScriptedTokens::new([TOKEN_A]); + let clock = ManualClock::at(0); + let mut core = core_with_sources( + idle_after(Duration::from_secs(2)), + tokens.clone(), + clock.clone(), + ); + let requested = params_at(SECOND); + + assert!(core + .handle_datagram(peer(), &open_request(&requested, None)) + .unwrap() + .is_some()); + assert_eq!(core.session_count(), 1); + assert_eq!(tokens.remaining(), 0); + + clock.set(2 * SECOND); + let packet = core + .handle_datagram(other_peer(), &no_test_request(&requested, None)) + .unwrap() + .expect("a no-test open is answered whatever the table holds"); + expect_no_test_reply(&packet, None); + + assert_eq!( + core.session_count(), + 0, + "the expired session was reclaimed, and no-test created none of its own" + ); +} + +#[test] +fn a_client_close_still_releases_its_session_without_a_reply() { + // Unchanged by this slice, and worth pinning next to the close the *server* + // sends: a client close is answered with nothing at all. + let (mut core, clock, token, negotiated) = manual_core( + idle_after(Duration::from_secs(60)), + ¶ms_at(SECOND), + ScriptedTokens::new([TOKEN_A]), + None, + ); + assert!(echo_at(&mut core, &clock, SECOND, token, 0, &negotiated, None).is_some()); + + clock.set(2 * SECOND); + assert_eq!( + core.handle_datagram(peer(), &close_request(token, None)) + .unwrap(), + None + ); + assert_eq!(core.session_count(), 0); +} + +/// A configuration whose sessions run for `maximum` and are never idle-expired +/// within a test's timescale. +fn max_duration_of(maximum: Duration) -> ServerConfig { + ServerConfig::default() + .with_max_test_duration(maximum) + .with_idle_timeout(Duration::from_secs(600)) +} + +#[test] +fn the_maximum_duration_deadline_starts_at_the_first_served_echo() { + // Measured behavior: opening a session well before its first echo does not + // move the deadline, and neither a rejected nor a rate-limited first request + // starts it. + let (mut core, clock, token, negotiated) = manual_core( + max_duration_of(Duration::from_secs(1)), + ¶ms_at(SECOND), + ScriptedTokens::new([TOKEN_A]), + None, + ); + assert_eq!( + negotiated.duration_ns, SECOND, + "the requested duration was reduced to the maximum" + ); + + // Ten seconds after the open — far past maximum + grace, had the open + // started the clock. + let packet = echo_at(&mut core, &clock, 10 * SECOND, token, 0, &negotiated, None) + .expect("the first echo must be answered"); + let reply = expect_echo_reply(&packet, &negotiated, None); + assert_eq!(reply.recv_count, Some(1)); + + // The deadline is 13 s: this echo, three seconds after the open but two + // before the origin's own deadline, is ordinary. + let packet = echo_at(&mut core, &clock, 12_999 * MS, token, 1, &negotiated, None) + .expect("an echo inside the deadline must be answered"); + assert_eq!( + expect_echo_reply(&packet, &negotiated, None).recv_count, + Some(2) + ); + assert_eq!(core.session_count(), 1); +} + +#[test] +fn the_first_echo_past_the_maximum_duration_carries_close_and_ends_the_session() { + let (mut core, clock, token, negotiated) = manual_core( + max_duration_of(Duration::from_secs(1)), + ¶ms_at(SECOND), + ScriptedTokens::new([TOKEN_A]), + None, + ); + + assert!(echo_at(&mut core, &clock, 10 * SECOND, token, 0, &negotiated, None).is_some()); + // One nanosecond short of maximum + 2 s grace, measured from that echo. + let packet = echo_at( + &mut core, + &clock, + 13 * SECOND - 1, + token, + 1, + &negotiated, + None, + ) + .expect("short of the deadline must be answered normally"); + let reply = expect_echo_reply(&packet, &negotiated, None); + assert_eq!(reply.flags & FLAG_CLOSE, 0, "not yet"); + + // At the deadline exactly. + let packet = echo_at(&mut core, &clock, 13 * SECOND, token, 2, &negotiated, None) + .expect("the deadline-crossing echo is answered, with Close"); + let reply = expect_closing_echo_reply(&packet, &negotiated, None); + assert_eq!(reply.token, token, "an ordinary reply in every other way"); + assert_eq!(reply.sequence, 2, "the triggering sequence, copied through"); + assert_eq!( + (reply.recv_count, reply.recv_window), + (Some(3), Some(0x7)), + "the triggering request is included in the statistics it reports" + ); + assert!(reply.payload.is_empty()); + assert_eq!( + packet.len(), + irtt_proto::echo_packet_len(false, &negotiated).unwrap(), + "the negotiated layout, unchanged by the flag" + ); + + // The token is unusable from that reply onward. + assert_eq!(core.session_count(), 0); + assert!( + echo_at( + &mut core, + &clock, + 13 * SECOND + 1, + token, + 3, + &negotiated, + None + ) + .is_none(), + "every later request is an unknown token" + ); +} + +#[test] +fn a_rate_limited_echo_at_the_deadline_defers_the_close_rather_than_losing_it() { + // The measured dropped-trigger case, and the reason rate allowance is + // judged before the deadline: a deadline-crossing echo with no allowance is + // silently dropped, the session stays usable, and the next echo that *is* + // served carries the close. + // + // A 5 s minimum interval against a 600 s idle timeout negotiates 5 s and + // refills every 5 s, so the echo three seconds after the first is limited. + let config = max_duration_of(Duration::from_secs(1)) + .with_min_send_interval(Duration::from_secs(5)) + .with_burst_allowance(1); + let (mut core, clock, token, negotiated) = manual_core( + config, + ¶ms_at(5 * SECOND), + ScriptedTokens::new([TOKEN_A]), + None, + ); + assert_eq!(negotiated.interval_ns, 5 * SECOND); + + assert_eq!( + count_of( + echo_at(&mut core, &clock, 10 * SECOND, token, 0, &negotiated, None), + &negotiated + ), + Some(1), + "the origin, which also spends the burst" + ); + + assert!( + echo_at(&mut core, &clock, 13 * SECOND, token, 1, &negotiated, None).is_none(), + "past the deadline but with no allowance: dropped, not closed" + ); + assert_eq!(core.session_count(), 1, "and the session is still usable"); + + // Two seconds later the allowance is back. + let packet = echo_at(&mut core, &clock, 15 * SECOND, token, 2, &negotiated, None) + .expect("the next served echo carries the close"); + let reply = expect_closing_echo_reply(&packet, &negotiated, None); + assert_eq!( + reply.recv_count, + Some(2), + "the rate-limited request was never counted" + ); + assert_eq!(core.session_count(), 0); +} + +#[test] +fn a_rejected_echo_past_the_deadline_neither_closes_nor_releases() { + // The close trigger runs only after ordinary admission and rate allowance + // succeed, so a datagram the server would refuse at any time refuses in the + // ordinary way here too. The specification records the corresponding + // upstream behavior as untested for classes other than rate limiting, so + // this is the deterministic `irtt-rs` reading rather than an observed one. + let (mut core, clock, token, negotiated) = manual_core( + max_duration_of(Duration::from_secs(1)), + ¶ms_at(SECOND), + ScriptedTokens::new([TOKEN_A]), + None, + ); + assert!(echo_at(&mut core, &clock, 10 * SECOND, token, 0, &negotiated, None).is_some()); + + clock.set(14 * SECOND); + assert_eq!( + core.handle_datagram( + other_peer(), + &echo_request(token, 1, &negotiated, &[], None) + ) + .unwrap(), + None, + "a foreign endpoint gets silence, not a close" + ); + assert_eq!(core.session_count(), 1, "and does not release the session"); + + let packet = echo_at(&mut core, &clock, 14_100 * MS, token, 2, &negotiated, None) + .expect("the owning endpoint still receives the close"); + let reply = expect_closing_echo_reply(&packet, &negotiated, None); + assert_eq!( + reply.recv_count, + Some(2), + "the foreign request advanced nothing" + ); + assert_eq!(core.session_count(), 0); +} + +#[test] +fn an_authenticated_session_closes_through_the_ordinary_reply_path() { + // No special close authentication exists, and none should: the flag rides on + // a reply the normal encoder and key path produced. + let config = max_duration_of(Duration::from_secs(1)).with_hmac_key(KEY); + let (mut core, clock, token, negotiated) = manual_core( + config, + ¶ms_at(SECOND), + ScriptedTokens::new([TOKEN_A]), + Some(KEY), + ); + + assert!(echo_at( + &mut core, + &clock, + 10 * SECOND, + token, + 0, + &negotiated, + Some(KEY) + ) + .is_some()); + + let packet = echo_at( + &mut core, + &clock, + 13 * SECOND, + token, + 1, + &negotiated, + Some(KEY), + ) + .expect("the deadline-crossing echo is answered"); + // Decoding verifies the MAC, and the flag comparison pins HMAC and Close + // together. + let reply = expect_closing_echo_reply(&packet, &negotiated, Some(KEY)); + assert_eq!(reply.recv_count, Some(2)); + irtt_proto::verify_packet_hmac(KEY, &packet).expect("the close-flagged reply is authenticated"); + assert_eq!(core.session_count(), 0); +} diff --git a/crates/irtt-server/src/tests/mod.rs b/crates/irtt-server/src/tests/mod.rs index b5bafdb..278e5d9 100644 --- a/crates/irtt-server/src/tests/mod.rs +++ b/crates/irtt-server/src/tests/mod.rs @@ -15,10 +15,12 @@ mod admission; mod close; mod echo; mod hmac; +mod lifecycle; mod negotiation; mod no_test; mod open; mod params; +mod rate; mod sessions; mod support; mod tokens; diff --git a/crates/irtt-server/src/tests/rate.rs b/crates/irtt-server/src/tests/rate.rs new file mode 100644 index 0000000..aa0486c --- /dev/null +++ b/crates/irtt-server/src/tests/rate.rs @@ -0,0 +1,257 @@ +//! Per-session reply allowance. +//! +//! The behavior pinned here is the clean specification's Section 9.3: a session +//! starts with its full burst, spends one unit per echo it is answered, +//! replenishes at one unit per refill interval, and drops an echo arriving with +//! no allowance without a reply and without advancing either reception +//! statistic. That last part is what the counts in the *later* replies prove. +//! +//! The token bucket itself is a black-box inference about the reference server, +//! not a protocol requirement, so nothing here reaches into the limiter's +//! representation: every assertion is about which datagrams came back. + +use std::time::Duration; + +use irtt_proto::{Clock, Params, ReceivedStats, StampAt}; + +use super::support::{ + core_with_sources, echo_at, echo_params, echo_request, expect_echo_reply, manual_core, + open_negotiated, other_peer, peer, ManualClock, ScriptedTokens, +}; +use crate::ServerConfig; + +const TOKEN_A: u64 = 0x0102_0304_0506_0708; +const TOKEN_B: u64 = 0x1112_1314_1516_1718; + +const MS: i64 = 1_000_000; +const SECOND: i64 = 1_000_000_000; + +/// Params reporting both statistics — the count is how a test tells a served +/// request from a dropped one — and requesting `interval_ns`. +fn params_at(interval_ns: i64) -> Params { + Params { + interval_ns, + ..echo_params( + ReceivedStats::Both, + StampAt::None, + Clock::Unspecified, + /* length */ 0, + ) + } +} + +/// The received count a reply reported, or `None` if there was no reply. +fn count_of(packet: Option>, params: &Params) -> Option { + let packet = packet?; + expect_echo_reply(&packet, params, None).recv_count +} + +#[test] +fn a_blast_is_answered_up_to_the_burst_and_then_replenishes() { + // The measured case: a 100 ms minimum interval and a burst of 5 answered + // exactly 5 of a blast of 12, and the next 3 requests after a 600 ms pause + // were all answered, with the count running 1..5 and then 6, 7, 8. + let config = ServerConfig::default() + .with_min_send_interval(Duration::from_millis(100)) + .with_burst_allowance(5); + let (mut core, clock, token, negotiated) = manual_core( + config, + ¶ms_at(SECOND), + ScriptedTokens::new([TOKEN_A]), + None, + ); + // The refill cadence is the configured minimum here, not the negotiated + // interval: the ordinary case, where the minimum is the shorter of the two. + assert_eq!(negotiated.interval_ns, SECOND); + + for sequence in 0..12 { + let count = count_of( + echo_at(&mut core, &clock, 0, token, sequence, &negotiated, None), + &negotiated, + ); + let expected = (sequence < 5).then_some(sequence + 1); + assert_eq!(count, expected, "request {sequence} of the blast"); + } + + // A pause long enough to refill the whole burst, then three requests spaced + // by one refill interval each. The counts continuing 6, 7, 8 is what proves + // the seven dropped requests advanced nothing. + for (offset, sequence, expected) in [(600, 12, 6), (700, 13, 7), (800, 14, 8)] { + let count = count_of( + echo_at( + &mut core, + &clock, + offset * MS, + token, + sequence, + &negotiated, + None, + ), + &negotiated, + ); + assert_eq!(count, Some(expected), "request at {offset} ms"); + } +} + +#[test] +fn a_burst_of_one_answers_one_request_per_interval() { + // The other measured case: with a burst of 1, a blast of 4 produced 1 reply. + let config = ServerConfig::default() + .with_min_send_interval(Duration::from_millis(100)) + .with_burst_allowance(1); + let (mut core, clock, token, negotiated) = manual_core( + config, + ¶ms_at(SECOND), + ScriptedTokens::new([TOKEN_A]), + None, + ); + + for sequence in 0..4 { + let count = count_of( + echo_at(&mut core, &clock, 0, token, sequence, &negotiated, None), + &negotiated, + ); + assert_eq!(count, (sequence == 0).then_some(1), "request {sequence}"); + } + + // Exactly one interval later the allowance is back, and the count says the + // three dropped requests were never counted. + let count = count_of( + echo_at(&mut core, &clock, 100 * MS, token, 4, &negotiated, None), + &negotiated, + ); + assert_eq!(count, Some(2)); +} + +#[test] +fn the_refill_cadence_is_never_slower_than_the_negotiated_interval() { + // The deliberate `irtt-rs` divergence, and the reason this test exists. + // + // A 5 s minimum interval against an 8 s idle timeout negotiates 2 s, because + // the idle cap is applied after the floor. The clean specification records + // the reference server continuing to replenish every 5 s in that + // configuration, so a fully conforming client sending at the 2 s interval it + // was handed is rate-limited on every request. `irtt-rs` refills at the + // interval it advertised instead. + let config = ServerConfig::default() + .with_min_send_interval(Duration::from_secs(5)) + .with_idle_timeout(Duration::from_secs(8)) + .with_burst_allowance(1); + let (mut core, clock, token, negotiated) = manual_core( + config, + ¶ms_at(10 * MS), + ScriptedTokens::new([TOKEN_A]), + None, + ); + assert_eq!( + negotiated.interval_ns, + 2 * SECOND, + "the idle cap wins over the larger configured minimum" + ); + + assert_eq!( + count_of( + echo_at(&mut core, &clock, 0, token, 0, &negotiated, None), + &negotiated + ), + Some(1), + "the first request spends the burst" + ); + + // One negotiated interval later — and well short of the 5 s the reference + // server would have made this client wait. + assert_eq!( + count_of( + echo_at(&mut core, &clock, 2 * SECOND, token, 1, &negotiated, None), + &negotiated + ), + Some(2), + "a client sending at the interval it was given must not be limited" + ); + assert_eq!( + count_of( + echo_at(&mut core, &clock, 4 * SECOND, token, 2, &negotiated, None), + &negotiated + ), + Some(3), + "and must keep being answered at that cadence" + ); +} + +#[test] +fn a_zero_burst_refuses_every_echo_and_a_zero_interval_refuses_none() { + // The two zero cases are deliberately not the same. A zero burst is no + // allowance at all; a zero interval is no waiting for allowance. + let (mut core, clock, token, negotiated) = manual_core( + ServerConfig::default().with_burst_allowance(0), + ¶ms_at(SECOND), + ScriptedTokens::new([TOKEN_A]), + None, + ); + for (offset, sequence) in [(0, 0), (SECOND, 1), (60 * SECOND, 2)] { + assert!( + echo_at( + &mut core, + &clock, + offset, + token, + sequence, + &negotiated, + None + ) + .is_none(), + "a zero burst answers nothing, however long the wait" + ); + } + + let (mut core, clock, token, negotiated) = manual_core( + ServerConfig::default().with_min_send_interval(Duration::ZERO), + ¶ms_at(SECOND), + ScriptedTokens::new([TOKEN_A]), + None, + ); + for sequence in 0..12 { + assert_eq!( + count_of( + echo_at(&mut core, &clock, 0, token, sequence, &negotiated, None), + &negotiated + ), + Some(sequence + 1), + "a zero interval never withholds allowance, whatever the burst" + ); + } +} + +#[test] +fn sibling_sessions_hold_their_allowance_independently() { + let config = ServerConfig::default() + .with_min_send_interval(Duration::from_millis(100)) + .with_burst_allowance(1); + let clock = ManualClock::at(0); + let mut core = core_with_sources( + config, + ScriptedTokens::new([TOKEN_A, TOKEN_B]), + clock.clone(), + ); + let requested = params_at(SECOND); + let (token_a, negotiated) = open_negotiated(&mut core, peer(), &requested, None); + let (token_b, _) = open_negotiated(&mut core, other_peer(), &requested, None); + + // The first session spends its whole allowance and is then limited. + assert!(echo_at(&mut core, &clock, 0, token_a, 0, &negotiated, None).is_some()); + assert!(echo_at(&mut core, &clock, 0, token_a, 1, &negotiated, None).is_none()); + + // The second still has its own, from its own endpoint. + clock.set(0); + let packet = core + .handle_datagram( + other_peer(), + &echo_request(token_b, 0, &negotiated, &[], None), + ) + .unwrap() + .expect("a sibling session's allowance is its own"); + assert_eq!( + expect_echo_reply(&packet, &negotiated, None).recv_count, + Some(1) + ); +} diff --git a/crates/irtt-server/src/tests/support.rs b/crates/irtt-server/src/tests/support.rs index 2f12853..068d5eb 100644 --- a/crates/irtt-server/src/tests/support.rs +++ b/crates/irtt-server/src/tests/support.rs @@ -94,8 +94,13 @@ impl TokenSource for ScriptedTokens { /// /// Running the script dry is a test bug, not a wrap-around: a test that asserts /// timestamps must know exactly which sample each field came from. The core -/// takes one sample when it classifies a datagram as an echo, and a second when -/// an admitted echo is about to be answered. +/// takes one sample per authenticated request — when it classifies a datagram as +/// an echo, or after authenticating an open or close — and one more when an +/// admitted echo is about to be answered. +/// +/// Tests about lifetime and rate want [`ManualClock`] instead: they care about +/// *when* a datagram arrives, not about how many times the core reads the clock +/// on the way. #[derive(Debug, Clone)] pub(crate) struct ScriptedClock { samples: Arc>>, @@ -131,6 +136,47 @@ pub(crate) fn sample(wall_ns: i64, mono_ns: i64) -> ClockSample { ClockSample { wall_ns, mono_ns } } +/// A clock standing still at an instant the test moves by hand. +/// +/// Lifetime and rate behavior is a question of when a datagram arrives, so these +/// tests set the time, send a packet, and set it again — with no sleeps, no +/// tolerances and nothing riding on how many times the core reads the clock. +/// [`ScriptedClock`] remains for the timestamp tests, where every individual +/// reading is the point. +/// +/// Both domains read the same value. Lifetime decisions are monotonic-only by +/// design, and no test here asserts a wall timestamp, so keeping the pair equal +/// makes the readings obviously ordered. +#[derive(Debug, Clone)] +pub(crate) struct ManualClock { + now_ns: Arc>, +} + +impl ManualClock { + pub(crate) fn at(now_ns: i64) -> Self { + Self { + now_ns: Arc::new(Mutex::new(now_ns)), + } + } + + /// Moves the clock to an exact instant. Deliberately absolute: a test that + /// pins a deadline reads better stating the instant than accumulating + /// offsets. + pub(crate) fn set(&self, now_ns: i64) { + *self.now_ns.lock().unwrap() = now_ns; + } +} + +impl ClockSource for ManualClock { + fn sample(&mut self) -> ClockSample { + let now_ns = *self.now_ns.lock().unwrap(); + ClockSample { + wall_ns: now_ns, + mono_ns: now_ns, + } + } +} + /// A core with deterministic tokens, so session identity is assertable. Its /// clock is the production one, which is fine for everything that does not /// assert timestamp values. @@ -138,11 +184,12 @@ pub(crate) fn core_with_tokens(config: ServerConfig, tokens: ScriptedTokens) -> ServerCore::with_token_source(config, Box::new(tokens)) } -/// A core with both nondeterministic sources scripted, for timestamp tests. +/// A core with both nondeterministic sources chosen, for timestamp, rate and +/// lifetime tests. pub(crate) fn core_with_sources( config: ServerConfig, tokens: ScriptedTokens, - clock: ScriptedClock, + clock: impl ClockSource + 'static, ) -> ServerCore { ServerCore::with_sources(config, Box::new(tokens), Box::new(clock)) } @@ -159,6 +206,40 @@ pub(crate) fn core_for(hmac_key: Option<&[u8]>, tokens: ScriptedTokens) -> Serve core_with_tokens(config, tokens) } +/// A core whose clock the test moves by hand, with one session already open. +/// +/// The clock starts at monotonic zero and the session is opened there, so every +/// instant a rate or lifetime test names reads as an offset from the open. +pub(crate) fn manual_core( + config: ServerConfig, + requested: &Params, + tokens: ScriptedTokens, + hmac_key: Option<&[u8]>, +) -> (ServerCore, ManualClock, u64, Params) { + let clock = ManualClock::at(0); + let mut core = core_with_sources(config, tokens, clock.clone()); + let (token, negotiated) = open_negotiated(&mut core, peer(), requested, hmac_key); + (core, clock, token, negotiated) +} + +/// Sends one echo as it would arrive at monotonic instant `at_ns`. +pub(crate) fn echo_at( + core: &mut ServerCore, + clock: &ManualClock, + at_ns: i64, + token: u64, + sequence: u32, + params: &Params, + hmac_key: Option<&[u8]>, +) -> Option> { + clock.set(at_ns); + core.handle_datagram( + peer(), + &echo_request(token, sequence, params, &[], hmac_key), + ) + .expect("a rejected or rate-limited echo is not an internal error") +} + /// Opens one ordinary session from `endpoint` and returns its token. pub(crate) fn open_session( core: &mut ServerCore, @@ -246,6 +327,24 @@ pub(crate) fn expect_echo_reply( reply } +/// Decodes a server-initiated close: an ordinary echo reply carrying Close. +/// +/// The flags are compared exactly, so Open is pinned clear — a reply with it set +/// makes an upstream client abort — and the reply is decoded and its MAC checked +/// by the same path as any other, which is the point: the close is not a +/// separate packet kind. +pub(crate) fn expect_closing_echo_reply( + packet: &[u8], + params: &Params, + hmac_key: Option<&[u8]>, +) -> EchoReply { + let reply = + decode_echo_reply(packet, params, hmac_key).expect("a closing echo reply must decode"); + let expected = FLAG_REPLY | FLAG_CLOSE | if hmac_key.is_some() { FLAG_HMAC } else { 0 }; + assert_eq!(reply.flags, expected, "closing echo reply flags"); + reply +} + /// Encodes a normal open request carrying `params`. pub(crate) fn open_request(params: &Params, hmac_key: Option<&[u8]>) -> Vec { encode_request(