Pass transmission errors to the future from {request, response}.new

crates/test-programs/src/bin/p3_http_outbound_request_invalid_dnsname.rs

@@ -6,7 +6,7 @@ test_programs::p3::export!(Component);
 
 impl test_programs::p3::exports::wasi::cli::run::Guest for Component {
     async fn run() -> Result<(), ()> {
-        let res = test_programs::p3::http::request(
+        let (transmit, _response) = test_programs::p3::http::request_with_transmit_result(
             Method::Get,
             Scheme::Http,
             "some.invalid.dnsname:3000",
@@ -17,9 +17,10 @@ impl test_programs::p3::exports::wasi::cli::run::Guest for Component {
             None,
             None,
         )
-        .await;
+        .await
+        .expect("failed to construct request");
 
-        let e = res.unwrap_err();
+        let e = transmit.expect_err("expected request transmission to fail");
         assert!(
             matches!(
                 e.downcast_ref::<ErrorCode>()

crates/test-programs/src/bin/p3_http_outbound_request_invalid_version.rs

@@ -10,7 +10,7 @@ impl test_programs::p3::exports::wasi::cli::run::Guest for Component {
             .into_iter()
             .find_map(|(k, v)| k.eq("HTTP_SERVER").then_some(v))
             .unwrap();
-        let res = test_programs::p3::http::request(
+        let (transmit, _response) = test_programs::p3::http::request_with_transmit_result(
             Method::Connect,
             Scheme::Http,
             &addr,
@@ -21,14 +21,15 @@ impl test_programs::p3::exports::wasi::cli::run::Guest for Component {
             Some(1_000_000_000),
             None,
         )
-        .await;
+        .await
+        .expect("failed to construct request");
 
         // The error seen during this test is mostly an `HttpProtocolError`, but
         // depending on scheduling it's possible to get stuck in hyper right now
         // where the server is indefinitely waiting on the client and the client
         // times out. Accept both kinds of errors here, and note the explicit 1s
         // timeout above to avoid this taking too long. in the timeout case.
-        let err = res.unwrap_err();
+        let err = transmit.expect_err("expected request transmission to fail");
         assert!(
             matches!(
                 err.downcast_ref::<ErrorCode>()

crates/test-programs/src/bin/p3_http_outbound_request_timeout.rs

@@ -1,4 +1,3 @@
-use anyhow::Context;
 use std::net::SocketAddr;
 use std::time::Duration;
 use test_programs::p3::wasi::http::types::{ErrorCode, Method, Scheme};
@@ -13,7 +12,7 @@ impl test_programs::p3::exports::wasi::cli::run::Guest for Component {
         let addr = SocketAddr::from(([203, 0, 113, 12], 80)).to_string();
         let timeout = Duration::from_millis(200);
         let connect_timeout: Option<u64> = Some(timeout.as_nanos() as u64);
-        let res = test_programs::p3::http::request(
+        let (transmit, _response) = test_programs::p3::http::request_with_transmit_result(
             Method::Get,
             Scheme::Http,
             &addr,
@@ -25,10 +24,9 @@ impl test_programs::p3::exports::wasi::cli::run::Guest for Component {
             None,
         )
         .await
-        .context("/get");
+        .expect("failed to construct request");
 
-        assert!(res.is_err());
-        let err = res.unwrap_err();
+        let err = transmit.expect_err("expected request transmission to fail");
         assert!(
             matches!(
                 err.downcast_ref::<ErrorCode>(),

crates/test-programs/src/bin/p3_http_outbound_request_unsupported_scheme.rs

@@ -6,7 +6,7 @@ test_programs::p3::export!(Component);
 
 impl test_programs::p3::exports::wasi::cli::run::Guest for Component {
     async fn run() -> Result<(), ()> {
-        let res = test_programs::p3::http::request(
+        let (transmit, _response) = test_programs::p3::http::request_with_transmit_result(
             Method::Get,
             Scheme::Other("WS".to_owned()),
             "localhost:3000",
@@ -17,10 +17,12 @@ impl test_programs::p3::exports::wasi::cli::run::Guest for Component {
             None,
             None,
         )
-        .await;
+        .await
+        .expect("failed to construct request");
 
         assert!(matches!(
-            res.unwrap_err()
+            transmit
+                .expect_err("expected request transmission to fail")
                 .downcast::<ErrorCode>()
                 .expect("expected a wasi-http ErrorCode"),
             ErrorCode::HttpProtocolError,

crates/test-programs/src/p3/http.rs

@@ -45,6 +45,36 @@ pub async fn request(
     first_by_timeout: Option<u64>,
     between_bytes_timeout: Option<u64>,
 ) -> Result<Response> {
+    let (transmit, response) = request_with_transmit_result(
+        method,
+        scheme,
+        authority,
+        path_with_query,
+        body,
+        additional_headers,
+        connect_timeout,
+        first_by_timeout,
+        between_bytes_timeout,
+    )
+    .await?;
+    transmit?;
+    response
+}
+
+/// Attempts to send the request. Return value will be an error if we were
+/// unable to properly construct the request. The inner values are the result of
+/// transmitting the request and the result of receiving the response.
+pub async fn request_with_transmit_result(
+    method: types::Method,
+    scheme: types::Scheme,
+    authority: &str,
+    path_with_query: &str,
+    body: Option<&[u8]>,
+    additional_headers: Option<&[(String, Vec<u8>)]>,
+    connect_timeout: Option<u64>,
+    first_by_timeout: Option<u64>,
+    between_bytes_timeout: Option<u64>,
+) -> Result<(Result<()>, Result<Response>)> {
     fn header_val(v: &str) -> Vec<u8> {
         v.to_string().into_bytes()
     }
@@ -116,7 +146,5 @@ pub async fn request(
             })
         },
     );
-    let response = handle?;
-    transmit?;
-    Ok(response)
+    Ok((transmit, handle))
 }

crates/wasi-http/src/p3/body.rs

@@ -568,39 +568,6 @@ where
     }
 }
 
-/// A wrapper around [http_body::Body], which allows attaching arbitrary state to it
-pub(crate) struct BodyWithState<T, U> {
-    body: T,
-    _state: U,
-}
-
-impl<T, U> http_body::Body for BodyWithState<T, U>
-where
-    T: http_body::Body + Unpin,
-    U: Unpin,
-{
-    type Data = T::Data;
-    type Error = T::Error;
-
-    #[inline]
-    fn poll_frame(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
-        Pin::new(&mut self.get_mut().body).poll_frame(cx)
-    }
-
-    #[inline]
-    fn is_end_stream(&self) -> bool {
-        self.body.is_end_stream()
-    }
-
-    #[inline]
-    fn size_hint(&self) -> http_body::SizeHint {
-        self.body.size_hint()
-    }
-}
-
 /// A wrapper around [http_body::Body], which validates `Content-Length`
 pub(crate) struct BodyWithContentLength<T, E> {
     body: T,
@@ -683,16 +650,6 @@ where
 }
 
 pub(crate) trait BodyExt {
-    fn with_state<T>(self, state: T) -> BodyWithState<Self, T>
-    where
-        Self: Sized,
-    {
-        BodyWithState {
-            body: self,
-            _state: state,
-        }
-    }
-
     fn with_content_length<E>(
         self,
         limit: u64,

crates/wasi-http/src/p3/host/handler.rs

@@ -1,12 +1,10 @@
 use crate::FieldMap;
 use crate::p3::bindings::http::client::{Host, HostWithStore};
 use crate::p3::bindings::http::types::{Request, Response};
-use crate::p3::body::{Body, BodyExt as _};
+use crate::p3::body::Body;
 use crate::p3::{HttpError, HttpResult};
 use crate::{Error, WasiHttp, WasiHttpCtxView};
 use core::task::{Context, Poll, Waker};
-use http_body_util::BodyExt as _;
-use std::sync::Arc;
 use tokio::sync::oneshot;
 use tokio::task::{self, JoinHandle};
 use tracing::debug;
@@ -23,27 +21,53 @@ impl Drop for AbortOnDropJoinHandle {
     }
 }
 
+const DROPPED_FUTURE_ERROR: &str =
+    "Future indicating transmission result dropped without being resolved.";
+
 async fn io_task_result(
     rx: oneshot::Receiver<(
-        Arc<AbortOnDropJoinHandle>,
+        Option<AbortOnDropJoinHandle>,
         oneshot::Receiver<Result<(), Error>>,
     )>,
 ) -> Result<(), Error> {
     let Ok((_io, io_result_rx)) = rx.await else {
-        return Ok(());
+        return Err(Error::InternalError(Some(DROPPED_FUTURE_ERROR.to_string())));
     };
-    io_result_rx.await.unwrap_or(Ok(()))
+    io_result_rx
+        .await
+        .unwrap_or_else(|_| Err(Error::InternalError(Some(DROPPED_FUTURE_ERROR.to_string()))))
+}
+
+fn send_dummy_io(
+    result: Result<(), Error>,
+    io_result_tx: oneshot::Sender<(
+        Option<AbortOnDropJoinHandle>,
+        oneshot::Receiver<Result<(), Error>>,
+    )>,
+) {
+    let (tx, rx) = oneshot::channel();
+    let _ = tx.send(result);
+    let _ = io_result_tx.send((None, rx));
+}
+
+fn send_dummy_io_err<T>(
+    store: &Accessor<T, WasiHttp>,
+    e: Error,
+    io_result_tx: oneshot::Sender<(
+        Option<AbortOnDropJoinHandle>,
+        oneshot::Receiver<Result<(), Error>>,
+    )>,
+) -> HttpError {
+    let err_code = store.with(|mut store| store.get().error_to_p3(&e));
+    send_dummy_io(Err(e), io_result_tx);
+    err_code.into()
 }
 
 impl<T> HostWithStore<T> for WasiHttp {
     async fn send(
         store: &Accessor<T, Self>,
         req: Resource<Request>,
     ) -> HttpResult<Resource<Response>> {
-        // A handle to the I/O task, if spawned, will be sent on this channel
-        // and kept as part of request body state
-        let (io_task_tx, io_task_rx) = oneshot::channel();
-
         // A handle to the I/O task, if spawned, will be sent on this channel
         // along with the result receiver
         let (io_result_tx, io_result_rx) = oneshot::channel();
@@ -61,7 +85,7 @@ impl<T> HostWithStore<T> for WasiHttp {
             let (req, options) =
                 req.into_http_with_getter(&mut store, io_task_result(io_result_rx), getter)?;
             HttpResult::Ok(store.get().hooks.send_request(
-                req.map(|body| body.with_state(io_task_rx).boxed_unsync()),
+                req,
                 options.as_deref().copied(),
                 Box::new(async {
                     // Forward the response processing result to `WasiHttpCtx` implementation
@@ -71,10 +95,26 @@ impl<T> HostWithStore<T> for WasiHttp {
                     Box::into_pin(fut).await
                 }),
             ))
-        })?;
-        let (res, io) = Box::into_pin(fut)
-            .await
-            .map_err(|e| store.with(|mut store| store.get().error_to_p3(&e)))?;
+        });
+        let fut = match fut {
+            Ok(fut) => fut,
+            Err(e) => match e.downcast() {
+                Ok(err_code) => {
+                    send_dummy_io(Err(err_code.clone().into()), io_result_tx);
+                    return Err(err_code.into());
+                }
+                Err(e) => {
+                    let e = Error::InternalError(Some(format!("{e}")));
+                    return Err(send_dummy_io_err(store, e, io_result_tx));
+                }
+            },
+        };
+        let (res, io) = match Box::into_pin(fut).await {
+            Ok(r) => r,
+            Err(e) => {
+                return Err(send_dummy_io_err(store, e, io_result_tx));
+            }
+        };
         let (
             http::response::Parts {
                 status, headers, ..
@@ -84,22 +124,23 @@ impl<T> HostWithStore<T> for WasiHttp {
 
         let mut io = Box::into_pin(io);
         let body = match io.as_mut().poll(&mut Context::from_waker(Waker::noop())) {
-            Poll::Ready(Ok(())) => body,
+            Poll::Ready(Ok(())) => {
+                send_dummy_io(Ok(()), io_result_tx);
+                body
+            }
             Poll::Ready(Err(e)) => {
-                return Err(store.with(|mut store| store.get().error_to_p3(&e)).into());
+                return Err(send_dummy_io_err(store, e, io_result_tx));
             }
             Poll::Pending => {
                 // I/O driver still needs to be polled, spawn a task and send handles to it
                 let (tx, rx) = oneshot::channel();
-                let io = task::spawn(async move {
+                let io = AbortOnDropJoinHandle(task::spawn(async move {
                     let res = io.await;
                     debug!(?res, "`send_request` I/O future finished");
                     _ = tx.send(res);
-                });
-                let io = Arc::new(AbortOnDropJoinHandle(io));
-                _ = io_result_tx.send((Arc::clone(&io), rx));
-                _ = io_task_tx.send(Arc::clone(&io));
-                body.with_state(io).boxed_unsync()
+                }));
+                _ = io_result_tx.send((Some(io), rx));
+                body
             }
         };
         store.with(|mut store| {