diff --git a/lib/_http_outgoing.js b/lib/_http_outgoing.js index 7694fe4b5b3f3e..6070eb79467c63 100644 --- a/lib/_http_outgoing.js +++ b/lib/_http_outgoing.js @@ -904,6 +904,8 @@ OutgoingMessage.prototype.write = function write(chunk, encoding, callback) { return ret; }; +const outgoingMessagePrototypeWrite = OutgoingMessage.prototype.write; + function onError(msg, err, callback) { if (msg.destroyed) { return; @@ -1260,4 +1262,5 @@ module.exports = { validateHeaderName, validateHeaderValue, OutgoingMessage, + outgoingMessagePrototypeWrite, }; diff --git a/lib/internal/streams/utils.js b/lib/internal/streams/utils.js index 45f55316104fe8..06b05bab01846f 100644 --- a/lib/internal/streams/utils.js +++ b/lib/internal/streams/utils.js @@ -345,6 +345,7 @@ module.exports = { isWritableEnded, isWritableFinished, isWritableErrored, + isOutgoingMessage, isServerRequest, isServerResponse, willEmitClose, diff --git a/lib/internal/streams/writable.js b/lib/internal/streams/writable.js index 47e003ea3a8a88..91e9c38ad631f6 100644 --- a/lib/internal/streams/writable.js +++ b/lib/internal/streams/writable.js @@ -512,6 +512,8 @@ Writable.prototype.write = function(chunk, encoding, cb) { return _write(this, chunk, encoding, cb) === true; }; +const writablePrototypeWrite = Writable.prototype.write; + Writable.prototype.cork = function() { const state = this._writableState; @@ -1150,7 +1152,11 @@ Writable.fromWeb = function(writableStream, options) { }; Writable.toWeb = function(streamWritable) { - return lazyWebStreams().newWritableStreamFromStreamWritable(streamWritable); + return lazyWebStreams().newWritableStreamFromStreamWritable( + streamWritable, + undefined, + writablePrototypeWrite, + ); }; Writable.prototype[SymbolAsyncDispose] = async function() { diff --git a/lib/internal/webstreams/adapters.js b/lib/internal/webstreams/adapters.js index d3f8212f1a135b..6617ac72fca753 100644 --- a/lib/internal/webstreams/adapters.js +++ b/lib/internal/webstreams/adapters.js @@ -1,25 +1,46 @@ 'use strict'; const { + ArrayBufferPrototypeGetDetached, ArrayPrototypeFilter, + BigInt64Array, + BigUint64Array, + DataView, + Float32Array, + Float64Array, + FunctionPrototypeCall, + Int16Array, + Int32Array, + Int8Array, ObjectKeys, PromisePrototypeThen, PromiseResolve, PromiseWithResolvers, SafePromiseAllReturnVoid, - SafePromisePrototypeFinally, SafeSet, StringPrototypeStartsWith, Symbol, + SymbolDispose, TypeError, TypedArrayPrototypeGetBuffer, TypedArrayPrototypeGetByteLength, TypedArrayPrototypeGetByteOffset, + TypedArrayPrototypeGetSymbolToStringTag, + TypedArrayPrototypeSet, + Uint16Array, + Uint32Array, Uint8Array, + Uint8ClampedArray, + globalThis: { + Float16Array, + }, } = primordials; const { TextEncoder } = require('internal/encoding'); +let addAbortListener; +let outgoingMessagePrototypeWrite; + const { ReadableStream, isReadableStream, @@ -27,6 +48,7 @@ const { const { WritableStream, + getWritableStreamDefaultControllerSignal, isWritableStream, } = require('internal/webstreams/writablestream'); @@ -44,6 +66,7 @@ const { const { isDestroyed, + isOutgoingMessage, isReadable, isWritable, isWritableEnded, @@ -55,8 +78,18 @@ const { const { isAnyArrayBuffer, + isArrayBufferView, + isDataView, + isSharedArrayBuffer, + isUint8Array, } = require('internal/util/types'); +const { + ArrayBufferViewGetBuffer, + ArrayBufferViewGetByteLength, + ArrayBufferViewGetByteOffset, +} = require('internal/webstreams/util'); + const { AbortError, ErrnoException, @@ -69,6 +102,7 @@ const { } = require('internal/errors'); const { + constructSharedArrayBuffer, getDeprecationWarningEmitter, kEmptyObject, normalizeEncoding, @@ -141,6 +175,62 @@ function handleKnownInternalErrors(cause) { const noop = () => {}; +function cloneArrayBufferView(view) { + const viewIsDataView = isDataView(view); + const buffer = ArrayBufferViewGetBuffer(view); + + // A detached backing buffer cannot be changed by the caller. Passing the + // original view through also preserves the native stream's validation. + if (!isSharedArrayBuffer(buffer) && + ArrayBufferPrototypeGetDetached(buffer)) { + return view; + } + + const byteLength = ArrayBufferViewGetByteLength(view); + const byteOffset = ArrayBufferViewGetByteOffset(view); + const copiedBytes = new Uint8Array( + isSharedArrayBuffer(buffer) ? + constructSharedArrayBuffer(byteLength) : + byteLength, + ); + TypedArrayPrototypeSet( + copiedBytes, + new Uint8Array(buffer, byteOffset, byteLength), + ); + const copiedBuffer = ArrayBufferViewGetBuffer(copiedBytes); + + if (viewIsDataView) { + return new DataView(copiedBuffer); + } + + switch (TypedArrayPrototypeGetSymbolToStringTag(view)) { + case 'BigInt64Array': + return new BigInt64Array(copiedBuffer); + case 'BigUint64Array': + return new BigUint64Array(copiedBuffer); + case 'Float16Array': + return new Float16Array(copiedBuffer); + case 'Float32Array': + return new Float32Array(copiedBuffer); + case 'Float64Array': + return new Float64Array(copiedBuffer); + case 'Int8Array': + return new Int8Array(copiedBuffer); + case 'Int16Array': + return new Int16Array(copiedBuffer); + case 'Int32Array': + return new Int32Array(copiedBuffer); + case 'Uint8Array': + return Buffer.isBuffer(view) ? Buffer.from(copiedBuffer) : copiedBytes; + case 'Uint8ClampedArray': + return new Uint8ClampedArray(copiedBuffer); + case 'Uint16Array': + return new Uint16Array(copiedBuffer); + case 'Uint32Array': + return new Uint32Array(copiedBuffer); + } +} + /** * @typedef {import('../../stream').Writable} Writable * @typedef {import('../../stream').Readable} Readable @@ -155,9 +245,14 @@ const noop = () => {}; /** * @param {Writable} streamWritable * @param {object} [options] + * @param {Function} [writablePrototypeWrite] * @returns {WritableStream} */ -function newWritableStreamFromStreamWritable(streamWritable, options = kEmptyObject) { +function newWritableStreamFromStreamWritable( + streamWritable, + options = kEmptyObject, + writablePrototypeWrite, +) { // Not using the internal/streams/utils isWritableNodeStream utility // here because it will return false if streamWritable is a Duplex // whose writable option is false. For a Duplex that is not writable, @@ -193,22 +288,188 @@ function newWritableStreamFromStreamWritable(streamWritable, options = kEmptyObj }; let controller; - let backpressurePromise; + let pendingWriteState; + let isTerminated = false; + let terminalReason; let closed; + let abortSignal; + let abortDisposable; + + function disposeAbortListener() { + abortDisposable?.[SymbolDispose](); + abortDisposable = undefined; + } + + function recordTerminalReason(reason) { + if (!isTerminated) { + isTerminated = true; + terminalReason = reason; + } + } + + function resolvePendingWrite(state) { + if (pendingWriteState !== state) { + return; + } + pendingWriteState = undefined; + disposeAbortListener(); + state.deferred.resolve(); + } + + function rejectPendingWrite(error) { + const state = pendingWriteState; + if (state === undefined) { + return; + } + pendingWriteState = undefined; + disposeAbortListener(); + state.deferred.reject(error); + } + + function listenForPendingAbort() { + addAbortListener ??= + require('internal/events/abort_listener').addAbortListener; + abortDisposable = addAbortListener(abortSignal, () => { + // Release an in-flight write so the sink's abort algorithm can run. + // The abort algorithm is solely responsible for destroying the stream. + recordTerminalReason(abortSignal.reason); + rejectPendingWrite(abortSignal.reason); + }); + } + + function maybeResolvePendingWrite(state) { + if (state.writeComplete && state.backpressureCleared) { + resolvePendingWrite(state); + } + } + + function markWriteComplete(state, error) { + if (pendingWriteState !== state) { + return; + } + if (error != null) { + rejectPendingWrite(error); + return; + } + state.writeComplete = true; + maybeResolvePendingWrite(state); + } + + function createPendingWriteState(waitForWriteCallback) { + const state = { + __proto__: null, + deferred: PromiseWithResolvers(), + writeComplete: !waitForWriteCallback, + backpressureCleared: false, + }; + pendingWriteState = state; + listenForPendingAbort(); + return state; + } function onDrain() { - backpressurePromise?.resolve(); + const state = pendingWriteState; + if (state === undefined) { + return; + } + state.backpressureCleared = true; + maybeResolvePendingWrite(state); + } + + function onWrite(error) { + const state = pendingWriteState; + if (state === undefined) { + return; + } + if (error != null) { + error = handleKnownInternalErrors(error); + } + markWriteComplete(state, error); + } + + function throwSyncWriteError(error) { + disposeAbortListener(); + // When the kDestroyOnSyncError flag is set (e.g. for + // CompressionStream), a sync throw must also destroy the + // stream so the readable side is errored too. Without this + // the readable side hangs forever. This replicates the + // TransformStream semantics: error both sides on any throw + // in the transform path. + if (options[kDestroyOnSyncError]) { + destroy(streamWritable, error); + } + throw error; + } + + function throwIfTerminated() { + if (abortSignal.aborted) { + recordTerminalReason(abortSignal.reason); + } + if (isTerminated) { + throw terminalReason; + } + } + + function writeChunk( + writeMethod, + chunk, + needDrainBeforeWrite, + waitForWriteCallback, + ) { + if (!waitForWriteCallback && !needDrainBeforeWrite) { + try { + const writeReturn = FunctionPrototypeCall( + writeMethod, + streamWritable, + chunk, + ); + throwIfTerminated(); + if (controller === undefined || + writeReturn || + !streamWritable.writableNeedDrain) { + return; + } + } catch (error) { + throwSyncWriteError(error); + } + + const state = createPendingWriteState(false); + return state.deferred.promise; + } + + const state = createPendingWriteState(waitForWriteCallback); + try { + // The return value only reports backpressure. The callback reports when + // the chunk has been handled and can safely be mutated by the caller. + const writeReturn = waitForWriteCallback ? + FunctionPrototypeCall(writeMethod, streamWritable, chunk, onWrite) : + FunctionPrototypeCall(writeMethod, streamWritable, chunk); + throwIfTerminated(); + const shouldWaitForDrain = + (needDrainBeforeWrite || !writeReturn) && + streamWritable.writableNeedDrain; + state.backpressureCleared = !shouldWaitForDrain; + maybeResolvePendingWrite(state); + } catch (error) { + resolvePendingWrite(state); + PromisePrototypeThen(state.deferred.promise, undefined, noop); + throwSyncWriteError(error); + } + + return state.deferred.promise; } const cleanup = eos(streamWritable, (error) => { error = handleKnownInternalErrors(error); cleanup(); + disposeAbortListener(); // This is a protection against non-standard, legacy streams // that happen to emit an error event again after finished is called. streamWritable.on('error', () => {}); if (error != null) { - backpressurePromise?.reject(error); + recordTerminalReason(error); + rejectPendingWrite(error); // If closed is not undefined, the error is happening // after the WritableStream close has already started. // We need to reject it here. @@ -226,57 +487,98 @@ function newWritableStreamFromStreamWritable(streamWritable, options = kEmptyObj closed = undefined; return; } - controller.error(new AbortError()); + const abortError = new AbortError(); + recordTerminalReason(abortError); + rejectPendingWrite(abortError); + controller.error(abortError); controller = undefined; }); streamWritable.on('drain', onDrain); return new WritableStream({ - start(c) { controller = c; }, + start(c) { + controller = c; + abortSignal = getWritableStreamDefaultControllerSignal(c); + }, write(chunk) { + let writeMethod; + let needDrainBeforeWrite; + let waitForWriteCallback; try { options[kValidateChunk]?.(chunk); - if (!streamWritable.writableObjectMode && isAnyArrayBuffer(chunk)) { + writeMethod = streamWritable.write; + const writableState = streamWritable._writableState; + const objectMode = streamWritable.writableObjectMode; + const isBufferSourceChunk = + isAnyArrayBuffer(chunk) || isArrayBufferView(chunk); + if (!objectMode && isAnyArrayBuffer(chunk)) { chunk = new Uint8Array(chunk); } - if (streamWritable.writableNeedDrain || !streamWritable.write(chunk)) { - backpressurePromise = PromiseWithResolvers(); - if (!streamWritable.writableNeedDrain) { - backpressurePromise.resolve(); + needDrainBeforeWrite = streamWritable.writableNeedDrain; + const isArrayBufferViewChunk = isArrayBufferView(chunk); + if (!objectMode && + isOutgoingMessage(streamWritable) && + isArrayBufferViewChunk && + !isUint8Array(chunk)) { + outgoingMessagePrototypeWrite ??= + require('_http_outgoing').outgoingMessagePrototypeWrite; + if (writeMethod === outgoingMessagePrototypeWrite) { + throw new ERR_INVALID_ARG_TYPE( + 'chunk', ['string', 'Buffer', 'Uint8Array'], chunk); } - return SafePromisePrototypeFinally( - backpressurePromise.promise, () => { - backpressurePromise = undefined; - }); } - } catch (error) { - // When the kDestroyOnSyncError flag is set (e.g. for - // CompressionStream), a sync throw must also destroy the - // stream so the readable side is errored too. Without this - // the readable side hangs forever. This replicates the - // TransformStream semantics: error both sides on any throw - // in the transform path. - if (options[kDestroyOnSyncError]) { - destroy(streamWritable, error); + const canAwaitWriteCallback = + streamWritable instanceof Writable && + writeMethod === writablePrototypeWrite && + streamWritable._readableState === undefined && + writableState?.corked === 0; + waitForWriteCallback = + isBufferSourceChunk && canAwaitWriteCallback; + if (!objectMode && + isArrayBufferViewChunk && + !waitForWriteCallback) { + // Duplex streams can retain chunks in their readable side or + // forward them after their own write callback. A corked stream does + // not invoke callbacks until uncork() or end(), and streams that + // override write() may not support callbacks at all. Preserve the + // existing completion timing for those cases by passing a private, + // same-type copy to the native stream. + chunk = cloneArrayBufferView(chunk); } - throw error; + } catch (error) { + throwSyncWriteError(error); } + + return writeChunk( + writeMethod, + chunk, + needDrainBeforeWrite, + waitForWriteCallback, + ); }, abort(reason) { + disposeAbortListener(); destroy(streamWritable, reason); }, close() { if (closed === undefined && !isWritableEnded(streamWritable)) { closed = PromiseWithResolvers(); - streamWritable.end(); + try { + streamWritable.end(); + } catch (error) { + closed = undefined; + disposeAbortListener(); + throw error; + } return closed.promise; } controller = undefined; + disposeAbortListener(); return PromiseResolve(); }, }, strategy); diff --git a/lib/internal/webstreams/writablestream.js b/lib/internal/webstreams/writablestream.js index f0eb687be129d1..3b570a0af12d10 100644 --- a/lib/internal/webstreams/writablestream.js +++ b/lib/internal/webstreams/writablestream.js @@ -596,6 +596,10 @@ const isWritableStreamDefaultWriter = const isWritableStreamDefaultController = isBrandCheck('WritableStreamDefaultController'); +function getWritableStreamDefaultControllerSignal(controller) { + return controller[kState].abortController.signal; +} + function createWritableStreamState() { return { __proto__: null, @@ -1357,6 +1361,7 @@ module.exports = { isWritableStreamDefaultController, isWritableStreamDefaultWriter, + getWritableStreamDefaultControllerSignal, isWritableStreamLocked, setupWritableStreamDefaultWriter, writableStreamAbort, diff --git a/test/parallel/test-stream-duplex.js b/test/parallel/test-stream-duplex.js index de67290a7e76cd..9908d000cb7ec8 100644 --- a/test/parallel/test-stream-duplex.js +++ b/test/parallel/test-stream-duplex.js @@ -120,7 +120,7 @@ process.on('exit', () => { this.push(null); }, write: common.mustCall((chunk) => { - assert.strictEqual(chunk, dataToWrite); + assert.deepStrictEqual(chunk, dataToWrite); }) }); @@ -143,7 +143,7 @@ process.on('exit', () => { this.push(null); }, write: common.mustCall((chunk) => { - assert.strictEqual(chunk, dataToWrite); + assert.deepStrictEqual(chunk, dataToWrite); }) }); diff --git a/test/parallel/test-webstreams-adapters-sync-write-error.js b/test/parallel/test-webstreams-adapters-sync-write-error.js index 748f682365ee93..ff6dbddf377f85 100644 --- a/test/parallel/test-webstreams-adapters-sync-write-error.js +++ b/test/parallel/test-webstreams-adapters-sync-write-error.js @@ -1,6 +1,6 @@ 'use strict'; // Flags: --no-warnings --expose-internals -require('../common'); +const common = require('../common'); const assert = require('assert'); const test = require('node:test'); const { Duplex, Writable } = require('stream'); @@ -9,6 +9,13 @@ const { newReadableWritablePairFromDuplex, } = require('internal/webstreams/adapters'); +function isSameError(expected) { + return common.mustCall((actual) => { + assert.strictEqual(actual, expected); + return true; + }); +} + // Verify that when the underlying Node.js stream throws synchronously from // write(), the writable web stream properly rejects but does not destroy // the stream (destroy-on-sync-throw is only used internally by @@ -34,6 +41,72 @@ test('WritableStream from Node.js stream handles sync write throw', async () => assert.strictEqual(writable.destroyed, false); }); +test('WritableStream from Node.js stream handles async write error', async () => { + const error = new Error('boom'); + const writable = new Writable({ + write(_chunk, _encoding, callback) { + setImmediate(callback, error); + }, + }); + const writer = Writable.toWeb(writable).getWriter(); + + await Promise.all([ + assert.rejects(writer.write(Buffer.from('hello')), isSameError(error)), + assert.rejects(writer.closed, isSameError(error)), + ]); +}); + +test('WritableStream aborts while a native write is pending', async () => { + const error = new Error('abort'); + let finishWrite; + let startWrite; + const writeStarted = new Promise((resolve) => { + startWrite = resolve; + }); + const writable = new Writable({ + write(_chunk, _encoding, callback) { + finishWrite = callback; + startWrite(); + }, + }); + const writer = Writable.toWeb(writable).getWriter(); + const writePromise = writer.write(Buffer.from('hello')); + await writeStarted; + + const writeRejected = assert.rejects(writePromise, isSameError(error)); + const closedRejected = assert.rejects(writer.closed, isSameError(error)); + await Promise.all([ + writer.abort(error), + writeRejected, + closedRejected, + ]); + + finishWrite(); + await new Promise(setImmediate); + assert.strictEqual(writable.destroyed, true); +}); + +test('WritableStream handles destruction while a write is pending', async () => { + const error = new Error('destroy'); + let startWrite; + const writeStarted = new Promise((resolve) => { + startWrite = resolve; + }); + const writable = new Writable({ + write(_chunk, _encoding, _callback) { + startWrite(); + }, + }); + const writer = Writable.toWeb(writable).getWriter(); + const writePromise = writer.write(Buffer.from('hello')); + await writeStarted; + + const writeRejected = assert.rejects(writePromise, isSameError(error)); + const closedRejected = assert.rejects(writer.closed, isSameError(error)); + writable.destroy(error); + await Promise.all([writeRejected, closedRejected]); +}); + test('Duplex-backed pair does NOT destroy on sync write throw', async () => { const error = new TypeError('invalid chunk'); const duplex = new Duplex({ diff --git a/test/parallel/test-webstreams-adapters-writable-buffer-sources.js b/test/parallel/test-webstreams-adapters-writable-buffer-sources.js index 995db97e747364..1a4a8eeb01f8ba 100644 --- a/test/parallel/test-webstreams-adapters-writable-buffer-sources.js +++ b/test/parallel/test-webstreams-adapters-writable-buffer-sources.js @@ -3,11 +3,46 @@ const common = require('../common'); const assert = require('assert'); const { Buffer } = require('buffer'); +const { ServerResponse } = require('http'); const { Duplex, Writable } = require('stream'); const { suite, test } = require('node:test'); const ctors = [ArrayBuffer, SharedArrayBuffer]; +function createServerResponse() { + const socket = new Writable({ + write(chunk, encoding, callback) { + callback(); + }, + }); + const response = new ServerResponse({ + method: 'GET', + httpVersionMajor: 1, + httpVersionMinor: 1, + }); + socket.on('error', common.mustNotCall()); + response.on('error', common.mustNotCall()); + response.assignSocket(socket); + return response; +} + +async function completesWithin(promise) { + let timer; + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error('write timed out')), + common.platformTimeout(1000), + ); + }), + ]); + } finally { + clearTimeout(timer); + } +} + suite('underlying Writable', () => { suite('in non-object mode', () => { for (const ctor of ctors) { @@ -26,6 +61,241 @@ suite('underlying Writable', () => { await writer.write(buffer); }); } + + for (const highWaterMark of [1, 16]) { + test(`waits for mutable chunks with highWaterMark ${highWaterMark}`, + async () => { + let finishWrite; + let consumed; + const writable = new Writable({ + highWaterMark, + write(chunk, encoding, callback) { + finishWrite = () => { + consumed = Buffer.from(chunk); + callback(); + }; + }, + }); + writable.on('error', common.mustNotCall()); + const writer = Writable.toWeb(writable).getWriter(); + const input = new Uint8Array([1, 2, 3, 4]); + let settled = false; + const writePromise = writer.write(input).then(() => { + settled = true; + }); + + await new Promise(setImmediate); + assert.strictEqual(settled, false); + finishWrite(); + await writePromise; + input.fill(9); + await writer.close(); + + assert.deepStrictEqual( + consumed, + Buffer.from([1, 2, 3, 4]), + ); + }); + } + + test('copies mutable chunks while the Writable is corked', async () => { + let consumed; + const writable = new Writable({ + write(chunk, encoding, callback) { + consumed = Buffer.from(chunk); + callback(); + }, + }); + writable.on('error', common.mustNotCall()); + writable.cork(); + const writer = Writable.toWeb(writable).getWriter(); + const input = new Uint8Array([1, 2, 3, 4]); + + await writer.write(input); + input.fill(9); + writable.uncork(); + await writer.close(); + + assert.deepStrictEqual(consumed, Buffer.from([1, 2, 3, 4])); + }); + + test('copies mutable chunks when write() is overridden', async () => { + let consumed; + let received; + let notifyConsumed; + const consumedPromise = new Promise((resolve) => { + notifyConsumed = resolve; + }); + const writable = new Writable({ + write(chunk, encoding, callback) { + callback(); + }, + }); + writable.on('error', common.mustNotCall()); + const writer = Writable.toWeb(writable).getWriter(); + writable.write = common.mustCall((chunk) => { + received = chunk; + setImmediate(() => { + consumed = Buffer.from(chunk); + notifyConsumed(); + }); + return true; + }); + const input = new Uint8Array([1, 2, 3, 4]); + + await writer.write(input); + input.fill(9); + await consumedPromise; + await writer.close(); + + assert.notStrictEqual(received.buffer, input.buffer); + assert.deepStrictEqual(consumed, Buffer.from([1, 2, 3, 4])); + }); + + test('does not trust a patched Writable.prototype.write', async () => { + const originalWrite = Writable.prototype.write; + let consumed; + let notifyConsumed; + const consumedPromise = new Promise((resolve) => { + notifyConsumed = resolve; + }); + const writable = new Writable({ + write(chunk, encoding, callback) { + setImmediate(() => { + consumed = Buffer.from(chunk); + callback(); + notifyConsumed(); + }); + }, + }); + writable.on('error', common.mustNotCall()); + let writer; + + Writable.prototype.write = function(chunk) { + return originalWrite.call(this, chunk); + }; + try { + writer = Writable.toWeb(writable).getWriter(); + const input = new Uint8Array([1, 2, 3, 4]); + await completesWithin(writer.write(input)); + input.fill(9); + await consumedPromise; + + assert.deepStrictEqual(consumed, Buffer.from([1, 2, 3, 4])); + } finally { + Writable.prototype.write = originalWrite; + } + await writer.close(); + }); + + test('reads write() once for classification and invocation', async () => { + const originalWrite = Writable.prototype.write; + let consumed; + const writable = new Writable({ + write(chunk, encoding, callback) { + setImmediate(() => { + consumed = Buffer.from(chunk); + callback(); + }); + }, + }); + writable.on('error', common.mustNotCall()); + const writer = Writable.toWeb(writable).getWriter(); + let writeAccesses = 0; + Object.defineProperty(writable, 'write', { + configurable: true, + get() { + writeAccesses++; + return writeAccesses === 1 ? + originalWrite : + function callbacklessWrite(chunk) { + return originalWrite.call(this, chunk); + }; + }, + }); + + const input = new Uint8Array([1, 2, 3, 4]); + await completesWithin(writer.write(input)); + input.fill(9); + + assert.strictEqual(writeAccesses, 1); + assert.deepStrictEqual(consumed, Buffer.from([1, 2, 3, 4])); + delete writable.write; + await writer.close(); + }); + + test('preserves cloned view brands and SharedArrayBuffer backing', + async () => { + const dataView = new DataView( + Uint8Array.from([0, 1, 2, 3, 4, 0]).buffer, + 1, + 4, + ); + const uint16Buffer = new ArrayBuffer(6); + const uint16 = new Uint16Array(uint16Buffer, 2, 2); + new Uint8Array(uint16Buffer, 2, 4).set([1, 2, 3, 4]); + const shared = new SharedArrayBuffer(6); + const sharedView = new Uint8Array(shared, 1, 4); + sharedView.set([1, 2, 3, 4]); + const inputs = [ + Buffer.from([1, 2, 3, 4]), + dataView, + uint16, + sharedView, + ]; + const expected = inputs.map((chunk) => ({ + brand: Buffer.isBuffer(chunk) ? + 'Buffer' : Object.prototype.toString.call(chunk), + bytes: Buffer.from(new Uint8Array( + chunk.buffer, + chunk.byteOffset, + chunk.byteLength, + )), + shared: chunk.buffer instanceof SharedArrayBuffer, + })); + const received = []; + const writable = new Writable({ + write(chunk, encoding, callback) { + callback(); + }, + }); + writable.on('error', common.mustNotCall()); + const writer = Writable.toWeb(writable).getWriter(); + writable.write = common.mustCall((chunk) => { + received.push(chunk); + return true; + }, inputs.length); + + for (const chunk of inputs) { + await writer.write(chunk); + new Uint8Array( + chunk.buffer, + chunk.byteOffset, + chunk.byteLength, + ).fill(9); + } + await writer.close(); + + for (let i = 0; i < received.length; i++) { + const actual = received[i]; + const actualBrand = Buffer.isBuffer(actual) ? + 'Buffer' : Object.prototype.toString.call(actual); + assert.strictEqual(actualBrand, expected[i].brand); + assert.notStrictEqual(actual.buffer, inputs[i].buffer); + assert.strictEqual( + actual.buffer instanceof SharedArrayBuffer, + expected[i].shared, + ); + assert.deepStrictEqual( + Buffer.from(new Uint8Array( + actual.buffer, + actual.byteOffset, + actual.byteLength, + )), + expected[i].bytes, + ); + } + }); }); suite('in object mode', () => { @@ -48,16 +318,71 @@ suite('underlying Writable', () => { }); }); +suite('underlying ServerResponse', () => { + test('rejects invalid view types before cloning', async () => { + const response = createServerResponse(); + const writer = Writable.toWeb(response).getWriter(); + + try { + await Promise.all([ + assert.rejects(writer.write(new DataView(new ArrayBuffer(4))), { + code: 'ERR_INVALID_ARG_TYPE', + }), + assert.rejects(writer.closed, { + code: 'ERR_INVALID_ARG_TYPE', + }), + ]); + } finally { + await new Promise((resolve) => response.end(resolve)); + } + }); + + test('preserves write() overrides', async () => { + const response = createServerResponse(); + const writer = Writable.toWeb(response).getWriter(); + const originalWrite = response.write; + let received; + response.write = common.mustCall((chunk) => { + received = chunk; + return true; + }); + const input = new DataView(Uint8Array.from([1, 2, 3, 4]).buffer); + + try { + await writer.write(input); + new Uint8Array(input.buffer).fill(9); + assert(received instanceof DataView); + assert.notStrictEqual(received.buffer, input.buffer); + assert.deepStrictEqual( + Buffer.from(received.buffer), + Buffer.from([1, 2, 3, 4]), + ); + } finally { + response.write = originalWrite; + await new Promise((resolve) => response.end(resolve)); + } + }); +}); + suite('underlying Duplex', () => { suite('in non-object mode', () => { for (const ctor of ctors) { - test(`converts ${ctor.name} chunks`, async () => { + test(`copies ${ctor.name} chunks`, async () => { const buffer = new ctor(4); + new Uint8Array(buffer).set([1, 2, 3, 4]); const duplex = new Duplex({ writableObjectMode: false, write: common.mustCall((chunk, encoding, callback) => { assert(Buffer.isBuffer(chunk)); - assert.strictEqual(chunk.buffer, buffer); + assert.notStrictEqual(chunk.buffer, buffer); + assert.strictEqual( + chunk.buffer instanceof SharedArrayBuffer, + buffer instanceof SharedArrayBuffer, + ); + assert.deepStrictEqual( + chunk, + Buffer.from([1, 2, 3, 4]), + ); callback(); }), read() { @@ -69,6 +394,37 @@ suite('underlying Duplex', () => { await writer.write(buffer); }); } + + test('copies mutable chunks without waiting for the write callback', + async () => { + let consumed; + let finishWrite; + const duplex = new Duplex({ + write(chunk, encoding, callback) { + finishWrite = () => { + consumed = Buffer.from(chunk); + callback(); + }; + }, + read() { + this.push(null); + }, + }); + duplex.on('error', common.mustNotCall()); + const writer = Duplex.toWeb(duplex).writable.getWriter(); + const input = new Uint8Array([1, 2, 3, 4]); + + await writer.write(input); + input.fill(9); + finishWrite(); + await new Promise(setImmediate); + + assert.deepStrictEqual( + consumed, + Buffer.from([1, 2, 3, 4]), + ); + await writer.close(); + }); }); suite('in object mode', () => { diff --git a/test/parallel/test-webstreams-compression-buffer-source.js b/test/parallel/test-webstreams-compression-buffer-source.js index 3304a8e64f3175..a81d4b674b3805 100644 --- a/test/parallel/test-webstreams-compression-buffer-source.js +++ b/test/parallel/test-webstreams-compression-buffer-source.js @@ -3,6 +3,7 @@ require('../common'); const assert = require('assert'); const test = require('node:test'); const { DecompressionStream, CompressionStream } = require('stream/web'); +const { gzipSync } = require('zlib'); // Minimal gzip-compressed bytes for "hello" const compressedGzip = new Uint8Array([ @@ -40,3 +41,31 @@ test('CompressionStream round-trip with ArrayBuffer input', async () => { const result = Buffer.concat(out.map((c) => Buffer.from(c))); assert.strictEqual(result.toString(), 'hello'); }); + +test('DecompressionStream writable completion is not coupled to readable ' + + 'backpressure', async () => { + const expected = Buffer.alloc(1024 * 1024, 0x61); + const compressed = gzipSync(expected); + const ds = new DecompressionStream('gzip'); + const writer = ds.writable.getWriter(); + let settled = false; + const writePromise = writer.write(compressed).then(() => { + settled = true; + }); + + await new Promise(setImmediate); + const settledBeforeRead = settled; + if (settledBeforeRead) { + compressed.fill(0); + } + + const outputPromise = Array.fromAsync(ds.readable); + await writePromise; + await writer.close(); + const output = Buffer.concat( + (await outputPromise).map((chunk) => Buffer.from(chunk)), + ); + + assert.strictEqual(settledBeforeRead, true); + assert.deepStrictEqual(output, expected); +});