Wasi-http: support inbound requests (proxy world)

Cargo.toml

@@ -48,6 +48,12 @@ serde_json = { workspace = true }
 wasmparser = { workspace = true }
 wasm-encoder = { workspace = true }
 
+async-trait = { workspace = true }
+tokio = { workspace = true, optional = true, features = [ "signal", "macros" ] }
+hyper = { workspace = true, optional = true }
+http-body = { workspace = true }
+http-body-util = { workspace = true }
+
 [target.'cfg(unix)'.dependencies]
 rustix = { workspace = true, features = ["mm", "param"] }
 
@@ -60,7 +66,7 @@ filecheck = { workspace = true }
 tempfile = { workspace = true }
 test-programs = { path = "crates/test-programs" }
 wasmtime-runtime = { workspace = true }
-tokio = { version = "1.8.0", features = ["rt", "time", "macros", "rt-multi-thread"] }
+tokio = { workspace = true, features = ["rt", "time", "macros", "rt-multi-thread"] }
 wast = { workspace = true }
 criterion = "0.5.0"
 num_cpus = "1.13.0"
@@ -101,6 +107,7 @@ members = [
   "crates/jit-icache-coherence",
   "crates/test-programs/wasi-tests",
   "crates/test-programs/wasi-http-tests",
+  "crates/test-programs/wasi-http-proxy-tests",
   "crates/test-programs/wasi-sockets-tests",
   "crates/test-programs/command-tests",
   "crates/test-programs/reactor-tests",
@@ -251,7 +258,11 @@ tempfile = "3.1.0"
 filecheck = "0.5.0"
 libc = "0.2.60"
 file-per-thread-logger = "0.2.0"
-tokio = { version = "1.26.0" }
+tokio = { version = "1.26.0", features = [ "rt", "time" ] }
+hyper = "=1.0.0-rc.3"
+http = "0.2.9"
+http-body = "1.0.0-rc.2"
+http-body-util = "0.1.0-rc.2"
 bytes = "1.4"
 futures = { version = "0.3.27", default-features = false }
 indexmap = "2.0.0"
@@ -275,7 +286,7 @@ jitdump = ["wasmtime/jitdump"]
 vtune = ["wasmtime/vtune"]
 wasi-nn = ["dep:wasmtime-wasi-nn"]
 wasi-threads = ["dep:wasmtime-wasi-threads"]
-wasi-http = ["dep:wasmtime-wasi-http", "wasmtime-wasi-http?/sync"]
+wasi-http = ["component-model", "dep:wasmtime-wasi-http", "dep:tokio", "dep:hyper", "wasmtime-wasi-http?/sync"]
 pooling-allocator = ["wasmtime/pooling-allocator", "wasmtime-cli-flags/pooling-allocator"]
 all-arch = ["wasmtime/all-arch"]
 component-model = [

crates/test-programs/Cargo.toml

@@ -15,6 +15,7 @@ heck = { workspace = true }
 
 [dependencies]
 anyhow = { workspace = true }
+bytes = { workspace = true }
 http = { version = "0.2.9" }
 http-body = "1.0.0-rc.2"
 http-body-util = "0.1.0-rc.2"

crates/test-programs/build.rs

@@ -33,6 +33,7 @@ fn build_and_generate_tests() {
     println!("cargo:rerun-if-changed=./wasi-sockets-tests");
     if BUILD_WASI_HTTP_TESTS {
         println!("cargo:rerun-if-changed=./wasi-http-tests");
+        println!("cargo:rerun-if-changed=./wasi-http-proxy-tests");
     } else {
         println!("cargo:rustc-cfg=skip_wasi_http_tests");
     }
@@ -50,6 +51,7 @@ fn build_and_generate_tests() {
         .env_remove("CARGO_ENCODED_RUSTFLAGS");
     if BUILD_WASI_HTTP_TESTS {
         cmd.arg("--package=wasi-http-tests");
+        cmd.arg("--package=wasi-http-proxy-tests");
     }
     let status = cmd.status().unwrap();
     assert!(status.success());
@@ -60,8 +62,14 @@ fn build_and_generate_tests() {
     components_rs(&meta, "wasi-tests", "bin", &command_adapter, &out_dir);
 
     if BUILD_WASI_HTTP_TESTS {
-        modules_rs(&meta, "wasi-http-tests", "bin", &out_dir);
         components_rs(&meta, "wasi-http-tests", "bin", &command_adapter, &out_dir);
+        components_rs(
+            &meta,
+            "wasi-http-proxy-tests",
+            "cdylib",
+            &reactor_adapter,
+            &out_dir,
+        );
     }
 
     components_rs(&meta, "command-tests", "bin", &command_adapter, &out_dir);

crates/test-programs/tests/wasi-http-components-sync.rs

@@ -4,9 +4,8 @@ use wasmtime::{
     Config, Engine, Store,
 };
 use wasmtime_wasi::preview2::{
-    command::sync::{add_to_linker, Command},
-    pipe::MemoryOutputPipe,
-    IsATTY, Table, WasiCtx, WasiCtxBuilder, WasiView,
+    command::sync::Command, pipe::MemoryOutputPipe, IsATTY, Table, WasiCtx, WasiCtxBuilder,
+    WasiView,
 };
 use wasmtime_wasi_http::{WasiHttpCtx, WasiHttpView};
 
@@ -60,8 +59,7 @@ fn instantiate_component(
     ctx: Ctx,
 ) -> Result<(Store<Ctx>, Command), anyhow::Error> {
     let mut linker = Linker::new(&ENGINE);
-    add_to_linker(&mut linker)?;
-    wasmtime_wasi_http::proxy::add_to_linker(&mut linker)?;
+    wasmtime_wasi_http::proxy::sync::add_to_linker(&mut linker)?;
 
     let mut store = Store::new(&ENGINE, ctx);
 

crates/test-programs/tests/wasi-http-components.rs

@@ -4,9 +4,7 @@ use wasmtime::{
     Config, Engine, Store,
 };
 use wasmtime_wasi::preview2::{
-    command::{add_to_linker, Command},
-    pipe::MemoryOutputPipe,
-    IsATTY, Table, WasiCtx, WasiCtxBuilder, WasiView,
+    command::Command, pipe::MemoryOutputPipe, IsATTY, Table, WasiCtx, WasiCtxBuilder, WasiView,
 };
 use wasmtime_wasi_http::{WasiHttpCtx, WasiHttpView};
 
@@ -60,7 +58,6 @@ async fn instantiate_component(
     ctx: Ctx,
 ) -> Result<(Store<Ctx>, Command), anyhow::Error> {
     let mut linker = Linker::new(&ENGINE);
-    add_to_linker(&mut linker)?;
     wasmtime_wasi_http::proxy::add_to_linker(&mut linker)?;
 
     let mut store = Store::new(&ENGINE, ctx);

crates/test-programs/tests/wasi-http-proxy.rs

@@ -0,0 +1,146 @@
+#![cfg(all(feature = "test_programs", not(skip_wasi_http_tests)))]
+use anyhow::Context;
+use wasmtime::{
+    component::{Component, Linker},
+    Config, Engine, Store,
+};
+use wasmtime_wasi::preview2::{
+    self, pipe::MemoryOutputPipe, IsATTY, Table, WasiCtx, WasiCtxBuilder, WasiView,
+};
+use wasmtime_wasi_http::{proxy::Proxy, WasiHttpCtx, WasiHttpView};
+
+lazy_static::lazy_static! {
+    static ref ENGINE: Engine = {
+        let mut config = Config::new();
+        config.wasm_backtrace_details(wasmtime::WasmBacktraceDetails::Enable);
+        config.wasm_component_model(true);
+        config.async_support(true);
+        let engine = Engine::new(&config).unwrap();
+        engine
+    };
+}
+// uses ENGINE, creates a fn get_module(&str) -> Module
+include!(concat!(
+    env!("OUT_DIR"),
+    "/wasi_http_proxy_tests_components.rs"
+));
+
+struct Ctx {
+    table: Table,
+    wasi: WasiCtx,
+    http: WasiHttpCtx,
+}
+
+impl WasiView for Ctx {
+    fn table(&self) -> &Table {
+        &self.table
+    }
+    fn table_mut(&mut self) -> &mut Table {
+        &mut self.table
+    }
+    fn ctx(&self) -> &WasiCtx {
+        &self.wasi
+    }
+    fn ctx_mut(&mut self) -> &mut WasiCtx {
+        &mut self.wasi
+    }
+}
+
+impl WasiHttpView for Ctx {
+    fn table(&mut self) -> &mut Table {
+        &mut self.table
+    }
+    fn ctx(&mut self) -> &mut WasiHttpCtx {
+        &mut self.http
+    }
+}
+
+async fn instantiate(component: Component, ctx: Ctx) -> Result<(Store<Ctx>, Proxy), anyhow::Error> {
+    let mut linker = Linker::new(&ENGINE);
+    wasmtime_wasi_http::proxy::add_to_linker(&mut linker)?;
+
+    let mut store = Store::new(&ENGINE, ctx);
+
+    let (proxy, _instance) = Proxy::instantiate_async(&mut store, &component, &linker).await?;
+    Ok((store, proxy))
+}
+
+#[test_log::test(tokio::test)]
+async fn wasi_http_proxy_tests() -> anyhow::Result<()> {
+    let stdout = MemoryOutputPipe::new(4096);
+    let stderr = MemoryOutputPipe::new(4096);
+
+    let mut table = Table::new();
+    let component = get_component("wasi_http_proxy_tests");
+
+    // Create our wasi context.
+    let mut builder = WasiCtxBuilder::new();
+    builder.stdout(stdout.clone(), IsATTY::No);
+    builder.stderr(stderr.clone(), IsATTY::No);
+    for (var, val) in test_programs::wasi_tests_environment() {
+        builder.env(var, val);
+    }
+    let wasi = builder.build(&mut table)?;
+    let http = WasiHttpCtx;
+
+    let ctx = Ctx { table, wasi, http };
+
+    let (mut store, proxy) = instantiate(component, ctx).await?;
+
+    let req = {
+        use http_body_util::{BodyExt, Empty};
+
+        let req = hyper::Request::builder().method(http::Method::GET).body(
+            Empty::<bytes::Bytes>::new()
+                .map_err(|e| anyhow::anyhow!(e))
+                .boxed(),
+        )?;
+        store.data_mut().new_incoming_request(req)?
+    };
+
+    let (sender, receiver) = tokio::sync::oneshot::channel();
+    let out = store.data_mut().new_response_outparam(sender)?;
+
+    let handle = preview2::spawn(async move {
+        proxy
+            .wasi_http_incoming_handler()
+            .call_handle(&mut store, req, out)
+            .await?;
+
+        Ok::<_, anyhow::Error>(())
+    });
+
+    let resp = match receiver.await {
+        Ok(Ok(resp)) => {
+            use http_body_util::BodyExt;
+            let (parts, body) = resp.into_parts();
+            let collected = BodyExt::collect(body).await?;
+            Ok(hyper::Response::from_parts(parts, collected))
+        }
+
+        Ok(Err(e)) => Err(e),
+
+        // This happens if the wasm never calls `set-response-outparam`
+        Err(e) => panic!("Failed to receive a response: {e:?}"),
+    };
+
+    // Now that the response has been processed, we can wait on the wasm to finish without
+    // deadlocking.
+    handle.await.context("Component execution")?;
+
+    let stdout = stdout.contents();
+    if !stdout.is_empty() {
+        println!("[guest] stdout:\n{}\n===", String::from_utf8_lossy(&stdout));
+    }
+    let stderr = stderr.contents();
+    if !stderr.is_empty() {
+        println!("[guest] stderr:\n{}\n===", String::from_utf8_lossy(&stderr));
+    }
+
+    match resp {
+        Ok(resp) => println!("response: {resp:?}"),
+        Err(e) => panic!("Error given in response: {e:?}"),
+    };
+
+    Ok(())
+}

crates/test-programs/wasi-http-proxy-tests/Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "wasi-http-proxy-tests"
+version = "0.0.0"
+readme = "README.md"
+edition = "2021"
+publish = false
+
+[lib]
+crate-type=["cdylib"]
+
+[dependencies]
+wit-bindgen = { workspace = true, features = ["macros", "realloc"] }

crates/test-programs/wasi-http-proxy-tests/src/lib.rs

@@ -0,0 +1,33 @@
+pub mod bindings {
+    use super::T;
+
+    wit_bindgen::generate!({
+        path: "../../wasi-http/wit",
+        world: "wasi:http/proxy",
+        exports: {
+            "wasi:http/incoming-handler": T,
+        },
+    });
+}
+
+use bindings::wasi::http::types::{IncomingRequest, ResponseOutparam};
+
+struct T;
+
+impl bindings::exports::wasi::http::incoming_handler::Guest for T {
+    fn handle(_request: IncomingRequest, outparam: ResponseOutparam) {
+        let hdrs = bindings::wasi::http::types::new_fields(&[]);
+        let resp = bindings::wasi::http::types::new_outgoing_response(200, hdrs);
+        let body =
+            bindings::wasi::http::types::outgoing_response_write(resp).expect("outgoing response");
+
+        bindings::wasi::http::types::set_response_outparam(outparam, Ok(resp));
+
+        let out = bindings::wasi::http::types::outgoing_body_write(body).expect("outgoing stream");
+        bindings::wasi::io::streams::blocking_write_and_flush(out, b"hello, world!")
+            .expect("writing response");
+
+        bindings::wasi::io::streams::drop_output_stream(out);
+        bindings::wasi::http::types::outgoing_body_finish(body, None);
+    }
+}

crates/test-programs/wasi-http-tests/src/lib.rs

@@ -119,7 +119,7 @@ pub async fn request(
     // TODO: The current implementation requires this drop after the request is sent.
     // The ownership semantics are unclear in wasi-http we should clarify exactly what is
     // supposed to happen here.
-    http_types::drop_outgoing_body(outgoing_body);
+    http_types::outgoing_body_finish(outgoing_body, None);
 
     let incoming_response = match http_types::future_incoming_response_get(future_response) {
         Some(result) => result.map_err(|_| anyhow!("incoming response errored"))?,