feat(p3): begin wasi:http implementation (#11439)

* doc: fix typo in p3 filesystem ref

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

* build: vendor `wasi:http@0.3.0` WIT

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

* feat(p3-http): scaffold p3 implementation

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

* test(p3-http): add tests

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

* chore(p3-http): add utils used by p3 tests

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

* chore: add `expect` reasons

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

* address review comments

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

* establish `conn` connections on drop

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

---------

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

Author: rvolosatovs

Cargo.lock

@@ -412,6 +412,7 @@ webpki-roots = "0.26.0"
 itertools = "0.14.0"
 base64 = "0.22.1"
 termcolor = "1.4.1"
+flate2 = "1.0.30"
 
 # =============================================================================
 #

Cargo.toml

@@ -76,6 +76,15 @@ make_vendor "wasi/src/p3" "
     sockets@bc1b8b1@wit-0.3.0-draft
 "
 
+make_vendor "wasi-http/src/p3" "
+    cli@939bd6d@wit-0.3.0-draft
+    clocks@13d1c82@wit-0.3.0-draft
+    filesystem@2007d36@wit-0.3.0-draft
+    http@376b674@wit-0.3.0-draft
+    random@4e94663@wit-0.3.0-draft
+    sockets@bc1b8b1@wit-0.3.0-draft
+"
+
 rm -rf $cache_dir
 
 # Separately (for now), vendor the `wasi-nn` WIT files since their retrieval is

ci/vendor-wit.sh

@@ -0,0 +1,64 @@
+use futures::join;
+use test_programs::p3::wasi::http::types::{ErrorCode, Headers, Request, Response};
+use test_programs::p3::{wit_future, wit_stream};
+use wit_bindgen_rt::async_support::spawn;
+
+struct T;
+
+test_programs::p3::proxy::export!(T);
+
+impl test_programs::p3::proxy::exports::wasi::http::handler::Guest for T {
+    async fn handle(request: Request) -> Result<Response, ErrorCode> {
+        assert!(request.get_scheme().is_some());
+        assert!(request.get_authority().is_some());
+        assert!(request.get_path_with_query().is_some());
+
+        // TODO: adapt below
+        //test_filesystem();
+
+        let header = String::from("custom-forbidden-header");
+        let req_hdrs = request.get_headers();
+
+        assert!(
+            !req_hdrs.has(&header),
+            "forbidden `custom-forbidden-header` found in request"
+        );
+
+        assert!(req_hdrs.delete(&header).is_err());
+        assert!(req_hdrs.append(&header, b"no".as_ref()).is_err());
+
+        assert!(
+            !req_hdrs.has(&header),
+            "append of forbidden header succeeded"
+        );
+
+        let hdrs = Headers::new();
+        let (mut contents_tx, contents_rx) = wit_stream::new();
+        let (trailers_tx, trailers_rx) = wit_future::new(|| todo!());
+        let (resp, transmit) = Response::new(hdrs, Some(contents_rx), trailers_rx);
+        spawn(async {
+            join!(
+                async {
+                    let remaining = contents_tx.write_all(b"hello, world!".to_vec()).await;
+                    assert!(remaining.is_empty());
+                    drop(contents_tx);
+                    trailers_tx
+                        .write(Ok(None))
+                        .await
+                        .expect("failed to write trailers");
+                },
+                async { transmit.await.unwrap() }
+            );
+        });
+        Ok(resp)
+    }
+}
+
+// Technically this should not be here for a proxy, but given the current
+// framework for tests it's required since this file is built as a `bin`
+fn main() {}
+
+// TODO: adapt below
+//fn test_filesystem() {
+//    assert!(std::fs::File::open(".").is_err());
+//}

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

@@ -0,0 +1,61 @@
+use {
+    test_programs::p3::{
+        proxy::exports::wasi::http::handler::Guest as Handler,
+        wasi::http::types::{ErrorCode, Request, Response},
+        wit_future, wit_stream,
+    },
+    wit_bindgen_rt::async_support::{self, StreamResult},
+};
+
+struct Component;
+
+test_programs::p3::proxy::export!(Component);
+
+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 (body, trailers) = request.consume_body().unwrap();
+
+        // let (headers, body) = Request::into_parts(request);
+
+        let (response, _result) = if false {
+            // This is the easy and efficient way to do it...
+            Response::new(headers, Some(body), trailers)
+        } else {
+            // ...but we do it the more difficult, less efficient way here to exercise various component model
+            // features (e.g. `future`s, `stream`s, and post-return asynchronous execution):
+            let (trailers_tx, trailers_rx) = wit_future::new(|| todo!());
+            let (mut pipe_tx, pipe_rx) = wit_stream::new();
+
+            async_support::spawn(async move {
+                let mut body_rx = body;
+                let mut chunk = Vec::with_capacity(1024);
+                loop {
+                    let (status, buf) = body_rx.read(chunk).await;
+                    chunk = buf;
+                    match status {
+                        StreamResult::Complete(_) => {
+                            chunk = pipe_tx.write_all(chunk).await;
+                            assert!(chunk.is_empty());
+                        }
+                        StreamResult::Dropped => break,
+                        StreamResult::Cancelled => unreachable!(),
+                    }
+                }
+
+                drop(pipe_tx);
+
+                trailers_tx.write(trailers.await).await.unwrap();
+                drop(request);
+            });
+
+            Response::new(headers, Some(pipe_rx), trailers_rx)
+        };
+
+        Ok(response)
+    }
+}
+
+// Unused function; required since this file is built as a `bin`:
+fn main() {}

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

