Add missing trap if attempting to transfer stream/future that's in a waitable set

design/mvp/CanonicalABI.md

@@ -2605,7 +2605,8 @@ Streams and futures are entirely symmetric, transferring ownership of the
 readable end from the lifting component to the host or lowering component and
 trapping if the readable end is in the middle of copying (which would create
 a dangling-pointer situation) or is in the `DONE` state (in which case the only
-valid operation is `{stream,future}.drop-{readable,writable}`).
+valid operation is `{stream,future}.drop-{readable,writable}`) or in a waitable
+set (in which case it must be removed first via `waitable.join(0)`).
 ```python
 def lift_stream(cx, i, t):
   return lift_async_value(ReadableStreamEnd, cx, i, t)
@@ -2619,6 +2620,7 @@ def lift_async_value(ReadableEndT, cx, i, t):
   trap_if(not isinstance(e, ReadableEndT))
   trap_if(e.shared.t != t)
   trap_if(e.state != CopyState.IDLE)
+  trap_if(e.in_waitable_set())
   return e.shared
 ```
 

design/mvp/canonical-abi/definitions.py

@@ -1507,6 +1507,7 @@ def lift_async_value(ReadableEndT, cx, i, t):
   trap_if(not isinstance(e, ReadableEndT))
   trap_if(e.shared.t != t)
   trap_if(e.state != CopyState.IDLE)
+  trap_if(e.in_waitable_set())
   return e.shared
 
 ## Storing

test/async/trap-if-transfer-in-waitable-set.wast

@@ -0,0 +1,51 @@
+;; This test contains a single component $Tester which creates a stream and
+;; future, joins the readable end into a waitable set, and then attempts to
+;; lift the readable end by returning it from an export. Because the readable
+;; end is in a waitable set, lifting must trap.
+
+(component definition $Tester
+  (core module $Memory (memory (export "mem") 1))
+  (core instance $memory (instantiate $Memory))
+  (core module $M
+    (import "" "waitable.join" (func $waitable.join (param i32 i32)))
+    (import "" "waitable-set.new" (func $waitable-set.new (result i32)))
+    (import "" "future.new" (func $future.new (result i64)))
+    (import "" "stream.new" (func $stream.new (result i64)))
+
+    (func $return-future-in-set (export "return-future-in-set") (result i32)
+      (local $ret64 i64) (local $rx i32) (local $ws i32)
+      (local.set $ret64 (call $future.new))
+      (local.set $rx (i32.wrap_i64 (local.get $ret64)))
+      (local.set $ws (call $waitable-set.new))
+      (call $waitable.join (local.get $rx) (local.get $ws))
+      (local.get $rx)
+    )
+    (func $return-stream-in-set (export "return-stream-in-set") (result i32)
+      (local $ret64 i64) (local $rx i32) (local $ws i32)
+      (local.set $ret64 (call $stream.new))
+      (local.set $rx (i32.wrap_i64 (local.get $ret64)))
+      (local.set $ws (call $waitable-set.new))
+      (call $waitable.join (local.get $rx) (local.get $ws))
+      (local.get $rx)
+    )
+  )
+  (type $FT (future u8))
+  (type $ST (stream u8))
+  (canon waitable.join (core func $waitable.join))
+  (canon waitable-set.new (core func $waitable-set.new))
+  (canon future.new $FT (core func $future.new))
+  (canon stream.new $ST (core func $stream.new))
+  (core instance $m (instantiate $M (with "" (instance
+    (export "waitable.join" (func $waitable.join))
+    (export "waitable-set.new" (func $waitable-set.new))
+    (export "future.new" (func $future.new))
+    (export "stream.new" (func $stream.new))
+  ))))
+  (func (export "return-future-in-set") async (result $FT) (canon lift (core func $m "return-future-in-set")))
+  (func (export "return-stream-in-set") async (result $ST) (canon lift (core func $m "return-stream-in-set")))
+)
+
+(component instance $i1 $Tester)
+(assert_trap (invoke "return-future-in-set") "cannot lift future in a waitable set")
+(component instance $i2 $Tester)
+(assert_trap (invoke "return-stream-in-set") "cannot lift stream in a waitable set")