CompressionStream/DecompressionStream: emit each chunk's output in bounded steps

packages/bun-types/globals.d.ts

@@ -93,7 +93,13 @@ declare var CompressionStream: Bun.__internal.UseLibDomIfAvailable<
   "CompressionStream",
   {
     prototype: CompressionStream;
-    new (format: Bun.CompressionFormat): CompressionStream;
+    /**
+     * @param strategy Bun extension. Its `highWaterMark` (bytes, default 64 KiB) bounds how much
+     * output one input chunk produces per step: the largest piece a reader receives per `read()`,
+     * and how far decoding runs ahead of a slow reader. A chunk larger than that may produce up to
+     * its own size per step.
+     */
+    new (format: Bun.CompressionFormat, strategy?: { highWaterMark?: number }): CompressionStream;
   }
 >;
 
@@ -102,7 +108,13 @@ declare var DecompressionStream: Bun.__internal.UseLibDomIfAvailable<
   "DecompressionStream",
   {
     prototype: DecompressionStream;
-    new (format: Bun.CompressionFormat): DecompressionStream;
+    /**
+     * @param strategy Bun extension. Its `highWaterMark` (bytes, default 64 KiB) bounds how much
+     * output one input chunk produces per step: the largest piece a reader receives per `read()`,
+     * and how far decoding runs ahead of a slow reader. A chunk larger than that may produce up to
+     * its own size per step.
+     */
+    new (format: Bun.CompressionFormat, strategy?: { highWaterMark?: number }): DecompressionStream;
   }
 >;
 

src/jsc/bindings/webcore/streams/BunStreamSource.cpp

@@ -1017,6 +1017,7 @@ static void rsisDetachNativeTransform(JSGlobalObject* globalObject, JSReadStream
         ts->m_nativeSinkReadyPromise.clear();
         resolvePromise(globalObject, ready, jsUndefined());
     }
+    nativeCodecAbandon(globalObject, ts);
     op->m_nativeTransform.clear();
 }
 
@@ -1542,6 +1543,9 @@ JSC_DEFINE_HOST_FUNCTION(jsWebStreamsHandler_boundReadStreamIntoSinkOnReady, (JS
             ts->m_nativeSinkReadyPromise.clear();
             Bun::WebStreams::resolvePromise(globalObject, ready, jsUndefined());
             scope.assertNoException();
+        } else if (ts->m_codecPromise) {
+            Bun::WebStreams::nativeCodecContinue(globalObject, ts);
+            RETURN_IF_EXCEPTION(scope, {});
         }
     }
     if (!op->m_waitingOnSink)

src/jsc/bindings/webcore/streams/JSCompressionStream.cpp

@@ -158,8 +158,10 @@ template<> JSC::EncodedJSValue JSC_HOST_CALL_ATTRIBUTES JSCompressionStreamConst
     auto format = parseCompressionFormat(lexicalGlobalObject, callFrame->argument(0));
     RETURN_IF_EXCEPTION(scope, {});
     ASSERT(format.has_value());
+    size_t highWaterMark = parseCodecHighWaterMark(lexicalGlobalObject, callFrame->argument(1));
+    RETURN_IF_EXCEPTION(scope, {});
 
