CABI: allow dropping a readable future end with a pending write

design/mvp/CanonicalABI.md

@@ -1751,7 +1751,7 @@ class BufferGuestImpl(Buffer):
   def is_zero_length(self):
     return self.length == 0
 
-class ReadableBufferGuestImpl(BufferGuestImpl):
+class ReadableBufferGuestImpl(BufferGuestImpl, ReadableBuffer):
   def read(self, n):
     assert(n <= self.remain())
     if self.t:
@@ -2124,7 +2124,7 @@ by `WritableFutureEnd.drop` so it can be asserted here:
     if not self.dropped:
       self.dropped = True
       if self.pending_buffer:
-        assert(isinstance(self.pending_buffer, WritableBuffer))
+        assert(isinstance(self.pending_buffer, ReadableBuffer))
         self.reset_and_notify_pending(CopyResult.DROPPED)
 ```
 Lastly, `read` and `write` work mostly like streams, but simplified based on

design/mvp/canonical-abi/definitions.py

@@ -962,7 +962,7 @@ def remain(self):
   def is_zero_length(self):
     return self.length == 0
 
-class ReadableBufferGuestImpl(BufferGuestImpl):
+class ReadableBufferGuestImpl(BufferGuestImpl, ReadableBuffer):
   def read(self, n):
     assert(n <= self.remain())
     if self.t:
@@ -1158,7 +1158,7 @@ def drop(self):
     if not self.dropped:
       self.dropped = True
       if self.pending_buffer:
-        assert(isinstance(self.pending_buffer, WritableBuffer))
+        assert(isinstance(self.pending_buffer, ReadableBuffer))
         self.reset_and_notify_pending(CopyResult.DROPPED)
 
   def read(self, inst, dst_buffer, on_copy_done):

design/mvp/canonical-abi/run_tests.py

@@ -2308,6 +2308,42 @@ def core_func(args):
   lift_and_run(lift_opts, inst, caller_ft, core_func, lambda:[], lambda _:())
 
 
+def test_future_drop_readable_with_pending_write():
+  store = Store()
+  inst = ComponentInstance(store)
+  mem = bytearray(24)
+  opts = mk_opts(memory=MemInst(mem, 'i32'), async_=True)
+  future_t = FutureType(U8Type())
+
+  def core_func(args):
+    assert(len(args) == 0)
+    [] = canon_task_return([], opts, [])
+    [packed] = canon_future_new(future_t)
+    rfi,wfi = unpack_new_ends(packed)
+
+    mem[0] = 42
+    [ret] = canon_future_write(future_t, opts, wfi, 0)
+    assert(ret == definitions.BLOCKED)
+
+    # The reader may drop its end before reading a value; the blocked write
+    # is notified that the readable end was dropped.
+    [] = canon_future_drop_readable(future_t, rfi)
+    retp = 16
+    [seti] = canon_waitable_set_new()
+    [] = canon_waitable_join(wfi, seti)
+    [event] = canon_waitable_set_wait(True, MemInst(mem, 'i32'), seti, retp)
+    assert(event == EventCode.FUTURE_WRITE)
+    assert(mem[retp+0] == wfi)
+    assert(mem[retp+4] == CopyResult.DROPPED)
+    [] = canon_waitable_join(wfi, 0)
+    [] = canon_waitable_set_drop(seti)
+    [] = canon_future_drop_writable(future_t, wfi)
+    return []
+
+  caller_ft = FuncType([], [], async_ = True)
+  lift_and_run(opts, inst, caller_ft, core_func, lambda:[], lambda _:())
+
+
 def test_cancel_subtask():
   store = Store()
   root_inst = ComponentInstance(store)
@@ -3039,6 +3075,7 @@ def core_consumer(args):
 test_wasm_to_wasm_stream_empty()
 test_cancel_copy()
 test_futures()
+test_future_drop_readable_with_pending_write()
 test_cancel_subtask()
 test_self_copy(None)
 test_self_copy(U8Type())