@@ -0,0 +1,163 @@
+use {
+    flate2::{
+        Compression,
+        write::{DeflateDecoder, DeflateEncoder},
+    },
+    std::{io::Write, mem},
+    test_programs::p3::{
+        proxy::exports::wasi::http::handler::Guest as Handler,
+        wasi::http::{
+            handler,
+            types::{ErrorCode, Headers, Request, Response},
+        },
+        wit_future, wit_stream,
+    },
+    wit_bindgen_rt::async_support::{self, StreamResult},
+};
+
+struct Component;
+
+test_programs::p3::proxy::export!(Component);
+
+impl Handler for Component {
+    /// Forward the specified request to the imported `wasi:http/handler`, transparently decoding the request body
+    /// if it is `deflate`d and then encoding the response body if the client has provided an `accept-encoding:
+    /// deflate` header.
+    async fn handle(request: Request) -> Result<Response, ErrorCode> {
+        // First, extract the parts of the request and check for (and remove) headers pertaining to body encodings.
+        let method = request.get_method();
+        let scheme = request.get_scheme();
+        let path_with_query = request.get_path_with_query();
+        let authority = request.get_authority();
+        let mut accept_deflated = false;
+        let mut content_deflated = false;
+        let headers = request.get_headers();
+        let mut headers = headers.copy_all();
+        headers.retain(|(k, v)| match (k.as_str(), v.as_slice()) {
+            ("accept-encoding", value)
+                if std::str::from_utf8(value)
+                    .map(|v| v.contains("deflate"))
+                    .unwrap_or(false) =>
+            {
+                accept_deflated = true;
+                false
+            }
+            ("content-encoding", b"deflate") => {
+                content_deflated = true;
+                false
+            }
+            _ => true,
+        });
+        let (mut body, trailers) = request.consume_body().unwrap();
+
+        let (body, trailers) = if content_deflated {
+            // Next, spawn a task to pipe and decode the original request body and trailers into a new request
+            // we'll create below.  This will run concurrently with any code in the imported `wasi:http/handler`.
+            let (trailers_tx, trailers_rx) = wit_future::new(|| todo!());
+            let (mut pipe_tx, pipe_rx) = wit_stream::new();
+
+            async_support::spawn(async move {
+                {
+                    let mut decoder = DeflateDecoder::new(Vec::new());
+
+                    let (mut status, mut chunk) = body.read(Vec::with_capacity(64 * 1024)).await;
+                    while let StreamResult::Complete(_) = status {
+                        decoder.write_all(&chunk).unwrap();
+                        let remaining = pipe_tx.write_all(mem::take(decoder.get_mut())).await;
+                        assert!(remaining.is_empty());
+                        *decoder.get_mut() = remaining;
+                        chunk.clear();
+                        (status, chunk) = body.read(chunk).await;
+                    }
+
+                    let remaining = pipe_tx.write_all(decoder.finish().unwrap()).await;
+                    assert!(remaining.is_empty());
+
+                    drop(pipe_tx);
+                }
+
+                trailers_tx.write(trailers.await).await.unwrap();
+
+                drop(request);
+            });
+
+            (pipe_rx, trailers_rx)
+        } else {
+            (body, trailers)
+        };
+
+        // 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(
+            Headers::from_list(&headers).unwrap(),
+            Some(body),
+            trailers,
+            None,
+        );
+        my_request.set_method(&method).unwrap();
+        my_request.set_scheme(scheme.as_ref()).unwrap();
+        my_request
+            .set_path_with_query(path_with_query.as_deref())
+            .unwrap();
+        my_request.set_authority(authority.as_deref()).unwrap();
+
+        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.
+        let status_code = response.get_status_code();
+        let mut headers = response.get_headers().copy_all();
+        if accept_deflated {
+            headers.push(("content-encoding".into(), b"deflate".into()));
+        }
+
+        let (mut body, trailers) = response.consume_body().unwrap();
+        let (body, trailers) = if accept_deflated {
+            headers.retain(|(name, _value)| name != "content-length");
+
+            // Spawn another task; this one is to pipe and encode the original response body and trailers into a
+            // new response we'll create below.  This will run concurrently with the caller's code (i.e. it won't
+            // necessarily complete before we return a value).
+            let (trailers_tx, trailers_rx) = wit_future::new(|| todo!());
+            let (mut pipe_tx, pipe_rx) = wit_stream::new();
+
+            async_support::spawn(async move {
+                {
+                    let mut encoder = DeflateEncoder::new(Vec::new(), Compression::fast());
+                    let (mut status, mut chunk) = body.read(Vec::with_capacity(64 * 1024)).await;
+
+                    while let StreamResult::Complete(_) = status {
+                        encoder.write_all(&chunk).unwrap();
+                        let remaining = pipe_tx.write_all(mem::take(encoder.get_mut())).await;
+                        assert!(remaining.is_empty());
+                        *encoder.get_mut() = remaining;
+                        chunk.clear();
+                        (status, chunk) = body.read(chunk).await;
+                    }
+
+                    let remaining = pipe_tx.write_all(encoder.finish().unwrap()).await;
+                    assert!(remaining.is_empty());
+
+                    drop(pipe_tx);
+                }
+
+                trailers_tx.write(trailers.await).await.unwrap();
+                drop(response);
+            });
+
+            (pipe_rx, trailers_rx)
+        } else {
+            (body, trailers)
+        };
+
+        // While the above tasks (if any) are running, synthesize a response from the parts collected above and
+        // return it.
+        let (my_response, _response_complete) =
+            Response::new(Headers::from_list(&headers).unwrap(), Some(body), trailers);
+        my_response.set_status_code(status_code).unwrap();
+
+        Ok(my_response)
+    }
+}
+
+// Unused function; required since this file is built as a `bin`:
+fn main() {}

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

