stream: end Readable.toWeb on half-open duplex

Author: ikeyan

lib/internal/webstreams/adapters.js

@@ -518,7 +518,9 @@ function newReadableStreamFromStreamReadable(streamReadable, options = kEmptyObj
 
   streamReadable.pause();
 
-  const cleanup = eos(streamReadable, (error) => {
+  // When adapting a Duplex as a ReadableStream, readable completion should not
+  // wait for a half-open writable side to finish as well.
+  const cleanup = eos(streamReadable, { writable: false }, (error) => {
     error = handleKnownInternalErrors(error);
 
     cleanup();
@@ -532,24 +534,15 @@ function newReadableStreamFromStreamReadable(streamReadable, options = kEmptyObj
       return;
     }
     controller.close();
+    if (isBYOB)
+      controller.byobRequest?.respond(0);
   });
 
   streamReadable.on('data', onData);
 
   return new ReadableStream({
     type: isBYOB ? 'bytes' : undefined,
-    start(c) {
-      controller = c;
-      if (isBYOB) {
-        streamReadable.once('end', () => {
-          // close the controller
-          controller.close();
-          // And unlock the last BYOB read request
-          controller.byobRequest?.respond(0);
-          wasCanceled = true;
-        });
-      }
-    },
+    start(c) { controller = c; },
 
     pull() { streamReadable.resume(); },
 

test/parallel/test-stream-readable-to-web-termination.js

@@ -1,6 +1,8 @@
 'use strict';
-require('../common');
-const { Readable } = require('stream');
+const common = require('../common');
+const assert = require('assert');
+const { Duplex, Readable } = require('stream');
+const { setTimeout: delay } = require('timers/promises');
 
 {
   const r = Readable.from([]);
@@ -10,3 +12,33 @@ const { Readable } = require('stream');
   const reader = Readable.toWeb(r).getReader();
   reader.read();
 }
+
+{
+  const duplex = new Duplex({
+    read() {
+      this.push(Buffer.from('x'));
+      this.push(null);
+    },
+    write(_chunk, _encoding, callback) {
+      callback();
+    },
+  });
+
+  const reader = Readable.toWeb(duplex).getReader();
+
+  (async () => {
+    const result = await reader.read();
+    assert.deepStrictEqual(result, {
+      value: new Uint8Array(Buffer.from('x')),
+      done: false,
+    });
+
+    const closeResult = await Promise.race([
+      reader.read(),
+      delay(common.platformTimeout(100)).then(() => 'timeout'),
+    ]);
+
+    assert.notStrictEqual(closeResult, 'timeout');
+    assert.deepStrictEqual(closeResult, { value: undefined, done: true });
+  })().then(common.mustCall());
+}