diff --git a/packages/bun-types/bun.d.ts b/packages/bun-types/bun.d.ts index 62140a7d5387..2f011ce04544 100644 --- a/packages/bun-types/bun.d.ts +++ b/packages/bun-types/bun.d.ts @@ -262,6 +262,10 @@ declare module "bun" { unref(): void; } + interface TransformerCancelCallback { + (reason: any): void | PromiseLike; + } + interface TransformerFlushCallback { (controller: TransformStreamDefaultController): void | PromiseLike; } diff --git a/packages/bun-types/globals.d.ts b/packages/bun-types/globals.d.ts index 4dd3dea89e07..59f2e3965611 100644 --- a/packages/bun-types/globals.d.ts +++ b/packages/bun-types/globals.d.ts @@ -868,6 +868,7 @@ interface QueuingStrategySize { } interface Transformer { + cancel?: Bun.TransformerCancelCallback; flush?: Bun.TransformerFlushCallback; readableType?: undefined; start?: Bun.TransformerStartCallback; diff --git a/src/js/builtins/BunBuiltinNames.h b/src/js/builtins/BunBuiltinNames.h index 4b5a89bcbb5d..54e9ab3b6d5d 100644 --- a/src/js/builtins/BunBuiltinNames.h +++ b/src/js/builtins/BunBuiltinNames.h @@ -24,6 +24,7 @@ using namespace JSC; #define BUN_COMMON_PRIVATE_IDENTIFIERS_EACH_PROPERTY_NAME(macro) \ macro(AbortSignal) \ macro(Buffer) \ + macro(CompressionStreamTransformer) \ macro(Loader) \ macro(ReadableByteStreamController) \ macro(ReadableStream) \ @@ -99,6 +100,7 @@ using namespace JSC; macro(fatal) \ macro(fd) \ macro(filename) \ + macro(finishPromise) \ macro(flushAlgorithm) \ macro(format) \ macro(fulfillModuleSync) \ diff --git a/src/js/builtins/CompressionStream.ts b/src/js/builtins/CompressionStream.ts index 5ca7520ddc4d..9b677c32a16a 100644 --- a/src/js/builtins/CompressionStream.ts +++ b/src/js/builtins/CompressionStream.ts @@ -1,21 +1,24 @@ export function initializeCompressionStream(this, format) { - const zlib = require("node:zlib"); - const { newBufferSourceTransformPairFromDuplex } = require("internal/webstreams_adapters"); - - const builders = { - "deflate": zlib.createDeflate, - "deflate-raw": zlib.createDeflateRaw, - "gzip": zlib.createGzip, - "brotli": zlib.createBrotliCompress, - "zstd": zlib.createZstdCompress, + // node:zlib NodeMode values (DEFLATE, GZIP, DEFLATERAW, BROTLI_ENCODE, + // ZSTD_COMPRESS) — the native transformer initializes the matching engine + // with node:zlib's defaults, so output bytes match the node-backed + // implementation this replaced. + const modes = { + __proto__: null, + "deflate": 1, + "deflate-raw": 5, + "gzip": 3, + "brotli": 9, + "zstd": 10, }; - if (!(format in builders)) - throw $ERR_INVALID_ARG_VALUE("format", format, "must be one of: " + Object.keys(builders).join(", ")); + if (!(format in modes)) + throw $ERR_INVALID_ARG_VALUE("format", format, "must be one of: " + Object.keys(modes).join(", ")); + + const transform = $createCompressionTransform(modes[format]); - const transform = newBufferSourceTransformPairFromDuplex(builders[format]()); - $putByIdDirectPrivate(this, "readable", transform.readable); - $putByIdDirectPrivate(this, "writable", transform.writable); + $putByIdDirectPrivate(this, "readable", $getByIdDirectPrivate(transform, "readable")); + $putByIdDirectPrivate(this, "writable", $getByIdDirectPrivate(transform, "writable")); return this; } diff --git a/src/js/builtins/DecompressionStream.ts b/src/js/builtins/DecompressionStream.ts index 0df175dc69fd..e4c48be2b2e7 100644 --- a/src/js/builtins/DecompressionStream.ts +++ b/src/js/builtins/DecompressionStream.ts @@ -1,21 +1,22 @@ export function initializeDecompressionStream(this, format) { - const zlib = require("node:zlib"); - const { newBufferSourceTransformPairFromDuplex } = require("internal/webstreams_adapters"); - - const builders = { - "deflate": zlib.createInflate, - "deflate-raw": zlib.createInflateRaw, - "gzip": zlib.createGunzip, - "brotli": zlib.createBrotliDecompress, - "zstd": zlib.createZstdDecompress, + // node:zlib NodeMode values (INFLATE, GUNZIP, INFLATERAW, BROTLI_DECODE, + // ZSTD_DECOMPRESS) — see CompressionStream for the encode-side table. + const modes = { + __proto__: null, + "deflate": 2, + "deflate-raw": 6, + "gzip": 4, + "brotli": 8, + "zstd": 11, }; - if (!(format in builders)) - throw $ERR_INVALID_ARG_VALUE("format", format, "must be one of: " + Object.keys(builders).join(", ")); + if (!(format in modes)) + throw $ERR_INVALID_ARG_VALUE("format", format, "must be one of: " + Object.keys(modes).join(", ")); + + const transform = $createCompressionTransform(modes[format]); - const transform = newBufferSourceTransformPairFromDuplex(builders[format]()); - $putByIdDirectPrivate(this, "readable", transform.readable); - $putByIdDirectPrivate(this, "writable", transform.writable); + $putByIdDirectPrivate(this, "readable", $getByIdDirectPrivate(transform, "readable")); + $putByIdDirectPrivate(this, "writable", $getByIdDirectPrivate(transform, "writable")); return this; } diff --git a/src/js/builtins/TransformStream.ts b/src/js/builtins/TransformStream.ts index f8bb7d34388e..0e8fca27523d 100644 --- a/src/js/builtins/TransformStream.ts +++ b/src/js/builtins/TransformStream.ts @@ -54,6 +54,10 @@ export function initializeTransformStream(this) { transformerDict["flush"] = transformer["flush"]; if (typeof transformerDict["flush"] !== "function") $throwTypeError("transformer.flush should be a function"); } + if ("cancel" in transformer) { + transformerDict["cancel"] = transformer["cancel"]; + if (typeof transformerDict["cancel"] !== "function") $throwTypeError("transformer.cancel should be a function"); + } if ("readableType" in transformer) throw new RangeError("TransformStream transformer has a readableType"); if ("writableType" in transformer) throw new RangeError("TransformStream transformer has a writableType"); diff --git a/src/js/builtins/TransformStreamInternals.ts b/src/js/builtins/TransformStreamInternals.ts index 833fafdc6b3e..4db8dc695176 100644 --- a/src/js/builtins/TransformStreamInternals.ts +++ b/src/js/builtins/TransformStreamInternals.ts @@ -65,7 +65,9 @@ export function createTransformStream( ); const controller = new TransformStreamDefaultController(); - $setUpTransformStreamDefaultController(stream, controller, transformAlgorithm, flushAlgorithm); + $setUpTransformStreamDefaultController(stream, controller, transformAlgorithm, flushAlgorithm, () => + Promise.$resolve(), + ); startAlgorithm().$then( () => { @@ -112,8 +114,7 @@ export function initializeTransformStream( return $transformStreamDefaultSourcePullAlgorithm(stream); }; const cancelAlgorithm = reason => { - $transformStreamErrorWritableAndUnblockWrite(stream, reason); - return Promise.$resolve(); + return $transformStreamDefaultSourceCancelAlgorithm(stream, reason); }; const underlyingSource = {}; $putByIdDirectPrivate(underlyingSource, "start", startAlgorithm); @@ -151,6 +152,10 @@ export function transformStreamErrorWritableAndUnblockWrite(stream, e) { const writable = $getByIdDirectPrivate(stream, "internalWritable"); $writableStreamDefaultControllerErrorIfNeeded($getByIdDirectPrivate(writable, "controller"), e); + $transformStreamUnblockWrite(stream); +} + +export function transformStreamUnblockWrite(stream) { if ($getByIdDirectPrivate(stream, "backpressure")) $transformStreamSetBackpressure(stream, false); } @@ -164,7 +169,13 @@ export function transformStreamSetBackpressure(stream, backpressure) { $putByIdDirectPrivate(stream, "backpressure", backpressure); } -export function setUpTransformStreamDefaultController(stream, controller, transformAlgorithm, flushAlgorithm) { +export function setUpTransformStreamDefaultController( + stream, + controller, + transformAlgorithm, + flushAlgorithm, + cancelAlgorithm, +) { $assert($isTransformStream(stream)); $assert($getByIdDirectPrivate(stream, "controller") === undefined); @@ -172,6 +183,8 @@ export function setUpTransformStreamDefaultController(stream, controller, transf $putByIdDirectPrivate(stream, "controller", controller); $putByIdDirectPrivate(controller, "transformAlgorithm", transformAlgorithm); $putByIdDirectPrivate(controller, "flushAlgorithm", flushAlgorithm); + $putByIdDirectPrivate(controller, "cancelAlgorithm", cancelAlgorithm); + $putByIdDirectPrivate(controller, "finishPromise", undefined); } export function setUpTransformStreamDefaultControllerFromTransformer(stream, transformer, transformerDict) { @@ -187,6 +200,9 @@ export function setUpTransformStreamDefaultControllerFromTransformer(stream, tra let flushAlgorithm = () => { return Promise.$resolve(); }; + let cancelAlgorithm = () => { + return Promise.$resolve(); + }; if ("transform" in transformerDict) transformAlgorithm = chunk => { @@ -199,13 +215,20 @@ export function setUpTransformStreamDefaultControllerFromTransformer(stream, tra }; } - $setUpTransformStreamDefaultController(stream, controller, transformAlgorithm, flushAlgorithm); + if ("cancel" in transformerDict) { + cancelAlgorithm = reason => { + return $promiseInvokeOrNoopMethod(transformer, transformerDict["cancel"], [reason]); + }; + } + + $setUpTransformStreamDefaultController(stream, controller, transformAlgorithm, flushAlgorithm, cancelAlgorithm); } export function transformStreamDefaultControllerClearAlgorithms(controller) { // We set transformAlgorithm to true to allow GC but keep the isTransformStreamDefaultController check. $putByIdDirectPrivate(controller, "transformAlgorithm", true); $putByIdDirectPrivate(controller, "flushAlgorithm", undefined); + $putByIdDirectPrivate(controller, "cancelAlgorithm", undefined); } export function transformStreamDefaultControllerEnqueue(controller, chunk) { @@ -236,9 +259,29 @@ export function transformStreamDefaultControllerError(controller, e) { } export function transformStreamDefaultControllerPerformTransform(controller, chunk) { + const transformAlgorithm = $getByIdDirectPrivate(controller, "transformAlgorithm"); + + // The algorithms are cleared as soon as teardown starts, but a write can + // still get here when it races reader.cancel(): the writable only errors + // once the cancel algorithm settles. Reject the write with the teardown + // outcome instead of invoking a cleared algorithm (the spec reference + // implementation rejects with an internal TypeError on this race). + if (transformAlgorithm === true) { + const stream = $getByIdDirectPrivate(controller, "stream"); + const writable = $getByIdDirectPrivate(stream, "internalWritable"); + const promiseCapability = $newPromiseCapability(Promise); + const rejectWithStoredError = () => { + promiseCapability.reject.$call(undefined, $getByIdDirectPrivate(writable, "storedError")); + }; + const finishPromise = $getByIdDirectPrivate(controller, "finishPromise"); + if (finishPromise !== undefined) finishPromise.promise.$then(rejectWithStoredError, rejectWithStoredError); + else rejectWithStoredError(); + return promiseCapability.promise; + } + const promiseCapability = $newPromiseCapability(Promise); - const transformPromise = $getByIdDirectPrivate(controller, "transformAlgorithm").$call(undefined, chunk); + const transformPromise = transformAlgorithm.$call(undefined, chunk); transformPromise.$then( () => { promiseCapability.resolve(); @@ -304,21 +347,60 @@ export function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) { } export function transformStreamDefaultSinkAbortAlgorithm(stream, reason) { - $transformStreamError(stream, reason); - return Promise.$resolve(); + const controller = $getByIdDirectPrivate(stream, "controller"); + const finishPromise = $getByIdDirectPrivate(controller, "finishPromise"); + if (finishPromise !== undefined) return finishPromise.promise; + + const cancelAlgorithm = $getByIdDirectPrivate(controller, "cancelAlgorithm"); + // A transform error can clear the algorithms while a write is in flight; + // the writable machinery still performs its abort steps once that write + // settles. The stream is fully errored by then — nothing left to cancel. + // (The spec reference implementation crashes on this race.) + if (cancelAlgorithm === undefined) return Promise.$resolve(); + + const readable = $getByIdDirectPrivate(stream, "readable"); + + const promiseCapability = $newPromiseCapability(Promise); + $putByIdDirectPrivate(controller, "finishPromise", promiseCapability); + + const cancelPromise = cancelAlgorithm.$call(undefined, reason); + $transformStreamDefaultControllerClearAlgorithms(controller); + + cancelPromise.$then( + () => { + if ($getByIdDirectPrivate(readable, "state") === $streamErrored) { + promiseCapability.reject.$call(undefined, $getByIdDirectPrivate(readable, "storedError")); + return; + } + + $readableStreamDefaultControllerError($getByIdDirectPrivate(readable, "readableStreamController"), reason); + promiseCapability.resolve.$call(); + }, + r => { + $readableStreamDefaultControllerError($getByIdDirectPrivate(readable, "readableStreamController"), r); + promiseCapability.reject.$call(undefined, r); + }, + ); + + return promiseCapability.promise; } export function transformStreamDefaultSinkCloseAlgorithm(stream) { - const readable = $getByIdDirectPrivate(stream, "readable"); const controller = $getByIdDirectPrivate(stream, "controller"); + const finishPromise = $getByIdDirectPrivate(controller, "finishPromise"); + if (finishPromise !== undefined) return finishPromise.promise; + + const readable = $getByIdDirectPrivate(stream, "readable"); const readableController = $getByIdDirectPrivate(readable, "readableStreamController"); + const promiseCapability = $newPromiseCapability(Promise); + $putByIdDirectPrivate(controller, "finishPromise", promiseCapability); + const flushAlgorithm = $getByIdDirectPrivate(controller, "flushAlgorithm"); $assert(flushAlgorithm !== undefined); - const flushPromise = $getByIdDirectPrivate(controller, "flushAlgorithm").$call(); + const flushPromise = flushAlgorithm.$call(); $transformStreamDefaultControllerClearAlgorithms(controller); - const promiseCapability = $newPromiseCapability(Promise); flushPromise.$then( () => { if ($getByIdDirectPrivate(readable, "state") === $streamErrored) { @@ -333,7 +415,11 @@ export function transformStreamDefaultSinkCloseAlgorithm(stream) { }, r => { $transformStreamError($getByIdDirectPrivate(controller, "stream"), r); - promiseCapability.reject.$call(undefined, $getByIdDirectPrivate(readable, "storedError")); + // Reject with r per spec. The readable's storedError is normally the + // same value, but when a concurrent reader.cancel() already closed the + // readable, transformStreamError cannot error it and storedError is + // undefined — the flush error must not be swallowed. + promiseCapability.reject.$call(undefined, r); }, ); return promiseCapability.promise; @@ -347,3 +433,210 @@ export function transformStreamDefaultSourcePullAlgorithm(stream) { return $getByIdDirectPrivate(stream, "backpressureChangePromise").promise; } + +export function transformStreamDefaultSourceCancelAlgorithm(stream, reason) { + const controller = $getByIdDirectPrivate(stream, "controller"); + const finishPromise = $getByIdDirectPrivate(controller, "finishPromise"); + if (finishPromise !== undefined) return finishPromise.promise; + + const cancelAlgorithm = $getByIdDirectPrivate(controller, "cancelAlgorithm"); + // controller.terminate() clears the algorithms while the readable can + // still hold queued chunks — and stay cancelable — without a finishPromise + // ever being set. The transformer is already torn down; there is nothing + // left to cancel. (The spec reference implementation crashes on this.) + if (cancelAlgorithm === undefined) return Promise.$resolve(); + + const writable = $getByIdDirectPrivate(stream, "internalWritable"); + + const promiseCapability = $newPromiseCapability(Promise); + $putByIdDirectPrivate(controller, "finishPromise", promiseCapability); + + const cancelPromise = cancelAlgorithm.$call(undefined, reason); + $transformStreamDefaultControllerClearAlgorithms(controller); + + cancelPromise.$then( + () => { + if ($getByIdDirectPrivate(writable, "state") === "errored") { + promiseCapability.reject.$call(undefined, $getByIdDirectPrivate(writable, "storedError")); + return; + } + + $writableStreamDefaultControllerErrorIfNeeded($getByIdDirectPrivate(writable, "controller"), reason); + $transformStreamUnblockWrite(stream); + promiseCapability.resolve.$call(); + }, + r => { + $writableStreamDefaultControllerErrorIfNeeded($getByIdDirectPrivate(writable, "controller"), r); + $transformStreamUnblockWrite(stream); + promiseCapability.reject.$call(undefined, r); + }, + ); + + return promiseCapability.promise; +} + +export function createCompressionTransform(mode) { + const { Buffer } = require("node:buffer"); + const { isArrayBuffer, isSharedArrayBuffer } = require("node:util/types"); + + const handle = new $CompressionStreamTransformer(mode); + const chunkSize = 16384; // node:zlib Z_DEFAULT_CHUNK — the native output granularity + let closed = false; + + function close() { + if (!closed) { + closed = true; + handle.close(); + } + } + + // Chunks at or past this many bytes run on the work pool instead of the JS + // thread. Below it the synchronous path is faster (no input copy, no thread + // hop, no promise allocation); above it the compression work dominates and + // offloading keeps the main thread responsive — the previous node:zlib + // adapter paid a threadpool round-trip per 16KB of output regardless. + // + // For DecompressionStream the input size says nothing about the work: a + // sub-threshold compressed chunk can expand to arbitrary output and stall + // the JS thread for its whole drive loop, so the decode modes (INFLATE, + // GUNZIP, INFLATERAW, BROTLI_DECODE, ZSTD_DECOMPRESS) always take the async + // path (the previous adapter was always async here too). + // + // BROTLI_ENCODE (mode 9) is the same: at the default quality 11, PROCESS + // only buffers input until the encoder's ring buffer fills (~256KB) and + // then entropy-codes the whole metablock in one call, so a stream of + // sub-threshold writes would run that encode on the JS thread every time + // cumulative input crosses a block boundary. zlib and zstd encode + // compress incrementally per call, so their sync fast path is sound. + const isDecode = mode === 2 || mode === 4 || mode === 6 || mode === 8 || mode === 11; + const asyncThreshold = isDecode || mode === 9 ? 1 : 4 * chunkSize; + + // node surfaces engine failures as a TypeError carrying the error code + // (its webstreams adapter wraps them); match the class and code but keep + // the engine's message, which the adapter dropped. + function wrapEngineError(error) { + const wrapped = $makeTypeError(error.message); + wrapped.code = error.code; + wrapped.errno = error.errno; + wrapped.cause = error; + return wrapped; + } + + function enqueueOutputs(controller, outputs) { + for (let i = 0; i < outputs.length; i++) $transformStreamDefaultControllerEnqueue(controller, outputs[i]); + } + + // The consume/produce loop lives in the native transformer; this wrapper + // only normalizes chunk types, wraps engine errors, and enqueues the + // returned output chunks (each an exact-size Uint8Array, ≤ chunkSize). + // Accepted chunk types follow node v26 (the Compression Streams spec): + // any BufferSource (ArrayBuffer or a view over one) is accepted; + // SharedArrayBuffer and SAB-backed views reject with + // ERR_INVALID_ARG_TYPE, null with its dedicated streams error. Strings + // are still accepted for compatibility with node's zlib Transform write + // path. + // + // Returns a promise when the chunk took the async path; the TransformStream + // gates the next write on it. + function drive(chunk, isFinish, controller) { + if (typeof chunk === "string") chunk = Buffer.from(chunk); + else if (ArrayBuffer.$isView(chunk)) { + if (isSharedArrayBuffer(chunk.buffer)) + throw $ERR_INVALID_ARG_TYPE("chunk", ["ArrayBuffer", "Buffer", "TypedArray", "DataView"], chunk); + // A view over a detached buffer reports byteLength 0 but must reject + // like node (whose Buffer.from copy throws on detached buffers). + if (chunk.buffer.detached) throw $makeTypeError("Cannot perform Construct on a detached ArrayBuffer"); + } else if (isArrayBuffer(chunk)) { + chunk = new Uint8Array(chunk); + } else if (chunk === null) throw $ERR_STREAM_NULL_VALUES(); + else throw $ERR_INVALID_ARG_TYPE("chunk", ["ArrayBuffer", "Buffer", "TypedArray", "DataView"], chunk); + + if (chunk.byteLength >= asyncThreshold) { + return handle.transformAsync(chunk, isFinish).$then( + outputs => { + enqueueOutputs(controller, outputs); + }, + error => { + close(); + throw wrapEngineError(error); + }, + ); + } + + let outputs; + try { + outputs = handle.transform(chunk, isFinish); + } catch (error) { + throw wrapEngineError(error); + } + enqueueOutputs(controller, outputs); + } + + const emptyChunk = new Uint8Array(0); + + return new TransformStream( + { + // Any failure (bad chunk type, engine error, enqueue on a torn-down + // readable) errors the stream and the transformer is never invoked + // again, so release the native handle on the way out. + transform(chunk, controller) { + try { + return drive(chunk, false, controller); + } catch (e) { + close(); + throw e; + } + }, + // The finish flush always takes the work pool: brotli at the default + // quality 11 buffers input during PROCESS and encodes the residual + // block (up to ~256KB) at FINISH, so the 0-byte input here says + // nothing about the work. For zlib/zstd the finish is just a trailer + // and the one thread hop is noise. close() is chained on settlement. + flush(controller) { + return handle.transformAsync(emptyChunk, true).$then( + outputs => { + // The engine is done once the outputs are extracted; close + // before enqueueing so a reader.cancel() that landed while + // the worker ran (it closes the readable and returns the + // already-set finishPromise without invoking cancel()) still + // releases the native context deterministically. The enqueue + // then throws on the closed readable — discard the outputs, + // the reader no longer wants them; sinkCloseAlgorithm sees + // the readable is not errored and resolves the close. + close(); + try { + enqueueOutputs(controller, outputs); + } catch {} + }, + error => { + close(); + throw wrapEngineError(error); + }, + ); + }, + // reader.cancel() / writer.abort() skip flush() — this is the + // teardown path that releases the native handle for them. + cancel() { + close(); + }, + }, + undefined, + // The readable side buffers output before signalling backpressure; a + // gated write only proceeds once a reader drains the queue. Two + // constraints pick the budget: + // + // - With the spec-default strategy (highWaterMark 0, initial + // backpressure), the first write would stall until a reader attaches. + // The Node-adapter implementation this replaces resolved writes while + // ~16KB of *input* was buffered — so a decompression write whose + // output expands far past its input (and every write after it) still + // resolved with no reader attached, and code in the wild awaits + // writes before reading. A budget of one chunkSize of *output* would + // deadlock the write that follows any >16KB expansion. + // - An unbounded queue would break producer throttling in piped flows. + // + // 64 chunkSizes (1MiB) of output covers the old input-side acceptance + // for typical expansion ratios while keeping piped flows bounded. + { highWaterMark: 64 * chunkSize, size: chunk => chunk.byteLength }, + ); +} diff --git a/src/js/internal/webstreams_adapters.ts b/src/js/internal/webstreams_adapters.ts index 002c257ff0e6..3ae84b149c4c 100644 --- a/src/js/internal/webstreams_adapters.ts +++ b/src/js/internal/webstreams_adapters.ts @@ -33,9 +33,6 @@ const SafePromisePrototypeFinally = $Promise.prototype.finally; const constants_zlib = $processBindingConstants.zlib; -const kValidateChunk = Symbol("kValidateChunk"); -const kDestroyOnSyncError = Symbol("kDestroyOnSyncError"); - function tryTransferToNativeReadable(stream, options) { const ptr = stream.$bunNativePtr; if (!ptr || ptr === -1) { @@ -220,7 +217,7 @@ function handleKnownInternalErrors(cause: Error | null): Error | null { const noop = () => {}; -function newWritableStreamFromStreamWritable(streamWritable, options = kEmptyObject) { +function newWritableStreamFromStreamWritable(streamWritable) { // 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, @@ -288,34 +285,20 @@ function newWritableStreamFromStreamWritable(streamWritable, options = kEmptyObj }, write(chunk) { - try { - options[kValidateChunk]?.(chunk); - if (!streamWritable.writableObjectMode && isAnyArrayBuffer(chunk)) { - chunk = new Uint8Array(chunk); - } - const needDrainBefore = streamWritable.writableNeedDrain; - if (needDrainBefore || !streamWritable.write(chunk)) { - backpressurePromise = PromiseWithResolvers(); - // write() may set writableNeedDrain; the post-write value is - // what decides whether we resolve immediately. - if (!streamWritable.writableNeedDrain) { - backpressurePromise.resolve(); - } - return SafePromisePrototypeFinally.$call(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]) { - destroyer(streamWritable, error); + if (!streamWritable.writableObjectMode && isAnyArrayBuffer(chunk)) { + chunk = new Uint8Array(chunk); + } + const needDrainBefore = streamWritable.writableNeedDrain; + if (needDrainBefore || !streamWritable.write(chunk)) { + backpressurePromise = PromiseWithResolvers(); + // write() may set writableNeedDrain; the post-write value is + // what decides whether we resolve immediately. + if (!streamWritable.writableNeedDrain) { + backpressurePromise.resolve(); } - throw error; + return SafePromisePrototypeFinally.$call(backpressurePromise.promise, () => { + backpressurePromise = undefined; + }); } }, @@ -654,15 +637,7 @@ function newReadableWritablePairFromDuplex(duplex, options = kEmptyObject) { return { readable, writable }; } - const writableOptions = { - __proto__: null, - [kValidateChunk]: options[kValidateChunk], - [kDestroyOnSyncError]: options[kDestroyOnSyncError], - }; - - const writable = isWritable(duplex) - ? newWritableStreamFromStreamWritable(duplex, writableOptions) - : new WritableStream(); + const writable = isWritable(duplex) ? newWritableStreamFromStreamWritable(duplex) : new WritableStream(); if (!isWritable(duplex)) writable.close(); @@ -860,22 +835,6 @@ function newStreamDuplexFromReadableWritablePair(pair = kEmptyObject, options = return duplex; } -// Shared by CompressionStream and DecompressionStream: per the Compression -// Streams spec, chunks must be BufferSource (ArrayBuffer or ArrayBufferView -// not backed by SharedArrayBuffer), and an invalid chunk must error both -// sides of the pair synchronously. -function newBufferSourceTransformPairFromDuplex(duplex) { - const { isArrayBufferView, isSharedArrayBuffer } = require("node:util/types"); - return newReadableWritablePairFromDuplex(duplex, { - [kValidateChunk]: function validateBufferSourceChunk(chunk) { - if (isSharedArrayBuffer(isArrayBufferView(chunk) ? chunk.buffer : chunk)) { - throw $ERR_INVALID_ARG_TYPE("chunk", ["ArrayBuffer", "Buffer", "TypedArray", "DataView"], chunk); - } - }, - [kDestroyOnSyncError]: true, - }); -} - export default { newWritableStreamFromStreamWritable, newReadableStreamFromStreamReadable, @@ -883,8 +842,5 @@ export default { newStreamReadableFromReadableStream, newReadableWritablePairFromDuplex, newStreamDuplexFromReadableWritablePair, - newBufferSourceTransformPairFromDuplex, - kValidateChunk, - kDestroyOnSyncError, _ReadableFromWeb: ReadableFromWeb, }; diff --git a/src/jsc/bindings/ZigGlobalObject.cpp b/src/jsc/bindings/ZigGlobalObject.cpp index 52e6d8df2f68..e45733fd9e76 100644 --- a/src/jsc/bindings/ZigGlobalObject.cpp +++ b/src/jsc/bindings/ZigGlobalObject.cpp @@ -2952,6 +2952,7 @@ void GlobalObject::addBuiltinGlobals(JSC::VM& vm) GlobalPropertyInfo(builtinNames.processBindingConstantsPrivateName(), this->processBindingConstants(), PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly), GlobalPropertyInfo(builtinNames.requireMapPrivateName(), this->requireMap(), PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly | 0), GlobalPropertyInfo(builtinNames.TextEncoderStreamEncoderPrivateName(), JSTextEncoderStreamEncoderConstructor(), PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly | 0), + GlobalPropertyInfo(builtinNames.CompressionStreamTransformerPrivateName(), JSCompressionStreamTransformerConstructor(), PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly | 0), GlobalPropertyInfo(builtinNames.makeErrorWithCodePrivateName(), JSFunction::create(vm, this, 2, String(), jsFunctionMakeErrorWithCode, ImplementationVisibility::Public), PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly), GlobalPropertyInfo(builtinNames.toClassPrivateName(), JSFunction::create(vm, this, 1, String(), jsFunctionToClass, ImplementationVisibility::Public), PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly), GlobalPropertyInfo(builtinNames.inheritsPrivateName(), JSFunction::create(vm, this, 1, String(), jsFunctionInherits, ImplementationVisibility::Public), PropertyAttribute::DontDelete | PropertyAttribute::ReadOnly), diff --git a/src/jsc/generated_classes_list.rs b/src/jsc/generated_classes_list.rs index 53eccf23d622..d0f701f3a567 100644 --- a/src/jsc/generated_classes_list.rs +++ b/src/jsc/generated_classes_list.rs @@ -104,6 +104,7 @@ pub mod Classes { pub use crate::webcore::TextDecoder; pub use crate::webcore::byte_blob_loader::Source as BlobInternalReadableStreamSource; pub use crate::webcore::byte_stream::Source as BytesInternalReadableStreamSource; + pub use crate::webcore::compression_stream_transformer::CompressionStreamTransformer; pub use crate::webcore::crypto::Crypto; pub use crate::webcore::file_reader::Source as FileInternalReadableStreamSource; pub use crate::webcore::text_encoder_stream_encoder::TextEncoderStreamEncoder; diff --git a/src/runtime/webcore.rs b/src/runtime/webcore.rs index 66c0a7324ab8..244a04f19243 100644 --- a/src/runtime/webcore.rs +++ b/src/runtime/webcore.rs @@ -13,6 +13,8 @@ pub mod bake_response; pub mod byte_blob_loader; #[path = "webcore/ByteStream.rs"] pub mod byte_stream; +#[path = "webcore/CompressionStreamTransformer.rs"] +pub mod compression_stream_transformer; #[path = "webcore/CookieMap.rs"] pub mod cookie_map; #[path = "webcore/Crypto.rs"] diff --git a/src/runtime/webcore/CompressionStreamTransformer.rs b/src/runtime/webcore/CompressionStreamTransformer.rs new file mode 100644 index 000000000000..28ee7a30fbc1 --- /dev/null +++ b/src/runtime/webcore/CompressionStreamTransformer.rs @@ -0,0 +1,509 @@ +use core::cell::Cell; + +use bun_jsc::any_task_job::{AnyTaskJob, AnyTaskJobCtx}; +use bun_jsc::{ + CallFrame, JSGlobalObject, JSPromiseStrong, JSUint8Array, JSValue, JsCell, JsResult, + StringJsc as _, Strong, +}; +use bun_zlib::NodeMode; + +use crate::node::node_zlib_binding::{CompressionContext, Error}; + +bun_output::declare_scope!(CompressionStreamTransformer, hidden); + +/// Streaming compression/decompression engine for `CompressionStream` / +/// `DecompressionStream`. One zlib/brotli/zstd context driven on the JS thread +/// for small chunks, or on a work-pool worker for large ones via +/// `transformAsync` — the builtins call `transform`/`transformAsync` per chunk +/// with no node:zlib stream object, Duplex machinery, or per-16KB threadpool +/// round-trips behind it. +pub(crate) enum Engine { + Zlib(crate::node::native_zlib_impl::Context), + Brotli(crate::node::native_brotli_impl::Context), + Zstd(crate::node::native_zstd_impl::Context), + /// Context released (explicit `close()` or post-flush teardown). + Closed, +} + +impl Engine { + fn ctx(&mut self) -> Option<&mut dyn CompressionContext> { + match self { + Engine::Zlib(ctx) => Some(ctx), + Engine::Brotli(ctx) => Some(ctx), + Engine::Zstd(ctx) => Some(ctx), + Engine::Closed => None, + } + } + + fn finish_flush(&self) -> i32 { + match self { + // Z_FINISH + Engine::Zlib(_) => 4, + // BROTLI_OPERATION_FINISH / ZSTD_e_end + Engine::Brotli(_) | Engine::Zstd(_) => 2, + Engine::Closed => 0, + } + } + + fn close(&mut self) { + if let Some(ctx) = self.ctx() { + ctx.close(); + } + // The replaced variant drops only its Rust-side fields (e.g. the + // zlib dictionary Vec) — the C state was just released above, and + // Engine deliberately has no Drop impl so this assignment cannot + // double-close it. + *self = Engine::Closed; + } +} + +impl Drop for CompressionStreamTransformer { + fn drop(&mut self) { + // GC'd without flush/cancel (stream abandoned) — release the native + // context here instead of leaking it. Idempotent for explicitly + // closed engines. + self.engine.with_mut(Engine::close); + } +} + +// R-2 (host-fn re-entrancy): every JS-exposed method takes `&self`; the +// engine lives in a `JsCell` and no JS is invoked while it is borrowed. +#[bun_jsc::JsClass] +pub struct CompressionStreamTransformer { + engine: JsCell, + /// Set while an `AsyncTransformCtx` is in flight on the work pool. + /// Serializes engine access between the JS thread and the worker — same + /// pattern as `NativeZlib::write_in_progress`. The TransformStream gates + /// the next write on the returned promise, so a second + /// `transform`/`transformAsync` while set is misuse and throws. + write_in_progress: Cell, + /// `close()` was called while an async transform was in flight; the + /// engine is closed in `then()` once the worker returns. + pending_close: Cell, + /// Per-mode native context footprint, fixed at construction. Kept outside + /// the `JsCell`: `estimated_size` runs on the GC marking thread, which + /// must not touch the JS-thread-owned cell. + context_size: usize, +} + +/// node:zlib's Z_DEFAULT_CHUNK — the per-output-buffer granularity. +const CHUNK: usize = 16384; + +/// The processChunkSync loop from node:zlib: run the context until it stops +/// filling the output window. Pure native — no JS is touched, so it is safe to +/// call off the JS thread once `write_in_progress` guarantees exclusive access +/// to the engine. +fn drive_loop(engine: &mut Engine, input: &[u8], is_finish: bool) -> Result>, Error> { + let finish_flush = engine.finish_flush(); + let Some(ctx) = engine.ctx() else { + return Err(Error::init( + c"transform after close".as_ptr(), + -1, + c"ERR_INVALID_STATE".as_ptr(), + )); + }; + let flush = if is_finish { finish_flush } else { 0 }; + + let mut input: &[u8] = input; + let mut outputs: Vec> = Vec::new(); + + // avail_out == 0 means more output is pending (regardless of input); + // avail_out > 0 means the engine consumed the input it was given and + // drained its output. + loop { + // The engine counters are u32, but the chunk length is user + // controlled and JSC allows >4GiB typed arrays on 64-bit: + // feed at most one u32 window per iteration instead of + // overflowing the casts. + let window_len = input.len().min(u32::MAX as usize); + let window = &input[..window_len]; + + // Zero-initialized so the window handed to the C engine is + // fully defined; full windows are adopted as-is below with no + // copy (len == capacity). + let mut out_vec = vec![0u8; CHUNK]; + ctx.set_buffers(Some(window), Some(&mut out_vec)); + ctx.set_flush(flush); + ctx.do_work(); + + #[expect(clippy::cast_possible_truncation)] // window_len <= u32::MAX + let mut avail_in = window_len as u32; + let mut avail_out = u32::try_from(CHUNK).expect("constant"); + ctx.update_write_result(&mut avail_in, &mut avail_out); + let err = ctx.get_error_info(); + if err.is_error() { + return Err(err); + } + + let written = CHUNK - avail_out as usize; + if written == CHUNK { + outputs.push(out_vec.into()); + } else if written > 0 { + outputs.push(out_vec[..written].into()); + } + + let consumed = window_len - avail_in as usize; + input = &input[consumed..]; + + if avail_out == 0 || (avail_in == 0 && !input.is_empty()) { + // Output window exhausted before the engine finished, or + // the engine consumed the whole u32 window and input it + // has not seen remains past it — keep driving. If the + // engine instead stopped mid-window with spare output, it + // reached stream end: node's drive loop ends the stream + // there and discards the trailing bytes (lib/zlib.js + // processCallback), and re-feeding them would spin + // forever on input the engine refuses to consume. + continue; + } + break; + } + + Ok(outputs) +} + +fn build_outputs_array(global: &JSGlobalObject, outputs: Vec>) -> JsResult { + JSValue::create_array_from_iter(global, outputs.into_iter(), |bytes| { + // `from_bytes` reaches `JSC::JSUint8Array::create`, which opens a + // throw scope (allocation can throw). Observe the exception here, + // before `put_index` opens the next scope — same pattern as + // `BunString::to_jsdomurl`. + bun_jsc::from_js_host_call(global, || JSUint8Array::from_bytes(global, bytes)) + }) +} + +/// `AnyTaskJob` payload for `transformAsync`: copies the input bytes, runs the +/// drive loop on the work pool against the transformer's in-place engine +/// (the zlib `z_stream` is self-referential and must not move), and `then` +/// settles the promise with the same `Array` shape as the +/// synchronous path. +struct AsyncTransformCtx { + /// Roots the JS wrapper so `transformer` stays live until `then`. + _this_value: Strong, + transformer: *const CompressionStreamTransformer, + input: Box<[u8]>, + is_finish: bool, + promise: JSPromiseStrong, + result: Result>, Error>, +} + +// SAFETY: `run` is the only off-thread access; it touches the engine through +// `transformer` while `write_in_progress` guarantees the JS thread does not. +// The C engine state is thread-agnostic; the JS-tied fields (`this_value`, +// `promise`) are not touched until `then` on the JS thread. +unsafe impl Send for AsyncTransformCtx {} + +impl AnyTaskJobCtx for AsyncTransformCtx { + fn run(&mut self, _global: *mut JSGlobalObject) { + // SAFETY: `this_value` roots the JS wrapper, which owns + // `*transformer`; `write_in_progress` is the only access to the + // engine cell while set, so the worker's borrow is exclusive — same + // pattern as `CompressionStream::::async_job_run_task`. + let transformer = unsafe { &*self.transformer }; + self.result = transformer + .engine + .with_mut(|engine| drive_loop(engine, &self.input, self.is_finish)); + } + + fn then(&mut self, global: &JSGlobalObject) -> JsResult<()> { + // SAFETY: as in `run`; `then` runs on the JS thread. + let transformer = unsafe { &*self.transformer }; + transformer.write_in_progress.set(false); + if transformer.pending_close.replace(false) { + transformer.engine.with_mut(Engine::close); + } + + let result = core::mem::replace(&mut self.result, Ok(Vec::new())); + match result { + Ok(outputs) => match build_outputs_array(global, outputs) { + Ok(array) => self.promise.resolve(global, array)?, + // Building the JS array threw (OOM in the Uint8Array/array + // allocation): settle the promise so the awaiting write + // rejects instead of hanging. + Err(e) => self.promise.reject(global, Err(e))?, + }, + Err(err) => { + let error = build_engine_error(global, err); + self.promise.reject(global, Ok(error))?; + } + } + Ok(()) + } +} + +impl CompressionStreamTransformer { + /// Native context footprint for the GC, mirroring the per-mode constants + /// the `NativeZlib`/`NativeBrotli`/`NativeZstd` handles report. + /// Called from any thread (concurrent GC marking). + pub fn estimated_size(&self) -> usize { + core::mem::size_of::() + self.context_size + } + + fn check_not_in_flight(&self, global: &JSGlobalObject) -> JsResult<()> { + if self.write_in_progress.get() { + return Err(throw_engine_error( + global, + Error::init( + c"transform already in progress".as_ptr(), + -1, + c"ERR_INVALID_STATE".as_ptr(), + ), + )); + } + Ok(()) + } + + // PORT NOTE: no `#[bun_jsc::host_fn]` — the `#[bun_jsc::JsClass]` derive + // already emits the construct shim that calls `::constructor`. + pub(crate) fn constructor(global: &JSGlobalObject, frame: &CallFrame) -> JsResult> { + let [mode_value] = frame.arguments_as_array::<1>(); + if !mode_value.is_number() { + return Err(global.throw_invalid_argument_type_value("mode", "number", mode_value)); + } + let mode_double = mode_value.as_number(); + if mode_double % 1.0 != 0.0 || !(1.0..=11.0).contains(&mode_double) { + return Err(global.throw_invalid_argument_type_value("mode", "integer", mode_value)); + } + #[expect(clippy::cast_possible_truncation, clippy::cast_sign_loss)] + let mode = NodeMode::from_int(mode_double as u8); + + let context_size: usize = match mode { + // deflate internal_state @ cloudflare/zlib (see NativeZlib) + NodeMode::DEFLATE + | NodeMode::INFLATE + | NodeMode::GZIP + | NodeMode::GUNZIP + | NodeMode::DEFLATERAW + | NodeMode::INFLATERAW + | NodeMode::UNZIP => 3309, + NodeMode::BROTLI_ENCODE => 5143, // sizeof(BrotliEncoderStateStruct) + NodeMode::BROTLI_DECODE => 855, // sizeof(BrotliDecoderStateStruct) + NodeMode::ZSTD_COMPRESS => 5272, // ZSTD_sizeof_CCtx estimate + NodeMode::ZSTD_DECOMPRESS => 95968, // ZSTD_sizeof_DCtx estimate + NodeMode::NONE => unreachable!("range-checked above"), + }; + + let engine = match mode { + NodeMode::DEFLATE + | NodeMode::INFLATE + | NodeMode::GZIP + | NodeMode::GUNZIP + | NodeMode::DEFLATERAW + | NodeMode::INFLATERAW + | NodeMode::UNZIP => Engine::Zlib(crate::node::native_zlib_impl::Context { + mode, + ..Default::default() + }), + NodeMode::BROTLI_ENCODE | NodeMode::BROTLI_DECODE => { + Engine::Brotli(crate::node::native_brotli_impl::Context { + mode, + ..Default::default() + }) + } + NodeMode::ZSTD_COMPRESS | NodeMode::ZSTD_DECOMPRESS => { + Engine::Zstd(crate::node::native_zstd_impl::Context { + mode, + ..Default::default() + }) + } + NodeMode::NONE => unreachable!("range-checked above"), + }; + + // Initialize only after the engine reaches its final heap address: + // zlib's z_stream is self-referential (deflateInit stores a + // state→strm back-pointer), so init-then-move leaves the stream + // "inconsistent" and every subsequent call fails with + // Z_STREAM_ERROR. node:zlib has the same invariant — its handles + // init() as a separate call on the already-boxed object. + let transformer = Box::new(CompressionStreamTransformer { + engine: JsCell::new(engine), + write_in_progress: Cell::new(false), + pending_close: Cell::new(false), + context_size, + }); + let err = transformer.engine.with_mut(|engine| { + let err = match engine { + Engine::Zlib(ctx) => { + // node:zlib defaults (zlib.ts): level Z_DEFAULT_COMPRESSION, + // windowBits 15, memLevel 8, strategy Z_DEFAULT_STRATEGY — + // CompressionStream exposes no options, so output bytes + // match the previous node:zlib-backed implementation. + ctx.init(-1, 15, 8, 0, None); + if ctx.mode == NodeMode::NONE { + Error::init( + c"Failed to initialize zlib stream".as_ptr(), + -1, + c"ERR_ZLIB_INITIALIZATION_FAILED".as_ptr(), + ) + } else { + Error::OK + } + } + Engine::Brotli(ctx) => ctx.init(), + // ZSTD_CONTENTSIZE_UNKNOWN — same as node:zlib with no + // pledgedSrcSize option. + Engine::Zstd(ctx) => ctx.init(u64::MAX), + Engine::Closed => unreachable!("just constructed"), + }; + if err.is_error() { + // An init-failed brotli/zstd context has `state: None`, and + // their `close()` doesn't tolerate that (brotli unwraps, + // zstd calls reset on null). Transition to Closed so the + // Drop this error-return is about to trigger is a no-op. + *engine = Engine::Closed; + } + err + }); + if err.is_error() { + return Err(throw_engine_error(global, err)); + } + + Ok(transformer) + } + + /// `transform(chunk, isFinish)` — run the full consume-input / + /// produce-output loop for one stream chunk and return a JS `Array` of + /// `Uint8Array`s, each its own exact-size allocation adopted without + /// copying. `isFinish` selects the family's finish operation (zlib + /// `Z_FINISH`, brotli `BROTLI_OPERATION_FINISH`, zstd `ZSTD_e_end`) for + /// the stream-end drain. Engine errors throw synchronously carrying + /// `message`/`code`/`errno`; the context stays open — the stream + /// teardown path (`cancel`/`close`) releases it. + #[bun_jsc::host_fn(method)] + pub(crate) fn transform( + &self, + global: &JSGlobalObject, + frame: &CallFrame, + ) -> JsResult { + if frame.arguments_count() < 2 { + return Err(global.throw_value( + bun_core::String::static_(b"transform(chunk, isFinish)").to_error_instance(global), + )); + } + let [chunk_value, is_finish_value] = frame.arguments_as_array::<2>(); + + let Some(in_buf) = chunk_value.as_array_buffer(global) else { + return Err(global.throw_invalid_argument_type_value( + "chunk", + "TypedArray or DataView", + chunk_value, + )); + }; + let is_finish = is_finish_value.to_boolean(); + + self.check_not_in_flight(global)?; + + // No JS runs while the engine is borrowed: the whole drive is pure + // native work; the output `Uint8Array`s and any error object are + // built after the borrow ends. + // `byte_slice` views the JS-owned backing store rooted via the + // argument value on the call stack. + let result = self + .engine + .with_mut(|engine| drive_loop(engine, in_buf.byte_slice(), is_finish)); + + match result { + Ok(outputs) => build_outputs_array(global, outputs), + Err(err) => Err(throw_engine_error(global, err)), + } + } + + /// `transformAsync(chunk, isFinish)` — same loop as `transform`, run on the + /// work pool. The input bytes are copied (the JS buffer is not pinned), the + /// engine stays in place (it is self-referential and must not move), and + /// the returned Promise resolves with the same `Array` shape + /// (or rejects carrying `message`/`code`/`errno`). The TransformStream + /// gates the next write on this promise, so at most one job is in flight + /// per transformer. + #[bun_jsc::host_fn(method)] + pub(crate) fn transform_async( + &self, + global: &JSGlobalObject, + frame: &CallFrame, + ) -> JsResult { + if frame.arguments_count() < 2 { + return Err(global.throw_value( + bun_core::String::static_(b"transformAsync(chunk, isFinish)") + .to_error_instance(global), + )); + } + let [chunk_value, is_finish_value] = frame.arguments_as_array::<2>(); + + let Some(in_buf) = chunk_value.as_array_buffer(global) else { + return Err(global.throw_invalid_argument_type_value( + "chunk", + "TypedArray or DataView", + chunk_value, + )); + }; + let is_finish = is_finish_value.to_boolean(); + + self.check_not_in_flight(global)?; + self.write_in_progress.set(true); + + let promise = JSPromiseStrong::init(global); + let promise_value = promise.value(); + + let ctx = AsyncTransformCtx { + _this_value: Strong::create(frame.this(), global), + transformer: core::ptr::from_ref(self), + input: Box::from(in_buf.byte_slice()), + is_finish, + promise, + result: Ok(Vec::new()), + }; + + match AnyTaskJob::create(global, ctx) { + Ok(job) => { + // SAFETY: `job` is the freshly-created live pointer. + unsafe { AnyTaskJob::schedule(job) }; + Ok(promise_value) + } + Err(e) => { + self.write_in_progress.set(false); + Err(e) + } + } + } + + /// Release the native context. Deferred when an async transform is in + /// flight (the returning `then()` closes it); otherwise idempotent. + /// Later `transform`/`transformAsync` calls throw. + #[bun_jsc::host_fn(method)] + pub(crate) fn close(&self, _global: &JSGlobalObject, _frame: &CallFrame) -> JsResult { + if self.write_in_progress.get() { + self.pending_close.set(true); + } else { + self.engine.with_mut(Engine::close); + } + Ok(JSValue::UNDEFINED) + } +} + +/// Build a JS `Error` carrying the engine's `message`/`code`/`errno` (the +/// node:zlib error triple). +fn build_engine_error(global: &JSGlobalObject, err: Error) -> JSValue { + let msg_bytes: &[u8] = if err.msg.is_null() { + b"Zlib error" + } else { + // SAFETY: non-null `Error::msg` points at a NUL-terminated C string + // (static literal or zlib/zstd-owned buffer valid for this call). + unsafe { bun_core::ffi::cstr(err.msg) }.to_bytes() + }; + let error_value = bun_core::String::clone_utf8(msg_bytes).to_error_instance(global); + + if !err.code.is_null() { + // SAFETY: same contract as `msg` above. + let code_bytes = unsafe { bun_core::ffi::cstr(err.code) }.to_bytes(); + if let Ok(code_value) = bun_core::String::clone_utf8(code_bytes).to_js(global) { + error_value.put(global, b"code", code_value); + } + } + error_value.put(global, b"errno", JSValue::js_number(f64::from(err.err))); + + error_value +} + +fn throw_engine_error(global: &JSGlobalObject, err: Error) -> bun_jsc::JsError { + global.throw_value(build_engine_error(global, err)) +} diff --git a/src/runtime/webcore/compression.classes.ts b/src/runtime/webcore/compression.classes.ts new file mode 100644 index 000000000000..830f12edb128 --- /dev/null +++ b/src/runtime/webcore/compression.classes.ts @@ -0,0 +1,27 @@ +import { define } from "../../codegen/class-definitions"; + +export default [ + define({ + name: "CompressionStreamTransformer", + construct: true, + finalize: true, + estimatedSize: true, + JSType: "0b11101110", + configurable: false, + klass: {}, + proto: { + transform: { + fn: "transform", + length: 2, + }, + transformAsync: { + fn: "transform_async", + length: 2, + }, + close: { + fn: "close", + length: 0, + }, + }, + }), +]; diff --git a/test/js/third_party/wpt-streams/cancel.any.js b/test/js/third_party/wpt-streams/cancel.any.js new file mode 100644 index 000000000000..fc5ef9570404 --- /dev/null +++ b/test/js/third_party/wpt-streams/cancel.any.js @@ -0,0 +1,205 @@ +// META: global=window,worker +// META: script=../resources/test-utils.js +'use strict'; + +const thrownError = new Error('bad things are happening!'); +thrownError.name = 'error1'; + +const originalReason = new Error('original reason'); +originalReason.name = 'error2'; + +promise_test(async t => { + let cancelled = undefined; + const ts = new TransformStream({ + cancel(reason) { + cancelled = reason; + } + }); + const res = await ts.readable.cancel(thrownError); + assert_equals(res, undefined, 'readable.cancel() should return undefined'); + assert_equals(cancelled, thrownError, 'transformer.cancel() should be called with the passed reason'); +}, 'cancelling the readable side should call transformer.cancel()'); + +promise_test(async t => { + const ts = new TransformStream({ + cancel(reason) { + assert_equals(reason, originalReason, 'transformer.cancel() should be called with the passed reason'); + throw thrownError; + } + }); + const writer = ts.writable.getWriter(); + const cancelPromise = ts.readable.cancel(originalReason); + await promise_rejects_exactly(t, thrownError, cancelPromise, 'readable.cancel() should reject with thrownError'); + await promise_rejects_exactly(t, thrownError, writer.closed, 'writer.closed should reject with thrownError'); +}, 'cancelling the readable side should reject if transformer.cancel() throws'); + +promise_test(async t => { + let aborted = undefined; + const ts = new TransformStream({ + cancel(reason) { + aborted = reason; + }, + flush: t.unreached_func('flush should not be called') + }); + const res = await ts.writable.abort(thrownError); + assert_equals(res, undefined, 'writable.abort() should return undefined'); + assert_equals(aborted, thrownError, 'transformer.abort() should be called with the passed reason'); +}, 'aborting the writable side should call transformer.abort()'); + +promise_test(async t => { + const ts = new TransformStream({ + cancel(reason) { + assert_equals(reason, originalReason, 'transformer.cancel() should be called with the passed reason'); + throw thrownError; + }, + flush: t.unreached_func('flush should not be called') + }); + const reader = ts.readable.getReader(); + const abortPromise = ts.writable.abort(originalReason); + await promise_rejects_exactly(t, thrownError, abortPromise, 'writable.abort() should reject with thrownError'); + await promise_rejects_exactly(t, thrownError, reader.closed, 'reader.closed should reject with thrownError'); +}, 'aborting the writable side should reject if transformer.cancel() throws'); + +promise_test(async t => { + const ts = new TransformStream({ + async cancel(reason) { + assert_equals(reason, originalReason, 'transformer.cancel() should be called with the passed reason'); + throw thrownError; + }, + flush: t.unreached_func('flush should not be called') + }); + const cancelPromise = ts.readable.cancel(originalReason); + const closePromise = ts.writable.close(); + await Promise.all([ + promise_rejects_exactly(t, thrownError, cancelPromise, 'cancelPromise should reject with thrownError'), + promise_rejects_exactly(t, thrownError, closePromise, 'closePromise should reject with thrownError'), + ]); +}, 'closing the writable side should reject if a parallel transformer.cancel() throws'); + +promise_test(async t => { + let controller; + const ts = new TransformStream({ + start(c) { + controller = c; + }, + async cancel(reason) { + assert_equals(reason, originalReason, 'transformer.cancel() should be called with the passed reason'); + controller.error(thrownError); + }, + flush: t.unreached_func('flush should not be called') + }); + const cancelPromise = ts.readable.cancel(originalReason); + const closePromise = ts.writable.close(); + await Promise.all([ + promise_rejects_exactly(t, thrownError, cancelPromise, 'cancelPromise should reject with thrownError'), + promise_rejects_exactly(t, thrownError, closePromise, 'closePromise should reject with thrownError'), + ]); +}, 'readable.cancel() and a parallel writable.close() should reject if a transformer.cancel() calls controller.error()'); + +promise_test(async t => { + let controller; + const ts = new TransformStream({ + start(c) { + controller = c; + }, + async cancel(reason) { + assert_equals(reason, originalReason, 'transformer.cancel() should be called with the passed reason'); + controller.error(thrownError); + }, + flush: t.unreached_func('flush should not be called') + }); + const cancelPromise = ts.writable.abort(originalReason); + await promise_rejects_exactly(t, thrownError, cancelPromise, 'cancelPromise should reject with thrownError'); + const closePromise = ts.readable.cancel(1); + await promise_rejects_exactly(t, thrownError, closePromise, 'closePromise should reject with thrownError'); +}, 'writable.abort() and readable.cancel() should reject if a transformer.cancel() calls controller.error()'); + +promise_test(async t => { + const cancelReason = new Error('cancel reason'); + let controller; + let cancelPromise; + let flushCalled = false; + const ts = new TransformStream({ + start(c) { + controller = c; + }, + flush() { + flushCalled = true; + cancelPromise = ts.readable.cancel(cancelReason); + }, + cancel: t.unreached_func('cancel should not be called') + }); + await flushAsyncEvents(); // ensure stream is started + await ts.writable.close(); + assert_true(flushCalled, 'flush() was called'); + await cancelPromise; +}, 'readable.cancel() should not call cancel() when flush() is already called from writable.close()'); + +promise_test(async t => { + const cancelReason = new Error('cancel reason'); + const abortReason = new Error('abort reason'); + let cancelCalls = 0; + let controller; + let cancelPromise; + const ts = new TransformStream({ + start(c) { + controller = c; + }, + cancel() { + if (++cancelCalls === 1) { + cancelPromise = ts.readable.cancel(cancelReason); + } + }, + flush: t.unreached_func('flush should not be called') + }); + await flushAsyncEvents(); // ensure stream is started + await ts.writable.abort(abortReason); + assert_equals(cancelCalls, 1); + await cancelPromise; + assert_equals(cancelCalls, 1); +}, 'readable.cancel() should not call cancel() again when already called from writable.abort()'); + +promise_test(async t => { + const cancelReason = new Error('cancel reason'); + let controller; + let closePromise; + let cancelCalled = false; + const ts = new TransformStream({ + start(c) { + controller = c; + }, + cancel() { + cancelCalled = true; + closePromise = ts.writable.close(); + }, + flush: t.unreached_func('flush should not be called') + }); + await flushAsyncEvents(); // ensure stream is started + await ts.readable.cancel(cancelReason); + assert_true(cancelCalled, 'cancel() was called'); + await closePromise; +}, 'writable.close() should not call flush() when cancel() is already called from readable.cancel()'); + +promise_test(async t => { + const cancelReason = new Error('cancel reason'); + const abortReason = new Error('abort reason'); + let cancelCalls = 0; + let controller; + let abortPromise; + const ts = new TransformStream({ + start(c) { + controller = c; + }, + cancel() { + if (++cancelCalls === 1) { + abortPromise = ts.writable.abort(abortReason); + } + }, + flush: t.unreached_func('flush should not be called') + }); + await flushAsyncEvents(); // ensure stream is started + await promise_rejects_exactly(t, abortReason, ts.readable.cancel(cancelReason)); + assert_equals(cancelCalls, 1); + await promise_rejects_exactly(t, abortReason, abortPromise); + assert_equals(cancelCalls, 1); +}, 'writable.abort() should not call cancel() again when already called from readable.cancel()'); diff --git a/test/js/third_party/wpt-streams/run.test.ts b/test/js/third_party/wpt-streams/run.test.ts new file mode 100644 index 000000000000..ee8df8d58cf0 --- /dev/null +++ b/test/js/third_party/wpt-streams/run.test.ts @@ -0,0 +1,45 @@ +// Runs vendored WPT streams .any.js tests against Bun's TransformStream. +// The .any.js files are byte-identical to upstream; this driver supplies the +// testharness globals they need. +// +// Vendored from web-platform-tests/wpt @ e4a4672e9e607fc2b28e7173b83ce4e38ef53071 +// streams/transform-streams/cancel.any.js + +import { describe } from "bun:test"; +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { wptTest } from "../wpt-h2/testharness-shim"; + +const g = globalThis as any; +g.self = globalThis; + +g.step_timeout = (fn: () => void, ms: number) => setTimeout(fn, ms); +g.delay = (ms: number) => new Promise(resolve => setTimeout(resolve, ms)); +g.flushAsyncEvents = () => + g + .delay(0) + .then(() => g.delay(0)) + .then(() => g.delay(0)) + .then(() => g.delay(0)); + +const wptTestObject = { + unreached_func(msg: string) { + return () => { + throw new Error(`unreached_func: ${msg}`); + }; + }, +}; + +g.promise_test = (fn: (t: unknown) => Promise, name: string) => { + wptTest(() => fn(wptTestObject), name); +}; + +// bun:test injects its own `test` binding into every imported module, which +// would shadow the WPT-style test(fn, name) global. Load each vendored file +// as text and run it inside a Function whose `test` parameter is the shim. +for (const file of ["cancel.any.js"]) { + const src = readFileSync(join(import.meta.dir, file), "utf8"); + describe(file, () => { + new Function("test", src)(wptTest); + }); +} diff --git a/test/js/web/streams/compression.test.ts b/test/js/web/streams/compression.test.ts index 7c0ddd035082..5af78d49a51f 100644 --- a/test/js/web/streams/compression.test.ts +++ b/test/js/web/streams/compression.test.ts @@ -1,4 +1,6 @@ import { describe, expect, test } from "bun:test"; +import { randomBytes } from "node:crypto"; +import zlib from "node:zlib"; describe("CompressionStream and DecompressionStream", () => { describe("brotli", () => { @@ -304,8 +306,8 @@ describe("CompressionStream chunk handling (Node v26 semantics)", () => { const reader = cs.readable.getReader(); const writeError = writer.write(42).catch(e => e); - // Without the kDestroyOnSyncError handling the readable side hangs - // forever here. + // A throw in transform() errors both sides of the TransformStream, so + // the readable side rejects instead of hanging. const readError = reader.read().catch(e => e); const [we, re] = await Promise.all([writeError, readError]); @@ -338,3 +340,702 @@ describe("CompressionStream chunk handling (Node v26 semantics)", () => { } }); }); + +describe("CompressionStream write/read ordering", () => { + // The implementation buffers output on the readable side: awaiting writes + // before any reader attaches must not deadlock. (A strictly spec-default + // TransformStream — readable highWaterMark 0 — would stall here; this pins + // Bun's long-standing buffered behavior.) + test("awaiting writes before reading does not deadlock", async () => { + const cs = new CompressionStream("gzip"); + const writer = cs.writable.getWriter(); + await writer.write(new TextEncoder().encode("hello")); + await writer.write(new TextEncoder().encode("world")); + await writer.close(); + + const chunks: Uint8Array[] = []; + for await (const chunk of cs.readable as unknown as AsyncIterable) chunks.push(chunk); + const ds = new DecompressionStream("gzip"); + const w2 = ds.writable.getWriter(); + await w2.write(Buffer.concat(chunks)); + await w2.close(); + const out: Uint8Array[] = []; + for await (const chunk of ds.readable as unknown as AsyncIterable) out.push(chunk); + expect(Buffer.concat(out).toString()).toBe("helloworld"); + }); + + test("writes after a chunk whose output expands past the buffer still resolve before reading", async () => { + // ~100 bytes of input inflating to 64KB of output blows straight through + // a single-chunkSize readable budget; the writes that follow must still + // resolve with no reader attached — the node-adapter implementation + // accepted ~16KB of *input* regardless of how large the buffered output + // grew, and this sequence resolved on it. + const big = zlib.gzipSync(Buffer.alloc(64 * 1024)); + const small = zlib.gzipSync(Buffer.from("hello")); + const ds = new DecompressionStream("gzip"); + const writer = ds.writable.getWriter(); + await writer.write(big); + for (let i = 0; i < 8; i++) await writer.write(small); + // close() cannot settle until the reader drains the buffered output, so + // capture its settlement and assert it after the drain loop. + const closed = writer.close().then( + () => "resolved", + e => `rejected: ${e}`, + ); + const out: Uint8Array[] = []; + for await (const chunk of ds.readable as unknown as AsyncIterable) out.push(chunk); + const total = Buffer.concat(out); + expect(total.length).toBe(64 * 1024 + 8 * 5); + expect(total.subarray(64 * 1024).toString()).toBe(Buffer.alloc(8 * 5, "hello").toString()); + expect(await closed).toBe("resolved"); + }); + + test("corrupt input rejects with Z_DATA_ERROR", async () => { + const ds = new DecompressionStream("gzip"); + const writer = ds.writable.getWriter(); + writer.write(new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8])).catch(() => {}); + writer.close().catch(() => {}); + try { + for await (const _ of ds.readable as unknown as AsyncIterable) { + } + expect.unreachable(); + } catch (e) { + expect((e as { code?: string }).code).toBe("Z_DATA_ERROR"); + } + }); +}); + +// Every behavior in this block was verified against Node v24 (except the +// brotli/zstd cases — formats Node doesn't support — which pin Bun's +// pre-existing behavior). +describe("CompressionStream Node.js compatibility", () => { + async function collect(readable: ReadableStream): Promise { + const chunks: Uint8Array[] = []; + for await (const chunk of readable as unknown as AsyncIterable) chunks.push(chunk); + return chunks; + } + + async function drain( + stream: CompressionStream | DecompressionStream, + inputs: Array, + ): Promise { + const writer = stream.writable.getWriter(); + const collected = collect(stream.readable); + try { + for (const input of inputs) await writer.write(input); + await writer.close(); + } catch (e) { + // The readable rejects with the same stream error; settle it so the + // write/close error is the one that propagates. + await collected.catch(() => {}); + throw e; + } + return Buffer.concat(await collected); + } + + async function decompress(format: string, inputs: Array): Promise { + return drain(new DecompressionStream(format as Bun.CompressionFormat), inputs); + } + + async function roundTrip(format: string, inputs: Array): Promise { + return decompress(format, [await drain(new CompressionStream(format as Bun.CompressionFormat), inputs)]); + } + + describe("input chunk types", () => { + test("accepts string, ArrayBuffer, DataView, TypedArray and offset subarray like node", async () => { + expect(await roundTrip("gzip", ["hello"])).toEqual(Buffer.from("hello")); + + const bytes = new TextEncoder().encode("hello"); + // A plain ArrayBuffer is a valid BufferSource. + expect(await roundTrip("gzip", [bytes.slice().buffer])).toEqual(Buffer.from("hello")); + + expect(await roundTrip("gzip", [new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength)])).toEqual( + Buffer.from("hello"), + ); + + // A non-byte TypedArray is interpreted as its underlying bytes. + expect(await roundTrip("gzip", [new Uint16Array([0x6568, 0x6c6c, 0x6f])])).toEqual( + Buffer.from([0x68, 0x65, 0x6c, 0x6c, 0x6f, 0x00]), + ); + + // Only the view's window, not its whole backing buffer. + const big = Buffer.alloc(32, 0xff); + big.set(bytes, 10); + expect(await roundTrip("gzip", [big.subarray(10, 15)])).toEqual(Buffer.from("hello")); + }); + + test.each([ + ["SharedArrayBuffer", () => new SharedArrayBuffer(8)], + ["SharedArrayBuffer-backed view", () => new Uint8Array(new SharedArrayBuffer(8))], + ["number", () => 42], + ["undefined", () => undefined], + ["plain object", () => ({})], + ["Blob", () => new Blob(["hello"])], + ] as Array<[string, () => unknown]>)("rejects %s with ERR_INVALID_ARG_TYPE like node", async (_label, make) => { + const cs = new CompressionStream("gzip"); + const writer = cs.writable.getWriter(); + const collected = collect(cs.readable).catch(() => []); + const err = (await writer.write(make() as Uint8Array).then( + () => null, + e => e, + )) as { code?: string; constructor: unknown } | null; + expect(err).not.toBeNull(); + expect(err!.constructor).toBe(TypeError); + expect(err!.code).toBe("ERR_INVALID_ARG_TYPE"); + // The failed write errors the whole stream. (Node instead leaves the + // stream wedged — later writes never settle — so this pins the only + // sane teardown.) + expect( + ( + await writer.closed.then( + () => null, + (e: { code?: string }) => e, + ) + )?.code, + ).toBe("ERR_INVALID_ARG_TYPE"); + await collected; + }); + + test("rejects null with ERR_STREAM_NULL_VALUES like node", async () => { + const cs = new CompressionStream("gzip"); + const writer = cs.writable.getWriter(); + const collected = collect(cs.readable).catch(() => []); + const err = (await writer.write(null as unknown as Uint8Array).then( + () => null, + e => e, + )) as { code?: string } | null; + expect(err?.code).toBe("ERR_STREAM_NULL_VALUES"); + await collected; + }); + + test("rejects a view over a detached ArrayBuffer with TypeError like node", async () => { + const buffer = new ArrayBuffer(5); + const view = new Uint8Array(buffer); + structuredClone(buffer, { transfer: [buffer] }); // detach + const cs = new CompressionStream("gzip"); + const writer = cs.writable.getWriter(); + const collected = collect(cs.readable).catch(() => []); + const err = await writer.write(view).then( + () => null, + e => e, + ); + expect(err).toBeInstanceOf(TypeError); + await collected; + }); + }); + + describe("empty input", () => { + test("empty chunks and an empty stream still produce a valid compressed stream", async () => { + for (const format of ["gzip", "deflate", "deflate-raw", "brotli", "zstd"]) { + expect(await roundTrip(format, [new Uint8Array(0)])).toEqual(Buffer.alloc(0)); + expect(await roundTrip(format, [])).toEqual(Buffer.alloc(0)); + } + }); + + test("closing an empty gzip/deflate/brotli DecompressionStream rejects with Z_BUF_ERROR like node", async () => { + for (const format of ["gzip", "deflate", "deflate-raw", "brotli"]) { + const err = (await decompress(format, []).then( + () => null, + e => e, + )) as { code?: string } | null; + expect(err?.code).toBe("Z_BUF_ERROR"); + } + }); + + test("closing an empty zstd DecompressionStream produces empty output", async () => { + expect(await decompress("zstd", [])).toEqual(Buffer.alloc(0)); + }); + }); + + describe("malformed compressed input", () => { + test("truncated gzip rejects with Z_BUF_ERROR on close like node", async () => { + const gzipped = zlib.gzipSync(Buffer.alloc(1000, "a")); + const err = (await decompress("gzip", [gzipped.subarray(0, gzipped.length - 5)]).then( + () => null, + e => e, + )) as { code?: string } | null; + expect(err?.code).toBe("Z_BUF_ERROR"); + }); + + test("trailing garbage after the gzip stream rejects with Z_DATA_ERROR like node", async () => { + const payload = Buffer.concat([zlib.gzipSync(Buffer.from("hello")), Buffer.from("garbage!")]); + const err = (await decompress("gzip", [payload]).then( + () => null, + e => e, + )) as { code?: string } | null; + expect(err?.code).toBe("Z_DATA_ERROR"); + }); + + test("trailing bytes the engine leaves unconsumed at stream end are discarded like node", async () => { + // The engine stops at stream end with the trailing bytes unconsumed + // and no error; node's drive loop treats leftover input with spare + // output as end-of-stream and discards it (lib/zlib.js + // processCallback) instead of re-feeding bytes the engine refuses. + // (gzip with NON-zero trailing bytes takes the multi-member path and + // rejects instead — pinned above; zero bytes are member padding.) + const payload = Buffer.from("hello"); + const cases: Array<[string, Buffer]> = [ + ["deflate", Buffer.concat([zlib.deflateSync(payload), Buffer.from([1])])], + ["deflate-raw", Buffer.concat([zlib.deflateRawSync(payload), Buffer.from([1, 2, 3])])], + ["gzip", Buffer.concat([zlib.gzipSync(payload), Buffer.alloc(8)])], + ["brotli", Buffer.concat([zlib.brotliCompressSync(payload), Buffer.from([1, 2, 3])])], + ["zstd", Buffer.concat([zlib.zstdCompressSync(payload), Buffer.from([1, 2, 3])])], + ]; + for (const [format, input] of cases) { + expect([format, (await decompress(format, [input])).toString()]).toEqual([format, "hello"]); + } + }); + + test("corrupt zstd input carries the zstd error code", async () => { + const err = (await decompress("zstd", [new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8])]).then( + () => null, + e => e, + )) as { code?: string } | null; + expect(err?.code).toBe("ZSTD_error_prefix_unknown"); + }); + + test("a corrupt chunk errors pending and subsequent operations like node", async () => { + const ds = new DecompressionStream("gzip"); + const writer = ds.writable.getWriter(); + const reader = ds.readable.getReader(); + // (Node resolves this write — its adapter buffers the chunk before the + // engine sees it — where the synchronous engine rejects it; the spec + // propagates transform errors to the write. Don't pin the timing, pin + // where the error surfaces and what it carries.) + await writer.write(new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8])).catch(() => {}); + const readErr = (await reader.read().then( + () => null, + e => e, + )) as { code?: string } | null; + expect(readErr?.constructor).toBe(TypeError); + expect(readErr?.code).toBe("Z_DATA_ERROR"); + expect( + await writer.close().then( + () => "resolved", + () => "rejected", + ), + ).toBe("rejected"); + }); + }); + + describe("gzip specifics", () => { + test("concatenated gzip members decompress to the concatenated payload like node", async () => { + const payload = Buffer.concat([zlib.gzipSync(Buffer.from("hello")), zlib.gzipSync(Buffer.from("world"))]); + expect((await decompress("gzip", [payload])).toString()).toBe("helloworld"); + }); + + test("decompressing byte-at-a-time yields the full payload", async () => { + const expected = Buffer.alloc(300, "x"); + const gzipped = zlib.gzipSync(expected) as Buffer; + const inputs = Array.from(gzipped, byte => new Uint8Array([byte])); + expect(await decompress("gzip", inputs)).toEqual(expected); + }); + }); + + describe("output", () => { + test("chunks are plain Uint8Arrays like node", async () => { + const cs = new CompressionStream("gzip"); + const writer = cs.writable.getWriter(); + const collected = collect(cs.readable); + await writer.write(Buffer.alloc(100, "a")); + await writer.close(); + const chunks = await collected; + expect(chunks.length).toBeGreaterThan(0); + for (const chunk of chunks) { + expect(Object.getPrototypeOf(chunk)).toBe(Uint8Array.prototype); + } + }); + + test("chunks expose no bytes beyond their view — no recycled heap memory reachable", async () => { + // Each output chunk is an exact-size allocation: chunk.buffer must not + // reach past the view's window at all, so there is no spare region + // that could disclose previous heap contents. + const cs = new CompressionStream("gzip"); + const writer = cs.writable.getWriter(); + const collected = collect(cs.readable); + await writer.write(Buffer.from("hello")); + await writer.close(); + const chunks = await collected; + for (const chunk of chunks) { + expect(chunk.byteOffset).toBe(0); + expect(chunk.buffer.byteLength).toBe(chunk.byteLength); + } + }); + + test("byte-at-a-time decompression yields bounded exact-size chunks", async () => { + // Incompressible input dribbled in byte-at-a-time makes the engine + // emit many small output chunks. Each is an independent exact-size + // allocation no larger than the native output granularity (16KB) — + // the native drive adopts the engine's output buffers instead of + // copying them into JS-side staging. + const payload = new Uint8Array(1024); + crypto.getRandomValues(payload); + const gzipped = zlib.gzipSync(payload) as Buffer; + + const ds = new DecompressionStream("gzip"); + const writer = ds.writable.getWriter(); + const collected = collect(ds.readable); + for (const byte of gzipped) await writer.write(new Uint8Array([byte])); + await writer.close(); + const chunks = await collected; + + expect(Buffer.concat(chunks)).toEqual(Buffer.from(payload)); + expect(chunks.length).toBeGreaterThan(1); + for (const chunk of chunks) { + expect(chunk.byteLength).toBeLessThanOrEqual(16384); + expect(chunk.buffer.byteLength).toBe(chunk.byteLength); + } + }); + + test("multi-chunk payloads round-trip for every format", async () => { + const incompressible = new Uint8Array(1024 * 1024); + crypto.getRandomValues(incompressible); + const compressible = Buffer.alloc(2 * 1024 * 1024, "abcdefgh"); + for (const format of ["gzip", "deflate", "deflate-raw", "brotli", "zstd"]) { + expect(await roundTrip(format, [incompressible])).toEqual(Buffer.from(incompressible)); + expect(await roundTrip(format, [compressible])).toEqual(compressible); + } + }); + + test("output bytes match node:zlib for gzip", async () => { + const payload = Buffer.alloc(100_000, "compression streams test "); + expect(await decompress("gzip", [zlib.gzipSync(payload)])).toEqual(payload); + const compressed = await drain(new CompressionStream("gzip"), [payload]); + expect(zlib.gunzipSync(compressed)).toEqual(payload); + }); + }); + + describe("teardown", () => { + test("reader.cancel() errors the writable with the cancel reason like node", async () => { + const cs = new CompressionStream("gzip"); + const writer = cs.writable.getWriter(); + const reader = cs.readable.getReader(); + await writer.write(Buffer.from("hello")); + await reader.cancel("because"); + expect( + await writer.write(Buffer.from("world")).then( + () => null, + e => e, + ), + ).toBe("because"); + expect( + await writer.closed.then( + () => null, + (e: unknown) => e, + ), + ).toBe("because"); + }); + + test("writer.abort() errors pending reads with the abort reason like node", async () => { + const cs = new CompressionStream("gzip"); + const writer = cs.writable.getWriter(); + const reader = cs.readable.getReader(); + await writer.write(Buffer.from("hello")); + const boom = new Error("boom"); + await writer.abort(boom); + expect( + await reader.read().then( + () => null, + e => e, + ), + ).toBe(boom); + }); + + test("a fresh stream can be torn down immediately", async () => { + // No writes at all — cancel/abort must not trip over the engine. + await new CompressionStream("gzip").readable.cancel("x"); + await new DecompressionStream("gzip").writable.abort("y"); + const cs = new CompressionStream("zstd"); + await Promise.all([cs.readable.cancel(), cs.writable.abort()].map(p => p.catch(() => {}))); + }); + }); +}); + +describe("engine lifecycle", () => { + // The native transformer holds a zlib/brotli/zstd context (~256KB for + // deflate). These loops create far more streams than would fit in memory + // if any teardown path leaked the context: completed, cancelled + // mid-stream, aborted, and abandoned-to-GC streams must all release it. + const RSS_BUDGET_MB = 256; + + async function rssGrowthMB(fn: () => Promise): Promise { + Bun.gc(true); + const before = process.memoryUsage.rss(); + await fn(); + Bun.gc(true); + return (process.memoryUsage.rss() - before) / 1024 / 1024; + } + + test("completed streams release the engine", async () => { + const growth = await rssGrowthMB(async () => { + for (let i = 0; i < 500; i++) { + const cs = new CompressionStream("gzip"); + const writer = cs.writable.getWriter(); + const reads = (async () => { + for await (const _ of cs.readable as unknown as AsyncIterable) { + } + })(); + await writer.write(new Uint8Array(1024)); + await writer.close(); + await reads; + } + }); + expect(growth).toBeLessThan(RSS_BUDGET_MB); + }); + + test("cancelled and aborted streams release the engine", async () => { + const growth = await rssGrowthMB(async () => { + for (let i = 0; i < 500; i++) { + const cs = new CompressionStream(i % 2 ? "gzip" : "zstd"); + const writer = cs.writable.getWriter(); + await writer.write(new Uint8Array(1024)).catch(() => {}); + if (i % 2) { + await cs.readable.cancel("done"); + await writer.abort("done").catch(() => {}); + } else { + await writer.abort("done").catch(() => {}); + await cs.readable.cancel("done").catch(() => {}); + } + } + }); + expect(growth).toBeLessThan(RSS_BUDGET_MB); + }); + + test("abandoned streams release the engine via GC", async () => { + const growth = await rssGrowthMB(async () => { + for (let i = 0; i < 1000; i++) { + // No flush, no cancel — the only release path is finalization. + const cs = new CompressionStream("deflate"); + const writer = cs.writable.getWriter(); + await writer.write(new Uint8Array(64)).catch(() => {}); + if (i % 100 === 99) Bun.gc(true); + } + }); + expect(growth).toBeLessThan(RSS_BUDGET_MB); + }); +}); + +// Chunks past createCompressionTransform's asyncThreshold (4 * 16KB) take the +// transformAsync path: input is copied, the drive loop runs on the work pool, +// and the write resolves once the worker completes. Output must be +// byte-identical to the synchronous path and to node:zlib's defaults. +describe("CompressionStream large-chunk work-pool offload", () => { + // Well past asyncThreshold; non-uniform so a corruption shows in the + // round-trip equality. + const big = (() => { + const b = new Uint8Array(256 * 1024); + for (let i = 0; i < b.length; i++) b[i] = (i * 131) & 0xff; + return b; + })(); + + async function collect(readable: ReadableStream) { + const chunks: Uint8Array[] = []; + for await (const c of readable) chunks.push(c); + return Buffer.concat(chunks); + } + + async function roundTrip(format: "gzip" | "deflate" | "deflate-raw" | "brotli" | "zstd") { + const cs = new CompressionStream(format); + const w = cs.writable.getWriter(); + w.write(big); + w.close(); + const compressed = await collect(cs.readable); + + const ds = new DecompressionStream(format); + const dw = ds.writable.getWriter(); + dw.write(compressed); + dw.close(); + const out = await collect(ds.readable); + + expect(out.equals(big)).toBe(true); + return compressed; + } + + for (const format of ["gzip", "deflate", "deflate-raw", "brotli", "zstd"] as const) { + test(`${format} round-trips a 256KB single-write chunk`, async () => { + await roundTrip(format); + }); + } + + test("gzip output is byte-identical to node:zlib's defaults", async () => { + const compressed = await roundTrip("gzip"); + expect(compressed.equals(zlib.gzipSync(big))).toBe(true); + }); + + test("multiple large writes in sequence are gated and round-trip", async () => { + const cs = new CompressionStream("gzip"); + const w = cs.writable.getWriter(); + // Each write awaits the work-pool round-trip before the next starts. + for (let i = 0; i < 4; i++) await w.write(big); + await w.close(); + const compressed = await collect(cs.readable); + + const ds = new DecompressionStream("gzip"); + const dw = ds.writable.getWriter(); + dw.write(compressed); + dw.close(); + const out = await collect(ds.readable); + + expect(out.length).toBe(4 * big.length); + for (let i = 0; i < 4; i++) { + expect(out.subarray(i * big.length, (i + 1) * big.length).equals(big)).toBe(true); + } + }); + + test("reader.cancel() while a large write is in flight defers the engine close", async () => { + const cs = new CompressionStream("gzip"); + const writer = cs.writable.getWriter(); + const reader = cs.readable.getReader(); + // The writable controller's started flag is set in a microtask; until + // then write() only queues the chunk and transformAsync is never + // reached, so pending_close would not be exercised. + await writer.ready; + // Do not await: the work-pool job is in flight when cancel runs. + const writePromise = writer.write(big); + await reader.cancel("stop"); + // The write took the async path and the stream is now torn down. The + // transformer's cancel() hook closed the handle with the worker still + // running, so the close is deferred via pending_close and the worker's + // then() releases the engine. The write then settles: the worker may + // have resolved before the writable was errored, or the writable + // errors it with the cancel reason, or enqueue throws on the closed + // readable — what must not happen is a hang or a crash from closing + // the engine under the worker. + const outcome = await writePromise.then( + () => "fulfilled", + e => (e === "stop" ? "rejected:stop" : `rejected:${e?.constructor?.name}`), + ); + expect(["fulfilled", "rejected:stop", "rejected:TypeError"]).toContain(outcome); + expect( + await writer.closed.then( + () => null, + e => e, + ), + ).toBe("stop"); + }); + + test("a corrupt large chunk rejects on the async path with the engine code", async () => { + const ds = new DecompressionStream("gzip"); + const writer = ds.writable.getWriter(); + const reader = ds.readable.getReader(); + const bad = Buffer.alloc(128 * 1024, 0xff); + const err = await writer.write(bad).then( + () => null, + e => e, + ); + expect(err).toBeInstanceOf(TypeError); + expect((err as any).code).toBe("Z_DATA_ERROR"); + expect( + await reader.read().then( + () => null, + e => e, + ), + ).toBe(err); + }); + + // brotli at the default quality 11 only buffers input during PROCESS + // and encodes a whole ~256KB metablock when the ring buffer fills, so a + // stream of sub-64KB writes (the common pipeThrough case from network + // or file sources) would run that encode on the JS thread if brotli + // encode kept the input-size threshold. Every non-empty brotli write + // takes the work pool. + test("brotli metablock encode from small writes runs on the work pool", async () => { + const cs = new CompressionStream("brotli"); + const writer = cs.writable.getWriter(); + const drain = (async () => { + for await (const _ of cs.readable); + })(); + await writer.ready; + + // 16KB incompressible chunks: each is below the 64KB zlib/zstd + // threshold. The 16th write fills the 256KB ring buffer and triggers + // the q11 metablock encode (hundreds of ms). If that write took the + // sync path the event loop would not turn before its promise settles. + for (let i = 0; i < 15; i++) await writer.write(randomBytes(16 * 1024)); + let eventLoopTurned = false; + setImmediate(() => { + eventLoopTurned = true; + }); + await writer.write(randomBytes(16 * 1024)); + expect(eventLoopTurned).toBe(true); + + await writer.close(); + await drain; + }); + + // brotli at the default quality 11 buffers input during PROCESS and + // encodes the residual block at FINISH, so the 0-byte finish chunk is + // where the work is. flush() must take the work-pool path so that + // encoding runs off the JS thread. + test("brotli finish-flush runs on the work pool, not the JS thread", async () => { + const cs = new CompressionStream("brotli"); + const writer = cs.writable.getWriter(); + const drain = (async () => { + for await (const _ of cs.readable); + })(); + + // 200KB incompressible so q11 FINISH does hundreds of ms of real + // encoding; < 256KB so PROCESS buffers it whole and emits nothing. + await writer.write(randomBytes(200 * 1024)); + + // If flush blocks the JS thread, the close promise settles via + // microtasks only and this macrotask never gets a turn before the + // await below resumes. If flush goes to the work pool, the close + // promise stays pending across the event-loop iteration that runs + // this callback. + let eventLoopTurned = false; + setImmediate(() => { + eventLoopTurned = true; + }); + + await writer.close(); + await drain; + + expect(eventLoopTurned).toBe(true); + }); + + test("an engine error on the async finish-flush rejects close with the wrapped code", async () => { + const ds = new DecompressionStream("gzip"); + const writer = ds.writable.getWriter(); + const reader = ds.readable.getReader(); + // Valid gzip header with no body: PROCESS accepts it, FINISH fails + // with Z_BUF_ERROR because the stream ended prematurely. + await writer.write(zlib.gzipSync(Buffer.from("hello")).subarray(0, 10)); + const closeErr = await writer.close().then( + () => null, + e => e, + ); + const readErr = await reader.read().then( + () => null, + e => e, + ); + expect(closeErr).toBeInstanceOf(TypeError); + expect((closeErr as any).code).toBe("Z_BUF_ERROR"); + expect(readErr).toBe(closeErr); + }); + + // reader.cancel() while the finish-flush worker is in flight closes + // the readable and short-circuits sourceCancelAlgorithm on the + // already-set finishPromise, so the transformer's cancel() hook never + // runs. The flush fulfillment must release the engine itself and + // discard the outputs rather than throwing on the closed readable + // (which would reject both close and cancel with an internal error + // and skip the release). + test("reader.cancel() racing the async finish-flush resolves cleanly", async () => { + const cs = new CompressionStream("brotli"); + const writer = cs.writable.getWriter(); + const reader = cs.readable.getReader(); + // 200KB at q11: PROCESS buffers it, FINISH encodes it on the pool, + // giving the cancel a wide window. + await writer.write(randomBytes(200 * 1024)); + const closeP = writer.close(); + const cancelP = reader.cancel("stop"); + const [closeR, cancelR] = await Promise.allSettled([closeP, cancelP]); + // Before the fix both reject with + // "TypeError: TransformStream.readable cannot close or enqueue". + expect({ close: closeR, cancel: cancelR }).toEqual({ + close: { status: "fulfilled", value: undefined }, + cancel: { status: "fulfilled", value: undefined }, + }); + }); +}); diff --git a/test/js/web/streams/streams-leak.test.ts b/test/js/web/streams/streams-leak.test.ts index 33eb0cae4274..edacb0d4abda 100644 --- a/test/js/web/streams/streams-leak.test.ts +++ b/test/js/web/streams/streams-leak.test.ts @@ -34,28 +34,32 @@ test("native ReadableStream reuses the pull buffer across small reads", async () const chunks: Uint8Array[] = []; for await (const chunk of resp.body!) chunks.push(chunk); - // Some chunks coalesce on the wire; we just need a meaningful sample - // of small reads through the native pull path. - expect(chunks.length).toBeGreaterThan(20); + // Chunks coalesce on the wire — heavily so on a loaded CI machine (runs + // have seen anywhere from 8 to hundreds of chunks) — so the assertions + // below must not depend on how many reads the 2KB arrived in; any two + // reads are enough to observe buffer reuse. + expect(chunks.length).toBeGreaterThan(1); // Consecutive small reads should land in the same backing buffer (the - // tail subarray is reused until a read fills it). 2KB of ~few-byte - // chunks fits well inside one 256KB buffer, so the whole stream should - // share a handful at most. Pre-fix every chunk had its own 256KB - // buffer, so this was ~chunks.length. + // tail subarray is reused until a read fills it). 2KB of small chunks + // fits well inside one 256KB buffer, so at least some chunks must share + // a buffer. Pre-fix every chunk had its own fresh 256KB buffer, making + // this exactly chunks.length regardless of coalescing. const distinctBuffers = new Set(chunks.map(c => c.buffer)); - expect(distinctBuffers.size).toBeLessThan(8); + expect(distinctBuffers.size).toBeLessThan(chunks.length); let backingBytes = 0; for (const buf of distinctBuffers) backingBytes += buf.byteLength; - // Pre-fix this was ~chunks.length * 256KB ≈ 25–250 MB. + // Pre-fix this was ~chunks.length * 256KB ≈ 2–250 MB. expect(backingBytes).toBeLessThan(4 * 1024 * 1024); }); const BYTES_TO_WRITE = 500_000; // https://github.com/oven-sh/bun/issues/12198 -test.skipIf(isWindows)( +// Windows: spawns `cat`. ASAN: quarantine + shadow memory push absolute RSS +// past any fixed budget; the relative-growth assertion is covered in release. +test.skipIf(isWindows || isASAN)( "Absolute memory usage remains relatively constant when reading and writing to a pipe", async () => { async function write(bytes: number) { @@ -107,9 +111,7 @@ test.skipIf(isWindows)( console.log(require("bun:jsc").heapStats()); console.log("RSS delta", ((after - before) | 0) / 1024 / 1024); console.log("RSS total", (after / 1024 / 1024) | 0, "MB"); - // ASAN's quarantine + shadow memory raise the absolute RSS floor and slow - // recycling of freed allocations; widen both bounds under bun-asan. - expect(after).toBeLessThan((isASAN ? 700 : 250) * 1024 * 1024); - expect(after).toBeLessThan(before * (isASAN ? 3 : 1.5)); + expect(after).toBeLessThan(250 * 1024 * 1024); + expect(after).toBeLessThan(before * 1.5); }, ); diff --git a/test/js/web/streams/streams.test.js b/test/js/web/streams/streams.test.js index 244d53c5454d..39e6546a8f19 100644 --- a/test/js/web/streams/streams.test.js +++ b/test/js/web/streams/streams.test.js @@ -1425,3 +1425,297 @@ it("ReadableStream BYOB read pending at cancel() resolves with undefined", async expect(value).toBeUndefined(); await reader.closed; }); + +// The transformer.cancel hook (https://streams.spec.whatwg.org/#dom-transformer-cancel, +// added in whatwg/streams#1283): invoked — instead of flush — when the +// readable side is canceled or the writable side is aborted. Semantics below +// verified against Node v24, which implements the same spec text. +describe("TransformStream transformer.cancel", () => { + test("reader.cancel() invokes cancel with the reason and skips flush", async () => { + const events = []; + const ts = new TransformStream({ + flush() { + events.push("flush"); + }, + cancel(reason) { + events.push(`cancel:${reason}`); + }, + }); + const writer = ts.writable.getWriter(); + await ts.readable.cancel("stop"); + expect(events).toEqual(["cancel:stop"]); + // The cancel reason propagates to the writable side. + expect( + await writer.closed.then( + () => null, + e => e, + ), + ).toBe("stop"); + }); + + test("writer.abort() invokes cancel with the reason and errors the readable", async () => { + const events = []; + const ts = new TransformStream({ + flush() { + events.push("flush"); + }, + cancel(reason) { + events.push(`cancel:${reason.message}`); + }, + }); + const reader = ts.readable.getReader(); + const pendingRead = reader.read().then( + () => null, + e => e, + ); + const boom = new Error("boom"); + await ts.writable.getWriter().abort(boom); + expect(events).toEqual(["cancel:boom"]); + expect(await pendingRead).toBe(boom); + }); + + test("cancel only runs once when both sides tear down", async () => { + const events = []; + const ts = new TransformStream({ + flush() { + events.push("flush"); + }, + cancel(reason) { + events.push(`cancel:${reason}`); + }, + }); + await ts.readable.cancel("first"); + await ts.writable.abort("second"); + expect(events).toEqual(["cancel:first"]); + }); + + test("cancel is not called on a normal close", async () => { + const events = []; + const ts = new TransformStream({ + transform(chunk, controller) { + controller.enqueue(chunk); + }, + flush() { + events.push("flush"); + }, + cancel() { + events.push("cancel"); + }, + }); + const writer = ts.writable.getWriter(); + const collected = (async () => { + const out = []; + for await (const chunk of ts.readable) out.push(chunk); + return out; + })(); + await writer.write("x"); + await writer.close(); + expect(await collected).toEqual(["x"]); + expect(events).toEqual(["flush"]); + }); + + test("the writable only errors after an async cancel settles", async () => { + const events = []; + const { promise: gate, resolve: release } = Promise.withResolvers(); + const ts = new TransformStream({ + async cancel() { + events.push("cancel-start"); + await gate; + events.push("cancel-end"); + }, + }); + const writer = ts.writable.getWriter(); + const closed = writer.closed.then( + () => events.push("closed-resolved"), + () => events.push("closed-rejected"), + ); + const cancelPromise = ts.readable.cancel("r").then(() => events.push("cancel()-resolved")); + await Bun.sleep(0); + events.push("release"); + release(); + await cancelPromise; + await closed; + expect(events).toEqual(["cancel-start", "release", "cancel-end", "closed-rejected", "cancel()-resolved"]); + }); + + test("a throwing cancel rejects readable.cancel() and errors the writable with that error", async () => { + const fail = new Error("cancel-fail"); + const ts = new TransformStream({ + cancel() { + throw fail; + }, + }); + const writer = ts.writable.getWriter(); + expect( + await ts.readable.cancel("x").then( + () => null, + e => e, + ), + ).toBe(fail); + expect( + await writer.closed.then( + () => null, + e => e, + ), + ).toBe(fail); + }); + + test("a flush rejecting after reader.cancel() closed the readable surfaces the flush error", async () => { + // transformStreamDefaultSinkCloseAlgorithm's rejection handler must reject + // with the flush error r, not readable.storedError — when reader.cancel() + // already closed the readable, storedError is undefined and would swallow + // the flush failure. Cancel from inside flush so flush is provably in + // flight when the readable closes. + const fail = new Error("flush-fail"); + let cancelResult; + const ts = new TransformStream({ + async flush() { + cancelResult = ts.readable.cancel("x").then( + () => null, + e => e, + ); + await Bun.sleep(0); + throw fail; + }, + }); + const writer = ts.writable.getWriter(); + await writer.ready; + const closeResult = writer.close().then( + () => null, + e => e, + ); + expect(await closeResult).toBe(fail); + // The cancel joins the in-flight close's finishPromise and so carries the + // same rejection. + expect(await cancelResult).toBe(fail); + }); + + test("a rejecting cancel rejects writer.abort() and errors the readable with that error", async () => { + const fail = new Error("cancel-fail"); + const ts = new TransformStream({ + cancel() { + return Promise.reject(fail); + }, + }); + const reader = ts.readable.getReader(); + expect( + await ts.writable + .getWriter() + .abort("reason") + .then( + () => null, + e => e, + ), + ).toBe(fail); + expect( + await reader.read().then( + () => null, + e => e, + ), + ).toBe(fail); + }); + + test("a non-function cancel member throws at construction", () => { + expect(() => new TransformStream({ cancel: 42 })).toThrow(TypeError); + }); + + test("reader.cancel() after terminate() with queued chunks settles cleanly", async () => { + // terminate() clears the transformer algorithms while the readable still + // holds the queued chunk and stays cancelable — and no hook may run for + // a transformer that is already torn down. (Node crashes here with an + // internal 'cancelAlgorithm is not a function' TypeError.) + const events = []; + const ts = new TransformStream( + { + transform(chunk, controller) { + controller.enqueue(chunk); + controller.terminate(); + }, + flush() { + events.push("flush"); + }, + cancel(reason) { + events.push(`cancel:${reason}`); + }, + }, + undefined, + { highWaterMark: 1 }, // let the write through with no reader attached + ); + const writer = ts.writable.getWriter(); + await writer.write("x"); + await ts.readable.cancel("stop"); + expect(events).toEqual([]); + }); + + test("a write racing reader.cancel() rejects with the cancel reason", async () => { + // The algorithms are already cleared while the cancel algorithm settles; + // a write slipping into that window must reject with the teardown + // outcome. (Node rejects it with an internal 'transformAlgorithm is not + // a function' TypeError here.) + const ts = new TransformStream({ + transform(chunk, controller) { + controller.enqueue(chunk); + }, + cancel() {}, + }); + const reader = ts.readable.getReader(); + const firstRead = reader.read().then( + r => `read:${r.done}`, + () => "read rejected", + ); + await Bun.sleep(0); // let the initial pull clear backpressure + const writer = ts.writable.getWriter(); + const cancelPromise = reader.cancel("stop"); // intentionally not awaited before the write + expect( + await writer.write("x").then( + () => null, + e => e, + ), + ).toBe("stop"); + await cancelPromise; + expect(await firstRead).toBe("read:true"); + }); + + test("abort during an in-flight failing transform settles every promise", async () => { + // The failing transform clears the algorithms while the abort's steps + // are still queued behind the in-flight write; nothing here may crash or + // hang. (The spec reference implementation crashes on this race; Node + // rejects the abort with an internal TypeError.) + const fail = new Error("transform-fail"); + const { promise: gate, resolve: release } = Promise.withResolvers(); + const events = []; + const ts = new TransformStream({ + async transform() { + events.push("transform"); + await gate; + throw fail; + }, + cancel() { + events.push("cancel"); + }, + }); + const reader = ts.readable.getReader(); + const pendingRead = reader.read().then( + () => null, + e => e, + ); + const writer = ts.writable.getWriter(); + const writeResult = writer.write("x").then( + () => null, + e => e, + ); + await Bun.sleep(0); + // Settling is what matters here; whether abort() resolves or rejects on + // an already-failed stream is not pinned. + const abortResult = writer.abort(new Error("abort-reason")).then( + () => "settled", + () => "settled", + ); + await Bun.sleep(0); + release(); + expect(await writeResult).toBe(fail); + expect(await pendingRead).toBe(fail); + expect(await abortResult).toBe("settled"); + expect(events).toEqual(["transform"]); + }); +});