Streams: settle reader.[[closedPromise]] before performing close/erro…

…r steps of read requests

Follows whatwg/streams#1102.

streams/readable-streams/async-iterator.any.js

@@ -625,3 +625,26 @@ for (const preventCancel of [false, true]) {
     rs.getReader();
   }, `return() should unlock the stream synchronously when preventCancel = ${preventCancel}`);
 }
+
+promise_test(async () => {
+  const rs = new ReadableStream({
+    async start(c) {
+      c.enqueue('a');
+      c.enqueue('b');
+      c.enqueue('c');
+      await flushAsyncEvents();
+      // At this point, the async iterator has a read request in the stream's queue for its pending next() promise.
+      // Closing the stream now causes two things to happen *synchronously*:
+      //  1. ReadableStreamClose resolves reader.[[closedPromise]] with undefined.
+      //  2. ReadableStreamClose calls the read request's close steps, which calls ReadableStreamReaderGenericRelease,
+      //     which replaces reader.[[closedPromise]] with a rejected promise.
+      c.close();
+    }
+  });
+
+  const chunks = [];
+  for await (const chunk of rs) {
+    chunks.push(chunk);
+  }
+  assert_array_equals(chunks, ['a', 'b', 'c']);
+}, 'close() while next() is pending');

streams/readable-streams/default-reader.any.js

@@ -154,6 +154,57 @@ promise_test(t => {
 
 }, 'closed should be rejected after reader releases its lock (multiple stream locks)');
 
+promise_test(t => {
+
+  let controller;
+  const rs = new ReadableStream({
+    start(c) {
+      controller = c;
+    }
+  });
+
+  const reader = rs.getReader();
+  const promise1 = reader.closed;
+
+  controller.close();
+
+  reader.releaseLock();
+  const promise2 = reader.closed;
+
+  assert_not_equals(promise1, promise2, '.closed should be replaced');
+  return Promise.all([
+    promise1,
+    promise_rejects_js(t, TypeError, promise2, '.closed after releasing lock'),
+  ]);
+
+}, 'closed is replaced when stream closes and reader releases its lock');
+
+promise_test(t => {
+
+  const theError = { name: 'unique error' };
+  let controller;
+  const rs = new ReadableStream({
+    start(c) {
+      controller = c;
+    }
+  });
+
+  const reader = rs.getReader();
+  const promise1 = reader.closed;
+
+  controller.error(theError);
+
+  reader.releaseLock();
+  const promise2 = reader.closed;
+
+  assert_not_equals(promise1, promise2, '.closed should be replaced');
+  return Promise.all([
+    promise_rejects_exactly(t, theError, promise1, '.closed before releasing lock'),
+    promise_rejects_js(t, TypeError, promise2, '.closed after releasing lock')
+  ]);
+
+}, 'closed is replaced when stream errors and reader releases its lock');
+
 promise_test(() => {
 
   const rs = new ReadableStream({

streams/readable-streams/general.any.js

@@ -210,14 +210,14 @@ promise_test(() => {
       assert_unreached('closed should be rejected');
     }, e => {
       closed = true;
-      assert_true(read);
+      assert_false(read);
       assert_equals(e, error, 'closed should be rejected with the thrown error');
     }),
     reader.read().then(() => {
       assert_unreached('read() should be rejected');
     }, e => {
       read = true;
-      assert_false(closed);
+      assert_true(closed);
       assert_equals(e, error, 'read() should be rejected with the thrown error');
     })
   ]);