From 438217c96229f4cd172d4e0a386f4546af7da026 Mon Sep 17 00:00:00 2001
From: macovedj <djmacovei@gmail.com>
Date: Thu, 22 Oct 2026 09:59:23 -0500
Subject: [PATCH] rust: mark streams done after Dropped(0)

---
 .../src/rt/async_support/stream_support.rs      | 10 ++++++++--
 tests/runtime/simple-stream-payload/runner.rs   | 17 +++++++++++++++++
 2 files changed, 25 insertions(+), 2 deletions(-)

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<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.
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::<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