From 7a5f790cace83e3d4d02074d74f52fc7d6747bb7 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 3 Nov 2026 13:45:36 +1300 Subject: [PATCH 1/6] Add spawned server contexts, split raw peers and a two-way streaming case to `base-http-fixtures`. Co-Authored-By: Claude Opus 5.5 --- crates/client/tests/protocol.rs | 30 ++++ crates/fixtures/readme.md | 4 +- crates/fixtures/src/context.rs | 277 +++++++++++++++++++++++++++++- crates/fixtures/src/lib.rs | 8 +- crates/fixtures/src/protocol.rs | 95 +++++++++- crates/fixtures/src/raw.rs | 79 +++++++++ crates/fixtures/src/scheduler.rs | 22 ++- crates/fixtures/tests/protocol.rs | 42 ++++- crates/fixtures/tests/raw.rs | 80 ++++++++- crates/fixtures/tests/spawned.rs | 230 +++++++++++++++++++++++++ design/fixtures.md | 37 +++- 11 files changed, 883 insertions(+), 21 deletions(-) create mode 100644 crates/fixtures/tests/spawned.rs diff --git a/crates/client/tests/protocol.rs b/crates/client/tests/protocol.rs index 830b608..04acbe0 100644 --- a/crates/client/tests/protocol.rs +++ b/crates/client/tests/protocol.rs @@ -5,6 +5,7 @@ use base_http::Version; use base_http_fixtures::{Context, ServerContext, protocol::App}; use base_scheduler_native::NativeScheduler; +use std::sync::Arc; fn serve(app: App, version: Version) -> ServerContext { ServerContext::new().handler(app).version(version) @@ -38,3 +39,32 @@ mod https { base_http_fixtures::protocol_tests!(run: super::https); base_http_fixtures::protocol_trailer_tests!(run: super::https); } + +// The same cases with the server spawned onto the scheduler's workers while the test drives the client from its own thread, so the two run in parallel. +fn http11_spawned(app: App, test: impl AsyncFnOnce(&Context<'_, NativeScheduler, Arc>)) { + serve(app, Version::Http11).run_spawned(test); +} + +mod http11_spawned { + base_http_fixtures::protocol_tests!(run: super::http11_spawned); + base_http_fixtures::protocol_trailer_tests!(run: super::http11_spawned); +} + +fn http10_spawned(app: App, test: impl AsyncFnOnce(&Context<'_, NativeScheduler, Arc>)) { + serve(app, Version::Http10).run_spawned(test); +} + +mod http10_spawned { + base_http_fixtures::protocol_tests!(run: super::http10_spawned); +} + +#[cfg(feature = "rustls")] +fn https_spawned(app: App, test: impl AsyncFnOnce(&Context<'_, NativeScheduler, Arc>)) { + serve(app, Version::Http11).tls().run_spawned(test); +} + +#[cfg(feature = "rustls")] +mod https_spawned { + base_http_fixtures::protocol_tests!(run: super::https_spawned); + base_http_fixtures::protocol_trailer_tests!(run: super::https_spawned); +} diff --git a/crates/fixtures/readme.md b/crates/fixtures/readme.md index 5e03708..f1070d5 100644 --- a/crates/fixtures/readme.md +++ b/crates/fixtures/readme.md @@ -21,6 +21,8 @@ ServerContext::new().run(async |context| { }); ``` -[`ClientContext`] runs a client of any endpoint, such as a [`RawServer`]. Raw peers in [`raw`] speak HTTP/1 over blocking `std::net` sockets, so they can misbehave in ways the Base server cannot: responding then closing, stalling after a head, truncating a body, or closing or writing to an idle connection. [`socket`] waits until a peer's kernel has acknowledged what was sent, so a test observes a close or unsolicited input without sleeping. +[`ServerContext::run`] polls the server, the client and the test together on the test's thread, which makes interleavings reproducible but cannot run them in parallel. [`ServerContext::run_spawned`] instead spawns the server onto the scheduler's workers and drives the client from the test's thread, with the same bounds and teardown, so cross-thread wake-ups and races can show up. [`ServerContext::spawn`] returns the spawned server's handle, a [`SpawnedServer`], for tests which manage it themselves. + +[`ClientContext`] runs a client of any endpoint, such as a [`RawServer`]. Raw peers in [`raw`] speak HTTP/1 over blocking `std::net` sockets, so they can misbehave in ways the Base server cannot: responding then closing, stalling after a head, truncating a body, or closing or writing to an idle connection. [`RawPeer::split`] gives a peer separate reading and writing halves, for uploading and downloading at once from two threads. [`socket`] waits until a peer's kernel has acknowledged what was sent, so a test observes a close or unsolicited input without sleeping. [`protocol_tests!`] and [`protocol_trailer_tests!`] generate the shared HTTP protocol cases for a runner, and [`readable_body_tests!`] generates the `ReadableBody` contract cases for a body type. diff --git a/crates/fixtures/src/context.rs b/crates/fixtures/src/context.rs index c625f6d..de3dc71 100644 --- a/crates/fixtures/src/context.rs +++ b/crates/fixtures/src/context.rs @@ -8,16 +8,22 @@ use base_http::{ }; use base_http_client::{Client, Driver, Endpoint}; use base_http_server::Handler; -use base_scheduler::{Cancellation, Clock, SocketType}; +use base_http_server::Statistics; +use base_scheduler::{Cancellation, Clock, SocketType, Spawn, TaskError}; use base_scheduler_native::NativeScheduler; -use futures_lite::future::{block_on, zip}; +use futures_lite::future::{block_on, or, zip}; use std::{ any::Any, convert::Infallible, + fmt, future::{Future, poll_fn}, net::SocketAddr, panic::{AssertUnwindSafe, catch_unwind, resume_unwind}, pin::pin, + sync::{ + Arc, + mpsc::{self, RecvTimeoutError}, + }, task::Poll, thread, time::Duration, @@ -278,6 +284,271 @@ impl ServerContext { } } +impl ServerContext { + /// Bind a listener and spawn the server onto `scheduler`'s workers as an owned task, returning its handle. + /// + /// Unlike [`ServerContext::run`], the server makes progress on the scheduler's workers rather than on the test's thread, so a client driven from the test's thread runs in parallel with it. That exposes cross-thread wake-ups and races which a single `block_on` hides, at the cost of reproducible interleavings. The fixture relies only on the server running on the scheduler's workers; how the server spreads its connections across them is the server's choice and may change. + /// + /// The listener is bound before this returns, so a client never races the server's start. Binding blocks the calling thread briefly, so call this from outside the scheduler's workers. Shut the server down with [`SpawnedServer::shutdown`]; see [`SpawnedServer`] for what dropping it does instead. + /// + /// The scheduler is shared through an [`Arc`] because the spawned task must own a `'static` handle to it, and the application is shared as an `Arc` between the server and the handle. + pub fn spawn(self, scheduler: &Arc) -> SpawnedServer + where + S: Scheduler + Spawn + Send + Sync + 'static, + S::Socket: Send, + { + let Self { + handler, + version, + server, + client, + limits, + #[cfg(feature = "rustls")] + tls, + } = self; + let listener = block_on(listen(&**scheduler)); + let address = scheduler + .socket_local_address(&listener) + .expect("listener address"); + let endpoint = Endpoint::new(address); + #[cfg(feature = "rustls")] + let endpoint = match &tls { + Some(tls) => endpoint.with_tls(tls.client.clone(), SERVER_NAME), + None => endpoint, + }; + let handler = Arc::new(handler); + let stop = Cancellation::new(); + let serving = { + let (scheduler, handler, stop) = (scheduler.clone(), handler.clone(), stop.clone()); + #[cfg(feature = "rustls")] + let tls = tls.map(|tls| tls.server); + async move { + #[cfg(feature = "rustls")] + if let Some(config) = tls { + return base_http_server::serve_tls( + &*scheduler, + listener, + config, + &*handler, + &server, + &stop, + ) + .await; + } + base_http_server::serve(&*scheduler, listener, &*handler, &server, &stop).await + } + }; + let task = scheduler.spawn(serving).expect("spawn the server"); + SpawnedServer { + scheduler: scheduler.clone(), + task: Some(task), + stop, + handler, + endpoint, + address, + version, + client, + limits, + } + } + + /// Run `test` against a server spawned onto a fresh Native scheduler's workers, which is shut down afterwards. See [`ServerContext::run_spawned_on`]. + pub fn run_spawned( + self, + test: impl AsyncFnOnce(&Context<'_, NativeScheduler, Arc>) -> T, + ) -> T { + crate::shared_native(|scheduler| self.run_spawned_on(scheduler, test)) + } + + /// Run `test` against a server spawned onto `scheduler`'s workers, with a client driven from the calling thread, then shut the server down. The caller owns and shuts down `scheduler`. + /// + /// This has the teardown guarantees of [`ServerContext::run`]: the test and teardown are bounded, the client retires its connections before the server stops, the server's shutdown is bounded by the teardown timeout, a server failure or panic fails the test, and a test failure is reported first, with any server failure printed beforehand. + pub fn run_spawned_on( + self, + scheduler: &Arc, + test: impl AsyncFnOnce(&Context<'_, S, Arc>) -> T, + ) -> T + where + S: Scheduler + Spawn + Send + Sync + 'static, + S::Socket: Send, + { + let mut server = self.spawn(scheduler); + let tested = catch_unwind(AssertUnwindSafe(|| server.run(test))); + let stopped = block_on(server.join()); + let output = match tested { + Ok(output) => output, + Err(panic) => { + // The test's failure is usually the cause of a server failure, so report it first. + if let Err(failure) = &stopped { + eprintln!("server failed: {failure}"); + } + resume_unwind(panic) + } + }; + stopped.unwrap_or_else(|failure| failure.raise()); + output + } +} + +/// A server spawned onto a scheduler's workers by [`ServerContext::spawn`]. +/// +/// Shut it down with [`SpawnedServer::shutdown`], which cancels the server, waits for it within the teardown timeout and re-raises its failure. Rust has no asynchronous destructor, so dropping the handle without shutting it down cancels the server, blocks the dropping thread until it stops or the teardown timeout expires, and then panics, naming any server failure, so that a forgotten shutdown cannot hide one. When the thread is already unwinding, it prints the server's failure to stderr instead of panicking again. Drop it outside the scheduler's workers, since a worker blocked in the drop cannot run tasks. +#[must_use = "shut the server down to observe its failures"] +pub struct SpawnedServer { + scheduler: Arc, + // `None` once the server has been stopped. + task: Option>, + stop: Cancellation, + handler: Arc, + endpoint: Endpoint, + address: SocketAddr, + version: Version, + client: base_http_client::Options, + limits: Limits, +} + +type Served = Result; + +// Why a spawned server did not stop cleanly. +enum Failure { + Server(base_http_server::Error), + Task(TaskError), + Stalled(Duration), +} + +impl Failure { + // Re-raise a server panic with its own payload, and fail for anything else. + fn raise(self) -> ! { + match self { + Self::Task(TaskError::Panicked(panic)) => resume_unwind(panic), + failure => panic!("server failed: {failure}"), + } + } +} + +impl fmt::Display for Failure { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Server(error) => error.fmt(formatter), + Self::Task(error) => error.fmt(formatter), + Self::Stalled(timeout) => write!(formatter, "it did not stop within {timeout:?}"), + } + } +} + +impl SpawnedServer { + /// The client's endpoint, including any TLS configuration, for constructing clients. + pub fn endpoint(&self) -> &Endpoint { + &self.endpoint + } + + /// The server's listening address. + pub fn address(&self) -> SocketAddr { + self.address + } + + /// The application being served, shared with the server. + pub fn handler(&self) -> &Arc { + &self.handler + } + + /// Cancel the server and wait for it to stop, returning its statistics. + /// + /// Panics if the server does not stop within the teardown timeout or failed, and resumes its panic if it panicked. + pub async fn shutdown(mut self) -> Statistics { + self.join().await.unwrap_or_else(|failure| failure.raise()) + } + + // Cancel the server and join it within the teardown timeout. A server which overruns is destroyed once its task handle is dropped and any poll in progress returns. + async fn join(&mut self) -> Result { + self.stop.cancel(); + let Some(task) = self.task.take() else { + unreachable!("a spawned server is stopped only once") + }; + let timeout = self.limits.teardown; + let joined = or(async { Some(task.await) }, async { + expiry(timeout).await; + None + }) + .await; + match joined { + Some(Ok(served)) => served.map_err(Failure::Server), + Some(Err(error)) => Err(Failure::Task(error)), + None => Err(Failure::Stalled(timeout)), + } + } +} + +impl SpawnedServer { + /// Run `test` with a fresh client of this server, driven on the calling thread while the server keeps running on the scheduler's workers. + /// + /// The test and the client's teardown are bounded as in [`ServerContext::run`], and a test panic propagates after the client has retired its connections. The server keeps running afterwards, so further tests can use it before it is shut down. + pub fn run(&self, test: impl AsyncFnOnce(&Context<'_, S, Arc>) -> T) -> T { + let scheduler = &*self.scheduler; + block_on(async { + let (client, driver) = + Client::new(scheduler, self.endpoint.clone(), self.client.clone()) + .expect("valid client options"); + let context = Context { + scheduler, + handler: self.handler.clone(), + client, + endpoint: self.endpoint.clone(), + address: self.address, + version: self.version, + }; + let stop = Cancellation::new(); + let supervised = supervise( + scheduler, + self.limits, + &context.client, + driver, + async {}, + &stop, + test(&context), + ) + .await; + finish( + supervised.tested, + supervised.forced, + &context.client, + self.limits, + ) + }) + } +} + +// Expire after `timeout`, measured on a thread of its own rather than by the scheduler's clock, so a server occupying every worker, such as a handler which blocks, still fails its shutdown instead of hanging it. Dropping the future stops the thread. +async fn expiry(timeout: Duration) { + let expired = Latch::new(); + let (_cancel, cancelled) = mpsc::channel::<()>(); + let release = expired.clone(); + thread::spawn(move || { + if let Err(RecvTimeoutError::Timeout) = cancelled.recv_timeout(timeout) { + release.release(); + } + }); + let _ = expired.wait(&Cancellation::never()).await; +} + +impl Drop for SpawnedServer { + fn drop(&mut self) { + if self.task.is_none() { + return; + } + let stopped = block_on(self.join()); + match (stopped, thread::panicking()) { + (Ok(_), true) => {} + (Err(failure), true) => eprintln!("server failed: {failure}"), + (Ok(_), false) => panic!("a spawned server was dropped without shutdown"), + (Err(failure), false) => { + eprintln!("a spawned server was dropped without shutdown"); + failure.raise() + } + } + } +} + async fn listen(scheduler: &S) -> S::Socket { let never = Cancellation::never(); let address: SocketAddr = ([127, 0, 0, 1], 0).into(); @@ -351,7 +622,7 @@ impl<'s, S, H> Context<'s, S, H> { /// Run `work` with a raw client connected to the server, for wire-level requests, on its own thread. It does not use TLS. /// - /// The server runs on the test's thread, so a blocking peer there would stall it; the test instead awaits the peer thread without blocking. A panic in `work` propagates. + /// The client's driver, and with [`ServerContext::run`] the server too, runs on the test's thread, so a blocking peer there would stall them; the test instead awaits the peer thread without blocking. A panic in `work` propagates. pub async fn raw( &self, work: impl FnOnce(RawPeer) -> T + Send + 'static, diff --git a/crates/fixtures/src/lib.rs b/crates/fixtures/src/lib.rs index 5a548a9..1cdfac0 100644 --- a/crates/fixtures/src/lib.rs +++ b/crates/fixtures/src/lib.rs @@ -15,10 +15,12 @@ pub mod socket; #[cfg(feature = "rustls")] pub use context::SERVER_NAME; -pub use context::{ClientContext, Context, Hello, ServerContext, TEARDOWN_TIMEOUT, TEST_TIMEOUT}; +pub use context::{ + ClientContext, Context, Hello, ServerContext, SpawnedServer, TEARDOWN_TIMEOUT, TEST_TIMEOUT, +}; pub use latch::{LATCH_TIMEOUT, Latch}; -pub use raw::{RawListener, RawPeer, RawServer}; -pub use scheduler::{Scheduler, native}; +pub use raw::{RawListener, RawPeer, RawReader, RawServer, RawWriter}; +pub use scheduler::{Scheduler, native, shared_native}; /// The messages of `error` and each of its sources, outermost first. pub fn error_chain(error: &dyn std::error::Error) -> Vec { diff --git a/crates/fixtures/src/protocol.rs b/crates/fixtures/src/protocol.rs index 9bfd8cb..bb478d7 100644 --- a/crates/fixtures/src/protocol.rs +++ b/crates/fixtures/src/protocol.rs @@ -25,6 +25,10 @@ use std::{ /// The size of the large download, as in `AProtocol`'s "huge body". pub const LARGE_BODY: u64 = 8 * 1024 * 1024; const LARGE_CHUNK: u64 = 64 * 1024; +/// The size of each direction of the two-way stream in [`duplex_stream`], larger than the loopback socket buffers in both directions together, so neither side can finish without the other reading. +pub const DUPLEX_BODY: u64 = 16 * 1024 * 1024; +/// The size of each chunk [`duplex_stream`] uploads. +pub const DUPLEX_CHUNK: usize = 16 * 1024; const LINK: &str = "; rel=preload; as=style"; /// The application serving every protocol case. @@ -33,6 +37,7 @@ const LINK: &str = "; rel=preload; as=style"; /// | --- | --- | /// | `GET /`, `HEAD /` | `200` with `my-header: my-value` and `{method} {version}` | /// | `POST /` | `200` streaming the request body back | +/// | `POST /duplex` | `200` streaming the request body back, for the two-way streaming case | /// | `POST /length` | `200` with `Content Length: {n}`, the request body's length | /// | `GET /interim` | `103 Early Hints` with a `link` field to HTTP/1.1 clients, then `200` with `Hello World` | /// | `POST /request-trailer` | `200` with the request's `etag` trailer, after releasing [`App::request_received`] | @@ -98,6 +103,9 @@ impl Handler for App { (method, "/") if *method == Method::POST => { return Ok(Response::new(Status::OK, Reply::Echo(request.body))); } + (_, "/duplex") => { + return Ok(Response::new(Status::OK, Reply::Echo(request.body))); + } (_, "/length") => { let length = request .body @@ -404,6 +412,77 @@ pub async fn post>(context: &Context<'_, S, H>) { ); } +// A streamed upload which declares that it has no trailers. A `BodyReader` may always carry them, and HTTP/1.0, which cannot frame trailers, refuses to send such a body. +struct Untrailed(BodyReader>); + +impl ReadableBody for Untrailed { + type Data = Vec; + type Error = BodyError; + + fn length(&self) -> Option { + self.0.length() + } + + fn may_have_trailers(&self) -> bool { + false + } + + fn close(&mut self, error: Option) { + self.0.close(error); + } + + async fn read(&mut self, cancellation: &Cancellation) -> Result>, BodyError> { + self.0.read(cancellation).await + } +} + +/// The data of the two-way stream from `offset`: a pattern which does not repeat at chunk boundaries, so a lost, repeated or reordered chunk changes it. +pub fn duplex_data(offset: u64, length: usize) -> Vec { + (offset..offset + length as u64) + .map(|index| (index % 251) as u8) + .collect() +} + +/// A request body streams up while its echo streams down at the same time, each [`DUPLEX_BODY`] long and under backpressure. +/// +/// The upload is produced through a small bounded queue while the response is read concurrently. The stream is larger than what the sockets between client and server can buffer, so a client which finished its upload before reading, or a server which did not echo as it read, would stall the exchange until the test's deadline. The request declares its length and no trailers, so HTTP/1.0 can carry it too. +pub async fn duplex_stream>(context: &Context<'_, S, H>) { + let (upload, mut writer) = BodyReader::with_length(4 * DUPLEX_CHUNK, 4, DUPLEX_BODY); + let request = context.request(Method::POST, "/duplex", Untrailed(upload)); + let produce = async { + let mut offset = 0; + while offset < DUPLEX_BODY { + writer + .write(duplex_data(offset, DUPLEX_CHUNK), &never()) + .await + .expect("write the request body"); + offset += DUPLEX_CHUNK as u64; + } + writer.close_write(None); + }; + let consume = async { + let mut response = context + .client() + .call(request, &never()) + .await + .expect("POST /duplex"); + assert_eq!(response.status, Status::OK); + let echoed = body::collect(&mut response.body, &never()) + .await + .map(|collected| collected.data); + let intact = echoed.as_ref().is_ok_and(|data| { + data.len() as u64 == DUPLEX_BODY && *data == duplex_data(0, data.len()) + }); + // Report the length or error rather than the data. + assert!( + intact, + "the echoed stream: {:?} of {DUPLEX_BODY} bytes", + echoed.as_ref().map(Vec::len) + ); + }; + zip(produce, consume).await; +} + /// The server observes a request body's length. pub async fn request_length>(context: &Context<'_, S, H>) { let upload = BufferedBody::from(b"Hello World!".to_vec()); @@ -601,12 +680,13 @@ pub async fn response_trailer>(context: &Context<'_ /// Generate `#[test]` functions for the version-independent protocol cases, ported from `async-http`'s `AProtocol`. /// -/// The runner receives a fresh [`App`] and an async closure taking the running [`Context`]. It chooses the version, options and transport, typically by serving the application with a [`ServerContext`](crate::ServerContext), and owns the deadline. Invoke it inside a dedicated module: +/// The runner receives a fresh [`App`] and an async closure taking the running [`Context`]. It chooses the version, options and transport, typically by serving the application with a [`ServerContext`](crate::ServerContext), and owns the deadline. [`ServerContext::run`](crate::ServerContext::run) runs the server, the client and the case on the test's thread, which is reproducible; [`ServerContext::run_spawned`](crate::ServerContext::run_spawned) runs the server on the scheduler's workers in parallel with the client, which can expose cross-thread wake-up bugs and races. Invoke it inside a dedicated module: /// /// ``` /// use base_http::Version; /// use base_http_fixtures::{Context, ServerContext, protocol::App}; /// use base_scheduler_native::NativeScheduler; +/// use std::sync::Arc; /// /// fn http10(app: App, test: impl AsyncFnOnce(&Context<'_, NativeScheduler, App>)) { /// ServerContext::new() @@ -618,6 +698,18 @@ pub async fn response_trailer>(context: &Context<'_ /// mod http10 { /// base_http_fixtures::protocol_tests!(run: super::http10); /// } +/// +/// // The same cases with the server spawned onto the scheduler's workers, in parallel with the client. +/// fn http10_spawned(app: App, test: impl AsyncFnOnce(&Context<'_, NativeScheduler, Arc>)) { +/// ServerContext::new() +/// .handler(app) +/// .version(Version::Http10) +/// .run_spawned(test); +/// } +/// +/// mod http10_spawned { +/// base_http_fixtures::protocol_tests!(run: super::http10_spawned); +/// } /// # fn main() {} /// ``` #[macro_export] @@ -632,6 +724,7 @@ macro_rules! protocol_tests { large_body, broken_server, incorrect_length, + duplex_stream, } }; } diff --git a/crates/fixtures/src/raw.rs b/crates/fixtures/src/raw.rs index 7f91dd1..dcc5295 100644 --- a/crates/fixtures/src/raw.rs +++ b/crates/fixtures/src/raw.rs @@ -144,6 +144,85 @@ impl RawPeer { } } +impl RawPeer { + /// Split this connection into a reader and a writer, each usable from its own thread, as a peer which uploads and downloads at the same time needs. + /// + /// The reader keeps this peer's buffered input; the writer is a clone of the socket, made with [`TcpStream::try_clone`], so the timeouts and options set on this peer apply to both halves. + pub fn split(self) -> (RawReader, RawWriter) { + let writer = self + .stream + .get_ref() + .try_clone() + .expect("clone the raw peer's socket"); + ( + RawReader { + stream: self.stream, + }, + RawWriter { stream: writer }, + ) + } +} + +/// The reading half of a [`RawPeer`], from [`RawPeer::split`]. It owns the peer's buffered input. +#[derive(Debug)] +pub struct RawReader { + stream: BufReader, +} + +impl RawReader { + /// Read a request or response head. See [`read_head`]. + pub fn read_head(&mut self) -> Head { + read_head(&mut self.stream) + } + + /// Read the body that follows `head`. See [`read_body`]. + pub fn read_body(&mut self, head: &Head) -> Body { + read_body(&mut self.stream, head) + } +} + +impl Read for RawReader { + fn read(&mut self, buffer: &mut [u8]) -> io::Result { + self.stream.read(buffer) + } +} + +impl BufRead for RawReader { + fn fill_buf(&mut self) -> io::Result<&[u8]> { + self.stream.fill_buf() + } + fn consume(&mut self, amount: usize) { + self.stream.consume(amount) + } +} + +/// The writing half of a [`RawPeer`], from [`RawPeer::split`]. +#[derive(Debug)] +pub struct RawWriter { + stream: TcpStream, +} + +impl RawWriter { + /// The underlying socket, for socket options. It is a clone of the reader's socket. + pub fn get_mut(&mut self) -> &mut TcpStream { + &mut self.stream + } + + /// Write all of `bytes` at once. + pub fn write(&mut self, bytes: impl AsRef<[u8]>) { + self.stream + .write_all(bytes.as_ref()) + .expect("write to the raw peer"); + } + + /// Close this side's direction of the connection; the reader stays readable. + pub fn shutdown_write(&mut self) { + self.stream + .shutdown(Shutdown::Write) + .expect("shut down raw peer output"); + } +} + impl Read for RawPeer { fn read(&mut self, buffer: &mut [u8]) -> io::Result { self.stream.read(buffer) diff --git a/crates/fixtures/src/scheduler.rs b/crates/fixtures/src/scheduler.rs index 481c7e1..8cc89d3 100644 --- a/crates/fixtures/src/scheduler.rs +++ b/crates/fixtures/src/scheduler.rs @@ -8,7 +8,10 @@ use base_scheduler::{ }; use base_scheduler_native::{NativeOptions, NativeScheduler}; use futures_lite::future::block_on; -use std::panic::{AssertUnwindSafe, catch_unwind, resume_unwind}; +use std::{ + panic::{AssertUnwindSafe, catch_unwind, resume_unwind}, + sync::Arc, +}; /// The scheduler capabilities a fixture needs to listen, serve, connect and bound its teardown. /// @@ -47,11 +50,18 @@ impl Scheduler for S where /// Run `work` with a fresh two-worker Native scheduler, then shut it down, also when `work` panics, before propagating the panic. pub fn native(work: impl FnOnce(&NativeScheduler) -> T) -> T { - let scheduler = NativeScheduler::with_options(NativeOptions { - workers: 2, - ..Default::default() - }) - .expect("construct a Native scheduler"); + shared_native(|scheduler| work(scheduler)) +} + +/// Run `work` with a fresh two-worker Native scheduler shared through an [`Arc`], as [`ServerContext::spawn`](crate::ServerContext::spawn) needs, then shut it down as [`native`] does. +pub fn shared_native(work: impl FnOnce(&Arc) -> T) -> T { + let scheduler = Arc::new( + NativeScheduler::with_options(NativeOptions { + workers: 2, + ..Default::default() + }) + .expect("construct a Native scheduler"), + ); let result = catch_unwind(AssertUnwindSafe(|| work(&scheduler))); block_on(scheduler.shutdown()); result.unwrap_or_else(|panic| resume_unwind(panic)) diff --git a/crates/fixtures/tests/protocol.rs b/crates/fixtures/tests/protocol.rs index 04cd092..4505976 100644 --- a/crates/fixtures/tests/protocol.rs +++ b/crates/fixtures/tests/protocol.rs @@ -4,7 +4,7 @@ //! Each protocol case fails, at the assertion it exists for, against an application which breaks that case. use base_http::{ InterimHandler, InterimResponses, Method, Request, Response, ResponseMetadata, Status, Version, - body::{BufferedBody, ReadableBody}, + body::{BodyReader, BufferedBody, ReadableBody}, }; use base_http_fixtures::{ ServerContext, body, @@ -12,6 +12,7 @@ use base_http_fixtures::{ }; use base_http_server::Handler; use base_scheduler::Cancellation; +use futures_lite::future::zip; use std::{borrow::Borrow, io, time::Duration}; /// The shared application, sabotaged on every route the protocol cases use. @@ -71,6 +72,11 @@ impl Handler for Sabotaged { .call(route(request, "/"), interim, cancellation) .await; } + // Lose the first chunk of the stream. + (_, "/duplex") => { + let _ = request.body.read(cancellation).await; + return Ok(Response::new(Status::OK, Reply::Echo(request.body))); + } // Send the whole declared length. (_, "/incorrect-length") => reply("Borked!!!!"), // Drop the request trailer. @@ -127,6 +133,7 @@ sabotaged! { large_body => "unexpected data", broken_server => "the status of a failing handler", incorrect_length => "a truncated body must fail", + duplex_stream => "the echoed stream", request_trailer => "the request data and trailer the handler received", response_trailer => "the response trailer", } @@ -197,3 +204,36 @@ fn a_truncated_trailer_upload_fails_the_handler() { .await; }); } + +// The two-way stream is larger than the sockets can buffer: a client which uploads it all before reading the echo stalls until the deadline, so the case can only pass by streaming both ways at once. +#[test] +#[should_panic(expected = "test exceeded")] +fn a_sequential_duplex_stream_stalls() { + ServerContext::new() + .handler(App::new()) + .timeout(Duration::from_secs(1)) + .run(async |context| { + let (upload, mut writer) = BodyReader::bounded(4 * protocol::DUPLEX_CHUNK, 4); + let request = context.request(Method::POST, "/duplex", upload); + let upload_all = async { + let mut offset = 0; + while offset < protocol::DUPLEX_BODY { + writer + .write( + protocol::duplex_data(offset, protocol::DUPLEX_CHUNK), + &Cancellation::never(), + ) + .await + .unwrap(); + offset += protocol::DUPLEX_CHUNK as u64; + } + writer.close_write(None); + }; + let (response, ()) = zip( + context.client().call(request, &Cancellation::never()), + upload_all, + ) + .await; + body::payload(&mut response.unwrap().body).await; + }); +} diff --git a/crates/fixtures/tests/raw.rs b/crates/fixtures/tests/raw.rs index 6e03b03..339fb37 100644 --- a/crates/fixtures/tests/raw.rs +++ b/crates/fixtures/tests/raw.rs @@ -7,13 +7,18 @@ use base_http::{ body::{EmptyBody, ReadableBody}, }; use base_http_fixtures::{ - ClientContext, Latch, RawPeer, RawServer, + ClientContext, Latch, RawPeer, RawServer, ServerContext, body::{self, payload}, error_chain, + protocol::{App, DUPLEX_BODY, DUPLEX_CHUNK, duplex_data}, raw::{self, Head}, }; use base_scheduler::Cancellation; -use std::io::{BufRead, Cursor, Read}; +use std::{ + io::{self, BufRead, Cursor, Read, Write}, + thread, + time::Duration, +}; fn get() -> Request { Request::new(Method::GET, "/", EmptyBody) @@ -155,3 +160,74 @@ fn respond_then_close_serves_one_response() { }); server.join(); } + +fn upload_head(length: u64) -> String { + format!( + "POST /duplex HTTP/1.1\r\nHost: localhost\r\nContent-Length: {length}\r\nConnection: close\r\n\r\n" + ) +} + +// The halves of a split peer upload and download at the same time from two threads, each direction larger than the sockets can buffer. Done sequentially on one thread, the upload would stall once the unread echo filled the buffers, as `a_sequential_raw_upload_stalls_under_backpressure` shows. +#[test] +fn a_split_raw_peer_uploads_and_downloads_at_once() { + ServerContext::new() + .handler(App::new()) + .run(async |context| { + context + .raw(|peer| { + let (mut reader, mut writer) = peer.split(); + writer.get_mut().set_nodelay(true).unwrap(); + let uploading = thread::spawn(move || { + writer.write(upload_head(DUPLEX_BODY)); + let mut offset = 0; + while offset < DUPLEX_BODY { + writer.write(duplex_data(offset, DUPLEX_CHUNK)); + offset += DUPLEX_CHUNK as u64; + } + writer.shutdown_write(); + }); + let head = reader.read_head(); + assert_eq!(head.status(), 200); + let body = reader.read_body(&head); + assert!( + body.data.len() as u64 == DUPLEX_BODY + && body.data == duplex_data(0, body.data.len()), + "the echoed stream" + ); + // The server closes the connection after the exchange. + assert!(reader.fill_buf().unwrap().is_empty()); + reader.consume(0); + assert_eq!(reader.read(&mut [0; 1]).unwrap(), 0); + uploading.join().unwrap(); + }) + .await; + }); +} + +// Writing the whole upload before reading the echo stalls: the server stops reading once its echo is not read, so the write cannot finish. +#[test] +fn a_sequential_raw_upload_stalls_under_backpressure() { + ServerContext::new() + .handler(App::new()) + .run(async |context| { + context + .raw(|mut peer| { + peer.get_mut() + .set_write_timeout(Some(Duration::from_millis(500))) + .unwrap(); + peer.write(upload_head(DUPLEX_BODY)); + let error = peer + .get_mut() + .write_all(&duplex_data(0, DUPLEX_BODY as usize)) + .expect_err("a sequential upload stalls"); + assert!( + matches!( + error.kind(), + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut + ), + "unexpected failure: {error}" + ); + }) + .await; + }); +} diff --git a/crates/fixtures/tests/spawned.rs b/crates/fixtures/tests/spawned.rs new file mode 100644 index 0000000..6ba9eb9 --- /dev/null +++ b/crates/fixtures/tests/spawned.rs @@ -0,0 +1,230 @@ +// Released under the MIT License. +// Copyright, 2026, by Samuel Williams. + +//! A spawned server runs on the scheduler's workers beside a client driven from the test's thread, shuts down within its bound, and reports each way it can fail, including when its handle is dropped. +use base_http::{ + InterimHandler, InterimResponses, Method, Request, Response, Status, body::EmptyBody, + body::ReadableBody, +}; +use base_http_fixtures::{ + Hello, Latch, ServerContext, body::payload, protocol::App, shared_native, +}; +use base_http_server::Handler; +use base_scheduler::Cancellation; +use futures_lite::future::block_on; +use std::{ + convert::Infallible, + io::{Read, Write}, + net::TcpStream, + panic::{AssertUnwindSafe, catch_unwind}, + sync::Arc, + time::Duration, +}; + +fn failing() -> ServerContext { + ServerContext::new().server_options(base_http_server::Options { + max_connections: 0, + ..Default::default() + }) +} + +async fn hello( + context: &base_http_fixtures::Context<'_, impl base_http_fixtures::Scheduler, Arc>, +) { + let mut response = context + .client() + .call( + context.request(Method::GET, "/", EmptyBody), + &Cancellation::never(), + ) + .await + .unwrap(); + assert_eq!(response.status, Status::OK); + assert_eq!(payload(&mut response.body).await, b"Hello World!"); +} + +#[test] +fn a_spawned_server_serves_several_clients_and_shuts_down() { + shared_native(|scheduler| { + let server = ServerContext::new().spawn(scheduler); + let _: &Arc = server.handler(); + assert_eq!(server.endpoint().authority, server.address().to_string()); + server.run(async |context| hello(context).await); + let output = server.run(async |context| { + hello(context).await; + context + .raw(|mut peer| { + peer.write("GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"); + let (head, _) = peer.read_message(); + head.status() + }) + .await + }); + assert_eq!(output, 200); + let statistics = block_on(server.shutdown()); + assert_eq!(statistics.requests, 3); + }); +} + +#[test] +fn a_spawned_run_shares_the_application_with_the_server() { + ServerContext::new() + .handler(App::new()) + .run_spawned(async |context| { + let app: &App = context.handler(); + base_http_fixtures::protocol::response_trailer(context).await; + assert!(app.response_released().is_released()); + }); +} + +#[test] +#[should_panic(expected = "the test failed")] +fn a_test_panic_propagates_from_a_spawned_run() { + ServerContext::new().run_spawned(async |_| panic!("the test failed")); +} + +#[test] +#[should_panic(expected = "server failed")] +fn a_spawned_server_failure_fails_the_test() { + failing().run_spawned(async |_| {}); +} + +// The test's own failure is reported ahead of the server failure it may have caused. +#[test] +#[should_panic(expected = "the test failed first")] +fn a_test_panic_is_reported_before_a_spawned_server_failure() { + failing().run_spawned(async |_| panic!("the test failed first")); +} + +/// Panics in every call. +struct Panicking; + +impl Handler for Panicking { + type Error = Infallible; + type Body = EmptyBody; + + async fn call( + &self, + _: Request, + _: &mut InterimResponses, + _: &Cancellation, + ) -> Result, Infallible> { + panic!("the handler panicked") + } +} + +// A request the server cannot answer, sent without a client so the test does not wait for a response. +fn request(server: std::net::SocketAddr) -> TcpStream { + let mut stream = TcpStream::connect(server).unwrap(); + stream + .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n") + .unwrap(); + stream +} + +#[test] +#[should_panic(expected = "the handler panicked")] +fn shutdown_resumes_a_server_panic() { + shared_native(|scheduler| { + let server = ServerContext::new().handler(Panicking).spawn(scheduler); + let mut stream = request(server.address()); + // The server's task ends when the handler panics, closing the connection. + let _ = stream.read(&mut [0; 1]); + block_on(server.shutdown()); + }); +} + +/// Blocks its worker in every call until released, so the server cannot observe cancellation. +struct Blocking { + started: Latch, + release: Latch, +} + +impl Handler for Blocking { + type Error = Infallible; + type Body = EmptyBody; + + async fn call( + &self, + _: Request, + _: &mut InterimResponses, + _: &Cancellation, + ) -> Result, Infallible> { + self.started.release(); + self.release.wait_blocking(); + Ok(Response::new(Status::OK, EmptyBody)) + } +} + +#[test] +fn a_server_which_does_not_stop_fails_its_shutdown() { + let (started, release) = (Latch::new(), Latch::new()); + shared_native(|scheduler| { + let server = ServerContext::new() + .handler(Blocking { + started: started.clone(), + release: release.clone(), + }) + .teardown_timeout(Duration::from_millis(50)) + .spawn(scheduler); + let _stream = request(server.address()); + started.wait_blocking(); + let failure = catch_unwind(AssertUnwindSafe(|| block_on(server.shutdown()))) + .expect_err("a stalled shutdown fails"); + release.release(); + let message = failure.downcast_ref::().unwrap(); + assert!( + message.contains("did not stop within 50ms"), + "unexpected failure: {message}" + ); + }); +} + +#[test] +#[should_panic(expected = "dropped without shutdown")] +fn dropping_a_spawned_server_without_shutdown_fails() { + shared_native(|scheduler| drop(ServerContext::new().spawn(scheduler))); +} + +#[test] +#[should_panic(expected = "server failed")] +fn dropping_a_failed_spawned_server_reports_its_failure() { + shared_native(|scheduler| drop(failing().spawn(scheduler))); +} + +// While a test unwinds, the dropped server is still cancelled and joined, and the test's own panic is reported. +#[test] +#[should_panic(expected = "the test failed")] +fn a_spawned_server_dropped_during_a_panic_is_stopped() { + shared_native(|scheduler| { + let _server = ServerContext::new().spawn(scheduler); + panic!("the test failed"); + }); +} + +#[test] +#[should_panic(expected = "the test failed")] +fn a_failed_spawned_server_dropped_during_a_panic_is_reported() { + shared_native(|scheduler| { + let _server = failing().spawn(scheduler); + panic!("the test failed"); + }); +} + +// A server panic caused by the test is reported after the test's own failure. +#[test] +#[should_panic(expected = "the test failed first")] +fn a_test_panic_is_reported_before_a_spawned_server_panic() { + ServerContext::new() + .handler(Panicking) + .run_spawned(async |context| { + context + .raw(|mut peer| { + peer.write("GET / HTTP/1.1\r\nHost: localhost\r\n\r\n"); + // The server's task ends when the handler panics, closing the connection. + let _ = peer.read(&mut [0; 1]); + }) + .await; + panic!("the test failed first"); + }); +} diff --git a/design/fixtures.md b/design/fixtures.md index 7012c88..39ac594 100644 --- a/design/fixtures.md +++ b/design/fixtures.md @@ -38,11 +38,40 @@ Rust has no asynchronous destructor, so a fixture which leaves teardown to the t `crates/fixtures/tests/context.rs` exercises a test panic, a test timeout, a forced teardown, a server failure and a raw peer panic, and checks each is reported. +## Spawned server context + +### Requirement + +`ServerContext::spawn` binds its listener before returning, then spawns the server's public entry point, `base_http_server::serve` or `serve_tls`, as one task through `base_scheduler::Spawn`, so the server runs on the scheduler's workers in parallel with a client driven from the test's thread. The fixture relies only on the server running on the scheduler's workers. It adds nothing to `base-http-server`'s public API and does not choose how connections are spread across tasks or workers. Its bounds are those a server which spawned its own connection tasks would need: the handler is `Handler + Send + Sync + 'static` and shared as an `Arc`, and the scheduler implements `Spawn`, is `Send + Sync + 'static` and is shared through an `Arc`. Native offers no other `'static` handle that can spawn. + +The returned `SpawnedServer` exposes the endpoint, address and handler. Its asynchronous `shutdown` cancels the server and waits for it for at most the teardown timeout. It resumes a server panic with its original payload, and fails for a server error or for overrunning the timeout. The timeout is measured on its own thread rather than by the scheduler's clock, so a server that occupies every worker still fails its shutdown instead of hanging it. Dropping the handle without `shutdown` cancels the server, waits for it as `shutdown` does, then panics and names any server failure. If the thread is already unwinding, the drop prints the failure to stderr instead of panicking. `run_spawned` and `run_spawned_on` run a test against a spawned server with the guarantees of `run`: the test and teardown are bounded, the client retires its connections before the server stops, a server failure fails the test, and a test failure is reported before any server failure. The scoped `run` and `run_on` are unchanged. + +### Rationale + +`run` polls the server, the client's driver and the test in one `block_on` on the test's thread. That makes results reproducible, but nothing runs in parallel, so cross-thread wake-up bugs and races in two-way streaming cannot show up. The user decided that the server's connection and task layout is an implementation detail. Relying only on the public entry points, with the bounds an internally spawning server would need, means that a later move to per-connection tasks inside `serve` does not break the fixtures. Rust has no asynchronous destructor, so a forgotten shutdown panics rather than silently discarding the server's outcome. + +### Scope + +`ServerContext::spawn`, `ServerContext::run_spawned`, `ServerContext::run_spawned_on` and `SpawnedServer` in `crates/fixtures/src/context.rs`, and `shared_native` in `crates/fixtures/src/scheduler.rs`. + +### Validation + +Inspect `spawn`: its task calls only `serve` or `serve_tls`, and the change leaves `crates/server/src` untouched. `crates/fixtures/tests/spawned.rs` covers the following, and checks how each is reported: + +- one spawned server serving several clients and a raw peer; +- an application shared with the server; +- a test panic; +- a server failure; +- a test failure reported before a server failure or panic; +- a server panic, resumed by `shutdown`; +- a shutdown that overruns its timeout while a handler blocks a worker; +- each drop path: clean or failed, and while unwinding or not. + ## Independent, bounded raw peers ### Requirement -Raw peers use blocking `std::net` sockets and no Base encoder, decoder or scheduler. Every read, write, accept, latch wait and acknowledgement wait is bounded by a timeout shorter than the default test timeout, so a stalled peer reports its own failure, and an accepted socket is explicitly made blocking, since it inherits the listener's nonblocking mode on some platforms. Tests sequence peers with latches or acknowledgement, not sleeps. A canned peer which closes a connection or writes outside an exchange waits until the other side's kernel has acknowledged it, using `socket::wait_acknowledged`, before continuing or signalling the test. Linux and macOS query the kernel; on other platforms the wait is a 50 ms sleep, which is documented on the peer helpers. Raw peers never block a scheduler thread: `Context::raw` runs one on its own thread. +Raw peers use blocking `std::net` sockets and no Base encoder, decoder or scheduler. Every read, write, accept, latch wait and acknowledgement wait is bounded by a timeout shorter than the default test timeout, so a stalled peer reports its own failure, and an accepted socket is explicitly made blocking, since it inherits the listener's nonblocking mode on some platforms. Tests sequence peers with latches or acknowledgement, not sleeps. A canned peer which closes a connection or writes outside an exchange waits until the other side's kernel has acknowledged it, using `socket::wait_acknowledged`, before continuing or signalling the test. Linux and macOS query the kernel; on other platforms the wait is a 50 ms sleep, which is documented on the peer helpers. Raw peers never block a scheduler thread: `Context::raw` runs one on its own thread. `RawPeer::split` divides a peer into a reader, which keeps the peer's buffered input, and a writer cloned from its socket with `TcpStream::try_clone`. Each half can be used from its own thread, so one peer can upload and download at the same time. ### Rationale @@ -54,13 +83,13 @@ Peers that share the implementation's codec can hide a shared bug. Unbounded blo ### Validation -Inspect peer construction for timeouts and the explicit blocking reset. Search tests built on the fixtures for `thread::sleep`. `crates/fixtures/tests/acknowledgement.rs` shows, on Linux and macOS, that a bounded wait expires while a peer leaves data unread, and that `wait_acknowledged` returns once the peer reads the data and the FIN. A loopback peer acknowledges a FIN as soon as it arrives, so no test can hold the window in which only the FIN is unacknowledged. Linux counts the FIN in the same queue as the data, but the separate connection-state check macOS needs is justified by inspection, not by a test. +Inspect peer construction for timeouts and the explicit blocking reset. Search tests built on the fixtures for `thread::sleep`. `crates/fixtures/tests/acknowledgement.rs` shows, on Linux and macOS, that a bounded wait expires while a peer leaves data unread, and that `wait_acknowledged` returns once the peer reads the data and the FIN. A loopback peer acknowledges a FIN as soon as it arrives, so no test can hold the window in which only the FIN is unacknowledged. Linux counts the FIN in the same queue as the data, but the separate connection-state check macOS needs is justified by inspection, not by a test. `crates/fixtures/tests/raw.rs` streams `DUPLEX_BODY` up and down at the same time through a split peer. It also shows that a sequential upload of the same size on one thread stalls until its write timeout. ## Behaviour suites ### Requirement -`protocol_tests!` and `protocol_trailer_tests!` generate the shared protocol cases for a runner chosen by the caller, and `readable_body_tests!` generates the `ReadableBody` contract cases for a body type. Every case is also a standalone public function, so a partial implementation can call a subset. The protocol suite runs between the Base client and server over HTTP/1.1, over HTTP/1.0 without the trailer cases, and over TLS with the `rustls` feature. The body suite runs against each body type `base-http` provides. Each assertion is shown to fail against an implementation that breaks it. +`protocol_tests!` and `protocol_trailer_tests!` generate the shared protocol cases for a runner chosen by the caller, and `readable_body_tests!` generates the `ReadableBody` contract cases for a body type. Every case is also a standalone public function, so a partial implementation can call a subset. The protocol suite includes a two-way streaming case, `duplex_stream`. The client streams a request body through a small bounded queue while it reads the streamed echo at the same time, and each direction is `DUPLEX_BODY` long, larger than the loopback sockets between client and server can buffer. The request declares its length and no trailers, so HTTP/1.0 carries it too. The protocol suite runs between the Base client and server over HTTP/1.1, over HTTP/1.0 without the trailer cases, and over TLS with the `rustls` feature. Each of these runs in two modes: scoped, with `run`, and spawned, with `run_spawned`, where the server runs on Native workers and the test drives the client from its own thread. The body suite runs against each body type `base-http` provides. Each assertion is shown to fail against an implementation that breaks it. ### Rationale @@ -72,4 +101,4 @@ Each new engine or body type should be checked against one shared definition of ### Validation -Check that the suites are applied as stated. `crates/fixtures/tests/protocol.rs` runs every protocol case against an application sabotaged on that case's route and requires the failure message of the assertion the case exists for; `crates/fixtures/tests/body.rs` runs every body assertion against a body with the corresponding fault and requires the expected failure. +Check that the suites are applied as stated. `crates/fixtures/tests/protocol.rs` runs every protocol case against an application sabotaged on that case's route and requires the failure message of the assertion the case exists for; `crates/fixtures/tests/body.rs` runs every body assertion against a body with the corresponding fault and requires the expected failure. `a_sequential_duplex_stream_stalls` in `crates/fixtures/tests/protocol.rs` shows that the two-way stream cannot complete when the client uploads it all before reading. The case can therefore pass only by streaming both ways at once. Whether it stalls depends on the platform's socket buffer sizes, so a platform with larger loopback buffers needs a larger `DUPLEX_BODY`. From fd35277cbf1d5c9426bc5d65da7dcc8c236dd916 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 3 Nov 2026 13:53:06 +1300 Subject: [PATCH 2/6] Show a stalled shutdown failing with the only worker blocked, and correct what the fixtures documentation claims about buffering and scheduler drains. Co-Authored-By: Claude Opus 5.5 --- crates/fixtures/src/context.rs | 2 +- crates/fixtures/src/protocol.rs | 6 +++--- crates/fixtures/tests/spawned.rs | 15 +++++++++++++-- design/fixtures.md | 8 ++++---- 4 files changed, 21 insertions(+), 10 deletions(-) diff --git a/crates/fixtures/src/context.rs b/crates/fixtures/src/context.rs index de3dc71..4e8d388 100644 --- a/crates/fixtures/src/context.rs +++ b/crates/fixtures/src/context.rs @@ -518,7 +518,7 @@ impl SpawnedServer { } } -// Expire after `timeout`, measured on a thread of its own rather than by the scheduler's clock, so a server occupying every worker, such as a handler which blocks, still fails its shutdown instead of hanging it. Dropping the future stops the thread. +// Expire after `timeout`, measured on a thread of its own rather than by the scheduler's clock, so a server occupying every worker, such as a handler which blocks, still fails its shutdown. Shutting the scheduler down afterwards still waits for that poll to return. Dropping the future stops the thread. async fn expiry(timeout: Duration) { let expired = Latch::new(); let (_cancel, cancelled) = mpsc::channel::<()>(); diff --git a/crates/fixtures/src/protocol.rs b/crates/fixtures/src/protocol.rs index bb478d7..01b0098 100644 --- a/crates/fixtures/src/protocol.rs +++ b/crates/fixtures/src/protocol.rs @@ -25,7 +25,7 @@ use std::{ /// The size of the large download, as in `AProtocol`'s "huge body". pub const LARGE_BODY: u64 = 8 * 1024 * 1024; const LARGE_CHUNK: u64 = 64 * 1024; -/// The size of each direction of the two-way stream in [`duplex_stream`], larger than the loopback socket buffers in both directions together, so neither side can finish without the other reading. +/// The size of each direction of the two-way stream in [`duplex_stream`], far more than the loopback path between client and server buffers in practice, so the upload cannot finish unless the echo is read at the same time. pub const DUPLEX_BODY: u64 = 16 * 1024 * 1024; /// The size of each chunk [`duplex_stream`] uploads. pub const DUPLEX_CHUNK: usize = 16 * 1024; @@ -436,7 +436,7 @@ impl ReadableBody for Untrailed { } } -/// The data of the two-way stream from `offset`: a pattern which does not repeat at chunk boundaries, so a lost, repeated or reordered chunk changes it. +/// The data of the two-way stream from `offset`: a pattern with a period of 251 bytes, which does not divide the chunk size, so a lost, repeated or reordered chunk changes the data or its length. pub fn duplex_data(offset: u64, length: usize) -> Vec { (offset..offset + length as u64) .map(|index| (index % 251) as u8) @@ -445,7 +445,7 @@ pub fn duplex_data(offset: u64, length: usize) -> Vec { /// A request body streams up while its echo streams down at the same time, each [`DUPLEX_BODY`] long and under backpressure. /// -/// The upload is produced through a small bounded queue while the response is read concurrently. The stream is larger than what the sockets between client and server can buffer, so a client which finished its upload before reading, or a server which did not echo as it read, would stall the exchange until the test's deadline. The request declares its length and no trailers, so HTTP/1.0 can carry it too. +/// The upload is produced through a small bounded queue while the response is read concurrently. The stream is far larger than the sockets between client and server buffer in practice, so a client which finished its upload before reading the echo would stall the exchange until the test's deadline. The request declares its length and no trailers, so HTTP/1.0 can carry it too. pub async fn duplex_stream>(context: &Context<'_, S, H>) { let (upload, mut writer) = BodyReader::with_length(4 * DUPLEX_CHUNK, 4, DUPLEX_BODY); let request = context.request(Method::POST, "/duplex", Untrailed(upload)); diff --git a/crates/fixtures/tests/spawned.rs b/crates/fixtures/tests/spawned.rs index 6ba9eb9..9903d9a 100644 --- a/crates/fixtures/tests/spawned.rs +++ b/crates/fixtures/tests/spawned.rs @@ -11,6 +11,7 @@ use base_http_fixtures::{ }; use base_http_server::Handler; use base_scheduler::Cancellation; +use base_scheduler_native::{NativeOptions, NativeScheduler}; use futures_lite::future::block_on; use std::{ convert::Infallible, @@ -159,7 +160,16 @@ impl Handler for Blocking { #[test] fn a_server_which_does_not_stop_fails_its_shutdown() { let (started, release) = (Latch::new(), Latch::new()); - shared_native(|scheduler| { + // With one worker, which the handler blocks, nothing on the scheduler can run, including its clock's timers. + let scheduler = Arc::new( + NativeScheduler::with_options(NativeOptions { + workers: 1, + ..Default::default() + }) + .unwrap(), + ); + { + let scheduler = &scheduler; let server = ServerContext::new() .handler(Blocking { started: started.clone(), @@ -177,7 +187,8 @@ fn a_server_which_does_not_stop_fails_its_shutdown() { message.contains("did not stop within 50ms"), "unexpected failure: {message}" ); - }); + } + block_on(scheduler.shutdown()); } #[test] diff --git a/design/fixtures.md b/design/fixtures.md index 39ac594..1f260e8 100644 --- a/design/fixtures.md +++ b/design/fixtures.md @@ -42,9 +42,9 @@ Rust has no asynchronous destructor, so a fixture which leaves teardown to the t ### Requirement -`ServerContext::spawn` binds its listener before returning, then spawns the server's public entry point, `base_http_server::serve` or `serve_tls`, as one task through `base_scheduler::Spawn`, so the server runs on the scheduler's workers in parallel with a client driven from the test's thread. The fixture relies only on the server running on the scheduler's workers. It adds nothing to `base-http-server`'s public API and does not choose how connections are spread across tasks or workers. Its bounds are those a server which spawned its own connection tasks would need: the handler is `Handler + Send + Sync + 'static` and shared as an `Arc`, and the scheduler implements `Spawn`, is `Send + Sync + 'static` and is shared through an `Arc`. Native offers no other `'static` handle that can spawn. +`ServerContext::spawn` binds its listener before returning, then spawns the server's public entry point, `base_http_server::serve` or `serve_tls`, as one task through `base_scheduler::Spawn`, so the server runs on the scheduler's workers in parallel with a client driven from the test's thread. The fixture relies only on the server running on the scheduler's workers. It adds nothing to `base-http-server`'s public API and does not choose how connections are spread across tasks or workers. Its bounds are those a server which spawned its own connection tasks would need: the handler is `Handler + Send + Sync + 'static` and shared as an `Arc`, and the scheduler implements `Spawn` and the fixture's `Scheduler` capabilities, is `Send + Sync + 'static` with a `Send` socket type, and is shared through an `Arc`. Native offers no other `'static` handle that can spawn. -The returned `SpawnedServer` exposes the endpoint, address and handler. Its asynchronous `shutdown` cancels the server and waits for it for at most the teardown timeout. It resumes a server panic with its original payload, and fails for a server error or for overrunning the timeout. The timeout is measured on its own thread rather than by the scheduler's clock, so a server that occupies every worker still fails its shutdown instead of hanging it. Dropping the handle without `shutdown` cancels the server, waits for it as `shutdown` does, then panics and names any server failure. If the thread is already unwinding, the drop prints the failure to stderr instead of panicking. `run_spawned` and `run_spawned_on` run a test against a spawned server with the guarantees of `run`: the test and teardown are bounded, the client retires its connections before the server stops, a server failure fails the test, and a test failure is reported before any server failure. The scoped `run` and `run_on` are unchanged. +The returned `SpawnedServer` exposes the endpoint, address and handler. Its asynchronous `shutdown` cancels the server and waits for it for at most the teardown timeout. It resumes a server panic with its original payload, and fails for a server error or for overrunning the timeout. The timeout is measured on its own thread rather than by the scheduler's clock, so a server that occupies every worker still fails its shutdown rather than hanging inside it. A scheduler shutdown afterwards, such as the one `run_spawned` performs, still waits for that server's poll to return, as Native drains its tasks; a poll that never returns therefore still hangs the test after reporting the failure. Dropping the handle without `shutdown` cancels the server, waits for it as `shutdown` does, then panics and names any server failure. If the thread is already unwinding, the drop prints the failure to stderr instead of panicking. `run_spawned` and `run_spawned_on` run a test against a spawned server with the guarantees of `run`: the test and teardown are bounded, the client retires its connections before the server stops, a server failure fails the test, and a test failure is reported before any server failure. The scoped `run` and `run_on` are unchanged. ### Rationale @@ -64,7 +64,7 @@ Inspect `spawn`: its task calls only `serve` or `serve_tls`, and the change leav - a server failure; - a test failure reported before a server failure or panic; - a server panic, resumed by `shutdown`; -- a shutdown that overruns its timeout while a handler blocks a worker; +- a shutdown that overruns its timeout while a handler blocks the only worker, so the scheduler's clock cannot run; - each drop path: clean or failed, and while unwinding or not. ## Independent, bounded raw peers @@ -89,7 +89,7 @@ Inspect peer construction for timeouts and the explicit blocking reset. Search t ### Requirement -`protocol_tests!` and `protocol_trailer_tests!` generate the shared protocol cases for a runner chosen by the caller, and `readable_body_tests!` generates the `ReadableBody` contract cases for a body type. Every case is also a standalone public function, so a partial implementation can call a subset. The protocol suite includes a two-way streaming case, `duplex_stream`. The client streams a request body through a small bounded queue while it reads the streamed echo at the same time, and each direction is `DUPLEX_BODY` long, larger than the loopback sockets between client and server can buffer. The request declares its length and no trailers, so HTTP/1.0 carries it too. The protocol suite runs between the Base client and server over HTTP/1.1, over HTTP/1.0 without the trailer cases, and over TLS with the `rustls` feature. Each of these runs in two modes: scoped, with `run`, and spawned, with `run_spawned`, where the server runs on Native workers and the test drives the client from its own thread. The body suite runs against each body type `base-http` provides. Each assertion is shown to fail against an implementation that breaks it. +`protocol_tests!` and `protocol_trailer_tests!` generate the shared protocol cases for a runner chosen by the caller, and `readable_body_tests!` generates the `ReadableBody` contract cases for a body type. Every case is also a standalone public function, so a partial implementation can call a subset. The protocol suite includes a two-way streaming case, `duplex_stream`. The client streams a request body through a small bounded queue while it reads the streamed echo at the same time, and each direction is `DUPLEX_BODY` long, far more than the loopback sockets between client and server buffer in practice. The request declares its length and no trailers, so HTTP/1.0 carries it too. The protocol suite runs between the Base client and server over HTTP/1.1, over HTTP/1.0 without the trailer cases, and over TLS with the `rustls` feature. Each of these runs in two modes: scoped, with `run`, and spawned, with `run_spawned`, where the server runs on Native workers and the test drives the client from its own thread. The body suite runs against each body type `base-http` provides. Each assertion is shown to fail against an implementation that breaks it. ### Rationale From 5965f50a5b10a813f330682ed6ad6dfa7e587912 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 3 Nov 2026 14:35:35 +1300 Subject: [PATCH 3/6] Keep `Arc` and the spawned server's handle out of the fixtures' API, and make the two-way stream and its stall tests independent of socket buffer sizes. `ServerContext::run_spawned` now owns its scheduler and the application, and the test receives the same `Context<'_, NativeScheduler, H>` as `run`. Spawning, joining and the shared scheduler are private, so `ServerContext::spawn`, `run_spawned_on`, `SpawnedServer` and `shared_native` are gone, along with the drop paths. A unit test runs a blocking handler on a one-worker scheduler and shows that the join still fails within its timeout. `duplex_stream` keeps its upload within `DUPLEX_WINDOW` of the echo received, so a server which buffers the request instead of streaming it stalls the case, as a new sabotage test shows. The sequential stall tests declare 1 GiB and assert that the upload stopped short. A new test shows that a split raw peer keeps its buffered input. The documentation and design now say that a poll which never returns cannot be preempted, so it hangs the scheduler's shutdown. Co-Authored-By: Claude Opus 5.5 --- crates/client/tests/protocol.rs | 7 +- crates/fixtures/readme.md | 2 +- crates/fixtures/src/context.rs | 213 ++++++++++++--------------- crates/fixtures/src/context/tests.rs | 60 ++++++++ crates/fixtures/src/lib.rs | 6 +- crates/fixtures/src/protocol.rs | 49 ++++-- crates/fixtures/src/scheduler.rs | 22 +-- crates/fixtures/tests/protocol.rs | 88 +++++++++-- crates/fixtures/tests/raw.rs | 47 +++++- crates/fixtures/tests/spawned.rs | 196 +++++------------------- design/fixtures.md | 28 ++-- 11 files changed, 366 insertions(+), 352 deletions(-) create mode 100644 crates/fixtures/src/context/tests.rs diff --git a/crates/client/tests/protocol.rs b/crates/client/tests/protocol.rs index 04acbe0..7a1c5c1 100644 --- a/crates/client/tests/protocol.rs +++ b/crates/client/tests/protocol.rs @@ -5,7 +5,6 @@ use base_http::Version; use base_http_fixtures::{Context, ServerContext, protocol::App}; use base_scheduler_native::NativeScheduler; -use std::sync::Arc; fn serve(app: App, version: Version) -> ServerContext { ServerContext::new().handler(app).version(version) @@ -41,7 +40,7 @@ mod https { } // The same cases with the server spawned onto the scheduler's workers while the test drives the client from its own thread, so the two run in parallel. -fn http11_spawned(app: App, test: impl AsyncFnOnce(&Context<'_, NativeScheduler, Arc>)) { +fn http11_spawned(app: App, test: impl AsyncFnOnce(&Context<'_, NativeScheduler, App>)) { serve(app, Version::Http11).run_spawned(test); } @@ -50,7 +49,7 @@ mod http11_spawned { base_http_fixtures::protocol_trailer_tests!(run: super::http11_spawned); } -fn http10_spawned(app: App, test: impl AsyncFnOnce(&Context<'_, NativeScheduler, Arc>)) { +fn http10_spawned(app: App, test: impl AsyncFnOnce(&Context<'_, NativeScheduler, App>)) { serve(app, Version::Http10).run_spawned(test); } @@ -59,7 +58,7 @@ mod http10_spawned { } #[cfg(feature = "rustls")] -fn https_spawned(app: App, test: impl AsyncFnOnce(&Context<'_, NativeScheduler, Arc>)) { +fn https_spawned(app: App, test: impl AsyncFnOnce(&Context<'_, NativeScheduler, App>)) { serve(app, Version::Http11).tls().run_spawned(test); } diff --git a/crates/fixtures/readme.md b/crates/fixtures/readme.md index f1070d5..9742574 100644 --- a/crates/fixtures/readme.md +++ b/crates/fixtures/readme.md @@ -21,7 +21,7 @@ ServerContext::new().run(async |context| { }); ``` -[`ServerContext::run`] polls the server, the client and the test together on the test's thread, which makes interleavings reproducible but cannot run them in parallel. [`ServerContext::run_spawned`] instead spawns the server onto the scheduler's workers and drives the client from the test's thread, with the same bounds and teardown, so cross-thread wake-ups and races can show up. [`ServerContext::spawn`] returns the spawned server's handle, a [`SpawnedServer`], for tests which manage it themselves. +[`ServerContext::run`] polls the server, the client and the test together on the test's thread, which makes interleavings reproducible but cannot run them in parallel. [`ServerContext::run_spawned`] instead spawns the server onto the workers of its own Native scheduler and drives the client from the test's thread, with the same bounds and teardown, so cross-thread wake-ups and races can show up. [`ClientContext`] runs a client of any endpoint, such as a [`RawServer`]. Raw peers in [`raw`] speak HTTP/1 over blocking `std::net` sockets, so they can misbehave in ways the Base server cannot: responding then closing, stalling after a head, truncating a body, or closing or writing to an idle connection. [`RawPeer::split`] gives a peer separate reading and writing halves, for uploading and downloading at once from two threads. [`socket`] waits until a peer's kernel has acknowledged what was sent, so a test observes a close or unsolicited input without sleeping. diff --git a/crates/fixtures/src/context.rs b/crates/fixtures/src/context.rs index 4e8d388..7095189 100644 --- a/crates/fixtures/src/context.rs +++ b/crates/fixtures/src/context.rs @@ -8,9 +8,8 @@ use base_http::{ }; use base_http_client::{Client, Driver, Endpoint}; use base_http_server::Handler; -use base_http_server::Statistics; use base_scheduler::{Cancellation, Clock, SocketType, Spawn, TaskError}; -use base_scheduler_native::NativeScheduler; +use base_scheduler_native::{NativeOptions, NativeScheduler}; use futures_lite::future::{block_on, or, zip}; use std::{ any::Any, @@ -233,7 +232,7 @@ impl ServerContext { Client::new(scheduler, endpoint.clone(), client).expect("valid client options"); let context = Context { scheduler, - handler, + handler: Arc::new(handler), client, endpoint, address, @@ -247,13 +246,14 @@ impl ServerContext { scheduler, listener, tls.server.clone(), - &context.handler, + &*context.handler, &server, &stop, ) .await; } - base_http_server::serve(scheduler, listener, &context.handler, &server, &stop).await + base_http_server::serve(scheduler, listener, &*context.handler, &server, &stop) + .await }; let supervised = supervise( scheduler, @@ -285,14 +285,49 @@ impl ServerContext { } impl ServerContext { - /// Bind a listener and spawn the server onto `scheduler`'s workers as an owned task, returning its handle. + /// Run `test` against a server spawned onto the workers of a fresh two-worker Native scheduler, with the client driven from the test's thread, then stop the server and shut the scheduler down. /// - /// Unlike [`ServerContext::run`], the server makes progress on the scheduler's workers rather than on the test's thread, so a client driven from the test's thread runs in parallel with it. That exposes cross-thread wake-ups and races which a single `block_on` hides, at the cost of reproducible interleavings. The fixture relies only on the server running on the scheduler's workers; how the server spreads its connections across them is the server's choice and may change. + /// Unlike [`ServerContext::run`], the server makes progress on the scheduler's workers rather than on the test's thread, so it runs in parallel with the client. That exposes cross-thread wake-ups and races which a single `block_on` hides, at the cost of reproducible interleavings. The fixture relies only on the server running on the scheduler's workers; how the server spreads its connections across them is the server's choice and may change. The context owns the scheduler and the application, so the application must be `Send + Sync + 'static`, as a server spawning its own tasks needs. /// - /// The listener is bound before this returns, so a client never races the server's start. Binding blocks the calling thread briefly, so call this from outside the scheduler's workers. Shut the server down with [`SpawnedServer::shutdown`]; see [`SpawnedServer`] for what dropping it does instead. + /// This has the guarantees of [`ServerContext::run`]: the listener is bound before the test starts, the test is bounded, the client retires its connections before the server stops, the server is cancelled and must stop within the teardown timeout, a server error fails the test, a server panic is resumed with its own payload, and a test failure is reported first, with any server failure printed beforehand. The teardown timeout is measured on a thread of its own, so a server occupying every worker still fails on time. /// - /// The scheduler is shared through an [`Arc`] because the spawned task must own a `'static` handle to it, and the application is shared as an `Arc` between the server and the handle. - pub fn spawn(self, scheduler: &Arc) -> SpawnedServer + /// Shutting the scheduler down afterwards waits for every task's current poll to return. A poll which never returns, such as a handler blocked forever, cannot be preempted, so it hangs the test after the failure is reported. + pub fn run_spawned( + self, + test: impl AsyncFnOnce(&Context<'_, NativeScheduler, H>) -> T, + ) -> T { + shared_native(2, |scheduler| self.run_spawned_on(scheduler, test)) + } + + // Spawn the server onto `scheduler`, run the test with a client on this thread, then join the server, reporting failures in order. + fn run_spawned_on( + self, + scheduler: &Arc, + test: impl AsyncFnOnce(&Context<'_, S, H>) -> T, + ) -> T + where + S: Scheduler + Spawn + Send + Sync + 'static, + S::Socket: Send, + { + let mut server = self.spawn(scheduler); + let tested = catch_unwind(AssertUnwindSafe(|| server.run(test))); + let stopped = block_on(server.join()); + let output = match tested { + Ok(output) => output, + Err(panic) => { + // The test's failure is usually the cause of a server failure, so report it first. + if let Err(failure) = &stopped { + eprintln!("server failed: {failure}"); + } + resume_unwind(panic) + } + }; + stopped.unwrap_or_else(|failure| failure.raise()); + output + } + + // Bind a listener, then spawn the server's public entry point as one task. The task must own a `'static` handle to the scheduler and the application, so both are shared through an `Arc`. + fn spawn(self, scheduler: &Arc) -> Spawned where S: Scheduler + Spawn + Send + Sync + 'static, S::Socket: Send, @@ -339,7 +374,7 @@ impl ServerContext { } }; let task = scheduler.spawn(serving).expect("spawn the server"); - SpawnedServer { + Spawned { scheduler: scheduler.clone(), task: Some(task), stop, @@ -351,52 +386,12 @@ impl ServerContext { limits, } } - - /// Run `test` against a server spawned onto a fresh Native scheduler's workers, which is shut down afterwards. See [`ServerContext::run_spawned_on`]. - pub fn run_spawned( - self, - test: impl AsyncFnOnce(&Context<'_, NativeScheduler, Arc>) -> T, - ) -> T { - crate::shared_native(|scheduler| self.run_spawned_on(scheduler, test)) - } - - /// Run `test` against a server spawned onto `scheduler`'s workers, with a client driven from the calling thread, then shut the server down. The caller owns and shuts down `scheduler`. - /// - /// This has the teardown guarantees of [`ServerContext::run`]: the test and teardown are bounded, the client retires its connections before the server stops, the server's shutdown is bounded by the teardown timeout, a server failure or panic fails the test, and a test failure is reported first, with any server failure printed beforehand. - pub fn run_spawned_on( - self, - scheduler: &Arc, - test: impl AsyncFnOnce(&Context<'_, S, Arc>) -> T, - ) -> T - where - S: Scheduler + Spawn + Send + Sync + 'static, - S::Socket: Send, - { - let mut server = self.spawn(scheduler); - let tested = catch_unwind(AssertUnwindSafe(|| server.run(test))); - let stopped = block_on(server.join()); - let output = match tested { - Ok(output) => output, - Err(panic) => { - // The test's failure is usually the cause of a server failure, so report it first. - if let Err(failure) = &stopped { - eprintln!("server failed: {failure}"); - } - resume_unwind(panic) - } - }; - stopped.unwrap_or_else(|failure| failure.raise()); - output - } } -/// A server spawned onto a scheduler's workers by [`ServerContext::spawn`]. -/// -/// Shut it down with [`SpawnedServer::shutdown`], which cancels the server, waits for it within the teardown timeout and re-raises its failure. Rust has no asynchronous destructor, so dropping the handle without shutting it down cancels the server, blocks the dropping thread until it stops or the teardown timeout expires, and then panics, naming any server failure, so that a forgotten shutdown cannot hide one. When the thread is already unwinding, it prints the server's failure to stderr instead of panicking again. Drop it outside the scheduler's workers, since a worker blocked in the drop cannot run tasks. -#[must_use = "shut the server down to observe its failures"] -pub struct SpawnedServer { +// A server spawned onto a scheduler's workers. `run_spawned_on` always joins it, so it needs no destructor. +struct Spawned { scheduler: Arc, - // `None` once the server has been stopped. + // `None` once the server has been joined. task: Option>, stop: Cancellation, handler: Arc, @@ -407,7 +402,7 @@ pub struct SpawnedServer { limits: Limits, } -type Served = Result; +type Served = Result; // Why a spawned server did not stop cleanly. enum Failure { @@ -436,54 +431,9 @@ impl fmt::Display for Failure { } } -impl SpawnedServer { - /// The client's endpoint, including any TLS configuration, for constructing clients. - pub fn endpoint(&self) -> &Endpoint { - &self.endpoint - } - - /// The server's listening address. - pub fn address(&self) -> SocketAddr { - self.address - } - - /// The application being served, shared with the server. - pub fn handler(&self) -> &Arc { - &self.handler - } - - /// Cancel the server and wait for it to stop, returning its statistics. - /// - /// Panics if the server does not stop within the teardown timeout or failed, and resumes its panic if it panicked. - pub async fn shutdown(mut self) -> Statistics { - self.join().await.unwrap_or_else(|failure| failure.raise()) - } - - // Cancel the server and join it within the teardown timeout. A server which overruns is destroyed once its task handle is dropped and any poll in progress returns. - async fn join(&mut self) -> Result { - self.stop.cancel(); - let Some(task) = self.task.take() else { - unreachable!("a spawned server is stopped only once") - }; - let timeout = self.limits.teardown; - let joined = or(async { Some(task.await) }, async { - expiry(timeout).await; - None - }) - .await; - match joined { - Some(Ok(served)) => served.map_err(Failure::Server), - Some(Err(error)) => Err(Failure::Task(error)), - None => Err(Failure::Stalled(timeout)), - } - } -} - -impl SpawnedServer { - /// Run `test` with a fresh client of this server, driven on the calling thread while the server keeps running on the scheduler's workers. - /// - /// The test and the client's teardown are bounded as in [`ServerContext::run`], and a test panic propagates after the client has retired its connections. The server keeps running afterwards, so further tests can use it before it is shut down. - pub fn run(&self, test: impl AsyncFnOnce(&Context<'_, S, Arc>) -> T) -> T { +impl Spawned { + // Run `test` with a fresh client of this server, driven on this thread, with the test and client teardown bounded as in `run`. + fn run(&self, test: impl AsyncFnOnce(&Context<'_, S, H>) -> T) -> T { let scheduler = &*self.scheduler; block_on(async { let (client, driver) = @@ -516,9 +466,28 @@ impl SpawnedServer { ) }) } + + // Cancel the server and join it within the teardown timeout. A server which overruns is destroyed once its task handle is dropped and any poll in progress returns. + async fn join(&mut self) -> Result<(), Failure> { + self.stop.cancel(); + let Some(task) = self.task.take() else { + unreachable!("a spawned server is joined only once") + }; + let timeout = self.limits.teardown; + let joined = or(async { Some(task.await) }, async { + expiry(timeout).await; + None + }) + .await; + match joined { + Some(Ok(served)) => served.map(drop).map_err(Failure::Server), + Some(Err(error)) => Err(Failure::Task(error)), + None => Err(Failure::Stalled(timeout)), + } + } } -// Expire after `timeout`, measured on a thread of its own rather than by the scheduler's clock, so a server occupying every worker, such as a handler which blocks, still fails its shutdown. Shutting the scheduler down afterwards still waits for that poll to return. Dropping the future stops the thread. +// Expire after `timeout`, measured on a thread of its own rather than by the scheduler's clock, so a server occupying every worker, such as a handler which blocks, still fails its join. Shutting the scheduler down afterwards still waits for that poll to return. Dropping the future stops the thread. async fn expiry(timeout: Duration) { let expired = Latch::new(); let (_cancel, cancelled) = mpsc::channel::<()>(); @@ -531,22 +500,18 @@ async fn expiry(timeout: Duration) { let _ = expired.wait(&Cancellation::never()).await; } -impl Drop for SpawnedServer { - fn drop(&mut self) { - if self.task.is_none() { - return; - } - let stopped = block_on(self.join()); - match (stopped, thread::panicking()) { - (Ok(_), true) => {} - (Err(failure), true) => eprintln!("server failed: {failure}"), - (Ok(_), false) => panic!("a spawned server was dropped without shutdown"), - (Err(failure), false) => { - eprintln!("a spawned server was dropped without shutdown"); - failure.raise() - } - } - } +// Run `work` with a fresh Native scheduler of `workers` workers, shared through an `Arc` so spawned tasks can own it, then shut it down, also when `work` panics. +fn shared_native(workers: usize, work: impl FnOnce(&Arc) -> T) -> T { + let scheduler = Arc::new( + NativeScheduler::with_options(NativeOptions { + workers, + ..Default::default() + }) + .expect("construct a Native scheduler"), + ); + let result = catch_unwind(AssertUnwindSafe(|| work(&scheduler))); + block_on(scheduler.shutdown()); + result.unwrap_or_else(|panic| resume_unwind(panic)) } async fn listen(scheduler: &S) -> S::Socket { @@ -567,10 +532,11 @@ async fn listen(scheduler: &S) -> S::Socket { socket } -/// What a test receives from [`ServerContext::run`]. +/// What a test receives from [`ServerContext::run`] or [`ServerContext::run_spawned`]. pub struct Context<'s, S, H> { scheduler: &'s S, - handler: H, + // Shared with a spawned server's task. + handler: Arc, client: Client<'s>, endpoint: Endpoint, address: SocketAddr, @@ -822,3 +788,6 @@ async fn bounded( }) .await } + +#[cfg(test)] +mod tests; diff --git a/crates/fixtures/src/context/tests.rs b/crates/fixtures/src/context/tests.rs new file mode 100644 index 0000000..0e35693 --- /dev/null +++ b/crates/fixtures/src/context/tests.rs @@ -0,0 +1,60 @@ +// Released under the MIT License. +// Copyright, 2026, by Samuel Williams. + +use super::*; +use base_http::{InterimHandler, InterimResponses, Request, Response, Status, body::EmptyBody}; +use std::{convert::Infallible, io::Write, net::TcpStream}; + +/// Blocks its worker in every call until released, so the server cannot observe cancellation. +struct Blocking { + started: Latch, + release: Latch, +} + +impl Handler for Blocking { + type Error = Infallible; + type Body = EmptyBody; + + async fn call( + &self, + _: Request, + _: &mut InterimResponses, + _: &Cancellation, + ) -> Result, Infallible> { + self.started.release(); + self.release.wait_blocking(); + Ok(Response::new(Status::OK, EmptyBody)) + } +} + +// With one worker, which the handler blocks, nothing on the scheduler can run, including its clock's timers, yet the join still fails on time. +#[test] +fn a_server_occupying_the_only_worker_fails_its_join() { + let (started, release) = (Latch::new(), Latch::new()); + let context = ServerContext::new() + .handler(Blocking { + started: started.clone(), + release: release.clone(), + }) + .teardown_timeout(Duration::from_millis(50)); + shared_native(1, |scheduler| { + let failure = catch_unwind(AssertUnwindSafe(|| { + context.run_spawned_on(scheduler, async |context| { + let mut stream = TcpStream::connect(context.address()).unwrap(); + stream + .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n") + .unwrap(); + started.wait_blocking(); + stream + }) + })) + .expect_err("a stalled join fails"); + // Let the handler return, so the scheduler can shut down. + release.release(); + let message = failure.downcast_ref::().unwrap(); + assert!( + message.contains("did not stop within 50ms"), + "unexpected failure: {message}" + ); + }); +} diff --git a/crates/fixtures/src/lib.rs b/crates/fixtures/src/lib.rs index 1cdfac0..7730ca3 100644 --- a/crates/fixtures/src/lib.rs +++ b/crates/fixtures/src/lib.rs @@ -15,12 +15,10 @@ pub mod socket; #[cfg(feature = "rustls")] pub use context::SERVER_NAME; -pub use context::{ - ClientContext, Context, Hello, ServerContext, SpawnedServer, TEARDOWN_TIMEOUT, TEST_TIMEOUT, -}; +pub use context::{ClientContext, Context, Hello, ServerContext, TEARDOWN_TIMEOUT, TEST_TIMEOUT}; pub use latch::{LATCH_TIMEOUT, Latch}; pub use raw::{RawListener, RawPeer, RawReader, RawServer, RawWriter}; -pub use scheduler::{Scheduler, native, shared_native}; +pub use scheduler::{Scheduler, native}; /// The messages of `error` and each of its sources, outermost first. pub fn error_chain(error: &dyn std::error::Error) -> Vec { diff --git a/crates/fixtures/src/protocol.rs b/crates/fixtures/src/protocol.rs index 01b0098..f3d27f7 100644 --- a/crates/fixtures/src/protocol.rs +++ b/crates/fixtures/src/protocol.rs @@ -13,9 +13,11 @@ use base_http::{ }; use base_http_server::Handler; use base_scheduler::Cancellation; +use event_listener::{Event, listener}; use futures_lite::future::zip; use std::{ borrow::Borrow, + cell::Cell, collections::VecDeque, convert::Infallible, io, @@ -25,7 +27,7 @@ use std::{ /// The size of the large download, as in `AProtocol`'s "huge body". pub const LARGE_BODY: u64 = 8 * 1024 * 1024; const LARGE_CHUNK: u64 = 64 * 1024; -/// The size of each direction of the two-way stream in [`duplex_stream`], far more than the loopback path between client and server buffers in practice, so the upload cannot finish unless the echo is read at the same time. +/// The size of each direction of the two-way stream in [`duplex_stream`], many times [`DUPLEX_WINDOW`]. pub const DUPLEX_BODY: u64 = 16 * 1024 * 1024; /// The size of each chunk [`duplex_stream`] uploads. pub const DUPLEX_CHUNK: usize = 16 * 1024; @@ -443,15 +445,26 @@ pub fn duplex_data(offset: u64, length: usize) -> Vec { .collect() } -/// A request body streams up while its echo streams down at the same time, each [`DUPLEX_BODY`] long and under backpressure. +/// How far, in bytes, [`duplex_stream`]'s upload may run ahead of the echo it has received: a few chunks, far less than the stream. +pub const DUPLEX_WINDOW: u64 = 4 * DUPLEX_CHUNK as u64; + +/// A request body streams up while its echo streams down at the same time, each [`DUPLEX_BODY`] long. /// -/// The upload is produced through a small bounded queue while the response is read concurrently. The stream is far larger than the sockets between client and server buffer in practice, so a client which finished its upload before reading the echo would stall the exchange until the test's deadline. The request declares its length and no trailers, so HTTP/1.0 can carry it too. +/// The upload never runs more than [`DUPLEX_WINDOW`] ahead of the echo the test has received, so the exchange can only finish if the client delivers the response while it is still sending the request, and the server echoes the request while it is still receiving it. A client which finished its upload before reading the echo, or a server which read the whole request before replying, stalls the exchange until the test's deadline, whatever the sockets buffer. The request declares its length and no trailers, so HTTP/1.0 can carry it too. pub async fn duplex_stream>(context: &Context<'_, S, H>) { let (upload, mut writer) = BodyReader::with_length(4 * DUPLEX_CHUNK, 4, DUPLEX_BODY); let request = context.request(Method::POST, "/duplex", Untrailed(upload)); + let received = Cell::new(0); + let progress = Event::new(); let produce = async { let mut offset = 0; while offset < DUPLEX_BODY { + // Wait for the echo to catch up to within the window. + // The consumer runs in this task, so the echo cannot advance between the check and the listener's registration. + while offset > received.get() + DUPLEX_WINDOW { + listener!(progress => echoed); + echoed.await; + } writer .write(duplex_data(offset, DUPLEX_CHUNK), &never()) .await @@ -467,17 +480,28 @@ pub async fn duplex_stream>(context: &Context<'_, S .await .expect("POST /duplex"); assert_eq!(response.status, Status::OK); - let echoed = body::collect(&mut response.body, &never()) - .await - .map(|collected| collected.data); - let intact = echoed.as_ref().is_ok_and(|data| { - data.len() as u64 == DUPLEX_BODY && *data == duplex_data(0, data.len()) - }); + let mut echoed = Vec::new(); + let read = loop { + match response.body.read(&never()).await { + Ok(frame) => { + let frame = frame.map_data(|data| echoed.extend_from_slice(&data)); + received.set(echoed.len() as u64); + progress.notify(usize::MAX); + if frame.is_end() { + break Ok(()); + } + } + Err(error) => break Err(error), + } + }; + let intact = read.is_ok() + && echoed.len() as u64 == DUPLEX_BODY + && echoed == duplex_data(0, echoed.len()); // Report the length or error rather than the data. assert!( intact, - "the echoed stream: {:?} of {DUPLEX_BODY} bytes", - echoed.as_ref().map(Vec::len) + "the echoed stream: {} of {DUPLEX_BODY} bytes, {read:?}", + echoed.len() ); }; zip(produce, consume).await; @@ -686,7 +710,6 @@ pub async fn response_trailer>(context: &Context<'_ /// use base_http::Version; /// use base_http_fixtures::{Context, ServerContext, protocol::App}; /// use base_scheduler_native::NativeScheduler; -/// use std::sync::Arc; /// /// fn http10(app: App, test: impl AsyncFnOnce(&Context<'_, NativeScheduler, App>)) { /// ServerContext::new() @@ -700,7 +723,7 @@ pub async fn response_trailer>(context: &Context<'_ /// } /// /// // The same cases with the server spawned onto the scheduler's workers, in parallel with the client. -/// fn http10_spawned(app: App, test: impl AsyncFnOnce(&Context<'_, NativeScheduler, Arc>)) { +/// fn http10_spawned(app: App, test: impl AsyncFnOnce(&Context<'_, NativeScheduler, App>)) { /// ServerContext::new() /// .handler(app) /// .version(Version::Http10) diff --git a/crates/fixtures/src/scheduler.rs b/crates/fixtures/src/scheduler.rs index 8cc89d3..481c7e1 100644 --- a/crates/fixtures/src/scheduler.rs +++ b/crates/fixtures/src/scheduler.rs @@ -8,10 +8,7 @@ use base_scheduler::{ }; use base_scheduler_native::{NativeOptions, NativeScheduler}; use futures_lite::future::block_on; -use std::{ - panic::{AssertUnwindSafe, catch_unwind, resume_unwind}, - sync::Arc, -}; +use std::panic::{AssertUnwindSafe, catch_unwind, resume_unwind}; /// The scheduler capabilities a fixture needs to listen, serve, connect and bound its teardown. /// @@ -50,18 +47,11 @@ impl Scheduler for S where /// Run `work` with a fresh two-worker Native scheduler, then shut it down, also when `work` panics, before propagating the panic. pub fn native(work: impl FnOnce(&NativeScheduler) -> T) -> T { - shared_native(|scheduler| work(scheduler)) -} - -/// Run `work` with a fresh two-worker Native scheduler shared through an [`Arc`], as [`ServerContext::spawn`](crate::ServerContext::spawn) needs, then shut it down as [`native`] does. -pub fn shared_native(work: impl FnOnce(&Arc) -> T) -> T { - let scheduler = Arc::new( - NativeScheduler::with_options(NativeOptions { - workers: 2, - ..Default::default() - }) - .expect("construct a Native scheduler"), - ); + let scheduler = NativeScheduler::with_options(NativeOptions { + workers: 2, + ..Default::default() + }) + .expect("construct a Native scheduler"); let result = catch_unwind(AssertUnwindSafe(|| work(&scheduler))); block_on(scheduler.shutdown()); result.unwrap_or_else(|panic| resume_unwind(panic)) diff --git a/crates/fixtures/tests/protocol.rs b/crates/fixtures/tests/protocol.rs index 4505976..4f34eaa 100644 --- a/crates/fixtures/tests/protocol.rs +++ b/crates/fixtures/tests/protocol.rs @@ -12,8 +12,9 @@ use base_http_fixtures::{ }; use base_http_server::Handler; use base_scheduler::Cancellation; -use futures_lite::future::zip; -use std::{borrow::Borrow, io, time::Duration}; +use base_scheduler::Clock; +use futures_lite::future::{or, zip}; +use std::{borrow::Borrow, cell::Cell, io, time::Duration}; /// The shared application, sabotaged on every route the protocol cases use. struct Sabotaged(App); @@ -205,35 +206,94 @@ fn a_truncated_trailer_upload_fails_the_handler() { }); } -// The two-way stream is larger than the sockets can buffer: a client which uploads it all before reading the echo stalls until the deadline, so the case can only pass by streaming both ways at once. +// Far more than any socket buffers, so a sequential upload of it stalls whatever the host's buffer sizes. +const SEQUENTIAL_BODY: u64 = 1 << 30; + +// A client which uploads the whole request before reading the echo stalls once the unread echo fills the buffers between it and the server, which is why the two-way case must read and write at once. #[test] -#[should_panic(expected = "test exceeded")] fn a_sequential_duplex_stream_stalls() { ServerContext::new() .handler(App::new()) - .timeout(Duration::from_secs(1)) .run(async |context| { - let (upload, mut writer) = BodyReader::bounded(4 * protocol::DUPLEX_CHUNK, 4); + let (upload, mut writer) = + BodyReader::with_length(4 * protocol::DUPLEX_CHUNK, 4, SEQUENTIAL_BODY); let request = context.request(Method::POST, "/duplex", upload); + let written = Cell::new(0); let upload_all = async { - let mut offset = 0; - while offset < protocol::DUPLEX_BODY { + while written.get() < SEQUENTIAL_BODY { writer .write( - protocol::duplex_data(offset, protocol::DUPLEX_CHUNK), + protocol::duplex_data(written.get(), protocol::DUPLEX_CHUNK), &Cancellation::never(), ) .await .unwrap(); - offset += protocol::DUPLEX_CHUNK as u64; + written.set(written.get() + protocol::DUPLEX_CHUNK as u64); } writer.close_write(None); }; - let (response, ()) = zip( - context.client().call(request, &Cancellation::never()), - upload_all, + let sequential = async { + let (response, ()) = zip( + context.client().call(request, &Cancellation::never()), + upload_all, + ) + .await; + body::payload(&mut response.unwrap().body).await; + }; + let stalled = or( + async { + sequential.await; + false + }, + async { + let _ = context + .scheduler() + .sleep(Duration::from_millis(500), &Cancellation::never()) + .await; + true + }, ) .await; - body::payload(&mut response.unwrap().body).await; + assert!( + stalled && written.get() < SEQUENTIAL_BODY, + "the upload did not stall: {} bytes written", + written.get() + ); }); } + +/// Reads the whole two-way stream's request before echoing it, as a server which buffers instead of streaming does. +struct Buffering(App); + +impl Borrow for Buffering { + fn borrow(&self) -> &App { + &self.0 + } +} + +impl Handler for Buffering { + type Error = io::Error; + type Body = Reply; + + async fn call( + &self, + mut request: Request, + _: &mut InterimResponses, + cancellation: &Cancellation, + ) -> Result>, io::Error> { + let collected = body::collect(&mut request.body, cancellation) + .await + .map_err(io::Error::other)?; + Ok(reply(collected.data)) + } +} + +// The upload stays within a few chunks of the echo, so a server which buffers the request never receives enough to reply, and the case cannot pass. +#[test] +#[should_panic(expected = "test exceeded")] +fn the_duplex_stream_stalls_against_a_buffering_server() { + ServerContext::new() + .handler(Buffering(App::new())) + .timeout(Duration::from_secs(1)) + .run(async |context| protocol::duplex_stream(context).await); +} diff --git a/crates/fixtures/tests/raw.rs b/crates/fixtures/tests/raw.rs index 339fb37..e0f6887 100644 --- a/crates/fixtures/tests/raw.rs +++ b/crates/fixtures/tests/raw.rs @@ -161,13 +161,38 @@ fn respond_then_close_serves_one_response() { server.join(); } +// Input a peer buffered before the split is read through the reader, not lost. +#[test] +fn a_split_raw_peer_keeps_its_buffered_input() { + let sent = Latch::new(); + let server = RawServer::spawn({ + let sent = sent.clone(); + move |listener| { + let mut peer = listener.accept(); + peer.write("first\r\nsecond\r\n"); + peer.close_write_acknowledged(); + sent.release(); + peer.expect_closed(); + } + }); + let mut peer = RawPeer::connect(server.address()); + // Both lines have arrived in one write, so reading the first buffers the second, and a reader which lost the buffer would read EOF. + sent.wait_blocking(); + assert_eq!(peer.read_line(), "first"); + let (mut reader, mut writer) = peer.split(); + assert_eq!(raw::read_line(&mut reader), "second"); + assert_eq!(reader.read(&mut [0; 1]).unwrap(), 0); + writer.shutdown_write(); + server.join(); +} + fn upload_head(length: u64) -> String { format!( "POST /duplex HTTP/1.1\r\nHost: localhost\r\nContent-Length: {length}\r\nConnection: close\r\n\r\n" ) } -// The halves of a split peer upload and download at the same time from two threads, each direction larger than the sockets can buffer. Done sequentially on one thread, the upload would stall once the unread echo filled the buffers, as `a_sequential_raw_upload_stalls_under_backpressure` shows. +// The halves of a split peer upload and download at the same time from two threads. Done sequentially on one thread, the upload stalls once the unread echo fills the buffers, as `a_sequential_raw_upload_stalls_under_backpressure` shows. #[test] fn a_split_raw_peer_uploads_and_downloads_at_once() { ServerContext::new() @@ -204,9 +229,10 @@ fn a_split_raw_peer_uploads_and_downloads_at_once() { }); } -// Writing the whole upload before reading the echo stalls: the server stops reading once its echo is not read, so the write cannot finish. +// Writing the whole upload before reading the echo stalls: the server stops reading once its echo is not read, so the writes cannot finish. The declared upload is far more than any socket buffers, so it stalls whatever the host's buffer sizes. #[test] fn a_sequential_raw_upload_stalls_under_backpressure() { + const DECLARED: u64 = 1 << 30; ServerContext::new() .handler(App::new()) .run(async |context| { @@ -215,11 +241,18 @@ fn a_sequential_raw_upload_stalls_under_backpressure() { peer.get_mut() .set_write_timeout(Some(Duration::from_millis(500))) .unwrap(); - peer.write(upload_head(DUPLEX_BODY)); - let error = peer - .get_mut() - .write_all(&duplex_data(0, DUPLEX_BODY as usize)) - .expect_err("a sequential upload stalls"); + peer.write(upload_head(DECLARED)); + let mut written = 0; + let error = loop { + assert!(written < DECLARED, "the upload did not stall"); + if let Err(error) = peer + .get_mut() + .write_all(&duplex_data(written, DUPLEX_CHUNK)) + { + break error; + } + written += DUPLEX_CHUNK as u64; + }; assert!( matches!( error.kind(), diff --git a/crates/fixtures/tests/spawned.rs b/crates/fixtures/tests/spawned.rs index 9903d9a..432b7f0 100644 --- a/crates/fixtures/tests/spawned.rs +++ b/crates/fixtures/tests/spawned.rs @@ -1,26 +1,16 @@ // Released under the MIT License. // Copyright, 2026, by Samuel Williams. -//! A spawned server runs on the scheduler's workers beside a client driven from the test's thread, shuts down within its bound, and reports each way it can fail, including when its handle is dropped. +//! A spawned server runs on the scheduler's workers beside a client driven from the test's thread, and reports each way it can fail, in order. use base_http::{ InterimHandler, InterimResponses, Method, Request, Response, Status, body::EmptyBody, body::ReadableBody, }; -use base_http_fixtures::{ - Hello, Latch, ServerContext, body::payload, protocol::App, shared_native, -}; +use base_http_fixtures::{Context, Hello, ServerContext, body::payload, protocol::App}; use base_http_server::Handler; use base_scheduler::Cancellation; -use base_scheduler_native::{NativeOptions, NativeScheduler}; -use futures_lite::future::block_on; -use std::{ - convert::Infallible, - io::{Read, Write}, - net::TcpStream, - panic::{AssertUnwindSafe, catch_unwind}, - sync::Arc, - time::Duration, -}; +use base_scheduler_native::NativeScheduler; +use std::{convert::Infallible, io::Read}; fn failing() -> ServerContext { ServerContext::new().server_options(base_http_server::Options { @@ -29,41 +19,28 @@ fn failing() -> ServerContext { }) } -async fn hello( - context: &base_http_fixtures::Context<'_, impl base_http_fixtures::Scheduler, Arc>, -) { - let mut response = context - .client() - .call( - context.request(Method::GET, "/", EmptyBody), - &Cancellation::never(), - ) - .await - .unwrap(); - assert_eq!(response.status, Status::OK); - assert_eq!(payload(&mut response.body).await, b"Hello World!"); -} - #[test] -fn a_spawned_server_serves_several_clients_and_shuts_down() { - shared_native(|scheduler| { - let server = ServerContext::new().spawn(scheduler); - let _: &Arc = server.handler(); - assert_eq!(server.endpoint().authority, server.address().to_string()); - server.run(async |context| hello(context).await); - let output = server.run(async |context| { - hello(context).await; - context - .raw(|mut peer| { - peer.write("GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"); - let (head, _) = peer.read_message(); - head.status() - }) - .await - }); - assert_eq!(output, 200); - let statistics = block_on(server.shutdown()); - assert_eq!(statistics.requests, 3); +fn a_spawned_server_serves_a_client_and_a_raw_peer() { + ServerContext::new().run_spawned(async |context: &Context<'_, NativeScheduler, Hello>| { + let _: &Hello = context.handler(); + let mut response = context + .client() + .call( + context.request(Method::GET, "/", EmptyBody), + &Cancellation::never(), + ) + .await + .unwrap(); + assert_eq!(response.status, Status::OK); + assert_eq!(payload(&mut response.body).await, b"Hello World!"); + let status = context + .raw(|mut peer| { + peer.write("GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"); + let (head, _) = peer.read_message(); + head.status() + }) + .await; + assert_eq!(status, 200); }); } @@ -114,112 +91,23 @@ impl Handler for Panicking { } } -// A request the server cannot answer, sent without a client so the test does not wait for a response. -fn request(server: std::net::SocketAddr) -> TcpStream { - let mut stream = TcpStream::connect(server).unwrap(); - stream - .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n") - .unwrap(); - stream -} - -#[test] -#[should_panic(expected = "the handler panicked")] -fn shutdown_resumes_a_server_panic() { - shared_native(|scheduler| { - let server = ServerContext::new().handler(Panicking).spawn(scheduler); - let mut stream = request(server.address()); - // The server's task ends when the handler panics, closing the connection. - let _ = stream.read(&mut [0; 1]); - block_on(server.shutdown()); - }); -} - -/// Blocks its worker in every call until released, so the server cannot observe cancellation. -struct Blocking { - started: Latch, - release: Latch, -} - -impl Handler for Blocking { - type Error = Infallible; - type Body = EmptyBody; - - async fn call( - &self, - _: Request, - _: &mut InterimResponses, - _: &Cancellation, - ) -> Result, Infallible> { - self.started.release(); - self.release.wait_blocking(); - Ok(Response::new(Status::OK, EmptyBody)) - } -} - -#[test] -fn a_server_which_does_not_stop_fails_its_shutdown() { - let (started, release) = (Latch::new(), Latch::new()); - // With one worker, which the handler blocks, nothing on the scheduler can run, including its clock's timers. - let scheduler = Arc::new( - NativeScheduler::with_options(NativeOptions { - workers: 1, - ..Default::default() +// Sent without a client, so the test does not wait for a response. +async fn panic_the_server(context: &Context<'_, NativeScheduler, H>) { + context + .raw(|mut peer| { + peer.write("GET / HTTP/1.1\r\nHost: localhost\r\n\r\n"); + // The server's task ends when the handler panics, closing the connection. + let _ = peer.read(&mut [0; 1]); }) - .unwrap(), - ); - { - let scheduler = &scheduler; - let server = ServerContext::new() - .handler(Blocking { - started: started.clone(), - release: release.clone(), - }) - .teardown_timeout(Duration::from_millis(50)) - .spawn(scheduler); - let _stream = request(server.address()); - started.wait_blocking(); - let failure = catch_unwind(AssertUnwindSafe(|| block_on(server.shutdown()))) - .expect_err("a stalled shutdown fails"); - release.release(); - let message = failure.downcast_ref::().unwrap(); - assert!( - message.contains("did not stop within 50ms"), - "unexpected failure: {message}" - ); - } - block_on(scheduler.shutdown()); + .await; } #[test] -#[should_panic(expected = "dropped without shutdown")] -fn dropping_a_spawned_server_without_shutdown_fails() { - shared_native(|scheduler| drop(ServerContext::new().spawn(scheduler))); -} - -#[test] -#[should_panic(expected = "server failed")] -fn dropping_a_failed_spawned_server_reports_its_failure() { - shared_native(|scheduler| drop(failing().spawn(scheduler))); -} - -// While a test unwinds, the dropped server is still cancelled and joined, and the test's own panic is reported. -#[test] -#[should_panic(expected = "the test failed")] -fn a_spawned_server_dropped_during_a_panic_is_stopped() { - shared_native(|scheduler| { - let _server = ServerContext::new().spawn(scheduler); - panic!("the test failed"); - }); -} - -#[test] -#[should_panic(expected = "the test failed")] -fn a_failed_spawned_server_dropped_during_a_panic_is_reported() { - shared_native(|scheduler| { - let _server = failing().spawn(scheduler); - panic!("the test failed"); - }); +#[should_panic(expected = "the handler panicked")] +fn a_spawned_server_panic_is_resumed() { + ServerContext::new() + .handler(Panicking) + .run_spawned(async |context| panic_the_server(context).await); } // A server panic caused by the test is reported after the test's own failure. @@ -229,13 +117,7 @@ fn a_test_panic_is_reported_before_a_spawned_server_panic() { ServerContext::new() .handler(Panicking) .run_spawned(async |context| { - context - .raw(|mut peer| { - peer.write("GET / HTTP/1.1\r\nHost: localhost\r\n\r\n"); - // The server's task ends when the handler panics, closing the connection. - let _ = peer.read(&mut [0; 1]); - }) - .await; + panic_the_server(context).await; panic!("the test failed first"); }); } diff --git a/design/fixtures.md b/design/fixtures.md index 1f260e8..ee44d4a 100644 --- a/design/fixtures.md +++ b/design/fixtures.md @@ -42,36 +42,36 @@ Rust has no asynchronous destructor, so a fixture which leaves teardown to the t ### Requirement -`ServerContext::spawn` binds its listener before returning, then spawns the server's public entry point, `base_http_server::serve` or `serve_tls`, as one task through `base_scheduler::Spawn`, so the server runs on the scheduler's workers in parallel with a client driven from the test's thread. The fixture relies only on the server running on the scheduler's workers. It adds nothing to `base-http-server`'s public API and does not choose how connections are spread across tasks or workers. Its bounds are those a server which spawned its own connection tasks would need: the handler is `Handler + Send + Sync + 'static` and shared as an `Arc`, and the scheduler implements `Spawn` and the fixture's `Scheduler` capabilities, is `Send + Sync + 'static` with a `Send` socket type, and is shared through an `Arc`. Native offers no other `'static` handle that can spawn. +`ServerContext::run_spawned` creates a two-worker Native scheduler, binds the listener, then spawns the server's public entry point, `base_http_server::serve` or `serve_tls`, as one task through `base_scheduler::Spawn`, so the server runs on the scheduler's workers in parallel with a client driven from the test's thread. It takes ownership of the application and owns the scheduler for the whole run; the test receives the same `Context<'_, NativeScheduler, H>` as `run`, with the application as `&H`. Any sharing between the test and the server's task, such as an `Arc`, stays inside the fixture and is not part of its public API. The fixture relies only on the server running on the scheduler's workers. It adds nothing to `base-http-server`'s public API and does not choose how connections are spread across tasks or workers. The application must be `Handler + Send + Sync + 'static`, as a server which spawned its own connection tasks would need. -The returned `SpawnedServer` exposes the endpoint, address and handler. Its asynchronous `shutdown` cancels the server and waits for it for at most the teardown timeout. It resumes a server panic with its original payload, and fails for a server error or for overrunning the timeout. The timeout is measured on its own thread rather than by the scheduler's clock, so a server that occupies every worker still fails its shutdown rather than hanging inside it. A scheduler shutdown afterwards, such as the one `run_spawned` performs, still waits for that server's poll to return, as Native drains its tasks; a poll that never returns therefore still hangs the test after reporting the failure. Dropping the handle without `shutdown` cancels the server, waits for it as `shutdown` does, then panics and names any server failure. If the thread is already unwinding, the drop prints the failure to stderr instead of panicking. `run_spawned` and `run_spawned_on` run a test against a spawned server with the guarantees of `run`: the test and teardown are bounded, the client retires its connections before the server stops, a server failure fails the test, and a test failure is reported before any server failure. The scoped `run` and `run_on` are unchanged. +`run_spawned` has the guarantees of `run`: the listener is bound before the test starts, the test is bounded, the client retires its connections before the server stops, a server error fails the test, a server panic is resumed with its original payload, and a test failure is reported before any server failure, which is printed to stderr first. After the test, the server is cancelled and joined for at most the teardown timeout, and fails the test if it overruns. That timeout is measured on its own thread rather than by the scheduler's clock, so a server occupying every worker, such as a handler which blocks, still fails on time. The scheduler is then shut down, and Native drains its tasks, so the call does not return until an overrunning server's current poll returns. A poll which never returns cannot be preempted and hangs the test after the failure is reported, so teardown is bounded only for servers whose polls return. The scoped `run` and `run_on` are unchanged. ### Rationale -`run` polls the server, the client's driver and the test in one `block_on` on the test's thread. That makes results reproducible, but nothing runs in parallel, so cross-thread wake-up bugs and races in two-way streaming cannot show up. The user decided that the server's connection and task layout is an implementation detail. Relying only on the public entry points, with the bounds an internally spawning server would need, means that a later move to per-connection tasks inside `serve` does not break the fixtures. Rust has no asynchronous destructor, so a forgotten shutdown panics rather than silently discarding the server's outcome. +`run` polls the server, the client's driver and the test in one `block_on` on the test's thread. That makes results reproducible, but nothing runs in parallel, so cross-thread wake-up bugs and races in two-way streaming cannot show up. The user decided that the server's connection and task layout is an implementation detail. Relying only on the public entry points, with the bounds an internally spawning server would need, means that a later move to per-connection tasks inside `serve` does not break the fixtures. The user decided to keep `Arc` and the spawned server's handle out of the public API: a context which owns its scheduler and application can join the server itself, so no forgotten shutdown or drop path exists for a test to get wrong. ### Scope -`ServerContext::spawn`, `ServerContext::run_spawned`, `ServerContext::run_spawned_on` and `SpawnedServer` in `crates/fixtures/src/context.rs`, and `shared_native` in `crates/fixtures/src/scheduler.rs`. +`ServerContext::run_spawned` and its private spawning, joining and scheduler helpers in `crates/fixtures/src/context.rs`. ### Validation -Inspect `spawn`: its task calls only `serve` or `serve_tls`, and the change leaves `crates/server/src` untouched. `crates/fixtures/tests/spawned.rs` covers the following, and checks how each is reported: +Inspect the private spawn: its task calls only `serve` or `serve_tls`, and the change leaves `crates/server/src` untouched. Check that `crates/fixtures/src/lib.rs` exports no `Arc`-based or spawned-server type. `crates/fixtures/tests/spawned.rs` covers the following, and checks how each is reported: -- one spawned server serving several clients and a raw peer; +- a spawned server serving a client and a raw peer; - an application shared with the server; - a test panic; - a server failure; -- a test failure reported before a server failure or panic; -- a server panic, resumed by `shutdown`; -- a shutdown that overruns its timeout while a handler blocks the only worker, so the scheduler's clock cannot run; -- each drop path: clean or failed, and while unwinding or not. +- a server panic, resumed with its payload; +- a test failure reported before a server failure or panic. + +The unit test in `crates/fixtures/src/context/tests.rs` runs a server on a one-worker scheduler whose handler blocks that worker, so the scheduler's clock cannot run, and shows that the join still fails within its timeout. ## Independent, bounded raw peers ### Requirement -Raw peers use blocking `std::net` sockets and no Base encoder, decoder or scheduler. Every read, write, accept, latch wait and acknowledgement wait is bounded by a timeout shorter than the default test timeout, so a stalled peer reports its own failure, and an accepted socket is explicitly made blocking, since it inherits the listener's nonblocking mode on some platforms. Tests sequence peers with latches or acknowledgement, not sleeps. A canned peer which closes a connection or writes outside an exchange waits until the other side's kernel has acknowledged it, using `socket::wait_acknowledged`, before continuing or signalling the test. Linux and macOS query the kernel; on other platforms the wait is a 50 ms sleep, which is documented on the peer helpers. Raw peers never block a scheduler thread: `Context::raw` runs one on its own thread. `RawPeer::split` divides a peer into a reader, which keeps the peer's buffered input, and a writer cloned from its socket with `TcpStream::try_clone`. Each half can be used from its own thread, so one peer can upload and download at the same time. +Raw peers use blocking `std::net` sockets and no Base encoder, decoder or scheduler. Every read, write, accept, latch wait and acknowledgement wait is bounded by a timeout shorter than the default test timeout, so a stalled peer reports its own failure, and an accepted socket is explicitly made blocking, since it inherits the listener's nonblocking mode on some platforms. Tests sequence peers with latches or acknowledgement, not sleeps. A canned peer which closes a connection or writes outside an exchange waits until the other side's kernel has acknowledged it, using `socket::wait_acknowledged`, before continuing or signalling the test. Linux and macOS query the kernel; on other platforms the wait is a 50 ms sleep, which is documented on the peer helpers. Raw peers never block a scheduler thread: `Context::raw` runs one on its own thread. `RawPeer::split` divides a peer into a reader, which keeps the peer's buffered input, as a test shows by buffering input before the split, and a writer cloned from its socket with `TcpStream::try_clone`. Each half can be used from its own thread, so one peer can upload and download at the same time. ### Rationale @@ -83,13 +83,13 @@ Peers that share the implementation's codec can hide a shared bug. Unbounded blo ### Validation -Inspect peer construction for timeouts and the explicit blocking reset. Search tests built on the fixtures for `thread::sleep`. `crates/fixtures/tests/acknowledgement.rs` shows, on Linux and macOS, that a bounded wait expires while a peer leaves data unread, and that `wait_acknowledged` returns once the peer reads the data and the FIN. A loopback peer acknowledges a FIN as soon as it arrives, so no test can hold the window in which only the FIN is unacknowledged. Linux counts the FIN in the same queue as the data, but the separate connection-state check macOS needs is justified by inspection, not by a test. `crates/fixtures/tests/raw.rs` streams `DUPLEX_BODY` up and down at the same time through a split peer. It also shows that a sequential upload of the same size on one thread stalls until its write timeout. +Inspect peer construction for timeouts and the explicit blocking reset. Search tests built on the fixtures for `thread::sleep`. `crates/fixtures/tests/acknowledgement.rs` shows, on Linux and macOS, that a bounded wait expires while a peer leaves data unread, and that `wait_acknowledged` returns once the peer reads the data and the FIN. A loopback peer acknowledges a FIN as soon as it arrives, so no test can hold the window in which only the FIN is unacknowledged. Linux counts the FIN in the same queue as the data, but the separate connection-state check macOS needs is justified by inspection, not by a test. `crates/fixtures/tests/raw.rs` streams `DUPLEX_BODY` up and down at the same time through a split peer. It also shows that a sequential upload on one thread stalls until its write timeout; that upload declares 1 GiB, far more than any socket buffers, so the stall does not depend on the host's buffer sizes. ## Behaviour suites ### Requirement -`protocol_tests!` and `protocol_trailer_tests!` generate the shared protocol cases for a runner chosen by the caller, and `readable_body_tests!` generates the `ReadableBody` contract cases for a body type. Every case is also a standalone public function, so a partial implementation can call a subset. The protocol suite includes a two-way streaming case, `duplex_stream`. The client streams a request body through a small bounded queue while it reads the streamed echo at the same time, and each direction is `DUPLEX_BODY` long, far more than the loopback sockets between client and server buffer in practice. The request declares its length and no trailers, so HTTP/1.0 carries it too. The protocol suite runs between the Base client and server over HTTP/1.1, over HTTP/1.0 without the trailer cases, and over TLS with the `rustls` feature. Each of these runs in two modes: scoped, with `run`, and spawned, with `run_spawned`, where the server runs on Native workers and the test drives the client from its own thread. The body suite runs against each body type `base-http` provides. Each assertion is shown to fail against an implementation that breaks it. +`protocol_tests!` and `protocol_trailer_tests!` generate the shared protocol cases for a runner chosen by the caller, and `readable_body_tests!` generates the `ReadableBody` contract cases for a body type. Every case is also a standalone public function, so a partial implementation can call a subset. The protocol suite includes a two-way streaming case, `duplex_stream`. The client streams a request body while it reads the streamed echo at the same time, each direction `DUPLEX_BODY` long, and the upload never runs more than `DUPLEX_WINDOW`, a few chunks, ahead of the echo received. The exchange can therefore finish only if the client delivers the response while still sending the request and the server echoes the request while still receiving it, whatever the sockets buffer. The request declares its length and no trailers, so HTTP/1.0 carries it too. The protocol suite runs between the Base client and server over HTTP/1.1, over HTTP/1.0 without the trailer cases, and over TLS with the `rustls` feature. Each of these runs in two modes: scoped, with `run`, and spawned, with `run_spawned`, where the server runs on Native workers and the test drives the client from its own thread. The body suite runs against each body type `base-http` provides. Each assertion is shown to fail against an implementation that breaks it. ### Rationale @@ -101,4 +101,4 @@ Each new engine or body type should be checked against one shared definition of ### Validation -Check that the suites are applied as stated. `crates/fixtures/tests/protocol.rs` runs every protocol case against an application sabotaged on that case's route and requires the failure message of the assertion the case exists for; `crates/fixtures/tests/body.rs` runs every body assertion against a body with the corresponding fault and requires the expected failure. `a_sequential_duplex_stream_stalls` in `crates/fixtures/tests/protocol.rs` shows that the two-way stream cannot complete when the client uploads it all before reading. The case can therefore pass only by streaming both ways at once. Whether it stalls depends on the platform's socket buffer sizes, so a platform with larger loopback buffers needs a larger `DUPLEX_BODY`. +Check that the suites are applied as stated. `crates/fixtures/tests/protocol.rs` runs every protocol case against an application sabotaged on that case's route and requires the failure message of the assertion the case exists for; `crates/fixtures/tests/body.rs` runs every body assertion against a body with the corresponding fault and requires the expected failure. `the_duplex_stream_stalls_against_a_buffering_server` in `crates/fixtures/tests/protocol.rs` shows that the two-way case cannot pass against a server which reads the whole request before replying. `a_sequential_duplex_stream_stalls` shows that a client which uploads a 1 GiB request before reading the echo stalls, and asserts that the upload stopped short, so the window is not what makes a sequential exchange fail. From 0a5b82825929c137cc1f76640e59efe3502e1302 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 3 Nov 2026 16:28:28 +1300 Subject: [PATCH 4/6] Drop the `Untrailed` wrapper and add a two-way case whose echo is chunked. --- crates/fixtures/src/protocol.rs | 93 ++++++++++++++++++++----------- crates/fixtures/tests/body.rs | 12 ++++ crates/fixtures/tests/protocol.rs | 16 ++++++ design/fixtures.md | 4 +- 4 files changed, 89 insertions(+), 36 deletions(-) diff --git a/crates/fixtures/src/protocol.rs b/crates/fixtures/src/protocol.rs index d808aab..9eac380 100644 --- a/crates/fixtures/src/protocol.rs +++ b/crates/fixtures/src/protocol.rs @@ -40,6 +40,7 @@ const LINK: &str = "; rel=preload; as=style"; /// | `GET /`, `HEAD /` | `200` with `my-header: my-value` and `{method} {version}` | /// | `POST /` | `200` streaming the request body back | /// | `POST /duplex` | `200` streaming the request body back, for the two-way streaming case | +/// | `POST /duplex-chunked` | `200` streaming the request body back with no length and possible trailers, so HTTP/1.1 chunks it | /// | `POST /length` | `200` with `Content Length: {n}`, the request body's length | /// | `GET /interim` | `103 Early Hints` with a `link` field to HTTP/1.1 clients, then `200` with `Hello World` | /// | `POST /request-trailer` | `200` with the request's `etag` trailer, after releasing [`App::request_received`] | @@ -108,6 +109,9 @@ impl Handler for App { (_, "/duplex") => { return Ok(Response::new(Status::OK, Reply::Echo(request.body))); } + (_, "/duplex-chunked") => { + return Ok(Response::new(Status::OK, Reply::Stream(request.body))); + } (_, "/length") => { let length = request .body @@ -252,6 +256,8 @@ pub enum Reply { Buffered(BufferedBody>), /// The request body, streamed back. Echo(B), + /// The request body, streamed back without its length and reporting that it may have trailers, so an HTTP/1.1 response is chunked. + Stream(B), /// A scripted body. Staged(Staged), /// Zero bytes, produced on demand. @@ -282,7 +288,7 @@ impl ReadableBody for Reply { fn is_ready(&self) -> bool { match self { Self::Buffered(body) => body.is_ready(), - Self::Echo(body) => body.is_ready(), + Self::Echo(body) | Self::Stream(body) => body.is_ready(), Self::Staged(_) => false, Self::Zeroes(_) => true, } @@ -291,7 +297,7 @@ impl ReadableBody for Reply { fn is_empty(&self) -> bool { match self { Self::Buffered(body) => body.is_empty(), - Self::Echo(body) => body.is_empty(), + Self::Echo(body) | Self::Stream(body) => body.is_empty(), Self::Staged(body) => body.is_empty(), Self::Zeroes(body) => body.remaining == 0, } @@ -301,6 +307,7 @@ impl ReadableBody for Reply { match self { Self::Buffered(body) => body.length(), Self::Echo(body) => body.length(), + Self::Stream(_) => None, Self::Staged(body) => body.length(), Self::Zeroes(body) => Some(body.length), } @@ -310,6 +317,7 @@ impl ReadableBody for Reply { match self { Self::Buffered(body) => body.may_have_trailers(), Self::Echo(body) => body.may_have_trailers(), + Self::Stream(_) => true, Self::Staged(body) => body.may_have_trailers(), Self::Zeroes(_) => false, } @@ -318,7 +326,7 @@ impl ReadableBody for Reply { fn close(&mut self, error: Option) { match self { Self::Buffered(body) => body.close(None), - Self::Echo(body) => body.close(None), + Self::Echo(body) | Self::Stream(body) => body.close(None), Self::Staged(body) => body.close(error), Self::Zeroes(body) => body.remaining = 0, } @@ -327,7 +335,7 @@ impl ReadableBody for Reply { async fn read(&mut self, cancellation: &Cancellation) -> Result>, BodyError> { match self { Self::Buffered(body) => Ok(body.read(cancellation).await?), - Self::Echo(body) => Ok(body + Self::Echo(body) | Self::Stream(body) => Ok(body .read(cancellation) .await .map_err(BodyError::new)? @@ -414,46 +422,61 @@ pub async fn post>(context: &Context<'_, S, H>) { ); } -// A streamed upload which declares that it has no trailers. A `BodyReader` may always carry them, and HTTP/1.0, which cannot frame trailers, refuses to send such a body. -struct Untrailed(BodyReader>); - -impl ReadableBody for Untrailed { - type Data = Vec; - type Error = BodyError; - - fn length(&self) -> Option { - self.0.length() - } - - fn may_have_trailers(&self) -> bool { - false - } - - fn close(&mut self, error: Option) { - self.0.close(error); - } - - async fn read(&mut self, cancellation: &Cancellation) -> Result>, BodyError> { - self.0.read(cancellation).await - } -} - /// The data of the two-way stream from `offset`: a pattern with a period of 251 bytes, which does not divide the chunk size, so a lost, repeated or reordered chunk changes the data or its length. pub fn duplex_data(offset: u64, length: usize) -> Vec { - (offset..offset + length as u64) - .map(|index| (index % 251) as u8) - .collect() + // Copy whole periods rather than computing each byte: per-byte work dominates the parallel two-way cases under coverage instrumentation. + let mut data = Vec::with_capacity(length); + let mut start = (offset % DUPLEX_PERIOD.len() as u64) as usize; + while data.len() < length { + let end = DUPLEX_PERIOD.len().min(start + length - data.len()); + data.extend_from_slice(&DUPLEX_PERIOD[start..end]); + start = 0; + } + data } +// One period of the two-way stream's data. +const DUPLEX_PERIOD: [u8; 251] = { + let mut period = [0; 251]; + let mut index = 0; + while index < period.len() { + period[index] = index as u8; + index += 1; + } + period +}; + /// How far, in bytes, [`duplex_stream`]'s upload may run ahead of the echo it has received: a few chunks, far less than the stream. pub const DUPLEX_WINDOW: u64 = 4 * DUPLEX_CHUNK as u64; /// A request body streams up while its echo streams down at the same time, each [`DUPLEX_BODY`] long. /// -/// The upload never runs more than [`DUPLEX_WINDOW`] ahead of the echo the test has received, so the exchange can only finish if the client delivers the response while it is still sending the request, and the server echoes the request while it is still receiving it. A client which finished its upload before reading the echo, or a server which read the whole request before replying, stalls the exchange until the test's deadline, whatever the sockets buffer. The request declares its length and no trailers, so HTTP/1.0 can carry it too. +/// The upload never runs more than [`DUPLEX_WINDOW`] ahead of the echo the test has received, so the exchange can only finish if the client delivers the response while it is still sending the request, and the server echoes the request while it is still receiving it. A client which finished its upload before reading the echo, or a server which read the whole request before replying, stalls the exchange until the test's deadline, whatever the sockets buffer. The request declares its length, so it is sent with `Content-Length` over either version, and so is the echo, an incoming body of known length without trailers. pub async fn duplex_stream>(context: &Context<'_, S, H>) { + duplex(context, "/duplex", |_, _| {}).await; +} + +/// As [`duplex_stream`], but the server streams the echo back as a body of unknown length which may have trailers, so over HTTP/1.1 the response is chunked while the fixed-length request is still being received. HTTP/1.0 has no chunked framing, so there the echo is close-delimited. +pub async fn duplex_chunked_stream>(context: &Context<'_, S, H>) { + let chunked = context.version() == Version::Http11; + duplex(context, "/duplex-chunked", |may_have_trailers, length| { + // Only a chunked incoming body may have trailers; neither a chunked nor a close-delimited one has a length. + assert!( + may_have_trailers == chunked && length.is_none(), + "the echo's framing: may have trailers {may_have_trailers}, length {length:?}" + ); + }) + .await; +} + +// The two-way stream through `path`, passing the response body's `may_have_trailers` and length to `framing` before reading it. +async fn duplex>( + context: &Context<'_, S, H>, + path: &str, + framing: impl FnOnce(bool, Option), +) { let (upload, mut writer) = BodyReader::with_length(4 * DUPLEX_CHUNK, 4, DUPLEX_BODY); - let request = context.request(Method::POST, "/duplex", Untrailed(upload)); + let request = context.request(Method::POST, path, upload); let received = Cell::new(0); let progress = Event::new(); let produce = async { @@ -478,8 +501,9 @@ pub async fn duplex_stream>(context: &Context<'_, S .client() .call(request, &never()) .await - .expect("POST /duplex"); + .expect("POST to the two-way stream"); assert_eq!(response.status, Status::OK); + framing(response.body.may_have_trailers(), response.body.length()); let mut echoed = Vec::new(); let read = loop { match response.body.read(&never()).await { @@ -760,6 +784,7 @@ macro_rules! protocol_tests { broken_server, incorrect_length, duplex_stream, + duplex_chunked_stream, } }; } diff --git a/crates/fixtures/tests/body.rs b/crates/fixtures/tests/body.rs index ea25615..efa1b54 100644 --- a/crates/fixtures/tests/body.rs +++ b/crates/fixtures/tests/body.rs @@ -60,6 +60,15 @@ mod echo { base_http_fixtures::readable_body_tests!(run: super::echo); } +fn stream(test: impl AsyncFnOnce(Sample>>)) { + let body = BufferedBody::new([b"Hello".as_slice(), b" World"]); + block_on(test(Sample::new(Reply::Stream(body), "Hello World"))); +} + +mod stream { + base_http_fixtures::readable_body_tests!(run: super::stream); +} + fn zeroes(test: impl AsyncFnOnce(Sample>)) { let body = Reply::Zeroes(Zeroes::new(LARGE_BODY)); block_on(test(Sample::new(body, vec![0; LARGE_BODY as usize]))); @@ -104,6 +113,9 @@ fn a_reply_reports_readiness_and_trailers_by_variant() { assert!(!staged.is_ready() && staged.may_have_trailers()); let echo = Reply::Echo(EmptyBody); assert!(echo.is_ready() && echo.is_empty() && !echo.may_have_trailers()); + let stream = Reply::Stream(EmptyBody); + assert!(stream.is_ready() && stream.is_empty() && stream.may_have_trailers()); + assert_eq!(stream.length(), None); let buffered = Reply::::Buffered(BufferedBody::default()); assert!(buffered.is_ready()); } diff --git a/crates/fixtures/tests/protocol.rs b/crates/fixtures/tests/protocol.rs index a775187..1c38060 100644 --- a/crates/fixtures/tests/protocol.rs +++ b/crates/fixtures/tests/protocol.rs @@ -78,6 +78,10 @@ impl Handler for Sabotaged { let _ = request.body.read(cancellation).await; return Ok(Response::new(Status::OK, Reply::Echo(request.body))); } + // Echo with the request's length, so the response is not chunked. + (_, "/duplex-chunked") => { + return Ok(Response::new(Status::OK, Reply::Echo(request.body))); + } // Send the whole declared length. (_, "/incorrect-length") => reply("Borked!!!!"), // Drop the request trailer, or over HTTP/1.0, which drops it, claim to have received it. @@ -139,6 +143,7 @@ sabotaged! { broken_server => "the status of a failing handler", incorrect_length => "a truncated body must fail", duplex_stream => "the echoed stream", + duplex_chunked_stream => "the echo's framing", request_trailer => "the request data and trailer the handler received", response_trailer => "the response trailer", } @@ -320,3 +325,14 @@ fn the_duplex_stream_stalls_against_a_buffering_server() { .timeout(Duration::from_secs(1)) .run(async |context| protocol::duplex_stream(context).await); } + +// The two-way stream's data is its offset modulo 251, whichever offset and length a chunk starts at. +#[test] +fn the_duplex_data_continues_its_pattern() { + for (offset, length) in [(0, 0), (0, 600), (250, 1), (250, 252), (1 << 32, 1000)] { + let expected: Vec = (offset..offset + length as u64) + .map(|index| (index % 251) as u8) + .collect(); + assert_eq!(protocol::duplex_data(offset, length), expected); + } +} diff --git a/design/fixtures.md b/design/fixtures.md index 185701d..a525d49 100644 --- a/design/fixtures.md +++ b/design/fixtures.md @@ -89,7 +89,7 @@ Inspect peer construction for timeouts and the explicit blocking reset. Search t ### Requirement -`protocol_tests!` and `protocol_trailer_tests!` generate the shared protocol cases for a runner chosen by the caller, and `readable_body_tests!` generates the `ReadableBody` contract cases for a body type. Every case is also a standalone public function, so a partial implementation can call a subset. The protocol suite includes a two-way streaming case, `duplex_stream`. The client streams a request body while it reads the streamed echo at the same time, each direction `DUPLEX_BODY` long, and the upload never runs more than `DUPLEX_WINDOW`, a few chunks, ahead of the echo received. The exchange can therefore finish only if the client delivers the response while still sending the request and the server echoes the request while still receiving it, whatever the sockets buffer. The request declares its length and no trailers, so HTTP/1.0 carries it too. The protocol suite runs between the Base client and server over HTTP/1.1, over HTTP/1.0, where the trailer cases check that trailers are dropped and the data arrives intact, and over TLS with the `rustls` feature. Each of these runs in two modes: scoped, with `run`, and spawned, with `run_spawned`, where the server runs on Native workers and the test drives the client from its own thread. The body suite runs against each body type `base-http` provides. Each assertion is shown to fail against an implementation that breaks it. +`protocol_tests!` and `protocol_trailer_tests!` generate the shared protocol cases for a runner chosen by the caller, and `readable_body_tests!` generates the `ReadableBody` contract cases for a body type. Every case is also a standalone public function, so a partial implementation can call a subset. The protocol suite includes two two-way streaming cases, `duplex_stream` and `duplex_chunked_stream`. In each, the client streams a request body while it reads the streamed echo at the same time, each direction `DUPLEX_BODY` long, and the upload never runs more than `DUPLEX_WINDOW`, a few chunks, ahead of the echo received. The exchange can therefore finish only if the client delivers the response while still sending the request and the server echoes the request while still receiving it, whatever the sockets buffer. The request declares its length, so it is sent with `Content-Length` over either version, as [HTTP/1 trailers](http1-trailers.md) requires for a request of known length, even though a streamed body may have trailers. In `duplex_stream` the server echoes the incoming body, whose known length and lack of trailers give the response `Content-Length`. In `duplex_chunked_stream` it echoes the body with no length and `may_have_trailers` true, so over HTTP/1.1 the response is chunked, and the case checks that the client's response body may have trailers and has no length; over HTTP/1.0 it checks that the echo is close-delimited. The protocol suite runs between the Base client and server over HTTP/1.1, over HTTP/1.0, where the trailer cases check that trailers are dropped and the data arrives intact, and over TLS with the `rustls` feature. Each of these runs in two modes: scoped, with `run`, and spawned, with `run_spawned`, where the server runs on Native workers and the test drives the client from its own thread. The body suite runs against each body type `base-http` provides. Each assertion is shown to fail against an implementation that breaks it. ### Rationale @@ -101,4 +101,4 @@ Each new engine or body type should be checked against one shared definition of ### Validation -Check that the suites are applied as stated. `crates/fixtures/tests/protocol.rs` runs every protocol case against an application sabotaged on that case's route and requires the failure message of the assertion the case exists for, and runs the trailer cases over HTTP/1.0 against an application which reports a dropped request trailer or sends the response trailer as data; `crates/fixtures/tests/body.rs` runs every body assertion against a body with the corresponding fault and requires the expected failure. `the_duplex_stream_stalls_against_a_buffering_server` in `crates/fixtures/tests/protocol.rs` shows that the two-way case cannot pass against a server which reads the whole request before replying. `a_sequential_duplex_stream_stalls` shows that a client which uploads a 1 GiB request before reading the echo stalls, and asserts that the upload stopped short, so the window is not what makes a sequential exchange fail. +Check that the suites are applied as stated. `crates/fixtures/tests/protocol.rs` runs every protocol case against an application sabotaged on that case's route and requires the failure message of the assertion the case exists for, and runs the trailer cases over HTTP/1.0 against an application which reports a dropped request trailer or sends the response trailer as data. The sabotaged `/duplex-chunked` route echoes with the request's length, so `duplex_chunked_stream` fails on the echo's framing. `crates/fixtures/tests/body.rs` runs every body assertion against a body with the corresponding fault and requires the expected failure. `the_duplex_stream_stalls_against_a_buffering_server` in `crates/fixtures/tests/protocol.rs` shows that the two-way case cannot pass against a server which reads the whole request before replying. `a_sequential_duplex_stream_stalls` shows that a client which uploads a 1 GiB request before reading the echo stalls, and asserts that the upload stopped short, so the window is not what makes a sequential exchange fail. From 8c871d15a46c2db13d60868b48e497d59e950254 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 3 Nov 2026 16:35:35 +1300 Subject: [PATCH 5/6] Check the framing of both two-way cases, including the chunked case over HTTP/1.0. --- crates/fixtures/src/protocol.rs | 11 +++++++++-- crates/fixtures/tests/protocol.rs | 10 ++++++++++ design/fixtures.md | 4 ++-- 3 files changed, 21 insertions(+), 4 deletions(-) diff --git a/crates/fixtures/src/protocol.rs b/crates/fixtures/src/protocol.rs index 9eac380..da21500 100644 --- a/crates/fixtures/src/protocol.rs +++ b/crates/fixtures/src/protocol.rs @@ -446,14 +446,21 @@ const DUPLEX_PERIOD: [u8; 251] = { period }; -/// How far, in bytes, [`duplex_stream`]'s upload may run ahead of the echo it has received: a few chunks, far less than the stream. +/// How far, in bytes, the upload of [`duplex_stream`] and [`duplex_chunked_stream`] may run ahead of the echo it has received: a few chunks, far less than the stream. pub const DUPLEX_WINDOW: u64 = 4 * DUPLEX_CHUNK as u64; /// A request body streams up while its echo streams down at the same time, each [`DUPLEX_BODY`] long. /// /// The upload never runs more than [`DUPLEX_WINDOW`] ahead of the echo the test has received, so the exchange can only finish if the client delivers the response while it is still sending the request, and the server echoes the request while it is still receiving it. A client which finished its upload before reading the echo, or a server which read the whole request before replying, stalls the exchange until the test's deadline, whatever the sockets buffer. The request declares its length, so it is sent with `Content-Length` over either version, and so is the echo, an incoming body of known length without trailers. pub async fn duplex_stream>(context: &Context<'_, S, H>) { - duplex(context, "/duplex", |_, _| {}).await; + duplex(context, "/duplex", |may_have_trailers, length| { + // The request's known length frames it, and the echo inherits it. + assert!( + !may_have_trailers && length == Some(DUPLEX_BODY), + "the echo's framing: may have trailers {may_have_trailers}, length {length:?}" + ); + }) + .await; } /// As [`duplex_stream`], but the server streams the echo back as a body of unknown length which may have trailers, so over HTTP/1.1 the response is chunked while the fixed-length request is still being received. HTTP/1.0 has no chunked framing, so there the echo is close-delimited. diff --git a/crates/fixtures/tests/protocol.rs b/crates/fixtures/tests/protocol.rs index 1c38060..8a237eb 100644 --- a/crates/fixtures/tests/protocol.rs +++ b/crates/fixtures/tests/protocol.rs @@ -167,6 +167,16 @@ fn response_trailer_http10() { .run(async |context| protocol::response_trailer(context).await); } +// Over HTTP/1.0 the chunked two-way case expects a close-delimited echo, so an echo with the request's length fails it too. +#[test] +#[should_panic(expected = "the echo's framing")] +fn duplex_chunked_stream_http10() { + ServerContext::new() + .handler(Sabotaged(App::new())) + .version(Version::Http10) + .run(async |context| protocol::duplex_chunked_stream(context).await); +} + /// The application behind a router that sends every request to a missing route. struct Misrouted(App); diff --git a/design/fixtures.md b/design/fixtures.md index a525d49..de8f902 100644 --- a/design/fixtures.md +++ b/design/fixtures.md @@ -89,7 +89,7 @@ Inspect peer construction for timeouts and the explicit blocking reset. Search t ### Requirement -`protocol_tests!` and `protocol_trailer_tests!` generate the shared protocol cases for a runner chosen by the caller, and `readable_body_tests!` generates the `ReadableBody` contract cases for a body type. Every case is also a standalone public function, so a partial implementation can call a subset. The protocol suite includes two two-way streaming cases, `duplex_stream` and `duplex_chunked_stream`. In each, the client streams a request body while it reads the streamed echo at the same time, each direction `DUPLEX_BODY` long, and the upload never runs more than `DUPLEX_WINDOW`, a few chunks, ahead of the echo received. The exchange can therefore finish only if the client delivers the response while still sending the request and the server echoes the request while still receiving it, whatever the sockets buffer. The request declares its length, so it is sent with `Content-Length` over either version, as [HTTP/1 trailers](http1-trailers.md) requires for a request of known length, even though a streamed body may have trailers. In `duplex_stream` the server echoes the incoming body, whose known length and lack of trailers give the response `Content-Length`. In `duplex_chunked_stream` it echoes the body with no length and `may_have_trailers` true, so over HTTP/1.1 the response is chunked, and the case checks that the client's response body may have trailers and has no length; over HTTP/1.0 it checks that the echo is close-delimited. The protocol suite runs between the Base client and server over HTTP/1.1, over HTTP/1.0, where the trailer cases check that trailers are dropped and the data arrives intact, and over TLS with the `rustls` feature. Each of these runs in two modes: scoped, with `run`, and spawned, with `run_spawned`, where the server runs on Native workers and the test drives the client from its own thread. The body suite runs against each body type `base-http` provides. Each assertion is shown to fail against an implementation that breaks it. +`protocol_tests!` and `protocol_trailer_tests!` generate the shared protocol cases for a runner chosen by the caller, and `readable_body_tests!` generates the `ReadableBody` contract cases for a body type. Every case is also a standalone public function, so a partial implementation can call a subset. The protocol suite includes two two-way streaming cases, `duplex_stream` and `duplex_chunked_stream`. In each, the client streams a request body while it reads the streamed echo at the same time, each direction `DUPLEX_BODY` long, and the upload never runs more than `DUPLEX_WINDOW`, a few chunks, ahead of the echo received. The exchange can therefore finish only if the client delivers the response while still sending the request and the server echoes the request while still receiving it, whatever the sockets buffer. The request declares its length, so it is sent with `Content-Length` over either version, as [HTTP/1 trailers](http1-trailers.md) requires for a request of known length, even though a streamed body may have trailers. In `duplex_stream` the server echoes the incoming body, whose known length and lack of trailers give the response `Content-Length`, and the case checks that the client's response body has the request's length and no trailers. In `duplex_chunked_stream` it echoes the body with no length and `may_have_trailers` true, so over HTTP/1.1 the response is chunked, and the case checks that the client's response body may have trailers and has no length; over HTTP/1.0 it checks that the echo is close-delimited. The protocol suite runs between the Base client and server over HTTP/1.1, over HTTP/1.0, where the trailer cases check that trailers are dropped and the data arrives intact, and over TLS with the `rustls` feature. Each of these runs in two modes: scoped, with `run`, and spawned, with `run_spawned`, where the server runs on Native workers and the test drives the client from its own thread. The body suite runs against each body type `base-http` provides. Each assertion is shown to fail against an implementation that breaks it. ### Rationale @@ -101,4 +101,4 @@ Each new engine or body type should be checked against one shared definition of ### Validation -Check that the suites are applied as stated. `crates/fixtures/tests/protocol.rs` runs every protocol case against an application sabotaged on that case's route and requires the failure message of the assertion the case exists for, and runs the trailer cases over HTTP/1.0 against an application which reports a dropped request trailer or sends the response trailer as data. The sabotaged `/duplex-chunked` route echoes with the request's length, so `duplex_chunked_stream` fails on the echo's framing. `crates/fixtures/tests/body.rs` runs every body assertion against a body with the corresponding fault and requires the expected failure. `the_duplex_stream_stalls_against_a_buffering_server` in `crates/fixtures/tests/protocol.rs` shows that the two-way case cannot pass against a server which reads the whole request before replying. `a_sequential_duplex_stream_stalls` shows that a client which uploads a 1 GiB request before reading the echo stalls, and asserts that the upload stopped short, so the window is not what makes a sequential exchange fail. +Check that the suites are applied as stated. `crates/fixtures/tests/protocol.rs` runs every protocol case against an application sabotaged on that case's route and requires the failure message of the assertion the case exists for, and runs the trailer cases over HTTP/1.0 against an application which reports a dropped request trailer or sends the response trailer as data. The sabotaged `/duplex-chunked` route echoes with the request's length, so `duplex_chunked_stream` fails on the echo's framing over HTTP/1.1 and, in `duplex_chunked_stream_http10`, over HTTP/1.0. `crates/fixtures/tests/body.rs` runs every body assertion against a body with the corresponding fault and requires the expected failure. `the_duplex_stream_stalls_against_a_buffering_server` in `crates/fixtures/tests/protocol.rs` shows that `duplex_stream` cannot pass against a server which reads the whole request before replying; `duplex_chunked_stream` shares its windowed exchange. `a_sequential_duplex_stream_stalls` shows that a client which uploads a 1 GiB request before reading the echo stalls, and asserts that the upload stopped short, so the window is not what makes a sequential exchange fail. From 2443ae3623f7a679735a7b19591c1762e4e23fe7 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Tue, 3 Nov 2026 16:57:11 +1300 Subject: [PATCH 6/6] Require the sequential upload to stop, check `duplex_stream`'s framing against a reframed echo, and mark the own-thread join deadline as temporary. --- crates/fixtures/src/context.rs | 4 +- crates/fixtures/src/context/tests.rs | 2 +- crates/fixtures/tests/protocol.rs | 67 +++++++++++++++++++++++----- crates/fixtures/tests/raw.rs | 2 - design/fixtures.md | 6 +-- 5 files changed, 64 insertions(+), 17 deletions(-) diff --git a/crates/fixtures/src/context.rs b/crates/fixtures/src/context.rs index 7095189..0c0c4ff 100644 --- a/crates/fixtures/src/context.rs +++ b/crates/fixtures/src/context.rs @@ -289,7 +289,7 @@ impl ServerContext { /// /// Unlike [`ServerContext::run`], the server makes progress on the scheduler's workers rather than on the test's thread, so it runs in parallel with the client. That exposes cross-thread wake-ups and races which a single `block_on` hides, at the cost of reproducible interleavings. The fixture relies only on the server running on the scheduler's workers; how the server spreads its connections across them is the server's choice and may change. The context owns the scheduler and the application, so the application must be `Send + Sync + 'static`, as a server spawning its own tasks needs. /// - /// This has the guarantees of [`ServerContext::run`]: the listener is bound before the test starts, the test is bounded, the client retires its connections before the server stops, the server is cancelled and must stop within the teardown timeout, a server error fails the test, a server panic is resumed with its own payload, and a test failure is reported first, with any server failure printed beforehand. The teardown timeout is measured on a thread of its own, so a server occupying every worker still fails on time. + /// This has the guarantees of [`ServerContext::run`]: the listener is bound before the test starts, the test is bounded, the client retires its connections before the server stops, the server is cancelled and must stop within the teardown timeout, a server error fails the test, a server panic is resumed with its own payload, and a test failure is reported first, with any server failure printed beforehand. The server's teardown timeout is measured on a thread of its own, so a server occupying every worker still fails its join on time. This is a temporary workaround for the Native scheduler stranding timers and I/O behind a busy worker, and is to be removed once that is fixed; see the design proposal in [base-rs/base-scheduler#93](https://github.com/base-rs/base-scheduler/pull/93). The test's timeout and the client's teardown are still measured by the scheduler's clock, so they are bounded only while the scheduler's timers can run. /// /// Shutting the scheduler down afterwards waits for every task's current poll to return. A poll which never returns, such as a handler blocked forever, cannot be preempted, so it hangs the test after the failure is reported. pub fn run_spawned( @@ -488,6 +488,8 @@ impl Spawned { } // Expire after `timeout`, measured on a thread of its own rather than by the scheduler's clock, so a server occupying every worker, such as a handler which blocks, still fails its join. Shutting the scheduler down afterwards still waits for that poll to return. Dropping the future stops the thread. +// +// Temporary: Native strands timers and I/O registered from outside a worker behind a busy worker, so the scheduler's clock cannot bound the join. Remove this once that is fixed; see https://github.com/base-rs/base-scheduler/pull/93. async fn expiry(timeout: Duration) { let expired = Latch::new(); let (_cancel, cancelled) = mpsc::channel::<()>(); diff --git a/crates/fixtures/src/context/tests.rs b/crates/fixtures/src/context/tests.rs index 0e35693..5924730 100644 --- a/crates/fixtures/src/context/tests.rs +++ b/crates/fixtures/src/context/tests.rs @@ -27,7 +27,7 @@ impl Handler for Blocking { } } -// With one worker, which the handler blocks, nothing on the scheduler can run, including its clock's timers, yet the join still fails on time. +// With one worker, which the handler blocks, Native strands the scheduler clock's timers behind that worker, so only the join's own-thread deadline can fail it on time. This guards the temporary workaround until base-rs/base-scheduler#93 is fixed. #[test] fn a_server_occupying_the_only_worker_fails_its_join() { let (started, release) = (Latch::new(), Latch::new()); diff --git a/crates/fixtures/tests/protocol.rs b/crates/fixtures/tests/protocol.rs index 8a237eb..8d769ce 100644 --- a/crates/fixtures/tests/protocol.rs +++ b/crates/fixtures/tests/protocol.rs @@ -278,28 +278,75 @@ fn a_sequential_duplex_stream_stalls() { .await; body::payload(&mut response.unwrap().body).await; }; - let stalled = or( + // Sample the upload twice once its buffers have had time to fill: a stalled upload makes no progress between the samples, whatever the host's buffer sizes. + let samples = or( async { sequential.await; - false + None }, async { - let _ = context - .scheduler() - .sleep(Duration::from_millis(500), &Cancellation::never()) - .await; - true + let never = Cancellation::never(); + let pause = || { + context + .scheduler() + .sleep(Duration::from_millis(250), &never) + }; + let _ = pause().await; + let first = written.get(); + let _ = pause().await; + Some((first, written.get())) }, ) .await; assert!( - stalled && written.get() < SEQUENTIAL_BODY, - "the upload did not stall: {} bytes written", - written.get() + matches!(samples, Some((first, second)) if first == second), + "the upload did not stall: {samples:?} bytes written" ); }); } +/// Answers the two-way stream with an echo of unknown length which may have trailers, so `duplex_stream`'s echo is not framed by the request's length. +struct Reframed(App); + +impl Borrow for Reframed { + fn borrow(&self) -> &App { + &self.0 + } +} + +impl Handler for Reframed { + type Error = io::Error; + type Body = Reply; + + async fn call( + &self, + request: Request, + _: &mut InterimResponses, + _: &Cancellation, + ) -> Result>, io::Error> { + Ok(Response::new(Status::OK, Reply::Stream(request.body))) + } +} + +// Over HTTP/1.1 the reframed echo is chunked, so it may have trailers. +#[test] +#[should_panic(expected = "the echo's framing")] +fn duplex_stream_reframed() { + ServerContext::new() + .handler(Reframed(App::new())) + .run(async |context| protocol::duplex_stream(context).await); +} + +// Over HTTP/1.0 the reframed echo is close-delimited, so it has no length. +#[test] +#[should_panic(expected = "the echo's framing")] +fn duplex_stream_reframed_http10() { + ServerContext::new() + .handler(Reframed(App::new())) + .version(Version::Http10) + .run(async |context| protocol::duplex_stream(context).await); +} + /// Reads the whole two-way stream's request before echoing it, as a server which buffers instead of streaming does. struct Buffering(App); diff --git a/crates/fixtures/tests/raw.rs b/crates/fixtures/tests/raw.rs index e0f6887..00b1aa5 100644 --- a/crates/fixtures/tests/raw.rs +++ b/crates/fixtures/tests/raw.rs @@ -220,8 +220,6 @@ fn a_split_raw_peer_uploads_and_downloads_at_once() { "the echoed stream" ); // The server closes the connection after the exchange. - assert!(reader.fill_buf().unwrap().is_empty()); - reader.consume(0); assert_eq!(reader.read(&mut [0; 1]).unwrap(), 0); uploading.join().unwrap(); }) diff --git a/design/fixtures.md b/design/fixtures.md index de8f902..1f5536b 100644 --- a/design/fixtures.md +++ b/design/fixtures.md @@ -44,7 +44,7 @@ Rust has no asynchronous destructor, so a fixture which leaves teardown to the t `ServerContext::run_spawned` creates a two-worker Native scheduler, binds the listener, then spawns the server's public entry point, `base_http_server::serve` or `serve_tls`, as one task through `base_scheduler::Spawn`, so the server runs on the scheduler's workers in parallel with a client driven from the test's thread. It takes ownership of the application and owns the scheduler for the whole run; the test receives the same `Context<'_, NativeScheduler, H>` as `run`, with the application as `&H`. Any sharing between the test and the server's task, such as an `Arc`, stays inside the fixture and is not part of its public API. The fixture relies only on the server running on the scheduler's workers. It adds nothing to `base-http-server`'s public API and does not choose how connections are spread across tasks or workers. The application must be `Handler + Send + Sync + 'static`, as a server which spawned its own connection tasks would need. -`run_spawned` has the guarantees of `run`: the listener is bound before the test starts, the test is bounded, the client retires its connections before the server stops, a server error fails the test, a server panic is resumed with its original payload, and a test failure is reported before any server failure, which is printed to stderr first. After the test, the server is cancelled and joined for at most the teardown timeout, and fails the test if it overruns. That timeout is measured on its own thread rather than by the scheduler's clock, so a server occupying every worker, such as a handler which blocks, still fails on time. The scheduler is then shut down, and Native drains its tasks, so the call does not return until an overrunning server's current poll returns. A poll which never returns cannot be preempted and hangs the test after the failure is reported, so teardown is bounded only for servers whose polls return. The scoped `run` and `run_on` are unchanged. +`run_spawned` has the guarantees of `run`: the listener is bound before the test starts, the test is bounded, the client retires its connections before the server stops, a server error fails the test, a server panic is resumed with its original payload, and a test failure is reported before any server failure, which is printed to stderr first. After the test, the server is cancelled and joined for at most the teardown timeout, and fails the test if it overruns. That timeout is measured on its own thread rather than by the scheduler's clock, so a server occupying every worker, such as a handler which blocks, still fails on time. The own-thread deadline is a temporary workaround for the Native scheduler stranding timers and I/O behind a busy worker, and is to be removed once that is fixed; see the design proposal in [base-rs/base-scheduler#93](https://github.com/base-rs/base-scheduler/pull/93). The test's timeout and the client's teardown, in `supervise` and `bounded`, are still measured by the scheduler's clock, so they are bounded only while the scheduler's timers can run. The scheduler is then shut down, and Native drains its tasks, so the call does not return until an overrunning server's current poll returns. A poll which never returns cannot be preempted and hangs the test after the failure is reported, so teardown is bounded only for servers whose polls return. The scoped `run` and `run_on` are unchanged. ### Rationale @@ -65,7 +65,7 @@ Inspect the private spawn: its task calls only `serve` or `serve_tls`, and the c - a server panic, resumed with its payload; - a test failure reported before a server failure or panic. -The unit test in `crates/fixtures/src/context/tests.rs` runs a server on a one-worker scheduler whose handler blocks that worker, so the scheduler's clock cannot run, and shows that the join still fails within its timeout. +The unit test in `crates/fixtures/src/context/tests.rs` runs a server on a one-worker scheduler whose handler blocks that worker, so Native strands the scheduler clock's timers behind it, and shows that the join still fails within its timeout. ## Independent, bounded raw peers @@ -101,4 +101,4 @@ Each new engine or body type should be checked against one shared definition of ### Validation -Check that the suites are applied as stated. `crates/fixtures/tests/protocol.rs` runs every protocol case against an application sabotaged on that case's route and requires the failure message of the assertion the case exists for, and runs the trailer cases over HTTP/1.0 against an application which reports a dropped request trailer or sends the response trailer as data. The sabotaged `/duplex-chunked` route echoes with the request's length, so `duplex_chunked_stream` fails on the echo's framing over HTTP/1.1 and, in `duplex_chunked_stream_http10`, over HTTP/1.0. `crates/fixtures/tests/body.rs` runs every body assertion against a body with the corresponding fault and requires the expected failure. `the_duplex_stream_stalls_against_a_buffering_server` in `crates/fixtures/tests/protocol.rs` shows that `duplex_stream` cannot pass against a server which reads the whole request before replying; `duplex_chunked_stream` shares its windowed exchange. `a_sequential_duplex_stream_stalls` shows that a client which uploads a 1 GiB request before reading the echo stalls, and asserts that the upload stopped short, so the window is not what makes a sequential exchange fail. +Check that the suites are applied as stated. `crates/fixtures/tests/protocol.rs` runs every protocol case against an application sabotaged on that case's route and requires the failure message of the assertion the case exists for, and runs the trailer cases over HTTP/1.0 against an application which reports a dropped request trailer or sends the response trailer as data. The sabotaged `/duplex-chunked` route echoes with the request's length, so `duplex_chunked_stream` fails on the echo's framing over HTTP/1.1 and, in `duplex_chunked_stream_http10`, over HTTP/1.0. `duplex_stream_reframed` and `duplex_stream_reframed_http10` answer `/duplex` with an echo of unknown length which may have trailers, and `duplex_stream` fails on the echo's framing over each version. `crates/fixtures/tests/body.rs` runs every body assertion against a body with the corresponding fault and requires the expected failure. `the_duplex_stream_stalls_against_a_buffering_server` in `crates/fixtures/tests/protocol.rs` shows that `duplex_stream` cannot pass against a server which reads the whole request before replying; `duplex_chunked_stream` shares its windowed exchange. `a_sequential_duplex_stream_stalls` shows that a client which uploads a 1 GiB request before reading the echo stalls: it samples the bytes written 250 ms and 500 ms after the call starts and requires that the upload made no progress between them, so the window is not what makes a sequential exchange fail, and a client which keeps uploading does not pass.