wasi-http: Pass transmission errors through middleware

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<Response, ErrorCode> {
         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()

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)
     }
 }

crates/wasi-http/tests/all/p3/mod.rs

@@ -361,6 +361,75 @@ async fn compose(a: &[u8], b: &[u8]) -> Result<Vec<u8>> {
     .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::<Result<_, ErrorCode>>(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