feat(p3-http): begin outgoing HTTP implementation

Signed-off-by: Roman Volosatovs <rvolosatovs@riseup.net>

Author: rvolosatovs

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

@@ -9,11 +9,11 @@ use http_body_util::combinators::BoxBody;
 use std::sync::Arc;
 use tokio::sync::{mpsc, oneshot};
 use tokio_util::sync::PollSender;
-use wasmtime::StoreContextMut;
 use wasmtime::component::{
     Accessor, FutureConsumer, FutureReader, Resource, Source, StreamConsumer, StreamReader,
     StreamResult,
 };
+use wasmtime::{AsContextMut, StoreContextMut};
 
 /// The concrete type behind a `wasi:http/types/body` resource.
 pub(crate) enum Body {
@@ -24,10 +24,14 @@ pub(crate) enum Body {
         /// Future, on which guest will write result and optional trailers
         trailers_rx: FutureReader<Result<Option<Resource<Trailers>>, ErrorCode>>,
         /// Channel, on which transmission result will be written
-        result_tx: oneshot::Sender<Result<(), ErrorCode>>,
+        result_tx: oneshot::Sender<Box<dyn Future<Output = Result<(), ErrorCode>> + Send>>,
     },
     /// Body constructed by the host.
-    Host(BoxBody<Bytes, ErrorCode>),
+    Host {
+        body: BoxBody<Bytes, ErrorCode>,
+        /// Channel, on which transmission result will be written
+        result_tx: oneshot::Sender<Box<dyn Future<Output = Result<(), ErrorCode>> + Send>>,
+    },
     /// Body is consumed.
     Consumed,
 }
@@ -72,6 +76,38 @@ pub(crate) struct GuestBody {
         Option<oneshot::Receiver<Result<Option<Arc<http::HeaderMap>>, ErrorCode>>>,
 }
 
+impl GuestBody {
+    pub fn new<T: 'static>(
+        mut store: impl AsContextMut<Data = T>,
+        contents_rx: Option<StreamReader<u8>>,
+        trailers_rx: FutureReader<Result<Option<Resource<Trailers>>, ErrorCode>>,
+        getter: for<'a> fn(&'a mut T) -> WasiHttpCtxView<'a>,
+    ) -> Self {
+        let (trailers_http_tx, trailers_http_rx) = oneshot::channel();
+        trailers_rx.pipe(
+            &mut store,
+            GuestTrailerConsumer {
+                tx: trailers_http_tx,
+                getter,
+            },
+        );
+        let contents_rx = contents_rx.map(|rx| {
+            let (http_tx, http_rx) = mpsc::channel(1);
+            rx.pipe(
+                store,
+                GuestBodyConsumer {
+                    tx: PollSender::new(http_tx),
+                },
+            );
+            http_rx
+        });
+        Self {
+            trailers_rx: Some(trailers_http_rx),
+            contents_rx,
+        }
+    }
+}
+
 impl http_body::Body for GuestBody {
     type Data = Bytes;
     type Error = ErrorCode;
@@ -181,3 +217,41 @@ where
         }
     }
 }
+
+pub(crate) struct IncomingResponseBody {
+    pub incoming: hyper::body::Incoming,
+    pub timeout: tokio::time::Interval,
+}
+
+impl http_body::Body for IncomingResponseBody {
+    type Data = <hyper::body::Incoming as http_body::Body>::Data;
+    type Error = ErrorCode;
+
+    fn poll_frame(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
+        match Pin::new(&mut self.as_mut().incoming).poll_frame(cx) {
+            Poll::Ready(None) => Poll::Ready(None),
+            Poll::Ready(Some(Err(err))) => {
+                Poll::Ready(Some(Err(ErrorCode::from_hyper_response_error(err))))
+            }
+            Poll::Ready(Some(Ok(frame))) => {
+                self.timeout.reset();
+                Poll::Ready(Some(Ok(frame)))
+            }
+            Poll::Pending => {
+                ready!(self.timeout.poll_tick(cx));
+                Poll::Ready(Some(Err(ErrorCode::ConnectionReadTimeout)))
+            }
+        }
+    }
+
+    fn is_end_stream(&self) -> bool {
+        self.incoming.is_end_stream()
+    }
+
+    fn size_hint(&self) -> http_body::SizeHint {
+        self.incoming.size_hint()
+    }
+}

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

@@ -1,12 +1,48 @@
 use crate::p3::bindings::http::types::{ErrorCode, Method, Scheme};
 use core::convert::Infallible;
+use core::error::Error as _;
+use tracing::warn;
 
 impl From<Infallible> for ErrorCode {
     fn from(x: Infallible) -> Self {
         match x {}
     }
 }
 
