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/2] 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/2] 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())