trap on read from and write to intra-component stream/future (#12117)

The spec says we need to do this for streams and futures that have a payload
type.

Note that `p3_http_middleware_host_to_host` triggers this trap, which is
expected given how it functions, so now we assert accordingly.

Thanks to Alex for the tests!

Fixes #12108

Signed-off-by: Joel Dice <[email protected]>
Co-authored-by: Alex Crichton <[email protected]>

Author: dicej

crates/wasi-http/tests/all/p3/mod.rs

@@ -330,8 +330,15 @@ async fn p3_http_middleware() -> Result<()> {
 }
 
 #[test_log::test(tokio::test(flavor = "multi_thread"))]
-async fn p3_http_middleware_host_to_host() -> Result<()> {
-    test_http_middleware(true).await
+async fn p3_http_middleware_host_to_host() {
+    let error = format!("{:?}", test_http_middleware(true).await.unwrap_err());
+
+    let expected = "cannot read from and write to intra-component stream with non-unit payload";
+
+    assert!(
+        error.contains(expected),
+        "expected `{expected}`; got `{error}`"
+    );
 }
 
 async fn test_http_middleware(host_to_host: bool) -> Result<()> {

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

@@ -3608,6 +3608,7 @@ impl<T: 'static> VMComponentAsyncStore for StoreInner<T> {
         instance
             .guest_write(
                 StoreContextMut(self),
+                caller,
                 TransmitIndex::Future(ty),
                 options,
                 None,
@@ -3631,6 +3632,7 @@ impl<T: 'static> VMComponentAsyncStore for StoreInner<T> {
         instance
             .guest_read(
                 StoreContextMut(self),
+                caller,
                 TransmitIndex::Future(ty),
                 options,
                 None,
@@ -3655,6 +3657,7 @@ impl<T: 'static> VMComponentAsyncStore for StoreInner<T> {
         instance
             .guest_write(
                 StoreContextMut(self),
+                caller,
                 TransmitIndex::Stream(ty),
                 options,
                 None,
@@ -3679,6 +3682,7 @@ impl<T: 'static> VMComponentAsyncStore for StoreInner<T> {
         instance
             .guest_read(
                 StoreContextMut(self),
+                caller,
                 TransmitIndex::Stream(ty),
                 options,
                 None,
@@ -3716,6 +3720,7 @@ impl<T: 'static> VMComponentAsyncStore for StoreInner<T> {
         instance
             .guest_write(
                 StoreContextMut(self),
+                caller,
                 TransmitIndex::Stream(ty),
                 options,
                 Some(FlatAbi {
@@ -3745,6 +3750,7 @@ impl<T: 'static> VMComponentAsyncStore for StoreInner<T> {
         instance
             .guest_read(
                 StoreContextMut(self),
+                caller,
                 TransmitIndex::Stream(ty),
                 options,
                 Some(FlatAbi {

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

@@ -1916,6 +1916,7 @@ enum WriteState {
     /// The write end is owned by a guest task and a write is pending.
     GuestReady {
         instance: Instance,
+        caller: RuntimeComponentInstanceIndex,
         ty: TransmitIndex,
         flat_abi: Option<FlatAbi>,
         options: OptionsIndex,
@@ -1953,6 +1954,7 @@ enum ReadState {
     /// The read end is owned by a guest task and a read is pending.
     GuestReady {
         ty: TransmitIndex,
+        caller: RuntimeComponentInstanceIndex,
         flat_abi: Option<FlatAbi>,
         instance: Instance,
         options: OptionsIndex,
@@ -2672,6 +2674,7 @@ async fn write<D: 'static, P: Send + 'static, T: func::Lower + 'static, B: Write
             count,
             handle,
             instance,
+            caller,
         } => {
             let guest_offset = guest_offset.unwrap();
 
@@ -2746,6 +2749,7 @@ async fn write<D: 'static, P: Send + 'static, T: func::Lower + 'static, B: Write
                     count,
                     handle,
                     instance,
+                    caller,
                 };
 
                 anyhow::Ok(())
@@ -2916,9 +2920,11 @@ impl Instance {
         self,
         mut store: StoreContextMut<T>,
         flat_abi: Option<FlatAbi>,
+        write_caller: RuntimeComponentInstanceIndex,
         write_ty: TransmitIndex,
         write_options: OptionsIndex,
         write_address: usize,
+        read_caller: RuntimeComponentInstanceIndex,
         read_ty: TransmitIndex,
         read_options: OptionsIndex,
         read_address: usize,
@@ -2930,8 +2936,15 @@ impl Instance {
             (TransmitIndex::Future(write_ty), TransmitIndex::Future(read_ty)) => {
                 assert_eq!(count, 1);
 
-                let val = types[types[write_ty].ty]
-                    .payload
+                let payload = types[types[write_ty].ty].payload;
+
+                if write_caller == read_caller && payload.is_some() {
+                    bail!(
+                        "cannot read from and write to intra-component future with non-unit payload"
+                    )
+                }
+
+                let val = payload
                     .map(|ty| {
                         let lift =
                             &mut LiftContext::new(store.0.store_opaque_mut(), write_options, self);
@@ -2968,6 +2981,12 @@ impl Instance {
             }
             (TransmitIndex::Stream(write_ty), TransmitIndex::Stream(read_ty)) => {
                 if let Some(flat_abi) = flat_abi {
+                    if write_caller == read_caller && types[types[write_ty].ty].payload.is_some() {
+                        bail!(
+                            "cannot read from and write to intra-component stream with non-unit payload"
+                        )
+                    }
+
                     // Fast path memcpy for "flat" (i.e. no pointers or handles) payloads:
                     let length_in_bytes = usize::try_from(flat_abi.size).unwrap() * count;
                     if length_in_bytes > 0 {
@@ -3086,6 +3105,7 @@ impl Instance {
     pub(super) fn guest_write<T: 'static>(
         self,
         mut store: StoreContextMut<T>,
+        caller: RuntimeComponentInstanceIndex,
         ty: TransmitIndex,
         options: OptionsIndex,
         flat_abi: Option<FlatAbi>,
@@ -3137,6 +3157,7 @@ impl Instance {
             );
             transmit.write = WriteState::GuestReady {
                 instance: self,
+                caller,
                 ty,
                 flat_abi,
                 options,
@@ -3156,6 +3177,7 @@ impl Instance {
                 count: read_count,
                 handle: read_handle,
                 instance: read_instance,
+                caller: read_caller,
             } => {
                 assert_eq!(flat_abi, read_flat_abi);
 
@@ -3195,9 +3217,11 @@ impl Instance {
                 self.copy(
                     store.as_context_mut(),
                     flat_abi,
+                    caller,
                     ty,
                     options,
                     address,
+                    read_caller,
                     read_ty,
                     read_options,
                     read_address,
@@ -3238,6 +3262,7 @@ impl Instance {
                         count: read_count - count,
                         handle: read_handle,
                         instance: read_instance,
+                        caller: read_caller,
                     };
                 }
 
@@ -3308,6 +3333,7 @@ impl Instance {
     pub(super) fn guest_read<T: 'static>(
         self,
         mut store: StoreContextMut<T>,
+        caller: RuntimeComponentInstanceIndex,
         ty: TransmitIndex,
         options: OptionsIndex,
         flat_abi: Option<FlatAbi>,
@@ -3362,6 +3388,7 @@ impl Instance {
                 count,
                 handle,
                 instance: self,
+                caller,
             };
             Ok::<_, crate::Error>(())
         };
@@ -3375,6 +3402,7 @@ impl Instance {
                 address: write_address,
                 count: write_count,
                 handle: write_handle,
+                caller: write_caller,
             } => {
                 assert_eq!(flat_abi, write_flat_abi);
 
@@ -3397,9 +3425,11 @@ impl Instance {
                 self.copy(
                     store.as_context_mut(),
                     flat_abi,
+                    write_caller,
                     write_ty,
                     write_options,
                     write_address,
+                    caller,
                     ty,
                     options,
                     address,
@@ -3440,6 +3470,7 @@ impl Instance {
                     let transmit = concurrent_state.get_mut(transmit_id)?;
                     transmit.write = WriteState::GuestReady {
                         instance: self,
+                        caller: write_caller,
                         ty: write_ty,
                         flat_abi: write_flat_abi,
                         options: write_options,

tests/misc_testsuite/component-model/async/intra-futures.wast

@@ -0,0 +1,48 @@
+;;! component_model_async = true
+;;! component_model_async_builtins = true
+
+(component
+  (core module $libc (memory (export "m") 1))
+  (core instance $libc (instantiate $libc))
+
+  (type $s (future u32))
+  (core func $future.new (canon future.new $s))
+  (core func $future.read (canon future.read $s async (memory $libc "m")))
+  (core func $future.write (canon future.write $s async (memory $libc "m")))
+
+  (core module $m
+    (import "" "future.new" (func $future.new (result i64)))
+    (import "" "future.read" (func $future.read (param i32 i32) (result i32)))
+    (import "" "future.write" (func $future.write (param i32 i32) (result i32)))
+
+    (func (export "run")
+      (local $tmp i64)
+      (local $r i32)
+      (local $w i32)
+      (local.set $tmp (call $future.new))
+
+      (local.set $r (i32.wrap_i64 (local.get $tmp)))
+      (local.set $w (i32.wrap_i64 (i64.shr_u (local.get $tmp) (i64.const 32))))
+
+      (call $future.read (local.get $r) (i32.const 0))
+      i32.const -1 ;; BLOCKED
+      i32.ne
+      if unreachable end
+
+      (call $future.write (local.get $w) (i32.const 0))
+      drop
+    )
+  )
+
+  (core instance $i (instantiate $m
+    (with "" (instance
+      (export "future.new" (func $future.new))
+      (export "future.read" (func $future.read))
+      (export "future.write" (func $future.write))
+    ))
+  ))
+
+  (func (export "run") (canon lift (core func $i "run")))
+)
+
+(assert_trap (invoke "run") "cannot read from and write to intra-component future with non-unit payload")

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

@@ -0,0 +1,48 @@
+;;! component_model_async = true
+;;! component_model_async_builtins = true
+
+(component
+  (core module $libc (memory (export "m") 1))
+  (core instance $libc (instantiate $libc))
+
+  (type $s (stream u32))
+  (core func $stream.new (canon stream.new $s))
+  (core func $stream.read (canon stream.read $s async (memory $libc "m")))
+  (core func $stream.write (canon stream.write $s async (memory $libc "m")))
+
+  (core module $m
+    (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)))
+
+    (func (export "run")
+      (local $tmp i64)
+      (local $r i32)
+      (local $w i32)
+      (local.set $tmp (call $stream.new))
+
+      (local.set $r (i32.wrap_i64 (local.get $tmp)))
+      (local.set $w (i32.wrap_i64 (i64.shr_u (local.get $tmp) (i64.const 32))))
+
+      (call $stream.read (local.get $r) (i32.const 0) (i32.const 4))
+      i32.const -1 ;; BLOCKED
+      i32.ne
+      if unreachable end
+
+      (call $stream.write (local.get $w) (i32.const 0) (i32.const 4))
+      drop
+    )
+  )
+
+  (core instance $i (instantiate $m
+    (with "" (instance
+      (export "stream.new" (func $stream.new))
+      (export "stream.read" (func $stream.read))
+      (export "stream.write" (func $stream.write))
+    ))
+  ))
+
+  (func (export "run") (canon lift (core func $i "run")))
+)
+
+(assert_trap (invoke "run") "cannot read from and write to intra-component stream with non-unit payload")