-    void* coder = CompressionStreamCoder__create(static_cast<uint8_t>(*format), false);
+    void* coder = CompressionStreamCoder__create(static_cast<uint8_t>(*format), false, highWaterMark);
     if (!coder) [[unlikely]] {
         throwTypeError(lexicalGlobalObject, scope, "failed to initialize compressor"_s);
         return {};

src/jsc/bindings/webcore/streams/JSCompressionStreamShared.h

@@ -5,18 +5,25 @@
 #include "root.h"
 #include "StreamsForward.h"
 
-// CompressionStreamCoder.rs
-extern "C" void* CompressionStreamCoder__create(uint8_t format, bool decompress);
-// Releases the cell's reference (in-flight async transforms hold their own).
+// CompressionStreamCoder.rs. Each transform call runs one bounded step; `more` means the coder
+// must be stepped again (with no input, it kept the tail) before the next chunk is fed.
+extern "C" void* CompressionStreamCoder__create(uint8_t format, bool decompress, size_t highWaterMark);
+// Releases the cell's reference (in-flight off-thread steps hold their own).
 extern "C" void CompressionStreamCoder__destroy(void* coder);
-extern "C" JSC::EncodedJSValue CompressionStreamCoder__transform(void* coder, JSC::JSGlobalObject* global, const uint8_t* input, size_t input_len, bool finish);
-extern "C" JSC::EncodedJSValue CompressionStreamCoder__transformInto(void* coder, JSC::JSGlobalObject* global, const uint8_t* input, size_t input_len, bool finish, uint8_t sinkId, void* sinkPtr);
+extern "C" JSC::EncodedJSValue CompressionStreamCoder__transform(void* coder, JSC::JSGlobalObject* global, const uint8_t* input, size_t input_len, bool finish, bool* more);
+extern "C" JSC::EncodedJSValue CompressionStreamCoder__transformInto(void* coder, JSC::JSGlobalObject* global, const uint8_t* input, size_t input_len, bool finish, uint8_t sinkId, void* sinkPtr, bool* more);
+// Off-thread step, completed by Bun__CompressionStream__deliverAsync.
 extern "C" void CompressionStreamCoder__transformAsync(void* coder, JSC::JSGlobalObject* global, JSC::EncodedJSValue streamCell, JSC::EncodedJSValue chunk, const uint8_t* input, size_t inputLen, bool finish);
 
 namespace Bun {
 namespace WebStreams {
 
 std::optional<CompressionFormat> parseCompressionFormat(JSC::JSGlobalObject*, JSC::JSValue formatValue);
+// The optional second constructor argument, read like a queuing strategy: its highWaterMark (bytes,
+// default 64 KiB) is the output bound of one codec step, i.e. the largest piece a consumer gets per
+// read() and how far the coder runs ahead of a slow consumer. Throws RangeError / TypeError as
+// ExtractHighWaterMark does; `size` is ignored.
+size_t parseCodecHighWaterMark(JSC::JSGlobalObject*, JSC::JSValue strategy);
 
 } // namespace WebStreams
 } // namespace Bun

src/jsc/bindings/webcore/streams/JSDecompressionStream.cpp

@@ -158,8 +158,10 @@ template<> JSC::EncodedJSValue JSC_HOST_CALL_ATTRIBUTES JSDecompressionStreamCon
     auto format = parseCompressionFormat(lexicalGlobalObject, callFrame->argument(0));
     RETURN_IF_EXCEPTION(scope, {});
     ASSERT(format.has_value());
+    size_t highWaterMark = parseCodecHighWaterMark(lexicalGlobalObject, callFrame->argument(1));
+    RETURN_IF_EXCEPTION(scope, {});
 
-    void* coder = CompressionStreamCoder__create(static_cast<uint8_t>(*format), true);
+    void* coder = CompressionStreamCoder__create(static_cast<uint8_t>(*format), true, highWaterMark);
     if (!coder) [[unlikely]] {
         throwTypeError(lexicalGlobalObject, scope, "failed to initialize decompressor"_s);
         return {};

src/jsc/bindings/webcore/streams/JSTransformStream.cpp

@@ -284,7 +284,7 @@ void JSTransformStream::visitChildrenImpl(JSCell* cell, Visitor& visitor)
     visitor.appendHidden(thisObject->m_pendingWriteChunk);
     visitor.appendHidden(thisObject->m_nativeSinkCell);
     visitor.appendHidden(thisObject->m_nativeSinkReadyPromise);
-    visitor.appendHidden(thisObject->m_asyncCodecPromise);
+    visitor.appendHidden(thisObject->m_codecPromise);
 }
 
 void JSTransformStream::analyzeHeap(JSCell* cell, HeapAnalyzer& analyzer)
@@ -299,7 +299,7 @@ void JSTransformStream::analyzeHeap(JSCell* cell, HeapAnalyzer& analyzer)
     analyzeBarrierEdge(vm, analyzer, cell, thisObject->m_pendingWriteChunk, "pendingWriteChunk"_s);
     analyzeBarrierEdge(vm, analyzer, cell, thisObject->m_nativeSinkCell, "nativeSinkCell"_s);
     analyzeBarrierEdge(vm, analyzer, cell, thisObject->m_nativeSinkReadyPromise, "nativeSinkReadyPromise"_s);
-    analyzeBarrierEdge(vm, analyzer, cell, thisObject->m_asyncCodecPromise, "asyncCodecPromise"_s);
+    analyzeBarrierEdge(vm, analyzer, cell, thisObject->m_codecPromise, "codecPromise"_s);
 }
 
 // Prototype host functions

