Fix double read/writes on stream/future handles (#12873)

* Fix double read/writes on stream/future handles

This should result in a first-class trap, not a bug.

* Fix CI

Author: alexcrichton

crates/c-api/include/wasmtime/trap.h

@@ -131,6 +131,9 @@ enum wasmtime_trap_code_enum {
   WASMTIME_TRAP_CODE_UNSUPPORTED_CALLBACK_CODE = 43,
   /// Cannot resume a thread which is not suspended.
   WASMTIME_TRAP_CODE_CANNOT_RESUME_THREAD = 44,
+  /// Cannot issue a read/write on a future/stream while there is a
+  /// pending operation already.
+  WASMTIME_TRAP_CODE_CONCURRENT_FUTURE_STREAM_OP = 45,
 };
 
 /**

crates/c-api/src/trap.rs

@@ -51,6 +51,7 @@ const _: () = {
     assert!(Trap::BackpressureOverflow as u8 == 42);
     assert!(Trap::UnsupportedCallbackCode as u8 == 43);
     assert!(Trap::CannotResumeThread as u8 == 44);
+    assert!(Trap::ConcurrentFutureStreamOp as u8 == 45);
 };
 
 #[repr(C)]

crates/environ/src/trap_encoding.rs

@@ -216,6 +216,10 @@ generate_trap_type! {
         /// Cannot resume a thread which is not suspended.
         CannotResumeThread = "cannot resume thread which is not suspended",
 
+        /// Cannot issue a read/write on a future/stream while there is a
+        /// pending operation already.
+        ConcurrentFutureStreamOp = "cannot have concurrent operations active on a future/stream",
+
         // if adding a variant here be sure to update `trap.rs` and `trap.h` as
         // mentioned above
     }

crates/wasmtime/src/runtime/component/concurrent.rs

@@ -142,7 +142,6 @@ impl Status {
 #[derive(Clone, Copy, Debug)]
 enum Event {
     None,
-    Cancelled,
     Subtask {
         status: Status,
     },
@@ -162,6 +161,7 @@ enum Event {
         code: ReturnCode,
         pending: Option<(TypeFutureTableIndex, u32)>,
     },
+    Cancelled,
 }
 
 impl Event {

crates/wasmtime/src/runtime/component/concurrent/futures_and_streams.rs

@@ -14,7 +14,7 @@ use crate::vm::component::{ComponentInstance, HandleTable, TransmitLocalState};
 use crate::vm::{AlwaysMut, VMStore};
 use crate::{AsContext, AsContextMut, StoreContextMut, ValRaw};
 use crate::{
-    Error, Result, bail, bail_bug, ensure,
+    Error, Result, Trap, bail, bail_bug, ensure,
     error::{Context as _, format_err},
 };
 use buffers::{Extender, SliceBuffer, UntypedWriteBuffer};
@@ -3423,10 +3423,7 @@ impl Instance {
         self.check_bounds(store.0, options, ty, address, count)?;
         let (rep, state) = self.id().get_mut(store.0).get_mut_by_index(ty, handle)?;
         let TransmitLocalState::Write { done } = *state else {
-            bail!(
-                "invalid handle {handle}; expected `Write`; got {:?}",
-                *state
-            );
+            bail!(Trap::ConcurrentFutureStreamOp);
         };
 
         if done {
@@ -3668,7 +3665,7 @@ impl Instance {
         self.check_bounds(store.0, options, ty, address, count)?;
         let (rep, state) = self.id().get_mut(store.0).get_mut_by_index(ty, handle)?;
         let TransmitLocalState::Read { done } = *state else {
-            bail_bug!("invalid handle {handle}; expected `Read`; got {:?}", *state);
+            bail!(Trap::ConcurrentFutureStreamOp);
         };
 
         if done {

tests/misc_testsuite/component-model/async/streams.wast

@@ -108,3 +108,81 @@
   (core func $stream-drop-writable (canon stream.drop-writable $stream-type))
   (core instance $i (instantiate $m (with "" (instance (export "stream.drop-writable" (func $stream-drop-writable))))))
 )
+
+;; Test which exercises an intra-component stream read/write to ensure that
+;; there's no Miri violations while doing this.
+(component definition $A
+  (core module $libc (memory (export "mem") 1))
+  (core instance $libc (instantiate $libc))
+  (core module $ics
+    (import "" "stream.new" (func $stream.new (result i64)))
+    (import "" "stream.read" (func $stream.read (param i32 i32 i32) (result i32)))
+    (import "" "stream.write" (func $stream.write (param i32 i32 i32) (result i32)))
+    (import "" "mem" (memory 1))
+
+    (global $r (mut i32) (i32.const 0))
+    (global $w (mut i32) (i32.const 0))
+
+    (func (export "read-twice")
+      call $init
+
+      (call $stream.read
+        (global.get $r)
+        (i32.const 100)
+        (i32.const 100))
+      i32.const -1 ;; BLOCKED
+      i32.ne
+      if unreachable end
+
+      (call $stream.read
+        (global.get $r)
+        (i32.const 0)
+        (i32.const 0))
+      unreachable
+    )
+
+    (func (export "write-twice")
+      call $init
+
+      (call $stream.write
+        (global.get $w)
+        (i32.const 100)
+        (i32.const 100))
+      i32.const -1 ;; BLOCKED
+      i32.ne
+      if unreachable end
+
+      (call $stream.write
+        (global.get $w)
+        (i32.const 0)
+        (i32.const 0))
+      unreachable
+    )
+
+    (func $init
+      (local $t64 i64)
+
+      (local.set $t64 (call $stream.new))
+      (global.set $r (i32.wrap_i64 (local.get $t64)))
+      (global.set $w (i32.wrap_i64 (i64.shr_u (local.get $t64) (i64.const 32))))
+    )
+  )
+  (type $s (stream u8))
+  (core func $stream.new (canon stream.new $s))
+  (core func $stream.read (canon stream.read $s async (memory $libc "mem")))
+  (core func $stream.write (canon stream.write $s async (memory $libc "mem")))
+  (core instance $ics (instantiate $ics
+    (with "" (instance
+      (export "stream.new" (func $stream.new))
+      (export "stream.read" (func $stream.read))
+      (export "stream.write" (func $stream.write))
+      (export "mem" (memory $libc "mem"))
+    ))
+  ))
+  (func (export "read-twice") async (canon lift (core func $ics "read-twice")))
+  (func (export "write-twice") async (canon lift (core func $ics "write-twice")))
+)
+(component instance $A $A)
+(assert_trap (invoke "read-twice") "cannot have concurrent operations active on a future/stream")
+(component instance $A $A)
+(assert_trap (invoke "write-twice") "cannot have concurrent operations active on a future/stream")