Fix handling of closed streams.

In Rust, `read` returning 0 bytes indicates end-of-stream, however in WASI
a `read` returning 0 bytes has no special meaning. So special-case a
0 coming from WASI to report it to Rust differently.

And, make writes on a closed WASI stream fail with `ConnectionReset`,
indicating that the stream has been closed.

Author: sunfishcode

src/io/streams.rs

@@ -40,7 +40,12 @@ impl AsyncRead for AsyncInputStream {
         // Ideally, the ABI would be able to read directly into buf. However, with the default
         // generated bindings, it returns a newly allocated vec, which we need to copy into buf.
         let read = match self.stream.read(buf.len() as u64) {
+            // 0 bytets from WASI's `read` just means we got zero bytes.
+            Ok(r) if r.is_empty() => {
+                return Err(std::io::Error::from(std::io::ErrorKind::Interrupted))
+            }
             Ok(r) => r,
+            // 0 bytes from Rust's `read` means end-of-stream.
             Err(StreamError::Closed) => return Ok(0),
             Err(StreamError::LastOperationFailed(err)) => {
                 return Err(std::io::Error::other(err.to_debug_string()))
@@ -97,13 +102,17 @@ impl AsyncWrite for AsyncOutputStream {
                     let writable = some.try_into().unwrap_or(usize::MAX).min(buf.len());
                     match self.stream.write(&buf[0..writable]) {
                         Ok(()) => return Ok(writable),
-                        Err(StreamError::Closed) => return Ok(0),
+                        Err(StreamError::Closed) => {
+                            return Err(std::io::Error::from(std::io::ErrorKind::ConnectionReset))
+                        }
                         Err(StreamError::LastOperationFailed(err)) => {
                             return Err(std::io::Error::other(err.to_debug_string()))
                         }
                     }
                 }
-                Err(StreamError::Closed) => return Ok(0),
+                Err(StreamError::Closed) => {
+                    return Err(std::io::Error::from(std::io::ErrorKind::ConnectionReset))
+                }
                 Err(StreamError::LastOperationFailed(err)) => {
                     return Err(std::io::Error::other(err.to_debug_string()))
                 }
@@ -116,7 +125,9 @@ impl AsyncWrite for AsyncOutputStream {
                 self.ready().await;
                 Ok(())
             }
-            Err(StreamError::Closed) => Ok(()),
+            Err(StreamError::Closed) => {
+                Err(std::io::Error::from(std::io::ErrorKind::ConnectionReset))
+            }
             Err(StreamError::LastOperationFailed(err)) => {
                 Err(std::io::Error::other(err.to_debug_string()))
             }