@@ -0,0 +1,51 @@
+mod bindings {
+    wit_bindgen::generate!({
+        path: "../wasi-http/src/p3/wit",
+        world: "local:local/middleware-with-chain",
+        inline: "
+package local:local;
+
+world middleware-with-chain {
+  include wasi:http/proxy@0.3.0-draft;
+
+  import chain-http;
+}
+
+interface chain-http {
+  use wasi:http/types@0.3.0-draft.{request, response, error-code};
+
+  handle: async func(request: request) -> result<response, error-code>;
+}
+        ",
+        generate_all,
+    });
+
+    use super::Component;
+    export!(Component);
+}
+
+use bindings::{
+    exports::wasi::http::handler::Guest as Handler,
+    local::local::chain_http,
+    wasi::clocks::monotonic_clock,
+    wasi::http::types::{ErrorCode, Request, Response},
+};
+use std::time::Duration;
+
+struct Component;
+
+impl Handler for Component {
+    async fn handle(request: Request) -> Result<Response, ErrorCode> {
+        // First, sleep briefly.  This will ensure the next call happens via a
+        // host->guest call to the `wit_bindgen_rt::async_support::callback`
+        // function, which exercises different code paths in both the host and
+        // the guest, which we want to test here.
+        let duration = Duration::from_millis(10);
+        monotonic_clock::wait_for(duration.as_nanos().try_into().unwrap()).await;
+
+        chain_http::handle(request).await
+    }
+}
+
+// Unused function; required since this file is built as a `bin`:
+fn main() {}