rust: mark streams done after Dropped(0)

crates/guest-rust/src/rt/async_support/stream_support.rs

@@ -402,7 +402,10 @@ where
     ) -> Result<Self::Result, Self::InProgress> {
         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.

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::<u8>();
+        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::<u8>();
+        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