From 6e7787f3fb0cd280aefb9d8d54f0b7c9a2b79a76 Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Fri, 14 Aug 2026 14:25:07 -0400 Subject: [PATCH 1/5] modify p3 http tests to check for transmission error --- ...3_http_outbound_request_invalid_dnsname.rs | 7 ++-- ...3_http_outbound_request_invalid_version.rs | 7 ++-- .../bin/p3_http_outbound_request_timeout.rs | 8 ++--- ...ttp_outbound_request_unsupported_scheme.rs | 8 +++-- crates/test-programs/src/p3/http.rs | 34 +++++++++++++++++-- 5 files changed, 47 insertions(+), 17 deletions(-) diff --git a/crates/test-programs/src/bin/p3_http_outbound_request_invalid_dnsname.rs b/crates/test-programs/src/bin/p3_http_outbound_request_invalid_dnsname.rs index 16a7d5baf017..16a64350931c 100644 --- a/crates/test-programs/src/bin/p3_http_outbound_request_invalid_dnsname.rs +++ b/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::() diff --git a/crates/test-programs/src/bin/p3_http_outbound_request_invalid_version.rs b/crates/test-programs/src/bin/p3_http_outbound_request_invalid_version.rs index e8211580bef6..dfd11d12cce5 100644 --- a/crates/test-programs/src/bin/p3_http_outbound_request_invalid_version.rs +++ b/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::() diff --git a/crates/test-programs/src/bin/p3_http_outbound_request_timeout.rs b/crates/test-programs/src/bin/p3_http_outbound_request_timeout.rs index c3f969c11f55..64f100fded4a 100644 --- a/crates/test-programs/src/bin/p3_http_outbound_request_timeout.rs +++ b/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 = 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::(), diff --git a/crates/test-programs/src/bin/p3_http_outbound_request_unsupported_scheme.rs b/crates/test-programs/src/bin/p3_http_outbound_request_unsupported_scheme.rs index 20f01700f181..2ce2d00d7d6f 100644 --- a/crates/test-programs/src/bin/p3_http_outbound_request_unsupported_scheme.rs +++ b/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::() .expect("expected a wasi-http ErrorCode"), ErrorCode::HttpProtocolError, diff --git a/crates/test-programs/src/p3/http.rs b/crates/test-programs/src/p3/http.rs index 1b21fc1c95a1..acea82711547 100644 --- a/crates/test-programs/src/p3/http.rs +++ b/crates/test-programs/src/p3/http.rs @@ -45,6 +45,36 @@ pub async fn request( first_by_timeout: Option, between_bytes_timeout: Option, ) -> Result { + 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)]>, + connect_timeout: Option, + first_by_timeout: Option, + between_bytes_timeout: Option, +) -> Result<(Result<()>, Result)> { fn header_val(v: &str) -> Vec { v.to_string().into_bytes() } @@ -116,7 +146,5 @@ pub async fn request( }) }, ); - let response = handle?; - transmit?; - Ok(response) + Ok((transmit, handle)) } From e13936abdab766f67f5063da092f9330db12c4ad Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Thu, 13 Aug 2026 10:19:01 -0400 Subject: [PATCH 2/5] Remove `BodyWithState` --- crates/wasi-http/src/p3/body.rs | 43 ------------------------- crates/wasi-http/src/p3/host/handler.rs | 24 ++++---------- 2 files changed, 7 insertions(+), 60 deletions(-) diff --git a/crates/wasi-http/src/p3/body.rs b/crates/wasi-http/src/p3/body.rs index b5795e86bfd1..909c27e03dfe 100644 --- a/crates/wasi-http/src/p3/body.rs +++ b/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 { - body: T, - _state: U, -} - -impl http_body::Body for BodyWithState -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, 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 { body: T, @@ -683,16 +650,6 @@ where } pub(crate) trait BodyExt { - fn with_state(self, state: T) -> BodyWithState - where - Self: Sized, - { - BodyWithState { - body: self, - _state: state, - } - } - fn with_content_length( self, limit: u64, diff --git a/crates/wasi-http/src/p3/host/handler.rs b/crates/wasi-http/src/p3/host/handler.rs index f8d0ab40a638..a6b9b0b6985f 100644 --- a/crates/wasi-http/src/p3/host/handler.rs +++ b/crates/wasi-http/src/p3/host/handler.rs @@ -1,12 +1,11 @@ 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; @@ -24,10 +23,7 @@ impl Drop for AbortOnDropJoinHandle { } async fn io_task_result( - rx: oneshot::Receiver<( - Arc, - oneshot::Receiver>, - )>, + rx: oneshot::Receiver<(AbortOnDropJoinHandle, oneshot::Receiver>)>, ) -> Result<(), Error> { let Ok((_io, io_result_rx)) = rx.await else { return Ok(()); @@ -40,10 +36,6 @@ impl HostWithStore for WasiHttp { store: &Accessor, req: Resource, ) -> HttpResult> { - // 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 +53,7 @@ impl HostWithStore 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.map(|body| body.boxed_unsync()), options.as_deref().copied(), Box::new(async { // Forward the response processing result to `WasiHttpCtx` implementation @@ -91,15 +83,13 @@ impl HostWithStore for WasiHttp { 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((io, rx)); + body.boxed_unsync() } }; store.with(|mut store| { From 1e6e8d4ff1874c28365860d67f78e6ccc356ebba Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Wed, 12 Aug 2026 15:40:49 -0400 Subject: [PATCH 3/5] fix by sending error --- crates/wasi-http/src/p3/host/handler.rs | 73 +++++++++++++++++++++---- 1 file changed, 63 insertions(+), 10 deletions(-) diff --git a/crates/wasi-http/src/p3/host/handler.rs b/crates/wasi-http/src/p3/host/handler.rs index a6b9b0b6985f..a579b9d21bd6 100644 --- a/crates/wasi-http/src/p3/host/handler.rs +++ b/crates/wasi-http/src/p3/host/handler.rs @@ -23,12 +23,46 @@ impl Drop for AbortOnDropJoinHandle { } async fn io_task_result( - rx: oneshot::Receiver<(AbortOnDropJoinHandle, oneshot::Receiver>)>, + rx: oneshot::Receiver<( + Option, + oneshot::Receiver>, + )>, ) -> Result<(), Error> { let Ok((_io, io_result_rx)) = rx.await else { - return Ok(()); + return Err(Error::InternalError(Some( + "Future indicating transmission result dropped without being resolved.".to_string(), + ))); }; - io_result_rx.await.unwrap_or(Ok(())) + io_result_rx.await.unwrap_or_else(|_| { + Err(Error::InternalError(Some( + "Future indicating transmission result dropped without being resolved.".to_string(), + ))) + }) +} + +fn send_dummy_io( + result: Result<(), Error>, + io_result_tx: oneshot::Sender<( + Option, + oneshot::Receiver>, + )>, +) { + let (tx, rx) = oneshot::channel(); + let _ = tx.send(result); + let _ = io_result_tx.send((None, rx)); +} + +fn send_dummy_io_err( + store: &Accessor, + e: Error, + io_result_tx: oneshot::Sender<( + Option, + oneshot::Receiver>, + )>, +) -> 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 HostWithStore for WasiHttp { @@ -63,10 +97,26 @@ impl HostWithStore 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, .. @@ -76,9 +126,12 @@ impl HostWithStore 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 @@ -88,7 +141,7 @@ impl HostWithStore for WasiHttp { debug!(?res, "`send_request` I/O future finished"); _ = tx.send(res); })); - _ = io_result_tx.send((io, rx)); + _ = io_result_tx.send((Some(io), rx)); body.boxed_unsync() } }; From 33b3f8894baf46decd8aa621ee4f8f58c554e2aa Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Mon, 17 Aug 2026 14:37:34 -0400 Subject: [PATCH 4/5] clippy --- crates/wasi-http/src/p3/host/handler.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/wasi-http/src/p3/host/handler.rs b/crates/wasi-http/src/p3/host/handler.rs index a579b9d21bd6..c96b384fd717 100644 --- a/crates/wasi-http/src/p3/host/handler.rs +++ b/crates/wasi-http/src/p3/host/handler.rs @@ -106,7 +106,7 @@ impl HostWithStore for WasiHttp { return Err(err_code.into()); } Err(e) => { - let e = Error::InternalError(Some(format!("{}", e))); + let e = Error::InternalError(Some(format!("{e}"))); return Err(send_dummy_io_err(store, e, io_result_tx)); } }, From cfa0eb86827c686f0bd7ec86442fdecf24e70564 Mon Sep 17 00:00:00 2001 From: Adam Bratschi-Kaye Date: Tue, 18 Aug 2026 14:47:53 +0000 Subject: [PATCH 5/5] review --- crates/wasi-http/src/p3/host/handler.rs | 20 +++++++++----------- 1 file changed, 9 insertions(+), 11 deletions(-) diff --git a/crates/wasi-http/src/p3/host/handler.rs b/crates/wasi-http/src/p3/host/handler.rs index c96b384fd717..d514eb5121bb 100644 --- a/crates/wasi-http/src/p3/host/handler.rs +++ b/crates/wasi-http/src/p3/host/handler.rs @@ -5,7 +5,6 @@ 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 tokio::sync::oneshot; use tokio::task::{self, JoinHandle}; use tracing::debug; @@ -22,6 +21,9 @@ 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<( Option, @@ -29,15 +31,11 @@ async fn io_task_result( )>, ) -> Result<(), Error> { let Ok((_io, io_result_rx)) = rx.await else { - return Err(Error::InternalError(Some( - "Future indicating transmission result dropped without being resolved.".to_string(), - ))); + return Err(Error::InternalError(Some(DROPPED_FUTURE_ERROR.to_string()))); }; - io_result_rx.await.unwrap_or_else(|_| { - Err(Error::InternalError(Some( - "Future indicating transmission result dropped without being resolved.".to_string(), - ))) - }) + io_result_rx + .await + .unwrap_or_else(|_| Err(Error::InternalError(Some(DROPPED_FUTURE_ERROR.to_string())))) } fn send_dummy_io( @@ -87,7 +85,7 @@ impl HostWithStore 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.boxed_unsync()), + req, options.as_deref().copied(), Box::new(async { // Forward the response processing result to `WasiHttpCtx` implementation @@ -142,7 +140,7 @@ impl HostWithStore for WasiHttp { _ = tx.send(res); })); _ = io_result_tx.send((Some(io), rx)); - body.boxed_unsync() + body } }; store.with(|mut store| {