+impl ErrorCode {
+    /// Translate a [`hyper::Error`] to a wasi-http [ErrorCode] in the context of a request.
+    pub fn from_hyper_request_error(err: hyper::Error) -> Self {
+        // If there's a source, we might be able to extract a wasi-http error from it.
+        if let Some(cause) = err.source() {
+            if let Some(err) = cause.downcast_ref::<Self>() {
+                return err.clone();
+            }
+        }
+
+        warn!("hyper request error: {err:?}");
+
+        Self::HttpProtocolError
+    }
+
+    /// Translate a [`hyper::Error`] to a wasi-http [ErrorCode] in the context of a response.
+    pub fn from_hyper_response_error(err: hyper::Error) -> Self {
+        if err.is_timeout() {
+            return ErrorCode::HttpResponseTimeout;
+        }
+
+        // If there's a source, we might be able to extract a wasi-http error from it.
+        if let Some(cause) = err.source() {
+            if let Some(err) = cause.downcast_ref::<Self>() {
+                return err.clone();
+            }
+        }
+
+        warn!("hyper response error: {err:?}");
+
+        ErrorCode::HttpProtocolError
+    }
+}
+
 impl From<http::Method> for Method {
     fn from(method: http::Method) -> Self {
         Self::from(&method)

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

@@ -1,15 +1,116 @@
 use crate::p3::bindings::http::handler::{Host, HostWithStore};
-use crate::p3::bindings::http::types::{Request, Response};
-use crate::p3::{HttpResult, WasiHttp, WasiHttpCtxView};
+use crate::p3::bindings::http::types::{ErrorCode, Request, Response};
+use crate::p3::body::{Body, ConsumedBody, GuestBody};
+use crate::p3::host::{delete_request, push_response};
+use crate::p3::{HttpError, HttpResult, WasiHttp, WasiHttpCtxView};
+use http::header::HOST;
+use http::{HeaderValue, Uri};
+use http_body_util::BodyExt as _;
+use std::sync::Arc;
+use tokio::sync::oneshot;
+use tracing::debug;
 use wasmtime::component::{Accessor, Resource};
 
 impl HostWithStore for WasiHttp {
-    #[expect(unused, reason = "work in progress")] // TODO: implement
     async fn handle<T>(
         store: &Accessor<T, Self>,
         req: Resource<Request>,
     ) -> HttpResult<Resource<Response>> {
-        todo!()
+        let getter = store.getter();
+        let (res_result_tx, res_result_rx) = oneshot::channel();
+        let (fut, req_result_tx) = store.with(|mut store| {
+            let WasiHttpCtxView { ctx, table } = store.get();
+            let Request {
+                method,
+                scheme,
+                authority,
+                path_with_query,
+                headers,
+                options,
+                body,
+            } = delete_request(table, req).map_err(HttpError::trap)?;
+            let mut headers = Arc::unwrap_or_clone(headers);
+            if ctx.set_host_header() {
+                let host = if let Some(authority) = authority.as_ref() {
+                    HeaderValue::try_from(authority.as_str())
+                        .map_err(|err| ErrorCode::InternalError(Some(err.to_string())))?
+                } else {
+                    HeaderValue::from_static("")
+                };
+                headers.insert(HOST, host);
+            }
+
+            let scheme = match scheme {
+                None => ctx.default_scheme().ok_or(ErrorCode::HttpProtocolError)?,
+                Some(scheme) if ctx.is_supported_scheme(&scheme) => scheme,
+                Some(..) => return Err(ErrorCode::HttpProtocolError.into()),
+            };
+            let mut uri = Uri::builder().scheme(scheme);
+            if let Some(authority) = authority {
+                uri = uri.authority(authority)
+            };
+            if let Some(path_with_query) = path_with_query {
+                uri = uri.path_and_query(path_with_query)
+            };
+            let uri = uri.build().map_err(|err| {
+                debug!(?err, "failed to build request URI");
+                ErrorCode::HttpRequestUriInvalid
+            })?;
+
+            let mut req = http::Request::builder();
+            *req.headers_mut().unwrap() = headers;
+            let (body, result_tx) = match body {
+                Body::Guest {
+                    contents_rx,
+                    trailers_rx,
+                    result_tx,
+                } => (
+                    GuestBody::new(&mut store, contents_rx, trailers_rx, getter).boxed(),
+                    Some(result_tx),
+                ),
+                Body::Host { body, result_tx } => (body, Some(result_tx)),
+                Body::Consumed => (ConsumedBody.boxed(), None),
+            };
+            let req = req
+                .method(method)
+                .uri(uri)
+                .body(body)
+                .map_err(|err| ErrorCode::InternalError(Some(err.to_string())))?;
+            HttpResult::Ok((
+                store.get().ctx.send_request(
+                    req,
+                    options.as_deref().copied(),
+                    Box::new(async {
+                        let Ok(fut) = res_result_rx.await else {
+                            return Ok(());
+                        };
+                        Box::into_pin(fut).await
+                    }),
+                ),
+                result_tx,
+            ))
+        })?;
+        let (res, io) = Box::into_pin(fut).await?;
+        if let Some(req_result_tx) = req_result_tx {
+            _ = req_result_tx.send(io);
+        } else {
+            Box::into_pin(io).await?;
+        }
+        let (
+            http::response::Parts {
+                status, headers, ..
+            },
+            body,
+        ) = res.into_parts();
+        let res = Response {
+            status,
+            headers: Arc::new(headers),
+            body: Body::Host {
+                body,
+                result_tx: res_result_tx,
+            },
+        };
+        store.with(|mut store| push_response(store.get().table, res).map_err(HttpError::trap))
     }
 }