implement new stream/future trap conditions

Cargo.toml

@@ -631,9 +631,12 @@ lto = true
 # wasm-wave = { git = "https://github.com/bytecodealliance/wasm-tools" }
 # wasm-compose = { git = "https://github.com/bytecodealliance/wasm-tools" }
 # wasm-metadata = { git = "https://github.com/bytecodealliance/wasm-tools" }
-wit-bindgen = { git = "https://github.com/bytecodealliance/wit-bindgen" }
-wit-bindgen-rt = { git = "https://github.com/bytecodealliance/wit-bindgen" }
-wit-bindgen-rust-macro = { git = "https://github.com/bytecodealliance/wit-bindgen" }
+# wit-bindgen = { git = "https://github.com/bytecodealliance/wit-bindgen" }
+# wit-bindgen-rt = { git = "https://github.com/bytecodealliance/wit-bindgen" }
+# wit-bindgen-rust-macro = { git = "https://github.com/bytecodealliance/wit-bindgen" }
+wit-bindgen = { git = "https://github.com/dicej/wit-bindgen", branch = "no-reads-writes-after-dropped" }
+wit-bindgen-rt = { git = "https://github.com/dicej/wit-bindgen", branch = "no-reads-writes-after-dropped" }
+wit-bindgen-rust-macro = { git = "https://github.com/dicej/wit-bindgen", branch = "no-reads-writes-after-dropped" }
 
 # wasmparser = { path = '../wasm-tools/crates/wasmparser' }
 # wat = { path = '../wasm-tools/crates/wat' }

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

@@ -66,6 +66,7 @@ impl Handler for Component {
                         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;
                     }
 

crates/wasi-http/Cargo.toml

@@ -34,6 +34,7 @@ rustls = { workspace = true }
 webpki-roots = { workspace = true }
 
 [dev-dependencies]
+env_logger = { workspace = true }
 test-programs-artifacts = { workspace = true }
 test-log = { workspace = true }
 tracing-subscriber = { workspace = true }

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

@@ -214,14 +214,14 @@ impl Response {
                 let fut = async move {
                     loop {
                         let (tail, mut rx_buffer) = contents.await;
-                        if let Some(tail) = tail {
-                            let buffer = rx_buffer.split();
-                            if !buffer.is_empty() {
-                                if let Err(..) = contents_tx.send(buffer.freeze()).await {
-                                    break;
-                                }
-                                rx_buffer.reserve(DEFAULT_BUFFER_CAPACITY);
+                        let buffer = rx_buffer.split();
+                        if !buffer.is_empty() {
+                            if let Err(..) = contents_tx.send(buffer.freeze()).await {
+                                break;
                             }
+                            rx_buffer.reserve(DEFAULT_BUFFER_CAPACITY);
+                        }
+                        if let Some(tail) = tail {
                             contents = tail.read(rx_buffer).boxed();
                         } else {
                             debug_assert!(rx_buffer.is_empty());

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

@@ -9,6 +9,8 @@ foreach_p3_http!(assert_test_exists);
 use super::proxy::{p3_http_echo, p3_http_middleware, p3_http_middleware_with_chain};
 
 async fn run(path: &str, server: &Server) -> anyhow::Result<()> {
+    _ = env_logger::try_init();
+
     let engine = test_programs_artifacts::engine(|config| {
         config.wasm_backtrace_details(wasmtime::WasmBacktraceDetails::Enable);
         config.async_support(true);

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

@@ -106,6 +106,8 @@ pub async fn p3_http_middleware_with_chain() -> Result<()> {
 }
 
 async fn test_http_echo(component: &str, use_compression: bool) -> Result<()> {
+    _ = env_logger::try_init();
+
     let body = b"And the mome raths outgrabe";
 
     // Prepare the raw body, optionally compressed if that's what we're

crates/wasi/Cargo.toml

@@ -38,6 +38,7 @@ futures = { workspace = true }
 url = { workspace = true }
 
 [dev-dependencies]
+env_logger = { workspace = true }
 tokio = { workspace = true, features = ["time", "sync", "io-std", "io-util", "rt", "rt-multi-thread", "net", "macros", "fs"] }
 test-log = { workspace = true }
 tracing-subscriber = { workspace = true }

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

@@ -104,6 +104,8 @@ impl WasiSocketsView for Ctx {
 }
 
 async fn run(path: &str) -> anyhow::Result<()> {
+    _ = env_logger::try_init();
+
     let path = Path::new(path);
     let engine = test_programs_artifacts::engine(|config| {
         config.async_support(true);

crates/wasmtime/src/runtime/component/concurrent.rs

@@ -4115,18 +4115,18 @@ impl Waitable {
                 assert_eq!(rep, self.rep());
                 assert_eq!(*state, StreamFutureState::Busy);
                 *state = match event {
-                    Event::FutureRead { .. } => StreamFutureState::Read,
-                    Event::FutureWrite { .. } => StreamFutureState::Write,
+                    Event::FutureRead { .. } => StreamFutureState::Read { done: false },
+                    Event::FutureWrite { .. } => StreamFutureState::Write { done: false },
                     _ => unreachable!(),
                 };
             }
             Event::StreamRead {
                 pending: Some((ty, handle)),
-                ..
+                code,
             }
             | Event::StreamWrite {
                 pending: Some((ty, handle)),
-                ..
+                code,
             } => {
                 let runtime_instance = instance.component().types()[ty].instance;
                 let (rep, WaitableState::Stream(actual_ty, state)) = instance.waitable_tables()
@@ -4139,9 +4139,10 @@ impl Waitable {
                 assert_eq!(*actual_ty, ty);
                 assert_eq!(rep, self.rep());
                 assert_eq!(*state, StreamFutureState::Busy);
+                let done = matches!(code, ReturnCode::Dropped(_));
                 *state = match event {
-                    Event::StreamRead { .. } => StreamFutureState::Read,
-                    Event::StreamWrite { .. } => StreamFutureState::Write,
+                    Event::StreamRead { .. } => StreamFutureState::Read { done },
+                    Event::StreamWrite { .. } => StreamFutureState::Write { done },
                     _ => unreachable!(),
                 };
             }