wasi/pipe: fix AsyncReadStream returning empty list when closed (#12816)

After the inner stream returns `0` the StreamError::Closed is sent to
the channel, and after picking it up in read() the stream is set as closed,
and the error is returned to the caller.

If the caller however tries to recv() after the socket is closed, AsyncReadStream
is going to return empty list and not StreamError::Closed which is ambiguous.

Author: vavrusa

crates/wasi/src/p2/pipe.rs

@@ -239,7 +239,15 @@ impl InputStream for AsyncReadStream {
                 self.closed = true;
                 Err(e)
             }
-            Err(TryRecvError::Empty) => Ok(Bytes::new()),
+            Err(TryRecvError::Empty) => {
+                if self.closed {
+                    // Note: if the stream is already closed it should return an error,
+                    //       returning empty list would break the wasi contract (returning 0 and ready)
+                    Err(StreamError::Closed)
+                } else {
+                    Ok(Bytes::new())
+                }
+            }
             Err(TryRecvError::Disconnected) => Err(StreamError::Trap(format_err!(
                 "AsyncReadStream sender died - should be impossible"
             ))),
@@ -382,6 +390,11 @@ mod test {
                 assert!(bs.is_empty());
                 resolves_immediately(reader.ready()).await;
                 assert!(matches!(reader.read(0), Err(StreamError::Closed)));
+
+                // Try again to make sure it keeps returning `StreamError::Closed`
+                resolves_immediately(reader.ready()).await;
+                assert!(matches!(reader.read(0), Err(StreamError::Closed)));
+                assert!(matches!(reader.read(0), Err(StreamError::Closed)));
             }
             res => panic!("unexpected: {res:?}"),
         }
@@ -445,6 +458,10 @@ mod test {
             }
             res => panic!("unexpected: {res:?}"),
         }
+
+        // Make sure it stays closed
+        resolves_immediately(reader.ready()).await;
+        assert!(matches!(reader.read(0), Err(StreamError::Closed)));
     }
 
     #[test_log::test(tokio::test(flavor = "multi_thread"))]