diff --git a/crates/guest-rust/src/rt/async_support/stream_support.rs b/crates/guest-rust/src/rt/async_support/stream_support.rs index 620ccf003..a622ea44e 100644 --- a/crates/guest-rust/src/rt/async_support/stream_support.rs +++ b/crates/guest-rust/src/rt/async_support/stream_support.rs @@ -402,7 +402,10 @@ where ) -> Result { match ReturnCode::decode(code) { ReturnCode::Blocked => Err(buf), - ReturnCode::Dropped(0) => Ok((StreamResult::Dropped, buf)), + ReturnCode::Dropped(0) => { + self.writer.done = true; + Ok((StreamResult::Dropped, buf)) + } ReturnCode::Cancelled(0) => Ok((StreamResult::Cancelled, buf)), code @ (ReturnCode::Completed(amt) | ReturnCode::Dropped(amt) @@ -648,7 +651,10 @@ unsafe impl<'a, O: StreamOps> WaitableOp for StreamReadOp<'a, O> { ReturnCode::Blocked => Err((buf, cleanup)), // Note that the `cleanup`, if any, is discarded here. - ReturnCode::Dropped(0) => Ok((StreamResult::Dropped, buf)), + ReturnCode::Dropped(0) => { + self.reader.done = true; + Ok((StreamResult::Dropped, buf)) + } // When an in-progress read is successfully cancelled then the // allocation that was being read into, if any, is just discarded. diff --git a/tests/runtime/simple-stream-payload/runner.rs b/tests/runtime/simple-stream-payload/runner.rs index b05bb574e..46f267667 100644 --- a/tests/runtime/simple-stream-payload/runner.rs +++ b/tests/runtime/simple-stream-payload/runner.rs @@ -11,6 +11,23 @@ export!(Component); impl Guest for Component { async fn run() { + // After the writer drops, repeated reads remain Dropped. + let (tx, mut rx) = wit_stream::new::(); + drop(tx); + for _ in 0..2 { + let (result, buf) = rx.read(Vec::with_capacity(1)).await; + assert_eq!(result, StreamResult::Dropped); + assert!(buf.is_empty()); + } + + // After the reader drops, repeated writes preserve unsent values. + let (mut tx, rx) = wit_stream::new::(); + drop(rx); + for _ in 0..2 { + let (result, buf) = tx.write(vec![42]).await; + assert_eq!(result, StreamResult::Dropped); + assert_eq!(buf.into_vec(), vec![42]); + } let (mut tx, rx) = wit_stream::new(); let test = async { // write one item