diff --git a/crates/client/tests/protocol.rs b/crates/client/tests/protocol.rs index 820d9ea..169d1a6 100644 --- a/crates/client/tests/protocol.rs +++ b/crates/client/tests/protocol.rs @@ -39,3 +39,33 @@ 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, App>)) { + 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, App>)) { + serve(app, Version::Http10).run_spawned(test); +} + +mod http10_spawned { + base_http_fixtures::protocol_tests!(run: super::http10_spawned); + base_http_fixtures::protocol_trailer_tests!(run: super::http10_spawned); +} + +#[cfg(feature = "rustls")] +fn https_spawned(app: App, test: impl AsyncFnOnce(&Context<'_, NativeScheduler, App>)) { + 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..9742574 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 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. [`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..0c0c4ff 100644 --- a/crates/fixtures/src/context.rs +++ b/crates/fixtures/src/context.rs @@ -8,16 +8,21 @@ use base_http::{ }; use base_http_client::{Client, Driver, Endpoint}; use base_http_server::Handler; -use base_scheduler::{Cancellation, Clock, SocketType}; -use base_scheduler_native::NativeScheduler; -use futures_lite::future::{block_on, zip}; +use base_scheduler::{Cancellation, Clock, SocketType, Spawn, TaskError}; +use base_scheduler_native::{NativeOptions, NativeScheduler}; +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, @@ -227,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, @@ -241,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, @@ -278,6 +284,238 @@ impl ServerContext { } } +impl ServerContext { + /// 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 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 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( + 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, + { + 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"); + Spawned { + scheduler: scheduler.clone(), + task: Some(task), + stop, + handler, + endpoint, + address, + version, + client, + limits, + } + } +} + +// 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 joined. + 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 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) = + 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, + ) + }) + } + + // 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 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::<()>(); + let release = expired.clone(); + thread::spawn(move || { + if let Err(RecvTimeoutError::Timeout) = cancelled.recv_timeout(timeout) { + release.release(); + } + }); + let _ = expired.wait(&Cancellation::never()).await; +} + +// 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 { let never = Cancellation::never(); let address: SocketAddr = ([127, 0, 0, 1], 0).into(); @@ -296,10 +534,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, @@ -351,7 +590,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, @@ -551,3 +790,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..5924730 --- /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, 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()); + 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 5a548a9..7730ca3 100644 --- a/crates/fixtures/src/lib.rs +++ b/crates/fixtures/src/lib.rs @@ -17,7 +17,7 @@ pub mod socket; pub use context::SERVER_NAME; pub use context::{ClientContext, Context, Hello, ServerContext, TEARDOWN_TIMEOUT, TEST_TIMEOUT}; pub use latch::{LATCH_TIMEOUT, Latch}; -pub use raw::{RawListener, RawPeer, RawServer}; +pub use raw::{RawListener, RawPeer, RawReader, RawServer, RawWriter}; pub use scheduler::{Scheduler, native}; /// The messages of `error` and each of its sources, outermost first. diff --git a/crates/fixtures/src/protocol.rs b/crates/fixtures/src/protocol.rs index 1b979a7..da21500 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,6 +27,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`], 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; const LINK: &str = "; rel=preload; as=style"; /// The application serving every protocol case. @@ -33,6 +39,8 @@ 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`] | @@ -98,6 +106,12 @@ 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))); + } + (_, "/duplex-chunked") => { + return Ok(Response::new(Status::OK, Reply::Stream(request.body))); + } (_, "/length") => { let length = request .body @@ -242,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. @@ -272,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, } @@ -281,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, } @@ -291,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), } @@ -300,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, } @@ -308,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, } @@ -317,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)? @@ -404,6 +422,122 @@ pub async fn post>(context: &Context<'_, S, H>) { ); } +/// 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 { + // 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, 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", |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. +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, path, 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 + .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 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 { + 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, {read:?}", + echoed.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()); @@ -613,7 +747,7 @@ 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; @@ -630,6 +764,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, App>)) { +/// 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] @@ -644,6 +790,8 @@ macro_rules! protocol_tests { large_body, broken_server, incorrect_length, + duplex_stream, + duplex_chunked_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/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 28b5ea5..8d769ce 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,7 +12,9 @@ use base_http_fixtures::{ }; use base_http_server::Handler; use base_scheduler::Cancellation; -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); @@ -71,6 +73,15 @@ 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))); + } + // 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. @@ -131,6 +142,8 @@ 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", + duplex_chunked_stream => "the echo's framing", request_trailer => "the request data and trailer the handler received", response_trailer => "the response trailer", } @@ -154,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); @@ -220,3 +243,153 @@ fn a_truncated_trailer_upload_fails_the_handler() { .await; }); } + +// 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] +fn a_sequential_duplex_stream_stalls() { + ServerContext::new() + .handler(App::new()) + .run(async |context| { + 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 { + while written.get() < SEQUENTIAL_BODY { + writer + .write( + protocol::duplex_data(written.get(), protocol::DUPLEX_CHUNK), + &Cancellation::never(), + ) + .await + .unwrap(); + written.set(written.get() + protocol::DUPLEX_CHUNK as u64); + } + writer.close_write(None); + }; + let sequential = async { + let (response, ()) = zip( + context.client().call(request, &Cancellation::never()), + upload_all, + ) + .await; + body::payload(&mut response.unwrap().body).await; + }; + // 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; + None + }, + async { + 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!( + 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); + +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); +} + +// 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/crates/fixtures/tests/raw.rs b/crates/fixtures/tests/raw.rs index 6e03b03..00b1aa5 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,105 @@ 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. 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() + .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_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 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| { + context + .raw(|mut peer| { + peer.get_mut() + .set_write_timeout(Some(Duration::from_millis(500))) + .unwrap(); + 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(), + 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..432b7f0 --- /dev/null +++ b/crates/fixtures/tests/spawned.rs @@ -0,0 +1,123 @@ +// 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, 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::{Context, Hello, ServerContext, body::payload, protocol::App}; +use base_http_server::Handler; +use base_scheduler::Cancellation; +use base_scheduler_native::NativeScheduler; +use std::{convert::Infallible, io::Read}; + +fn failing() -> ServerContext { + ServerContext::new().server_options(base_http_server::Options { + max_connections: 0, + ..Default::default() + }) +} + +#[test] +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); + }); +} + +#[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") + } +} + +// 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]); + }) + .await; +} + +#[test] +#[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. +#[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| { + panic_the_server(context).await; + panic!("the test failed first"); + }); +} diff --git a/design/fixtures.md b/design/fixtures.md index 3ccb64f..1f5536b 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::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 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 + +`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::run_spawned` and its private spawning, joining and scheduler helpers in `crates/fixtures/src/context.rs`. + +### Validation + +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: + +- a spawned server serving a client and a raw peer; +- an application shared with the server; +- a test panic; +- a server failure; +- 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 Native strands the scheduler clock's timers behind it, 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. +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 @@ -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 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 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. 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 @@ -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, 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. +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.