implement new stream/future trap conditions

As of WebAssembly/component-model#524, we now trap if a
guest tries to use a stream or future handle (besides dropping it) after it is
"done" (see that PR for what "done" means).  This required more careful tracking
of the state of each end of the stream or future.

In the process of debugging this into shape, I added some trace logging and
addressed a couple of issues I noticed:

- `wasi-http`'s `Response::into_http` wasn't noticing a non-empty buffer at the end of a stream
- the `p3_http_middleware` test was not clearing its buffer before each read

Signed-off-by: Joel Dice <[email protected]>

Author: dicej

Cargo.lock

@@ -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' }

Cargo.toml

@@ -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/test-programs/src/bin/p3_http_middleware.rs

@@ -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/Cargo.toml

@@ -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/src/p3/response.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/outgoing.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-http/tests/all/p3/proxy.rs

@@ -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/Cargo.toml

@@ -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/wasi/tests/all/p3/mod.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!(),
                 };
             }