Fix cross-component-instance streams being flagged as intra-instance (#14018)

* Fix cross-component-instance streams being flagged as intra-instance

The comparison key before this commit was
`RuntimeComponentInstanceIndex` which is a component-local index and
can't be compared across component instance. The fix is to use
`RuntimeInstance` which has the store-local component instance id plus
the `RuntimeComponentInstanceIndex` within that instance.

* Fix compile

Author: alexcrichton

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

@@ -6,8 +6,8 @@ use crate::component::matching::InstanceType;
 use crate::component::types;
 use crate::component::values::ErrorContextAny;
 use crate::component::{
-    AsAccessor, ComponentInstanceId, ComponentType, FutureAny, Instance, Lift, Lower, StreamAny,
-    Val, WasmList,
+    AsAccessor, ComponentInstanceId, ComponentType, FutureAny, Instance, Lift, Lower,
+    RuntimeInstance, StreamAny, Val, WasmList,
 };
 use crate::prelude::*;
 use crate::store::{StoreOpaque, StoreToken};
@@ -3265,20 +3265,20 @@ impl Instance {
     fn copy<T: 'static>(
         store: StoreContextMut<T>,
         flat_abi: Option<FlatAbi>,
-        write_instance: Instance,
-        write_caller_instance: RuntimeComponentInstanceIndex,
+        write_runtime_instance: RuntimeInstance,
         write_ty: TransmitIndex,
         write_options: OptionsIndex,
         write_address: usize,
-        read_instance: Instance,
-        read_caller_instance: RuntimeComponentInstanceIndex,
+        read_runtime_instance: RuntimeInstance,
         read_caller_thread: QualifiedThreadId,
         read_ty: TransmitIndex,
         read_options: OptionsIndex,
         read_address: usize,
         count: ItemCount,
         rep: u32,
     ) -> Result<()> {
+        let write_instance = Instance::from_runtime_instance(store.0, write_runtime_instance);
+        let read_instance = Instance::from_runtime_instance(store.0, read_runtime_instance);
         let (write_component, store) = write_instance.component_and_store_mut(store.0);
         let (read_component, mut store) = read_instance.component_and_store_mut(store);
         let write_types = write_component.types();
@@ -3327,7 +3327,7 @@ impl Instance {
                 .ok_or_else(|| crate::format_err!("read pointer out of bounds"))?;
         }
 
-        if write_caller_instance == read_caller_instance
+        if write_runtime_instance == read_runtime_instance
             && !allow_intra_component_read_write(write_payload_ty)
         {
             bail!(
@@ -3610,13 +3610,11 @@ impl Instance {
                 Instance::copy(
                     store.as_context_mut(),
                     flat_abi,
-                    self,
-                    caller,
+                    self.runtime_instance(caller),
                     ty,
                     options,
                     address,
-                    read_instance,
-                    read_caller_instance,
+                    read_instance.runtime_instance(read_caller_instance),
                     read_caller_thread,
                     read_ty,
                     read_options,
@@ -3848,13 +3846,11 @@ impl Instance {
                 Instance::copy(
                     store.as_context_mut(),
                     flat_abi,
-                    write_instance,
-                    write_caller,
+                    write_instance.runtime_instance(write_caller),
                     write_ty,
                     write_options,
                     write_address,
-                    self,
-                    caller_instance,
+                    self.runtime_instance(caller_instance),
                     caller_thread,
                     ty,
                     options,

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

@@ -572,6 +572,12 @@ impl Instance {
             index,
         }
     }
+
+    #[cfg(feature = "component-model-async")]
+    pub(crate) fn from_runtime_instance(store: &StoreOpaque, instance: RuntimeInstance) -> Self {
+        let id = StoreComponentInstanceId::new(store.id(), instance.instance);
+        Instance { id }
+    }
 }
 
 /// Translates a `CoreDef`, a definition of a core wasm item, to an

tests/all/component_model/async.rs

@@ -1062,3 +1062,123 @@ async fn async_call_stack() -> Result<()> {
         .await??;
     Ok(())
 }
+
+#[test]
+#[cfg_attr(miri, ignore)]
+fn inter_component_stream_is_not_intra_component() -> Result<()> {
+    let engine = Engine::default();
+
+    let writer = Component::new(
+        &engine,
+        r#"
+(component
+  (core module $libc
+    (memory (export "m") 1)
+    (data (i32.const 0x100) "hello")
+  )
+  (core instance $libc (instantiate $libc))
+
+  (type $s (stream string))
+  (core func $stream.new (canon stream.new $s))
+  (core func $stream.write (canon stream.write $s async (memory (core memory $libc "m"))))
+
+  (core module $m
+    (import "" "m" (memory 1))
+    (import "" "stream.new" (func $stream.new (result i64)))
+    (import "" "stream.write" (func $stream.write (param i32 i32 i32) (result i32)))
+
+    (global $w (mut i32) (i32.const 0))
+
+    (func (export "mk") (result i32)
+      (local $tmp i64)
+      (local.set $tmp (call $stream.new))
+      (global.set $w (i32.wrap_i64 (i64.shr_u (local.get $tmp) (i64.const 32))))
+      (i32.wrap_i64 (local.get $tmp))
+    )
+
+    (func (export "write")
+      ;; store ptr/len at 0x10
+      (i32.store (i32.const 0x10) (i32.const 0x100))
+      (i32.store (i32.const 0x14) (i32.const 5))
+
+      (call $stream.write (global.get $w) (i32.const 0x10) (i32.const 1))
+      i32.const -1 ;; BLOCKED
+      i32.ne
+      if unreachable end
+    )
+  )
+
+  (core instance $i (instantiate $m
+    (with "" (instance
+      (export "m" (memory $libc "m"))
+      (export "stream.new" (func $stream.new))
+      (export "stream.write" (func $stream.write))
+    ))
+  ))
+
+  (func (export "mk") (result $s) (canon lift (core func $i "mk")))
+  (func (export "write") (canon lift (core func $i "write")))
+)
+        "#,
+    )?;
+
+    let reader = Component::new(
+        &engine,
+        r#"
+(component
+  (core module $libc
+    (memory (export "m") 1)
+    (func (export "realloc") (param i32 i32 i32 i32) (result i32)
+      (i32.const 0x200)
+    )
+  )
+  (core instance $libc (instantiate $libc))
+
+  (type $s (stream string))
+  (core func $stream.read
+    (canon stream.read $s async
+      (memory (core memory $libc "m"))
+      (realloc (core func $libc "realloc"))))
+
+  (core module $m
+    (import "" "m" (memory 1))
+    (import "" "stream.read" (func $stream.read (param i32 i32 i32) (result i32)))
+
+    (func (export "read") (param $r i32) (result i32)
+      (call $stream.read (local.get $r) (i32.const 0x10) (i32.const 1))
+      i32.const 0x10 ;; COMPLETED | (1 << 4)
+      i32.ne
+      if unreachable end
+
+      i32.const 0x10
+    )
+  )
+
+  (core instance $i (instantiate $m
+    (with "" (instance
+      (export "m" (memory $libc "m"))
+      (export "stream.read" (func $stream.read))
+    ))
+  ))
+
+  (func (export "read") (param "s" $s) (result string)
+    (canon lift (core func $i "read") (memory (core memory $libc "m"))))
+)
+        "#,
+    )?;
+
+    let linker = Linker::new(&engine);
+    let mut store = Store::new(&engine, ());
+    let writer = linker.instantiate(&mut store, &writer)?;
+    let reader = linker.instantiate(&mut store, &reader)?;
+
+    let mk = writer.get_typed_func::<(), (StreamReader<String>,)>(&mut store, "mk")?;
+    let write = writer.get_typed_func::<(), ()>(&mut store, "write")?;
+    let read = reader.get_typed_func::<(StreamReader<String>,), (String,)>(&mut store, "read")?;
+
+    let (stream,) = mk.call(&mut store, ())?;
+    write.call(&mut store, ())?;
+    assert_eq!(read.call(&mut store, (stream,))?, ("hello".to_string(),));
+
+    Ok(())
+}