From 6e9b0073e5577447496895a9d287f692a3dff0fe Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Tue, 27 Oct 2026 16:46:29 -0400 Subject: [PATCH 1/4] wasi-http: Clean up `run_http` test helper The `run_http` test helper currently takes a channel and sends the uncollected http request down that channel for the caller to see. But only a few tests actually use that and it can otherwise be difficult because the request body many never actually be read and a guest program can be left hanging if it awaits that. This change hopefully makes `run_http` more intuitive by not taking the channel and in general reading the full request body. The new version `run_http_get_forwarded_request` can be used if the caller needs to see how the request was passed along and the new endpoint `p3-test-immediate-response` can be used when the test needs a response to be sent without the body being read. --- .../src/bin/p3_http_api_bench.rs | 4 +- crates/wasi-http/tests/all/p3/mod.rs | 134 ++++++++++-------- 2 files changed, 77 insertions(+), 61 deletions(-) diff --git a/crates/test-programs/src/bin/p3_http_api_bench.rs b/crates/test-programs/src/bin/p3_http_api_bench.rs index 382a81747cc6..05bbe7d694aa 100644 --- a/crates/test-programs/src/bin/p3_http_api_bench.rs +++ b/crates/test-programs/src/bin/p3_http_api_bench.rs @@ -143,7 +143,9 @@ fn new_outgoing_requests(n: usize) -> (Vec, Vec) { let (request, writers) = new_request(); request.set_path_with_query(Some("/")).unwrap(); request.set_scheme(Some(&Scheme::Http)).unwrap(); - request.set_authority(Some("p3-test")).unwrap(); + request + .set_authority(Some("p3-test-immediate-response")) + .unwrap(); (request, writers) }) .unzip() diff --git a/crates/wasi-http/tests/all/p3/mod.rs b/crates/wasi-http/tests/all/p3/mod.rs index 9f436f5c4f8a..d5acac0b5bc2 100644 --- a/crates/wasi-http/tests/all/p3/mod.rs +++ b/crates/wasi-http/tests/all/p3/mod.rs @@ -32,19 +32,33 @@ foreach_p3_http!(assert_test_exists); #[derive(Default)] struct TestHooks { - request_tx: Option>>, + request_tx: Option>>>, io_dropped_tx: Option>, } impl TestHooks { - fn new(request_tx: oneshot::Sender>) -> Self { + fn new(request_tx: Option>>>) -> Self { Self { - request_tx: Some(request_tx), + request_tx, ..Self::default() } } } +/// Collects the body of `request`, and then sends it over the channel if +/// present. +async fn collect_request( + request: http::Request, + request_tx: Option>>>, +) -> Result<(), Error> { + let (parts, body) = request.into_parts(); + let body = body.collect().await?; + if let Some(request_tx) = request_tx { + _ = request_tx.send(http::Request::from_parts(parts, body)); + } + Ok(()) +} + impl WasiHttpHooks for TestHooks { fn forbidden_header_filter(&mut self) -> ForbiddenHeaderFilter { Arc::new(|name| { @@ -70,7 +84,22 @@ impl WasiHttpHooks for TestHooks { > { _ = fut; if let Some("p3-test") = request.uri().authority().map(|v| v.as_str()) { - _ = self.request_tx.take().unwrap().send(request); + // The default test case reads the entire request before responding. + let request_tx = self.request_tx.take(); + Box::new(async move { + collect_request(request, request_tx).await?; + Ok(( + http::Response::new(Default::default()), + Box::new(async { Ok(()) }) as Box + Send>, + )) + }) + } else if let Some("p3-test-immediate-response") = + request.uri().authority().map(|v| v.as_str()) + { + // Respond immediately, collecting the request body in the + // background after the response is returned. + let request_tx = self.request_tx.take(); + tokio::spawn(collect_request(request, request_tx)); Box::new(async { Ok(( http::Response::new(Default::default()), @@ -95,12 +124,13 @@ impl WasiHttpHooks for TestHooks { )) }) } else if let Some("p3-test-trailers") = request.uri().authority().map(|v| v.as_str()) { - _ = self.request_tx.take().unwrap().send(request); + let request_tx = self.request_tx.take(); let body = http_body_util::StreamBody::new(futures::stream::iter([ Ok(http_body::Frame::data(Bytes::from_static(b"response"))), Ok(http_body::Frame::trailers(forbidden_trailers())), ])); - Box::new(async { + Box::new(async move { + collect_request(request, request_tx).await?; Ok(( http::Response::new(body.boxed_unsync()), Box::new(async { Ok(()) }) as Box + Send>, @@ -128,8 +158,8 @@ struct Ctx { } impl Ctx { - fn new(request_tx: oneshot::Sender>) -> Self { - Self::with_hooks(TestHooks::new(request_tx)) + fn new() -> Self { + Self::with_hooks(TestHooks::new(None)) } fn with_hooks(hooks: TestHooks) -> Self { @@ -174,7 +204,7 @@ async fn run_cli(path: &str, server: &Server) -> wasmtime::Result<()> { .env("HTTP_SERVER", server.addr()) .inherit_stdio() .build(), - ..Ctx::new(oneshot::channel().0) + ..Ctx::new() }, ); let mut linker = Linker::new(&engine); @@ -209,7 +239,7 @@ async fn p3_http_api_bench() -> Result<()> { .context("failed to link `wasi:http@0.3.x`")?; for api in ["request.new", "response.consume-body", "client.send"] { - // `TestHooks` only accepts a single request to `p3-test`, so use a + // `TestHooks` only accepts a single request to `p3-test*`, so use a // fresh store and a single iteration for each API. let mut store = Store::new( &engine, @@ -218,7 +248,7 @@ async fn p3_http_api_bench() -> Result<()> { .args(&["p3_http_api_bench", api, "1"]) .inherit_stdio() .build(), - ..Ctx::new(oneshot::channel().0) + ..Ctx::new() }, ); let command = Command::instantiate_async(&mut store, &component, &linker).await?; @@ -235,9 +265,18 @@ async fn p3_http_api_bench() -> Result<()> { async fn run_http + 'static>( component_filename: &str, req: http::Request + Send + Sync + 'static>, - request_tx: oneshot::Sender>, ) -> wasmtime::Result>, Option>> { - run_http_with_hooks(component_filename, req, TestHooks::new(request_tx)).await + run_http_with_hooks(component_filename, req, TestHooks::new(None)).await +} + +/// Like [`run_http`], but sends the request forwarded by the guest to one of +/// the `p3-test*` endpoints on `request_tx`. +async fn run_http_get_forwarded_request + 'static>( + component_filename: &str, + req: http::Request + Send + Sync + 'static>, + request_tx: oneshot::Sender>>, +) -> wasmtime::Result>, Option>> { + run_http_with_hooks(component_filename, req, TestHooks::new(Some(request_tx))).await } async fn run_http_with_hooks + 'static>( @@ -405,12 +444,7 @@ async fn wasi_http_proxy_tests() -> wasmtime::Result<()> { .uri("http://example.com:8080/test-path") .method(http::Method::GET); - let res = run_http( - P3_API_PROXY_COMPONENT, - req.body(Empty::new())?, - oneshot::channel().0, - ) - .await?; + let res = run_http(P3_API_PROXY_COMPONENT, req.body(Empty::new())?).await?; match res { Ok(res) => println!("response: {res:?}"), @@ -439,7 +473,7 @@ async fn p3_http_echo_host_to_host_content_length() -> Result<()> { config.wasm_component_model_async(true); }); let component = Component::from_file(&engine, P3_HTTP_ECHO_COMPONENT)?; - let mut store = Store::new(&engine, Ctx::new(oneshot::channel().0)); + let mut store = Store::new(&engine, Ctx::new()); let mut linker = Linker::new(&engine); wasmtime_wasi::p2::add_to_linker_async(&mut linker)?; wasmtime_wasi::p3::add_to_linker(&mut linker)?; @@ -575,7 +609,6 @@ async fn p3_http_middleware_error() -> Result<()> { let result = run_http( path.to_str().unwrap(), request.body(http_body_util::StreamBody::new(body_rx))?, - oneshot::channel().0, ) .await; result @@ -718,7 +751,6 @@ async fn test_http_echo(component: &str, use_compression: bool, host_to_host: bo run_http( component, request.body(http_body_util::StreamBody::new(body_rx))?, - oneshot::channel().0 ), async { body_tx @@ -790,19 +822,20 @@ async fn p3_http_proxy() -> Result<()> { let (mut body_tx, body_rx) = futures::channel::mpsc::channel::>(1); - // Tell the guest to forward the request to `http://p3-test/`, which we - // handle specially in `TestHttpCtx::send_request` above, sending the - // request body to the oneshot sender we specify below and then immediately - // returning a dummy response. We won't start sending the request body - // until after the guest has exited and we've dropped the store. + // Tell the guest to forward the request to + // `http://p3-test-immediate-response/`, which we handle specially in + // `TestHooks::send_request` above by immediately returning a dummy response + // and then collecting the request in the background before sending it to + // the oneshot sender we specify below. We won't start sending the request + // body until after the guest has exited and we've dropped the store. let request = http::Request::builder() .uri("http://localhost/") .method(http::Method::GET) - .header("url", "http://p3-test/"); + .header("url", "http://p3-test-immediate-response/"); let (request_body_tx, request_body_rx) = oneshot::channel(); - let response = run_http( + let response = run_http_get_forwarded_request( P3_HTTP_PROXY_COMPONENT, request.body(http_body_util::StreamBody::new(body_rx))?, request_body_tx, @@ -812,7 +845,8 @@ async fn p3_http_proxy() -> Result<()> { assert!(response.status().as_u16() == 200); // The guest has exited and the store has been dropped; now we finally send - // the request body and assert that we've received the entire thing. + // the request body and assert that it was all successfully passed to the + // server on the other end. let ((), request_body) = futures::join!( async { @@ -822,16 +856,7 @@ async fn p3_http_proxy() -> Result<()> { .unwrap(); drop(body_tx); }, - async { - request_body_rx - .await - .unwrap() - .into_body() - .collect() - .await - .unwrap() - .to_bytes() - } + async { request_body_rx.await.unwrap().into_body().to_bytes() } ); assert_eq!(request_body, body.as_slice()); @@ -925,15 +950,16 @@ async fn p3_http_forwarded_trailers_filtered() -> Result<()> { ])); let (request_tx, request_rx) = oneshot::channel(); - let response = run_http(P3_HTTP_PROXY_COMPONENT, request.body(body)?, request_tx) - .await? - .unwrap(); + let response = + run_http_get_forwarded_request(P3_HTTP_PROXY_COMPONENT, request.body(body)?, request_tx) + .await? + .unwrap(); assert_eq!(response.status().as_u16(), 200); let response = response.into_body(); assert_trailers_filtered(response.trailers(), "response to client"); assert_eq!(response.to_bytes(), b"response".as_slice()); - let upstream = request_rx.await.unwrap().into_body().collect().await?; + let upstream = request_rx.await.unwrap().into_body(); assert_trailers_filtered(upstream.trailers(), "request to upstream"); assert_eq!(upstream.to_bytes(), b"request".as_slice()); Ok(()) @@ -950,7 +976,7 @@ async fn p3_http_forbidden_headers() -> Result<()> { .header("foo", "bar"); let (request_tx, request_rx) = oneshot::channel(); - let response = run_http( + let response = run_http_get_forwarded_request( P3_HTTP_FORBIDDEN_HEADERS_COMPONENT, request.body(Empty::new())?, request_tx, @@ -1059,11 +1085,7 @@ async fn p3_http_empty_frame_at_end_of_stream() -> Result<()> { // Use the echo component which actually reads from the stream let response = futures::join!( - run_http( - P3_HTTP_ECHO_COMPONENT, - request.body(wrapped_body)?, - oneshot::channel().0 - ), + run_http(P3_HTTP_ECHO_COMPONENT, request.body(wrapped_body)?), async { body_tx .send(Ok(http_body::Frame::data(raw_body))) @@ -1111,11 +1133,7 @@ async fn p3_http_data_frame_at_end_of_stream() -> Result<()> { // Use the echo component which actually reads from the stream let response = futures::join!( - run_http( - P3_HTTP_ECHO_COMPONENT, - request.body(wrapped_body)?, - oneshot::channel().0 - ), + run_http(P3_HTTP_ECHO_COMPONENT, request.body(wrapped_body)?), async { body_tx .send(Ok(http_body::Frame::data(raw_body))) @@ -1199,11 +1217,7 @@ async fn p3_http_empty_frames_interleaved() -> Result<()> { .method(http::Method::GET); let response = futures::join!( - run_http( - P3_HTTP_ECHO_COMPONENT, - request.body(wrapped_body)?, - oneshot::channel().0 - ), + run_http(P3_HTTP_ECHO_COMPONENT, request.body(wrapped_body)?), async { body_tx .send(Ok(http_body::Frame::data(Bytes::from_static(b"hello ")))) From 48aeac7c4fc66c0db19a811af3067126d71577a7 Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Thu, 29 Oct 2026 15:12:50 -0400 Subject: [PATCH 2/4] code review --- crates/wasi-http/tests/all/p3/mod.rs | 63 ++++++++++++++++++---------- 1 file changed, 41 insertions(+), 22 deletions(-) diff --git a/crates/wasi-http/tests/all/p3/mod.rs b/crates/wasi-http/tests/all/p3/mod.rs index d5acac0b5bc2..8fd45515de23 100644 --- a/crates/wasi-http/tests/all/p3/mod.rs +++ b/crates/wasi-http/tests/all/p3/mod.rs @@ -30,14 +30,19 @@ use wasmtime_wasi_http::{ foreach_p3_http!(assert_test_exists); +/// Channel over which a request forwarded to one of the `p3-test*` endpoints is +/// sent once its body has been collected, or the error encountered while +/// collecting it. +type ForwardedRequestTx = oneshot::Sender>, Error>>; + #[derive(Default)] struct TestHooks { - request_tx: Option>>>, + request_tx: Option, io_dropped_tx: Option>, } impl TestHooks { - fn new(request_tx: Option>>>) -> Self { + fn new(request_tx: Option) -> Self { Self { request_tx, ..Self::default() @@ -45,16 +50,24 @@ impl TestHooks { } } -/// Collects the body of `request`, and then sends it over the channel if -/// present. +/// Collects the body of `request`, and then sends it, or the error encountered +/// while collecting it, over the channel if present. async fn collect_request( request: http::Request, - request_tx: Option>>>, + request_tx: Option, ) -> Result<(), Error> { let (parts, body) = request.into_parts(); - let body = body.collect().await?; - if let Some(request_tx) = request_tx { - _ = request_tx.send(http::Request::from_parts(parts, body)); + let result = body.collect().await; + let Some(request_tx) = request_tx else { + return result.map(|_body| ()); + }; + match result { + Ok(body) => { + _ = request_tx.send(Ok(http::Request::from_parts(parts, body))); + } + Err(err) => { + _ = request_tx.send(Err(err)); + } } Ok(()) } @@ -270,11 +283,12 @@ async fn run_http + 'static>( } /// Like [`run_http`], but sends the request forwarded by the guest to one of -/// the `p3-test*` endpoints on `request_tx`. -async fn run_http_get_forwarded_request + 'static>( +/// the `p3-test*` endpoints on `request_tx`, or the error encountered while +/// collecting its body. +async fn run_http_forwarded + 'static>( component_filename: &str, req: http::Request + Send + Sync + 'static>, - request_tx: oneshot::Sender>>, + request_tx: ForwardedRequestTx, ) -> wasmtime::Result>, Option>> { run_http_with_hooks(component_filename, req, TestHooks::new(Some(request_tx))).await } @@ -835,7 +849,7 @@ async fn p3_http_proxy() -> Result<()> { .header("url", "http://p3-test-immediate-response/"); let (request_body_tx, request_body_rx) = oneshot::channel(); - let response = run_http_get_forwarded_request( + let response = run_http_forwarded( P3_HTTP_PROXY_COMPONENT, request.body(http_body_util::StreamBody::new(body_rx))?, request_body_tx, @@ -856,7 +870,14 @@ async fn p3_http_proxy() -> Result<()> { .unwrap(); drop(body_tx); }, - async { request_body_rx.await.unwrap().into_body().to_bytes() } + async { + request_body_rx + .await + .unwrap() + .unwrap() + .into_body() + .to_bytes() + } ); assert_eq!(request_body, body.as_slice()); @@ -950,16 +971,15 @@ async fn p3_http_forwarded_trailers_filtered() -> Result<()> { ])); let (request_tx, request_rx) = oneshot::channel(); - let response = - run_http_get_forwarded_request(P3_HTTP_PROXY_COMPONENT, request.body(body)?, request_tx) - .await? - .unwrap(); + let response = run_http_forwarded(P3_HTTP_PROXY_COMPONENT, request.body(body)?, request_tx) + .await? + .unwrap(); assert_eq!(response.status().as_u16(), 200); let response = response.into_body(); assert_trailers_filtered(response.trailers(), "response to client"); assert_eq!(response.to_bytes(), b"response".as_slice()); - let upstream = request_rx.await.unwrap().into_body(); + let upstream = request_rx.await.unwrap().unwrap().into_body(); assert_trailers_filtered(upstream.trailers(), "request to upstream"); assert_eq!(upstream.to_bytes(), b"request".as_slice()); Ok(()) @@ -976,7 +996,7 @@ async fn p3_http_forbidden_headers() -> Result<()> { .header("foo", "bar"); let (request_tx, request_rx) = oneshot::channel(); - let response = run_http_get_forwarded_request( + let response = run_http_forwarded( P3_HTTP_FORBIDDEN_HEADERS_COMPONENT, request.body(Empty::new())?, request_tx, @@ -987,7 +1007,7 @@ async fn p3_http_forbidden_headers() -> Result<()> { // Grab the headers that actually reached the backend and assert that none // of the forbidden inbound headers leaked across the trust boundary. - let outbound = request_rx.await.unwrap(); + let outbound = request_rx.await.unwrap().unwrap(); let headers = outbound.headers(); // Forbidden headers shouldn't have gotten forwarded. @@ -1293,8 +1313,7 @@ mod p3_http_forward_body { .body(http_body_util::StreamBody::new(futures::stream::iter( frames.map(Ok::<_, ErrorCode>), )))?; - let (tx, _) = oneshot::channel(); - let res = run_http(P3_HTTP_FORWARD_BODY_COMPONENT, req, tx) + let res = run_http(P3_HTTP_FORWARD_BODY_COMPONENT, req) .await? .unwrap(); Ok(res.into_body()) From baf5c04df7b0f101554d3171d7c5e5fa84845dcd Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Tue, 27 Oct 2026 15:56:43 -0400 Subject: [PATCH 3/4] wasi-http: Send error to guest from forwarded body Ensure that an error encountered when sending a request or response body is delivered to the associated `transmit` future when the body is coming from another host body which has been forwarded. Previously a guest which forwards a request or response would have no way to see if the request/response body was delivered without error. For example, if the original sender passed a body with incorrect content length or if it dropped the connection before completion, the guest would still receive `Ok(())` on the transmission future. --- crates/test-programs/src/bin/p3_http_proxy.rs | 35 ++- crates/wasi-http/src/default_send_request.rs | 4 +- crates/wasi-http/src/error.rs | 18 +- crates/wasi-http/src/p3/body.rs | 199 +++++++++++++----- crates/wasi-http/src/p3/request.rs | 25 +-- crates/wasi-http/src/p3/response.rs | 25 +-- crates/wasi-http/tests/all/http_server.rs | 38 +++- crates/wasi-http/tests/all/p3/mod.rs | 50 +++++ 8 files changed, 302 insertions(+), 92 deletions(-) diff --git a/crates/test-programs/src/bin/p3_http_proxy.rs b/crates/test-programs/src/bin/p3_http_proxy.rs index df7737ee80f7..0396d7f07e98 100644 --- a/crates/test-programs/src/bin/p3_http_proxy.rs +++ b/crates/test-programs/src/bin/p3_http_proxy.rs @@ -16,8 +16,15 @@ test_programs::p3::service::export!(Component); impl Handler for Component { // Forward the request body and trailers to a URL specified in a header. + // + // If `expect-transmit-error` is set, also assert that the `response.new` + // transmit future of the returned response reports an error, and then send + // a request to `http://p3-test/` to signal that the assertion has run. async fn handle(request: Request) -> Result { let headers = request.get_headers().copy_all(); + let expect_transmit_error = headers + .iter() + .any(|(k, v)| k == "expect-transmit-error" && v == b"true"); Ok( if let Some(url) = headers.iter().find_map(|(k, v)| { (k == "url") @@ -42,7 +49,12 @@ impl Handler for Component { outgoing_request .set_authority(Some(url.authority())) .unwrap(); - client::send(outgoing_request).await? + let response = client::send(outgoing_request).await?; + if expect_transmit_error { + forward_expecting_transmit_error(response) + } else { + response + } } else { bad_request() }, @@ -50,6 +62,27 @@ impl Handler for Component { } } +fn forward_expecting_transmit_error(upstream: Response) -> Response { + let (body, trailers) = Response::consume_body(upstream, wit_future::new(|| Ok(())).1); + let (response, transmit) = Response::new(Fields::new(), Some(body), trailers); + wit_bindgen::spawn_local(async move { + // In the background, assert that the response fails to complete. + let result = transmit.await; + assert!( + result.is_err(), + "transmitting the response should have failed, got {result:?}" + ); + + // Send a request to indicate that the assertion passed. + let done = Request::new(Fields::new(), None, wit_future::new(|| Ok(None)).1, None).0; + done.set_scheme(Some(&Scheme::Http)).unwrap(); + done.set_authority(Some("p3-test")).unwrap(); + done.set_path_with_query(Some("/")).unwrap(); + client::send(done).await.unwrap(); + }); + response +} + fn bad_request() -> Response { respond(400) } diff --git a/crates/wasi-http/src/default_send_request.rs b/crates/wasi-http/src/default_send_request.rs index de42d3489b63..46ee8d1e9e0e 100644 --- a/crates/wasi-http/src/default_send_request.rs +++ b/crates/wasi-http/src/default_send_request.rs @@ -68,7 +68,7 @@ pub async fn default_send_request( let connect = async { let stream = TcpStream::connect(&authority) .await - .map_err(Error::Connect)?; + .map_err(|e| Error::Connect(e.into()))?; let stream = if use_tls { // derived from https://github.com/rustls/rustls/blob/main/examples/src/bin/simpleclient.rs let root_cert_store = rustls::RootCertStore { @@ -82,7 +82,7 @@ pub async fn default_send_request( let stream = connector .connect(domain, stream) .await - .map_err(Error::Tls)?; + .map_err(|e| Error::Tls(e.into()))?; stream.boxed() } else { stream.boxed() diff --git a/crates/wasi-http/src/error.rs b/crates/wasi-http/src/error.rs index 4cdab5f8e2f3..48bc2c0cf3b5 100644 --- a/crates/wasi-http/src/error.rs +++ b/crates/wasi-http/src/error.rs @@ -1,19 +1,23 @@ use std::fmt; use std::io; +use std::sync::Arc; /// Error type for `wasi:http` embedder interactions. /// /// This error type is a superset of the wasip2/wasip3 error types for example /// and is the error that the embedder interacts with for all operations. -#[derive(Debug)] +/// +/// Errors from other crates which cannot be cloned are wrapped in [`Arc`] so +/// that this error can be. +#[derive(Clone, Debug)] #[non_exhaustive] #[expect(missing_docs, reason = "too cluttered")] pub enum Error { - Hyper(hyper::Error), - Connect(io::Error), - Tls(io::Error), + Hyper(Arc), + Connect(Arc), + Tls(Arc), #[cfg(feature = "default-send-request")] - InvalidDnsNameError(rustls::pki_types::InvalidDnsNameError), + InvalidDnsNameError(Arc), HttpRequestBodySize(Option), DnsTimeout, @@ -79,7 +83,7 @@ pub type Result = std::result::Result; impl From for Error { fn from(err: hyper::Error) -> Self { - Self::Hyper(err) + Self::Hyper(Arc::new(err)) } } @@ -92,7 +96,7 @@ impl From for Error { #[cfg(feature = "default-send-request")] impl From for Error { fn from(err: rustls::pki_types::InvalidDnsNameError) -> Self { - Self::InvalidDnsNameError(err) + Self::InvalidDnsNameError(Arc::new(err)) } } diff --git a/crates/wasi-http/src/p3/body.rs b/crates/wasi-http/src/p3/body.rs index a35c7de1d8d8..5d34e76a1b68 100644 --- a/crates/wasi-http/src/p3/body.rs +++ b/crates/wasi-http/src/p3/body.rs @@ -697,78 +697,113 @@ where } } -/// A wrapper around [http_body::Body], which validates `Content-Length` -pub(crate) struct BodyWithContentLength { +/// A wrapper around [http_body::Body], which optionally validates +/// `Content-Length` and sends a copy of the first error on a channel. The +/// channel is closed without an error once the body finishes successfully. +pub(crate) struct BodyWithErrorReport { body: T, - error_tx: Option>, - make_error: fn(Option) -> E, + error_tx: Option>, + content_length: Option, +} + +/// `Content-Length` validation state of a [BodyWithErrorReport] +struct ContentLength { + make_error: fn(Option) -> Error, /// Limit of bytes to be sent limit: u64, /// Number of bytes sent sent: u64, } -impl BodyWithContentLength { - /// Sends the error constructed by [Self::make_error] on [Self::error_tx]. - /// Does nothing if an error has already been sent on [Self::error_tx]. - fn send_error(&mut self, sent: Option) -> Poll>> { - if let Some(error_tx) = self.error_tx.take() { - _ = error_tx.send((self.make_error)(sent)); +impl ContentLength { + /// Records `n` more bytes as sent, returning an error if that exceeds + /// [Self::limit]. + fn record(&mut self, n: usize) -> Result<(), Error> { + let sent = self + .sent + .saturating_add(u64::try_from(n).unwrap_or(u64::MAX)); + if sent > self.limit { + return Err((self.make_error)(Some(sent))); } - Poll::Ready(Some(Err((self.make_error)(sent)))) + self.sent = sent; + Ok(()) } } -impl http_body::Body for BodyWithContentLength +impl http_body::Body for BodyWithErrorReport where - T: http_body::Body + Unpin, + T: http_body::Body + Unpin, { type Data = T::Data; type Error = T::Error; #[inline] fn poll_frame( - mut self: Pin<&mut Self>, + self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll, Self::Error>>> { - match ready!(Pin::new(&mut self.as_mut().body).poll_frame(cx)) { + let this = self.get_mut(); + let res = match ready!(Pin::new(&mut this.body).poll_frame(cx)) { Some(Ok(frame)) => { - let Some(data) = frame.data_ref() else { - return Poll::Ready(Some(Ok(frame))); - }; - let Ok(sent) = data.len().try_into() else { - return self.send_error(None); - }; - let Some(sent) = self.sent.checked_add(sent) else { - return self.send_error(None); - }; - if sent > self.limit { - return self.send_error(Some(sent)); + // If we got a data frame have content length set, track the + // size. + if let (Some(content_length), Some(data)) = + (&mut this.content_length, frame.data_ref()) + { + match content_length.record(data.len()) { + Ok(()) => Some(Ok(frame)), + Err(err) => Some(Err(err)), + } + // Otherwise this wasn't a data frame, or we don't have content + // length set, so just pass the the frame as is. + } else { + Some(Ok(frame)) } - self.sent = sent; - Poll::Ready(Some(Ok(frame))) } - Some(Err(err)) => Poll::Ready(Some(Err(err))), - None if self.limit != self.sent => { - // short write - let sent = self.sent; - self.send_error(Some(sent)) + Some(Err(err)) => Some(Err(err)), + // If we've hit the end, check the content length + None => { + if let Some(content_length) = &this.content_length + && content_length.sent != content_length.limit + { + // short write + Some(Err((content_length.make_error)(Some(content_length.sent)))) + } else { + None + } } - None => Poll::Ready(None), + }; + match &res { + Some(Err(err)) => { + if let Some(error_tx) = this.error_tx.take() { + _ = error_tx.send(err.clone()); + } + } + None => this.error_tx = None, + Some(Ok(..)) => {} } + Poll::Ready(res) } #[inline] fn is_end_stream(&self) -> bool { - // Do not report end of stream before `limit` bytes were sent, otherwise - // the consumer might stop polling before a short write is detected in - // `poll_frame`. - self.sent == self.limit && self.body.is_end_stream() + match &self.content_length { + // Do not report end of stream before `limit` bytes were sent, + // otherwise the consumer might stop polling before a short write is + // detected in `poll_frame`. + Some(content_length) if content_length.sent != content_length.limit => false, + _ => self.body.is_end_stream(), + } } #[inline] fn size_hint(&self) -> http_body::SizeHint { - http_body::SizeHint::with_exact(self.limit.saturating_sub(self.sent)) + match &self.content_length { + Some(content_length) => http_body::SizeHint::with_exact( + content_length.limit.saturating_sub(content_length.sent), + ), + None => self.body.size_hint(), + } } } @@ -839,21 +874,26 @@ pub(crate) trait BodyExt { } } - fn with_content_length( + /// Reports the first error of this body on `error_tx`. If `content_length` + /// is set, the body is also validated against it, with `make_error` + /// constructing the error for a mismatch. + fn with_error_report( self, - limit: u64, - error_tx: oneshot::Sender, - make_error: fn(Option) -> E, - ) -> BodyWithContentLength + error_tx: oneshot::Sender, + content_length: Option, + make_error: fn(Option) -> Error, + ) -> BodyWithErrorReport where Self: Sized, { - BodyWithContentLength { + BodyWithErrorReport { body: self, error_tx: Some(error_tx), - make_error, - limit, - sent: 0, + content_length: content_length.map(|limit| ContentLength { + make_error, + limit, + sent: 0, + }), } } } @@ -880,7 +920,7 @@ mod tests { let (error_tx, mut error_rx) = oneshot::channel(); let body = Full::new(Bytes::from_static(b"hello")) .map_err(|never| match never {}) - .with_content_length(10, error_tx, Error::HttpResponseBodySize); + .with_error_report(error_tx, Some(10), Error::HttpResponseBodySize); let mut body = pin!(body); assert!(!body.is_end_stream()); assert_eq!(body.size_hint().exact(), Some(10)); @@ -910,7 +950,7 @@ mod tests { let (error_tx, mut error_rx) = oneshot::channel(); let body = Full::new(Bytes::from_static(b"helloworld")) .map_err(|never| match never {}) - .with_content_length(3, error_tx, Error::HttpResponseBodySize); + .with_error_report(error_tx, Some(3), Error::HttpResponseBodySize); let mut body = pin!(body); assert!(!body.is_end_stream()); assert_eq!(body.size_hint().exact(), Some(3)); @@ -932,7 +972,7 @@ mod tests { let (error_tx, mut error_rx) = oneshot::channel(); let body = Full::new(Bytes::from_static(b"hello")) .map_err(|never| match never {}) - .with_content_length(5, error_tx, Error::HttpResponseBodySize); + .with_error_report(error_tx, Some(5), Error::HttpResponseBodySize); let mut body = pin!(body); assert!(!body.is_end_stream()); assert_eq!(body.size_hint().exact(), Some(5)); @@ -951,6 +991,63 @@ mod tests { assert!(error_rx.try_recv().is_err()); } + /// An error returned by the body is passed through, and a copy of it is + /// reported on the channel. + #[test] + fn error_report_body_error() { + let (error_tx, mut error_rx) = oneshot::channel(); + let body = http_body_util::StreamBody::new(futures::stream::iter([ + Ok(http_body::Frame::data(Bytes::from_static(b"hello"))), + Err(Error::ConnectionTerminated), + ])) + .with_error_report(error_tx, None, Error::HttpResponseBodySize); + let mut body = pin!(body); + + let mut cx = Context::from_waker(Waker::noop()); + match body.as_mut().poll_frame(&mut cx) { + Poll::Ready(Some(Ok(frame))) => assert_eq!(frame.into_data().unwrap(), "hello"), + other => panic!("unexpected poll result: {other:?}"), + } + assert!(matches!( + error_rx.try_recv(), + Err(oneshot::error::TryRecvError::Empty) + )); + match body.as_mut().poll_frame(&mut cx) { + Poll::Ready(Some(Err(Error::ConnectionTerminated))) => {} + other => panic!("unexpected poll result: {other:?}"), + } + assert!(matches!( + error_rx.try_recv(), + Ok(Error::ConnectionTerminated) + )); + } + + /// A body without a `Content-Length` which finishes successfully closes the + /// channel without an error. + #[test] + fn error_report_success() { + let (error_tx, mut error_rx) = oneshot::channel(); + let body = Full::new(Bytes::from_static(b"hello")) + .map_err(|never| match never {}) + .with_error_report(error_tx, None, Error::HttpResponseBodySize); + let mut body = pin!(body); + assert_eq!(body.size_hint().exact(), Some(5)); + + let mut cx = Context::from_waker(Waker::noop()); + match body.as_mut().poll_frame(&mut cx) { + Poll::Ready(Some(Ok(frame))) => assert_eq!(frame.into_data().unwrap(), "hello"), + other => panic!("unexpected poll result: {other:?}"), + } + assert!(matches!( + body.as_mut().poll_frame(&mut cx), + Poll::Ready(None) + )); + assert!(matches!( + error_rx.try_recv(), + Err(oneshot::error::TryRecvError::Closed) + )); + } + /// `mpsc::channel` panics above `Semaphore::MAX_PERMITS` and at zero, so the /// capacity handed to it must always be clamped - including for the huge /// values an embedder or `-Shttp-outgoing-body-buffer-chunks` can supply. diff --git a/crates/wasi-http/src/p3/request.rs b/crates/wasi-http/src/p3/request.rs index 4f2b41a0c879..7c1472b5e83a 100644 --- a/crates/wasi-http/src/p3/request.rs +++ b/crates/wasi-http/src/p3/request.rs @@ -190,21 +190,16 @@ impl Request { let is_forbidden = getter(store.as_context_mut().data_mut()) .hooks .forbidden_header_filter(); - let body = body.with_filtered_trailers(is_forbidden).boxed_unsync(); - if let Some(limit) = content_length { - let (http_result_tx, http_result_rx) = oneshot::channel(); - _ = result_tx.send(Box::new(async move { - if let Ok(err) = http_result_rx.await { - return Err(err); - }; - fut.await - })); - body.with_content_length(limit, http_result_tx, Error::HttpRequestBodySize) - .boxed_unsync() - } else { - _ = result_tx.send(Box::new(fut)); - body - } + let (body_result_tx, body_result_rx) = oneshot::channel(); + _ = result_tx.send(Box::new(async move { + if let Ok(err) = body_result_rx.await { + return Err(err); + }; + fut.await + })); + body.with_filtered_trailers(is_forbidden) + .with_error_report(body_result_tx, content_length, Error::HttpRequestBodySize) + .boxed_unsync() } }; let mut store = store.as_context_mut(); diff --git a/crates/wasi-http/src/p3/response.rs b/crates/wasi-http/src/p3/response.rs index 65a927951319..7578fc70a97b 100644 --- a/crates/wasi-http/src/p3/response.rs +++ b/crates/wasi-http/src/p3/response.rs @@ -83,21 +83,16 @@ impl Response { let is_forbidden = getter(store.as_context_mut().data_mut()) .hooks .forbidden_header_filter(); - let body = body.with_filtered_trailers(is_forbidden); - if let Some(limit) = content_length { - let (http_result_tx, http_result_rx) = oneshot::channel(); - _ = result_tx.send(Box::new(async move { - if let Ok(err) = http_result_rx.await { - return Err(err); - }; - fut.await - })); - body.with_content_length(limit, http_result_tx, Error::HttpResponseBodySize) - .boxed_unsync() - } else { - _ = result_tx.send(Box::new(fut)); - body.boxed_unsync() - } + let (body_result_tx, body_result_rx) = oneshot::channel(); + _ = result_tx.send(Box::new(async move { + if let Ok(err) = body_result_rx.await { + return Err(err); + }; + fut.await + })); + body.with_filtered_trailers(is_forbidden) + .with_error_report(body_result_tx, content_length, Error::HttpResponseBodySize) + .boxed_unsync() } }; Ok(http::Response::from_parts(res, body)) diff --git a/crates/wasi-http/tests/all/http_server.rs b/crates/wasi-http/tests/all/http_server.rs index 5bc87311107c..88d81541f5e1 100644 --- a/crates/wasi-http/tests/all/http_server.rs +++ b/crates/wasi-http/tests/all/http_server.rs @@ -38,6 +38,16 @@ impl Server { conns: usize, run: impl Fn(TokioIo) -> F + Send + 'static, ) -> Result + where + F: Future>, + { + Self::new_raw(conns, move |stream| run(TokioIo::new(stream))) + } + + fn new_raw( + conns: usize, + run: impl Fn(tokio::net::TcpStream) -> F + Send + 'static, + ) -> Result where F: Future>, { @@ -62,7 +72,7 @@ impl Server { match listener.accept().await { Ok((stream, ..)) => { debug!(i, "accepted connection"); - if let Err(err) = run(TokioIo::new(stream)).await { + if let Err(err) = run(stream).await { warn!(i, ?err, "failed to serve connection"); } } @@ -142,6 +152,32 @@ impl Server { }) } + /// Send response headers declaring a `content-length` of + /// 100000 plus only part of the body, then close the connection. + pub fn http1_drop_mid_body(conns: usize) -> Result { + debug!("initializing drop-mid-body http1 server"); + Self::new_raw(conns, |mut stream| async move { + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; + + let mut request = Vec::new(); + let mut buf = [0; 1024]; + while !request.windows(4).any(|w| w == b"\r\n\r\n") { + let n = stream.read(&mut buf).await?; + if n == 0 { + return Ok(()); + } + request.extend_from_slice(&buf[..n]); + } + stream + .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 100000\r\n\r\n") + .await?; + stream.write_all(&[b'a'; 5000]).await?; + stream.flush().await?; + drop(stream); + Ok(()) + }) + } + pub fn http2(conns: usize) -> Result { debug!("initializing http2 server"); Self::new(conns, |io| async move { diff --git a/crates/wasi-http/tests/all/p3/mod.rs b/crates/wasi-http/tests/all/p3/mod.rs index 8fd45515de23..e4def2ab0fd5 100644 --- a/crates/wasi-http/tests/all/p3/mod.rs +++ b/crates/wasi-http/tests/all/p3/mod.rs @@ -884,6 +884,56 @@ async fn p3_http_proxy() -> Result<()> { Ok(()) } +/// A guest forwards a response body from an upstream server which closes the +/// connection partway through it. The client must see an error in the response +/// body, and the guest must see an error from the `response.new` transmit +/// future. +#[test_log::test(tokio::test(flavor = "multi_thread"))] +async fn p3_http_proxy_response_body_error() -> Result<()> { + let server = Server::http1_drop_mid_body(1)?; + + let engine = test_programs_artifacts::engine(|config| { + config.wasm_backtrace_details(wasmtime::WasmBacktraceDetails::Enable); + config.wasm_component_model_async(true); + }); + let component = Component::from_file(&engine, P3_HTTP_PROXY_COMPONENT)?; + // The guest sends a request to `http://p3-test/` once it has checked the + // transmit result to signal successful completion. Hook up `done_rx` to + // receive that request. + let (done_tx, done_rx) = oneshot::channel(); + let mut store = Store::new(&engine, Ctx::new(done_tx)); + let mut linker = Linker::new(&engine); + wasmtime_wasi::p2::add_to_linker_async(&mut linker)?; + wasmtime_wasi::p3::add_to_linker(&mut linker)?; + wasmtime_wasi_http::p3::add_to_linker(&mut linker)?; + let service = Service::instantiate_async(&mut store, &component, &linker).await?; + + let req = http::Request::builder() + .uri("http://localhost/") + .header("url", format!("http://{}/", server.addr())) + .header("expect-transmit-error", "true") + .body(Empty::::new())?; + let (req, _io) = Request::from_http(&mut store.data_mut().hooks, req); + store + .run_concurrent(async |store| { + let res = service.handle(store, req).await?.unwrap(); + // As in `wasmtime serve`, report that the incoming request was + // handled successfully. + let res = store.with(|store| res.into_http(store, async { Ok(()) }))?; + assert_eq!(res.status(), 200); + let body = res.into_body().collect().await; + assert!( + body.is_err(), + "client should see an error in the response body" + ); + done_rx + .await + .context("guest did not check the transmit result")?; + wasmtime::error::Ok(()) + }) + .await? +} + #[test_log::test(tokio::test(flavor = "multi_thread"))] async fn p3_http_proxy_store_drop_cancels_io() -> Result<()> { let (_body_tx, body_rx) = futures::channel::mpsc::channel::>(1); From 9dd3d9bb1c6ab80666f8bfbfae3cc20aae93112c Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Thu, 29 Oct 2026 08:57:40 -0400 Subject: [PATCH 4/4] fix test --- crates/wasi-http/tests/all/p3/mod.rs | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/crates/wasi-http/tests/all/p3/mod.rs b/crates/wasi-http/tests/all/p3/mod.rs index e4def2ab0fd5..e97985a8c707 100644 --- a/crates/wasi-http/tests/all/p3/mod.rs +++ b/crates/wasi-http/tests/all/p3/mod.rs @@ -901,7 +901,7 @@ async fn p3_http_proxy_response_body_error() -> Result<()> { // transmit result to signal successful completion. Hook up `done_rx` to // receive that request. let (done_tx, done_rx) = oneshot::channel(); - let mut store = Store::new(&engine, Ctx::new(done_tx)); + let mut store = Store::new(&engine, Ctx::with_hooks(TestHooks::new(Some(done_tx)))); let mut linker = Linker::new(&engine); wasmtime_wasi::p2::add_to_linker_async(&mut linker)?; wasmtime_wasi::p3::add_to_linker(&mut linker)?; @@ -928,8 +928,9 @@ async fn p3_http_proxy_response_body_error() -> Result<()> { ); done_rx .await - .context("guest did not check the transmit result")?; - wasmtime::error::Ok(()) + .context("guest did not check the transmit result")? + .context("failed to collect the guest's completion request")?; + Ok(()) }) .await? }