stream: enhance internal pipe handling with additional error management and tests

Author: mertcanaltin

lib/internal/webstreams/readablestream.js

@@ -31,6 +31,7 @@ const {
 
 const {
   AbortError,
+  ErrnoException,
   codes: {
     ERR_ILLEGAL_CONSTRUCTOR,
     ERR_INVALID_ARG_TYPE,
@@ -49,6 +50,13 @@ const {
   StreamPipe,
 } = internalBinding('stream_pipe');
 
+const {
+  kReadBytesOrError,
+  streamBaseState,
+} = internalBinding('stream_wrap');
+
+const { UV_EOF } = internalBinding('uv');
+
 const {
   isArrayBufferView,
   isDataView,
@@ -130,6 +138,7 @@ const {
   isWritableStreamDefaultWriter,
 
   writableStreamAbort,
+  writableStreamClose,
   writableStreamCloseQueuedOrInFlight,
   writableStreamDefaultWriterCloseWithErrorPropagation,
   writableStreamDefaultWriterRelease,
@@ -1383,18 +1392,67 @@ function readableStreamPipeTo(
       !preventClose &&
       !preventAbort &&
       !preventCancel) {
-    // Use native piping
+    // Native C++ StreamPipe path for internal-to-internal piping.
+    // Ref: https://github.com/nodejs/performance/issues/134
     const promise = PromiseWithResolvers();
 
     source[kState].disturbed = true;
 
+    let pipeError = null;
+    let isComplete = false;
+    const originalSourceOnread = sourceStreamBase.onread;
+
+    sourceStreamBase.onread = (arrayBuffer) => {
+      const nread = streamBaseState[kReadBytesOrError];
+      if (nread < 0 && nread !== UV_EOF) {
+        pipeError = new ErrnoException(nread, 'read');
+      }
+      if (originalSourceOnread) {
+        return originalSourceOnread(arrayBuffer);
+      }
+    };
+
+    function finalize(error) {
+      if (isComplete) return;
+      isComplete = true;
+      sourceStreamBase.onread = originalSourceOnread;
+
+      if (error) {
+        if (source[kState].state === 'readable') {
+          readableStreamError(source, error);
+        }
+        if (dest[kState].state === 'writable') {
+          writableStreamAbort(dest, error);
+        }
+        promise.reject(error);
+      } else {
+        if (source[kState].state === 'readable') {
+          readableStreamClose(source);
+        }
+        if (dest[kState].state === 'writable' &&
+            !writableStreamCloseQueuedOrInFlight(dest)) {
+          PromisePrototypeThen(
+            writableStreamClose(dest),
+            () => promise.resolve(),
+            (err) => promise.reject(err),
+          );
+        } else {
+          promise.resolve();
+        }
+      }
+    }
+
     try {
       const pipe = new StreamPipe(sourceStreamBase, destStreamBase);
       pipe.onunpipe = () => {
-        promise.resolve();
+        if (pipeError) {
+          finalize(pipeError);
+        }
       };
+      pipe.oncomplete = () => finalize(pipeError);
       pipe.start();
     } catch (error) {
+      sourceStreamBase.onread = originalSourceOnread;
       promise.reject(error);
     }
 

test/parallel/test-whatwg-webstreams-internal-pipe.js

@@ -1,42 +1,25 @@
 // Flags: --expose-internals --no-warnings
 'use strict';
 
-// Tests for the internal StreamBase pipe optimization infrastructure
-// described in nodejs/performance#134
-//
-// Note(mertcanaltin): Full fast-path testing requires real StreamBase implementations
-// (like HTTP/2 streams or TCP sockets), not JSStream mocks.
-// These tests verify the marker attachment and fallback behavior.
+// Tests for internal StreamBase pipe optimization (nodejs/performance#134)
 
 const common = require('../common');
-
 const assert = require('assert');
-
-const {
-  internalBinding,
-} = require('internal/test/binding');
+const { internalBinding } = require('internal/test/binding');
 
 const {
   newWritableStreamFromStreamBase,
   newReadableStreamFromStreamBase,
 } = require('internal/webstreams/adapters');
 
-const {
-  kStreamBase,
-} = require('internal/webstreams/util');
-
-const {
-  JSStream,
-} = internalBinding('js_stream');
+const { kStreamBase } = require('internal/webstreams/util');
+const { JSStream } = internalBinding('js_stream');
 
 // kStreamBase marker is attached to ReadableStream
 {
   const stream = new JSStream();
   const readable = newReadableStreamFromStreamBase(stream);
-
   assert.strictEqual(readable[kStreamBase], stream);
-
-  // Cleanup
   stream.emitEOF();
 }
 
@@ -47,10 +30,7 @@ const {
   stream.onshutdown = (req) => req.oncomplete();
 
   const writable = newWritableStreamFromStreamBase(stream);
-
   assert.strictEqual(writable[kStreamBase], stream);
-
-  // Cleanup
   writable.close();
 }
 
@@ -65,18 +45,15 @@ const {
     },
   });
 
-  const ws = new WritableStream({
-    write() {},
-  });
+  const ws = new WritableStream({ write() {} });
 
   assert.strictEqual(rs[kStreamBase], undefined);
   assert.strictEqual(ws[kStreamBase], undefined);
 
-  // Pipe should still work (standard path)
   rs.pipeTo(ws).then(common.mustCall());
 }
 
-// Mixed streams (one internal, one JS) use standard path
+// Mixed streams use standard JS path
 {
   const stream = new JSStream();
   stream.onshutdown = (req) => req.oncomplete();
@@ -85,17 +62,13 @@ const {
   const { WritableStream } = require('stream/web');
   const chunks = [];
   const ws = new WritableStream({
-    write(chunk) {
-      chunks.push(chunk);
-    },
+    write(chunk) { chunks.push(chunk); },
   });
 
-  // Readable has kStreamBase, ws does not - should use standard path
   assert.ok(readable[kStreamBase]);
   assert.strictEqual(ws[kStreamBase], undefined);
 
   const pipePromise = readable.pipeTo(ws);
-
   stream.readBuffer(Buffer.from('hello'));
   stream.emitEOF();
 
@@ -104,12 +77,47 @@ const {
   }));
 }
 
-// Verify kStreamBase is the correct symbol from util
+// Verify kStreamBase symbol identity
 {
-  const {
-    kStreamBase: kStreamBase2,
-  } = require('internal/webstreams/util');
-
-  // Should be the same symbol
+  const { kStreamBase: kStreamBase2 } = require('internal/webstreams/util');
   assert.strictEqual(kStreamBase, kStreamBase2);
 }
+
+// FileHandle.readableWebStream() uses async reads, not StreamBase
+{
+  const fs = require('fs/promises');
+  const path = require('path');
+  const os = require('os');
+
+  async function testFileStreamPipe() {
+    const tmpDir = os.tmpdir();
+    const testFile = path.join(tmpDir, `test-webstream-pipe-${process.pid}.txt`);
+    const testData = 'Hello, WebStreams pipe!';
+
+    await fs.writeFile(testFile, testData);
+
+    try {
+      const fileHandle = await fs.open(testFile, 'r');
+      const readable = fileHandle.readableWebStream();
+
+      assert.strictEqual(readable[kStreamBase], undefined);
+
+      const chunks = [];
+      const writable = new (require('stream/web').WritableStream)({
+        write(chunk) { chunks.push(chunk); },
+      });
+
+      await readable.pipeTo(writable);
+      await fileHandle.close();
+
+      const result = Buffer.concat(chunks.map((c) =>
+        (c instanceof Uint8Array ? Buffer.from(c) : Buffer.from(c)),
+      )).toString();
+      assert.strictEqual(result, testData);
+    } finally {
+      await fs.unlink(testFile).catch(() => {});
+    }
+  }
+
+  testFileStreamPipe().then(common.mustCall());
+}