[pull] main from nodejs:main

lib/internal/streams/iter/broadcast.js

@@ -58,6 +58,7 @@ const {
   kResolvedPromise,
   clampHWM,
   convertChunks,
+  getWriterSignal,
   getMinCursor,
   hasProtocol,
   onSignalAbort,
@@ -526,41 +527,41 @@ class BroadcastWriter {
     return this.#isClosedOrAborted() ? null : this.#broadcast[kGetDesiredSize]();
   }
 
-  #canUseWriteFastPath(options) {
-    return !options?.signal && !this.#isClosed() && !this.#aborted &&
+  #canUseWriteFastPath(signal) {
+    return !signal && !this.#isClosed() && !this.#aborted &&
         this.#broadcast[kCanWrite]();
   }
 
   write(chunk, options) {
+    const signal = getWriterSignal(options);
     // Fast path: no signal, writer open, buffer has space
-    if (this.#canUseWriteFastPath(options)) {
+    if (this.#canUseWriteFastPath(signal)) {
       const converted = toUint8Array(chunk);
       this.#broadcast[kWrite]([converted]);
       this.#totalBytes += TypedArrayPrototypeGetByteLength(converted);
       return kResolvedPromise;
     }
-    return this.#writevSlow([chunk], options);
+    return this.#writevSlow([chunk], signal);
   }
 
   writev(chunks, options) {
     if (!ArrayIsArray(chunks)) {
       throw new ERR_INVALID_ARG_TYPE('chunks', 'Array', chunks);
     }
+    const signal = getWriterSignal(options);
     // Fast path: no signal, writer open, buffer has space
-    if (this.#canUseWriteFastPath(options)) {
+    if (this.#canUseWriteFastPath(signal)) {
       const converted = convertChunks(chunks);
       this.#broadcast[kWrite](converted);
       for (let i = 0; i < converted.length; i++) {
         this.#totalBytes += TypedArrayPrototypeGetByteLength(converted[i]);
       }
       return kResolvedPromise;
     }
-    return this.#writevSlow(chunks, options);
+    return this.#writevSlow(chunks, signal);
   }
 
-  async #writevSlow(chunks, options) {
-    const signal = options?.signal;
-
+  async #writevSlow(chunks, signal) {
     // Check for pre-aborted
     signal?.throwIfAborted();
 
@@ -623,6 +624,7 @@ class BroadcastWriter {
 
   // end() is synchronous internally - signal accepted for interface compliance.
   end(options) {
+    getWriterSignal(options);
     if (this.#isClosed()) return this.#closed;
     this.#closed = PromiseResolve(this.#totalBytes);
     this.#broadcast[kEnd]();

lib/internal/streams/iter/classic.js

@@ -61,6 +61,7 @@ const {
 } = require('internal/streams/iter/types');
 
 const {
+  getWriterSignal,
   validateBackpressure,
   toUint8Array,
 } = require('internal/streams/iter/utils');
@@ -572,10 +573,11 @@ function fromWritable(writable, options = kNullPrototype) {
     // as 'error' events caught by our generic error handler, rejecting
     // the next pending operation rather than the already-resolved one.
     //
-    // The options.signal parameter from the Writer interface is ignored.
-    // Classic stream.Writable has no per-write abort signal support;
-    // cancellation should be handled at the pipeline level instead.
-    write(chunk) {
+    // The options.signal parameter from the Writer interface is validated but
+    // otherwise ignored. Classic stream.Writable has no per-write abort signal
+    // support; cancellation should be handled at the pipeline level instead.
+    write(chunk, options) {
+      getWriterSignal(options);
       if (!isWritable()) {
         return PromiseReject(new ERR_STREAM_WRITE_AFTER_END());
       }
@@ -617,10 +619,11 @@ function fromWritable(writable, options = kNullPrototype) {
       return PromiseResolve();
     },
 
-    writev(chunks) {
+    writev(chunks, options) {
       if (!ArrayIsArray(chunks)) {
         throw new ERR_INVALID_ARG_TYPE('chunks', 'Array', chunks);
       }
+      getWriterSignal(options);
       if (!isWritable()) {
         return PromiseReject(new ERR_STREAM_WRITE_AFTER_END());
       }
@@ -666,8 +669,10 @@ function fromWritable(writable, options = kNullPrototype) {
       return -1;
     },
 
-    // options.signal is ignored for the same reason as write().
-    end() {
+    // options.signal is validated but otherwise ignored for the same reason as
+    // write().
+    end(options) {
+      getWriterSignal(options);
       if ((writable.writableFinished ?? false) ||
           (writable.destroyed ?? false)) {
         cleanup();

lib/internal/streams/iter/push.js

@@ -42,6 +42,7 @@ const {
   onSignalAbort,
   toUint8Array,
   convertChunks,
+  getWriterSignal,
   parsePullArgs,
   validateBackpressure,
 } = require('internal/streams/iter/utils');
@@ -565,26 +566,28 @@ class PushWriter {
   }
 
   write(chunk, options) {
-    if (!options?.signal && this.#queue.canWriteSync()) {
+    const signal = getWriterSignal(options);
+    if (!signal && this.#queue.canWriteSync()) {
       const bytes = toUint8Array(chunk);
       this.#queue.writeSync([bytes]);
       return kResolvedPromise;
     }
     const bytes = toUint8Array(chunk);
-    return this.#queue.writeAsync([bytes], options?.signal);
+    return this.#queue.writeAsync([bytes], signal);
   }
 
   writev(chunks, options) {
     if (!ArrayIsArray(chunks)) {
       throw new ERR_INVALID_ARG_TYPE('chunks', 'Array', chunks);
     }
-    if (!options?.signal && this.#queue.canWriteSync()) {
+    const signal = getWriterSignal(options);
+    if (!signal && this.#queue.canWriteSync()) {
       const bytes = convertChunks(chunks);
       this.#queue.writeSync(bytes);
       return kResolvedPromise;
     }
     const bytes = convertChunks(chunks);
-    return this.#queue.writeAsync(bytes, options?.signal);
+    return this.#queue.writeAsync(bytes, signal);
   }
 
   writeSync(chunk) {
@@ -601,6 +604,7 @@ class PushWriter {
   }
 
   end(options) {
+    getWriterSignal(options);
     const result = this.#queue.end();
     if (result === -2) {
       // Errored: reject with stored error
@@ -668,7 +672,7 @@ function createReadable(queue) {
         },
         async throw(error) {
           queue.consumerThrow(error);
-          return { __proto__: null, done: true, value: undefined };
+          throw error;
         },
       };
     },

lib/internal/streams/iter/utils.js

@@ -35,7 +35,10 @@ const { isError } = require('internal/util');
 
 const { isSharedArrayBuffer, isUint8Array } = require('internal/util/types');
 
-const { validateOneOf } = require('internal/validators');
+const {
+  validateAbortSignal,
+  validateOneOf,
+} = require('internal/validators');
 
 // Cached resolved promise to avoid allocating a new one on every sync fast-path.
 const kResolvedPromise = PromiseResolve();
@@ -267,6 +270,17 @@ function convertChunks(chunks) {
   return result;
 }
 
+/**
+ * Validate Writer options and return options.signal.
+ * @param {object|undefined} options
+ * @returns {AbortSignal|undefined}
+ */
+function getWriterSignal(options) {
+  const signal = options?.signal;
+  validateAbortSignal(signal, 'options.signal');
+  return signal;
+}
+
 /**
  * Wrap a caught value as an Error, converting non-Error values.
  * @param {unknown} error
@@ -378,6 +392,7 @@ module.exports = {
   clampHWM,
   concatBytes,
   convertChunks,
+  getWriterSignal,
   getMinCursor,
   hasProtocol,
   isPullOptions,

test/parallel/test-stream-iter-push-writer.js

@@ -147,10 +147,16 @@ async function testOndrainRejectsOnConsumerThrow() {
   // Consumer throws via iterator.throw() before draining enough
   // to clear backpressure. The drain should reject.
   const iter = readable[Symbol.asyncIterator]();
-  await iter.throw(new Error('consumer error'));
+  const err = new Error('consumer error');
+  const drainRejects = assert.rejects(drainPromise, (e) => e === err);
+  const pendingWriteRejects = pendingWrite.catch(() => {});
+  await assert.rejects(
+    () => iter.throw(err),
+    (e) => e === err,
+  );
 
-  await assert.rejects(drainPromise, /consumer error/);
-  await pendingWrite.catch(() => {}); // Ignore write rejection
+  await drainRejects;
+  await pendingWriteRejects; // Ignore write rejection
 }
 
 async function testWritev() {
@@ -303,7 +309,11 @@ async function testConsumerThrowRejectsWrites() {
   writer.writeSync('a');
 
   const iter = readable[Symbol.asyncIterator]();
-  await iter.throw(new Error('consumer boom'));
+  const err = new Error('consumer boom');
+  await assert.rejects(
+    () => iter.throw(err),
+    (e) => e === err,
+  );
 
   // Subsequent async writes should reject with the consumer's error
   await assert.rejects(
@@ -312,6 +322,18 @@ async function testConsumerThrowRejectsWrites() {
   );
 }
 
+async function testConsumerThrowRejectsWithThrownError() {
+  const { readable } = push();
+
+  const iter = readable[Symbol.asyncIterator]();
+  const err = new Error('boom');
+
+  await assert.rejects(
+    () => iter.throw(err),
+    (e) => e === err,
+  );
+}
+
 // end() resolves a pending read as done:true
 async function testEndResolvesPendingRead() {
   const { writer, readable } = push();
@@ -373,14 +395,16 @@ async function testConsumerThrowRejectsPendingRead() {
   await new Promise(setImmediate);
 
   const err = new Error('consumer read boom');
-  const throwResult = await iter.throw(err);
-  assert.strictEqual(throwResult.value, undefined);
-  assert.strictEqual(throwResult.done, true);
-
-  await assert.rejects(
+  const readRejects = assert.rejects(
     () => readPromise,
     (e) => e === err,
   );
+  await assert.rejects(
+    () => iter.throw(err),
+    (e) => e === err,
+  );
+
+  await readRejects;
 }
 
 // end() while writes are pending rejects those writes
@@ -527,6 +551,7 @@ Promise.all([
   testWriteUint8Array(),
   testOndrainWaitsForDrain(),
   testConsumerThrowRejectsWrites(),
+  testConsumerThrowRejectsWithThrownError(),
   testEndResolvesPendingRead(),
   testFailRejectsPendingRead(),
   testFailRejectsFutureReadWithFalsyReason(),

test/parallel/test-stream-iter-validation.js

@@ -3,8 +3,9 @@
 
 const common = require('../common');
 const assert = require('assert');
+const { Writable } = require('stream');
 const {
-  from, fromSync, pull, pullSync, pipeTo,
+  from, fromSync, pull, pullSync, pipeTo, fromWritable,
   push, duplex, broadcast, Broadcast, share, shareSync,
   Share, SyncShare,
   bytes, bytesSync, text, textSync,
@@ -42,6 +43,19 @@ assert.throws(() => push({ signal: {} }), { code: 'ERR_INVALID_ARG_TYPE' });
 assert.throws(() => push(42, {}), { code: 'ERR_INVALID_ARG_TYPE' });
 assert.throws(() => push('bad', {}), { code: 'ERR_INVALID_ARG_TYPE' });
 
+// Writer options.signal must be AbortSignal
+{
+  const { writer } = push();
+  const badOptions = { signal: 'bad' };
+  assert.throws(() => writer.write('a', badOptions),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  assert.throws(() => writer.writev(['b'], badOptions),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  assert.throws(() => writer.end(badOptions),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  writer.endSync();
+}
+
 // Writer.writev requires array
 {
   const { writer } = push();
@@ -147,6 +161,19 @@ assert.throws(() => broadcast({ highWaterMark: Number.MAX_SAFE_INTEGER + 1 }),
 assert.throws(() => broadcast({ signal: {} }), { code: 'ERR_INVALID_ARG_TYPE' });
 assert.throws(() => broadcast({ backpressure: 'bad' }), { code: 'ERR_INVALID_ARG_VALUE' });
 
+// BroadcastWriter options.signal must be AbortSignal
+{
+  const { writer } = broadcast();
+  const badOptions = { signal: 'bad' };
+  assert.throws(() => writer.write('a', badOptions),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  assert.throws(() => writer.writev(['b'], badOptions),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  assert.throws(() => writer.end(badOptions),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  writer.endSync();
+}
+
 // BroadcastWriter.writev requires array
 {
   const { writer } = broadcast();
@@ -160,6 +187,24 @@ assert.throws(() => broadcast({ backpressure: 'bad' }), { code: 'ERR_INVALID_ARG
 // Broadcast.from rejects non-streamable input
 assert.throws(() => Broadcast.from(42), { code: 'ERR_INVALID_ARG_TYPE' });
 
+// fromWritable Writer options.signal must be AbortSignal
+{
+  const writable = new Writable({
+    write(chunk, encoding, callback) {
+      callback();
+    },
+  });
+  const writer = fromWritable(writable);
+  const badOptions = { signal: 'bad' };
+  assert.throws(() => writer.write('a', badOptions),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  assert.throws(() => writer.writev(['b'], badOptions),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  assert.throws(() => writer.end(badOptions),
+                { code: 'ERR_INVALID_ARG_TYPE' });
+  writable.destroy();
+}
+
 // =============================================================================
 // share() / shareSync() validation
 // =============================================================================