src/jsc/bindings/webcore/streams/JSTransformStream.h

@@ -59,9 +59,11 @@ class JSTransformStream : public JSC::JSNonFinalObject {
     // ClearAlgorithms defers the eager free to the arm's epilogue instead.
     bool m_nativeStateInUse : 1 { false };
     bool m_nativeStateReleasePending : 1 { false };
-    // An off-thread codec task holds the coder; ClearAlgorithms / runNativeArm must defer
-    // the free until the task's JS-thread completion clears this.
+    // An off-thread codec step holds the coder; ClearAlgorithms / runNativeArm must defer
+    // the free until the step's JS-thread completion clears this.
     bool m_asyncCodecInFlight : 1 { false };
+    // The chunk behind m_codecPromise runs its steps on the thread pool.
+    bool m_codecChunkOffThread : 1 { false };
 
     // Native byte-producing subclasses only: when `readStreamIntoSink` attaches a
     // native JSSink controller to this transform, the transform arms write coder
@@ -72,10 +74,11 @@ class JSTransformStream : public JSC::JSNonFinalObject {
     // sink backpressure; the sink's onReady resolves it.
     JSC::WriteBarrier<JSC::JSObject> m_nativeSinkCell;
     JSC::WriteBarrier<JSC::JSPromise> m_nativeSinkReadyPromise;
-    // Pending transform-algorithm promise for the off-thread codec step; the
-    // WorkTask's single `Strong` roots this cell and this barrier keeps the
-    // promise alive until deliverAsync settles it.
-    JSC::WriteBarrier<JSC::JSPromise> m_asyncCodecPromise;
+    // Compression/Decompression only: transform-algorithm promise of a chunk whose codec
+    // steps span turns (off-thread, or waiting for the consumer to take the output so far);
+    // the consumer drives it on (WebStreamsInternals.h: nativeCodecContinue / Abandon). While
+    // set, the coder holds that chunk's state and ClearAlgorithms defers the coder release.
+    JSC::WriteBarrier<JSC::JSPromise> m_codecPromise;
     void* m_nativeSinkPtr { nullptr };
     uint8_t m_nativeSinkId { 0 };
 

src/jsc/bindings/webcore/streams/JSTransformStreamDefaultController.cpp

@@ -401,20 +401,23 @@ void nativeTransformReleaseState(JSTransformStream* stream)
         TextDecoder__destroyForStream(std::exchange(s->m_decoder, nullptr));
 }
 
-// ClearAlgorithms is the one shared terminal (post-flush, error, cancel), but a
-// re-entrant reader.cancel() from user JS inside a native arm's chunk coercion reaches
-// it while the coder is still in use on the stack. Defer when m_nativeStateInUse; the
-// runNativeArm epilogue frees it once control unwinds.
+void nativeTransformReleaseStateIfIdle(JSTransformStream* stream)
+{
+    if (!stream->m_nativeStateReleasePending || stream->m_nativeStateInUse || stream->m_asyncCodecInFlight || stream->m_codecPromise)
+        return;
+    nativeTransformReleaseState(stream);
+}
+
+// ClearAlgorithms (post-flush, error, cancel) can reach a coder that is still busy: an arm on
+// the stack, an off-thread step, or a chunk parked across turns (the close algorithm clears
+// algorithms as soon as the flush arm returns). Whoever finishes that work releases it.
 static void nativeTransformReleaseStateOrDefer(JSTransformStreamDefaultController* controller)
 {
     auto* stream = dynamicDowncast<JSTransformStream>(controller->m_algorithmContext.get());
     if (!stream)
         return;
-    if (stream->m_nativeStateInUse || stream->m_asyncCodecInFlight) {
-        stream->m_nativeStateReleasePending = true;
-        return;
-    }
-    nativeTransformReleaseState(stream);
+    stream->m_nativeStateReleasePending = true;
+    nativeTransformReleaseStateIfIdle(stream);
 }
 
 void transformStreamDefaultControllerClearAlgorithms(JSTransformStreamDefaultController* controller)

src/jsc/bindings/webcore/streams/JSWritableStreamDefaultController.cpp

@@ -564,8 +564,10 @@ void writableStreamDefaultControllerError(JSGlobalObject* globalObject, JSWritab
     auto scope = DECLARE_THROW_SCOPE(vm);
     auto* stream = controller->m_stream.get();
     ASSERT(stream->m_state == WritableStreamState::Writable);
+    writableStreamStartErroring(globalObject, stream, error);
+    // After StartErroring, not before as in the spec: it reaches an in-flight codec chunk through the algorithms.
     writableStreamDefaultControllerClearAlgorithms(controller);
-    RELEASE_AND_RETURN(scope, writableStreamStartErroring(globalObject, stream, error));
+    RETURN_IF_EXCEPTION(scope, );
 }
 
 void writableStreamDefaultControllerErrorIfNeeded(JSGlobalObject* globalObject, JSWritableStreamDefaultController* controller, JSValue error)

src/jsc/bindings/webcore/streams/TransformStreamOperations.cpp

@@ -277,6 +277,9 @@ JSPromise* transformStreamDefaultSourceCancelAlgorithm(JSGlobalObject* globalObj
     auto& vm = getVM(globalObject);
     auto scope = DECLARE_THROW_SCOPE(vm);
     auto* controller = stream->m_controller.get();
+    // The readable is closed: a codec chunk (or, with a close in flight, flush) still being
+    // drained into it can no longer finish.
+    nativeCodecAbandon(globalObject, stream);
     if (auto* finishPromise = controller->m_finishPromise.get())
         return finishPromise;
     auto* finishPromise = JSPromise::create(vm, globalObject->promiseStructure());
@@ -294,9 +297,21 @@ JSPromise* transformStreamDefaultSourceCancelAlgorithm(JSGlobalObject* globalObj
 
 JSPromise* transformStreamDefaultSourcePullAlgorithm(JSGlobalObject* globalObject, JSTransformStream* stream)
 {
+    auto& vm = getVM(globalObject);
+    auto scope = DECLARE_THROW_SCOPE(vm);
     ASSERT(stream->m_backpressure);
     ASSERT(stream->m_backpressureChangePromise);
     transformStreamSetBackpressure(globalObject, stream, false);
+    scope.assertNoException();
+    if (stream->m_codecPromise) [[unlikely]] {
+        // The readable is asking for the next piece of a native codec chunk. Its enqueues may
+        // set [[backpressure]] again right here; a null result (pull done) keeps the next pull
+        // valid in that case, and otherwise the usual promise does.
+        nativeCodecContinue(globalObject, stream);
+        RETURN_IF_EXCEPTION(scope, nullptr);
+        if (stream->m_backpressure)
+            return nullptr;
+    }
     return stream->m_backpressureChangePromise.get();
 }
 

src/jsc/bindings/webcore/streams/WebStreamsInternals.h

@@ -454,7 +454,7 @@ JSC::JSPromise* transformStreamDefaultSinkWriteAlgorithm(JSC::JSGlobalObject*, J
 JSC::JSPromise* transformStreamDefaultSinkAbortAlgorithm(JSC::JSGlobalObject*, JSTransformStream*, JSC::JSValue reason); // userJS: yes — TransformStreamOperations.cpp
 JSC::JSPromise* transformStreamDefaultSinkCloseAlgorithm(JSC::JSGlobalObject*, JSTransformStream*); // userJS: yes (user flush) — TransformStreamOperations.cpp
 JSC::JSPromise* transformStreamDefaultSourceCancelAlgorithm(JSC::JSGlobalObject*, JSTransformStream*, JSC::JSValue reason); // userJS: yes — TransformStreamOperations.cpp
-JSC::JSPromise* transformStreamDefaultSourcePullAlgorithm(JSC::JSGlobalObject*, JSTransformStream*); // userJS: no — TransformStreamOperations.cpp
+JSC::JSPromise* transformStreamDefaultSourcePullAlgorithm(JSC::JSGlobalObject*, JSTransformStream*); // userJS: yes (steps a pending codec chunk, whose enqueue fulfills read requests) — TransformStreamOperations.cpp
 
 // JSTransformStreamDefaultController.cpp
 
@@ -463,6 +463,8 @@ void transformStreamDefaultControllerClearAlgorithms(JSTransformStreamDefaultCon
 // completion, errors the writable, then throws stream.[[readable]].[[storedError]]).
 void transformStreamDefaultControllerEnqueue(JSC::JSGlobalObject*, JSTransformStreamDefaultController*, JSC::JSValue chunk); // userJS: yes; throws — JSTransformStreamDefaultController.cpp
 void nativeTransformReleaseState(JSTransformStream*); // userJS: no — JSTransformStreamDefaultController.cpp
+// Performs a release ClearAlgorithms deferred, once nothing holds the native state any more.
+void nativeTransformReleaseStateIfIdle(JSTransformStream*); // userJS: no — JSTransformStreamDefaultController.cpp
 
 // Rust-side single dispatch for the native-transform → native-JSSink byte write, routed
 // through SinkHandle::write (src/runtime/webcore/Sink.rs). Returns a negative number for
@@ -482,8 +484,8 @@ JSC::JSPromise* runNativeArm(JSC::JSCell* context, Arm&& arm)
     stream->m_nativeStateInUse = true;
     JSC::JSPromise* result = arm(stream);
     stream->m_nativeStateInUse = false;
-    if (stream->m_nativeStateReleasePending && !stream->m_asyncCodecInFlight) [[unlikely]]
-        nativeTransformReleaseState(stream);
+    if (stream->m_nativeStateReleasePending) [[unlikely]]
+        nativeTransformReleaseStateIfIdle(stream);
     return result;
 }
 void transformStreamDefaultControllerError(JSC::JSGlobalObject*, JSTransformStreamDefaultController*, JSC::JSValue error); // userJS: yes — JSTransformStreamDefaultController.cpp
@@ -511,6 +513,12 @@ JSC::JSPromise* compressionStreamTransform(JSC::JSGlobalObject*, JSCompressionSt
 JSC::JSPromise* compressionStreamFlush(JSC::JSGlobalObject*, JSCompressionStream*, JSTransformStreamDefaultController*); // userJS: yes — JSCompressionStreamShared.cpp
 JSC::JSPromise* decompressionStreamTransform(JSC::JSGlobalObject*, JSDecompressionStream*, JSTransformStreamDefaultController*, JSC::JSValue chunk); // userJS: yes — JSCompressionStreamShared.cpp
 JSC::JSPromise* decompressionStreamFlush(JSC::JSGlobalObject*, JSDecompressionStream*, JSTransformStreamDefaultController*); // userJS: yes — JSCompressionStreamShared.cpp
+// A codec chunk whose output is still pending (stream->m_codecPromise set) is driven by its
+// consumer: the readable's pull algorithm / the native sink's onReady continue it; the writable
+// starting to error with the write in flight, a readable cancel, or a sink detach abandon it
+// (no-ops when nothing is pending).
+void nativeCodecContinue(JSC::JSGlobalObject*, JSTransformStream*); // userJS: yes (enqueues) — JSCompressionStreamShared.cpp
+void nativeCodecAbandon(JSC::JSGlobalObject*, JSTransformStream*); // userJS: no — JSCompressionStreamShared.cpp
 
 // CrossRealmTransform.cpp — transferable streams are NOT implemented. These signatures are
 // FROZEN, but the .cpp may be a stub whose entry points assert / throw; the per-class

src/jsc/bindings/webcore/streams/WritableStreamOperations.cpp

@@ -7,6 +7,7 @@
 #include "JSDOMGlobalObject.h"
 #include "JSDOMWrapperCache.h"
 #include "JSStreamsRuntime.h"
+#include "JSTransformStream.h"
 #include "JSWritableStream.h"
 #include "JSWritableStreamDefaultController.h"
 #include "JSWritableStreamDefaultWriter.h"
@@ -310,6 +311,11 @@ void writableStreamStartErroring(JSGlobalObject* globalObject, JSWritableStream*
         writableStreamDefaultWriterEnsureReadyPromiseRejected(globalObject, writer, reason);
         RETURN_IF_EXCEPTION(scope, );
     }
+    // The in-flight write may be a native codec chunk still being drained into the readable,
+    // which nothing on this side would ever finish; give it up so the erroring can complete.
+    // An in-flight close (a flush) is left to finish: a close in progress wins over the abort.
+    if (controller->m_algorithms.kind == SinkKind::Transform && stream->m_inFlightWriteRequest)
+        nativeCodecAbandon(globalObject, uncheckedDowncast<JSTransformStream>(controller->m_algorithms.algorithmContext.get()));
     if (!writableStreamHasOperationMarkedInFlight(stream) && controller->m_started)
         RELEASE_AND_RETURN(scope, writableStreamFinishErroring(globalObject, stream));
 }

src/jsc/rare_data.rs

@@ -270,6 +270,8 @@ pub struct RareData {
 
     /// `node:http2` PADDED DATA scratch; see [`Self::take_h2_padded_frame_buffer`].
     h2_padded_frame_buffer: Option<Box<H2PaddedFrameBuffer>>,
+    /// Output scratch for one JS-thread `CompressionStream` step; see [`Self::take_compression_scratch`].
+    compression_scratch: Option<Vec<u8>>,
 
     // There is intentionally no `aws_signature_cache` field — storage lives in
     // `bun_s3_signing::credentials::AWS_SIGNATURE_CACHE` (process static; it
@@ -330,6 +332,7 @@ impl Default for RareData {
             listening_sockets_for_watch_mode: Mutex::new(Vec::new()),
             temp_pipe_read_buffer: None,
             h2_padded_frame_buffer: None,
+            compression_scratch: None,
             s3_default_client: Strong::empty(),
             node_quic_callbacks: Strong::empty(),
             default_csrf_secret: Box::default(),
@@ -682,6 +685,20 @@ impl RareData {
         self.h2_padded_frame_buffer.get_or_insert(buffer);
     }
 
+    /// Empty `Vec` with whatever capacity the last step left behind.
+    pub fn take_compression_scratch(&mut self) -> Vec<u8> {
+        self.compression_scratch.take().unwrap_or_default()
+    }
+
+    /// Hand a taken buffer back; the slot keeps the first one returned and lets an oversized one go.
+    pub fn put_back_compression_scratch(&mut self, mut buffer: Vec<u8>) {
+        const KEEP: usize = 256 * 1024;
+        if self.compression_scratch.is_none() && buffer.capacity() <= KEEP {
+            buffer.clear();
+            self.compression_scratch = Some(buffer);
+        }
+    }
+
     pub fn boring_engine(&mut self) -> *mut boring::ENGINE {
         // The raw `ENGINE_new()` result is cached without a null check:
         // `EVP_DigestInit_ex` tolerates a NULL engine, so OOM here degrades to