diff --git a/crates/test-programs/src/bin/p3_http_echo.rs b/crates/test-programs/src/bin/p3_http_echo.rs index f6198a4e44ad..90589f899269 100644 --- a/crates/test-programs/src/bin/p3_http_echo.rs +++ b/crates/test-programs/src/bin/p3_http_echo.rs @@ -15,9 +15,25 @@ impl Handler for Component { /// Return a response which echoes the request headers, body, and trailers. async fn handle(request: Request) -> Result { let headers = request.get_headers(); - let (_, result_rx) = wit_future::new(|| Ok(())); + let (result_tx, result_rx) = wit_future::new(|| Ok(())); let (body, trailers) = Request::consume_body(request, result_rx); + // If `inject-transmission-error` is set, report a transmission error + // and then loop without replying. + if headers + .get("inject-transmission-error") + .into_iter() + .any(|v| v == b"true") + { + result_tx + .write(Err(ErrorCode::InternalError(Some( + "Injected error by echo service".to_string(), + )))) + .await + .unwrap(); + futures::future::pending::<()>().await; + } + let (response, _result) = if headers .get("x-host-to-host") .into_iter() diff --git a/crates/test-programs/src/bin/p3_http_middleware.rs b/crates/test-programs/src/bin/p3_http_middleware.rs index c83bcf90f541..49420cef6565 100644 --- a/crates/test-programs/src/bin/p3_http_middleware.rs +++ b/crates/test-programs/src/bin/p3_http_middleware.rs @@ -67,7 +67,7 @@ impl exports::wasi::http::handler::Guest for Component { } _ => true, }); - let (_, result_rx) = wit_future::new(|| Ok(())); + let (result_tx, result_rx) = wit_future::new(|| Ok(())); let (mut body, trailers) = Request::consume_body(request, result_rx); let (body, trailers) = if content_deflated { @@ -107,7 +107,7 @@ impl exports::wasi::http::handler::Guest for Component { // While the above task (if any) is running, synthesize a request from the parts collected above and pass // it to the imported `wasi:http/handler`. - let (my_request, _request_complete) = Request::new( + let (my_request, request_complete) = Request::new( Headers::from_list(&headers).unwrap(), Some(body), trailers, @@ -120,6 +120,11 @@ impl exports::wasi::http::handler::Guest for Component { .unwrap(); my_request.set_authority(authority.as_deref()).unwrap(); + // Forward completion or transmission error back to the caller. + wit_bindgen::spawn_local(async move { + _ = result_tx.write(request_complete.await).await; + }); + let response = handler::handle(my_request).await?; // Now that we have the response, extract the parts, adding an extra header if we'll be encoding the body. @@ -129,7 +134,7 @@ impl exports::wasi::http::handler::Guest for Component { headers.push(("content-encoding".into(), b"deflate".into())); } - let (_, result_rx) = wit_future::new(|| Ok(())); + let (result_tx, result_rx) = wit_future::new(|| Ok(())); let (mut body, trailers) = Response::consume_body(response, result_rx); let (body, trailers) = if accept_deflated { headers.retain(|(name, _value)| name != "content-length"); @@ -171,10 +176,17 @@ impl exports::wasi::http::handler::Guest for Component { // While the above tasks (if any) are running, synthesize a response from the parts collected above and // return it. - let (my_response, _response_complete) = + let (my_response, response_complete) = Response::new(Headers::from_list(&headers).unwrap(), Some(body), trailers); my_response.set_status_code(status_code).unwrap(); + // Mirror the request path: forward the transmission result of the response we + // created into the response we consumed above, so response-body errors also + // propagate back toward their producer. + wit_bindgen::spawn_local(async move { + _ = result_tx.write(response_complete.await).await; + }); + Ok(my_response) } } diff --git a/crates/wasi-http/tests/all/p3/mod.rs b/crates/wasi-http/tests/all/p3/mod.rs index a97ef9d1c725..87616dfd24ef 100644 --- a/crates/wasi-http/tests/all/p3/mod.rs +++ b/crates/wasi-http/tests/all/p3/mod.rs @@ -361,6 +361,75 @@ async fn compose(a: &[u8], b: &[u8]) -> Result> { .to_wasmtime_result() } +/// Test that an error reported by `consume-body` is propagated up through +/// the middleware. +#[test_log::test(tokio::test(flavor = "multi_thread"))] +async fn p3_http_middleware_error() -> Result<()> { + _ = env_logger::try_init(); + + let echo = &fs::read(P3_HTTP_ECHO_COMPONENT).await?; + let middleware = &fs::read(P3_HTTP_MIDDLEWARE_COMPONENT).await?; + let inner = compose(middleware, echo).await?; + let composed = compose(middleware, &inner).await?; + + let tempdir = tempfile::tempdir()?; + let path = tempdir.path().join("temp.wasm"); + fs::write(&path, &composed).await?; + + let (mut body_tx, body_rx) = futures::channel::mpsc::channel::>(1); + // Add a header which will trigger the echo service to record a transmission + // error and never deliver a response. + let request = http::Request::builder() + .uri("http://localhost/") + .method(http::Method::GET) + .header("inject-transmission-error", "true"); + + let response = futures::join!( + async { + let result = run_http( + path.to_str().unwrap(), + request.body(http_body_util::StreamBody::new(body_rx))?, + oneshot::channel().0, + ) + .await; + result + }, + async { + body_tx + .send(Ok(http_body::Frame::data(Bytes::from_static( + b"And the mome raths outgrabe", + )))) + .await + .unwrap(); + body_tx + .send(Ok(http_body::Frame::trailers({ + let mut trailers = http::HeaderMap::new(); + assert!( + trailers + .insert("fizz", http::HeaderValue::from_static("buzz")) + .is_none() + ); + trailers + }))) + .await + .unwrap(); + drop(body_tx); + } + ) + .0; + + let err = response.unwrap_err(); + let expected_err = "Injected error by echo service"; + // Even though the echo service never replied, we can read the transmission + // error on the future from `request.new`. + assert!( + format!("{err:#}").contains(expected_err), + "Resulting error {err:#} does not contain expected message {expected_err}" + ); + + Ok(()) +} + #[test_log::test(tokio::test(flavor = "multi_thread"))] async fn p3_http_middleware_with_chain() -> Result<()> { test_http_middleware_with_chain(false).await