CABI: tidy stream code a bit (no change)

Author: lukewagner

design/mvp/canonical-abi/definitions.py

@@ -751,7 +751,7 @@ def drop(self):
 
 RevokeBuffer = Callable[[], None]
 OnPartialCopy = Callable[[RevokeBuffer], None]
-OnCopyDone = Callable[[Literal['completed','cancelled']], None]
+OnCopyDone = Callable[[Literal['completed','cancelled','closed']], None]
 
 class ReadableStream:
   t: ValType
@@ -793,7 +793,7 @@ def close(self):
     if not self.closed_:
       self.closed_ = True
       if self.pending_buffer:
-        self.reset_and_notify_pending('completed')
+        self.reset_and_notify_pending('closed')
 
   def closed(self):
     return self.closed_
@@ -2197,6 +2197,7 @@ async def copy(EndT, BufferT, event_code, stream_or_future_t, opts, task, i, ptr
 
   def copy_event(why, revoke_buffer):
     revoke_buffer()
+    assert(e.copying)
     e.copying = False
     return (event_code, i, pack_copy_result(task, e, buffer, why))
 
@@ -2209,13 +2210,13 @@ def on_copy_done(why):
   if e.copy(task.inst, buffer, on_partial_copy, on_copy_done) == 'done':
     return [pack_copy_result(task, e, buffer, 'completed')]
   else:
+    e.copying = True
     if opts.sync:
       await task.wait_on(e.wait_for_pending_event(), sync = True)
       code,index,payload = e.get_event()
       assert(code == event_code and index == i)
       return [payload]
     else:
-      e.copying = True
       return [BLOCKED]
 
 BLOCKED   = 0xffff_ffff