fix(p3-http): close stream handles on drop

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

Author: rvolosatovs

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

@@ -25,11 +25,9 @@ impl test_programs::p3::exports::wasi::cli::run::Guest for Component {
             request
                 .set_authority(Some("www.example.com"))
                 .expect("setting authority");
-            let (remaining, ()) =
-                futures::join!(contents_tx.write_all(b"request-body".to_vec()), async {
-                    drop(request);
-                },);
-            assert!(!remaining.is_empty());
+            drop(request);
+            let remaining = contents_tx.write_all(b"request-body".to_vec()).await;
+            assert_eq!(String::from_utf8_lossy(&remaining), "request-body");
         }
         {
             let headers = Headers::from_list(&[(

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

@@ -7,6 +7,8 @@ mod generated {
         world: "wasi:http/proxy",
         imports: {
             "wasi:http/handler/[async]handle": async | store | trappable | tracing,
+            "wasi:http/types/[drop]request": store | trappable | tracing,
+            "wasi:http/types/[drop]response": store | trappable | tracing,
             "wasi:http/types/[method]request.consume-body": async | store | trappable | tracing,
             "wasi:http/types/[method]response.consume-body": async | store | trappable | tracing,
             "wasi:http/types/[static]request.new": async | store | trappable | tracing,

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

@@ -36,6 +36,22 @@ pub(crate) enum Body {
     Consumed,
 }
 
+impl Body {
+    pub(crate) fn drop(self, mut store: impl AsContextMut) {
+        if let Body::Guest {
+            contents_rx,
+            mut trailers_rx,
+            ..
+        } = self
+        {
+            if let Some(mut contents_rx) = contents_rx {
+                contents_rx.close(&mut store);
+            }
+            trailers_rx.close(store);
+        }
+    }
+}
+
 pub(crate) enum GuestBodyKind {
     Request,
     Response,

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

@@ -19,11 +19,11 @@ use http_body_util::combinators::BoxBody;
 use std::io::Cursor;
 use std::sync::Arc;
 use tokio::sync::oneshot;
-use wasmtime::StoreContextMut;
 use wasmtime::component::{
     Accessor, Destination, FutureProducer, FutureReader, Resource, ResourceTable, StreamProducer,
     StreamReader, StreamResult,
 };
+use wasmtime::{StoreContextMut, component::Access};
 use wasmtime_wasi::p3::{FutureOneshotProducer, StreamEmptyProducer};
 
 fn get_fields<'a>(
@@ -481,6 +481,16 @@ impl HostRequestWithStore for WasiHttp {
             }
         })
     }
+
+    fn drop<T>(mut store: Access<'_, T, Self>, req: Resource<Request>) -> wasmtime::Result<()> {
+        let Request { body, .. } = store
+            .get()
+            .table
+            .delete(req)
+            .context("failed to delete request from table")?;
+        body.drop(store);
+        Ok(())
+    }
 }
 
 impl HostRequest for WasiHttpCtxView<'_> {
@@ -594,13 +604,6 @@ impl HostRequest for WasiHttpCtxView<'_> {
         let Request { headers, .. } = get_request(self.table, &req)?;
         push_fields(self.table, Fields::new_immutable(Arc::clone(headers)))
     }
-
-    fn drop(&mut self, req: Resource<Request>) -> wasmtime::Result<()> {
-        self.table
-            .delete(req)
-            .context("failed to delete request from table")?;
-        Ok(())
-    }
 }
 
 impl HostRequestOptions for WasiHttpCtxView<'_> {
@@ -799,6 +802,16 @@ impl HostResponseWithStore for WasiHttp {
             }
         })
     }
+
+    fn drop<T>(mut store: Access<'_, T, Self>, res: Resource<Response>) -> wasmtime::Result<()> {
+        let Response { body, .. } = store
+            .get()
+            .table
+            .delete(res)
+            .context("failed to delete response from table")?;
+        body.drop(store);
+        Ok(())
+    }
 }
 
 impl HostResponse for WasiHttpCtxView<'_> {
@@ -824,13 +837,6 @@ impl HostResponse for WasiHttpCtxView<'_> {
         let Response { headers, .. } = get_response(self.table, &res)?;
         push_fields(self.table, Fields::new_immutable(Arc::clone(headers)))
     }
-
-    fn drop(&mut self, res: Resource<Response>) -> wasmtime::Result<()> {
-        self.table
-            .delete(res)
-            .context("failed to delete response from table")?;
-        Ok(())
-    }
 }
 
 impl Host for WasiHttpCtxView<'_> {

crates/wasi-http/tests/all/p3/mod.rs

@@ -216,7 +216,6 @@ async fn p3_http_outbound_request_invalid_dnsname() -> anyhow::Result<()> {
     run_cli(P3_HTTP_OUTBOUND_REQUEST_INVALID_DNSNAME_COMPONENT, &server).await
 }
 
-#[ignore = "unimplemented"] // TODO: implement
 #[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn p3_http_outbound_request_response_build() -> anyhow::Result<()> {
     let server = Server::http1(1)?;