stream: cover synchronous Readable.toWeb eos path

Author: ikeyan

lib/internal/streams/end-of-stream.js

@@ -107,6 +107,8 @@ function getEosOnCloseError(stream, readable, readableFinished, writable, writab
   return null;
 }
 
+// Internal only: if eos() can settle immediately, invoke the callback before
+// returning cleanup. Callers must tolerate cleanup yet to be assigned.
 const kEosNodeSyncronousCallback = Symbol('kEosNodeSynchronousCallback');
 
 function eos(stream, options, callback) {
@@ -199,11 +201,10 @@ function eos(stream, options, callback) {
   };
   if (immediateResult !== undefined) {
     if (options.error !== false) {
-      const onerror = () => {};
-      stream.on('error', onerror);
+      stream.on('error', nop);
       cleanup = () => {
         callback = nop;
-        stream.removeListener('error', onerror);
+        stream.removeListener('error', nop);
       };
     }
     returnImmediately(immediateResult);

lib/internal/webstreams/adapters.js

@@ -454,6 +454,8 @@ function newStreamWritableFromWritableStream(writableStream, options = kEmptyObj
   return writable;
 }
 
+const kErrorSentinelAttached = Symbol('kErrorSentinelAttached');
+
 /**
  * @typedef {import('./queuingstrategies').QueuingStrategy} QueuingStrategy
  * @param {Readable} streamReadable
@@ -521,9 +523,12 @@ function newReadableStreamFromStreamReadable(streamReadable, options = kEmptyObj
     // If eos calls the callback synchronously, cleanup is still a no-op here.
     cleanup();
 
-    // This is a protection against non-standard, legacy streams
-    // that happen to emit an error event again after finished is called.
-    streamReadable.on('error', noop);
+    if (!(kErrorSentinelAttached in streamReadable)) {
+      // This is a protection against non-standard, legacy streams
+      // that happen to emit an error event again after finished is called.
+      streamReadable.on('error', noop);
+      streamReadable[kErrorSentinelAttached] = true;
+    }
     if (wasCanceled) {
       return;
     }

test/parallel/test-whatwg-webstreams-adapters-to-readablestream.js

@@ -286,6 +286,27 @@ const {
   );
 }
 
+{
+  const readable = new PassThrough();
+  readable.end();
+  readable.destroy();
+
+  (async () => {
+    await new Promise((resolve) => readable.once('close', resolve));
+    assert.strictEqual(readable.listenerCount('error'), 0);
+
+    const readableStream = newReadableStreamFromStreamReadable(readable);
+    // Only one error listener from the adapter should be added
+    assert.strictEqual(readable.listenerCount('error'), 1);
+    newReadableStreamFromStreamReadable(readable);
+    // No duplicate listeners should be added.
+    assert.strictEqual(readable.listenerCount('error'), 1);
+
+    const readResult = await readableStream.getReader().read();
+    assert.deepStrictEqual(readResult, { value: undefined, done: true });
+  })().then(common.mustCall());
+}
+
 {
   const duplex = new Duplex({ readable: false });
   duplex.destroy();