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

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

@@ -5,12 +5,17 @@
 #include "root.h"
 #include "StreamsForward.h"
 
-// CompressionStreamCoder.rs
+// CompressionStreamCoder.rs. A chunk (or the flush) is transformed in steps of bounded
+// output: each call below runs one step and reports through `more` whether the coder
+// stopped at its cap, in which case it must be stepped again (with no input; it keeps the
+// chunk's unconsumed tail) before the next chunk is fed.
 extern "C" void* CompressionStreamCoder__create(uint8_t format, bool decompress);
-// Releases the cell's reference (in-flight async transforms hold their own).
+// 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; completes through Bun__CompressionStream__deliverAsync. A continuation
+// step passes an undefined chunk and no input.
 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 {

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

@@ -147,6 +147,12 @@ namespace WebCore {
 #define FOR_EACH_WEB_STREAMS_REACTION_HANDLER_TS_CONTROLLER(V) \
     V(onTSPerformTransformRejected)
 
+// owner: JSCompressionStreamShared.cpp. context = the JSCompressionStream / JSDecompressionStream
+// (a JSTransformStream) whose codec chunk parked mid-way; registered on the readable side's
+// backpressureChangePromise or on the native sink's ready promise.
+#define FOR_EACH_WEB_STREAMS_REACTION_HANDLER_CODEC(V) \
+    V(onCodecChunkResume)
+
 // owner: CrossRealmTransform.cpp (transferable streams are not implemented; the handler may
 // assert-not-reached). context = the JSCrossRealmTransformState.
 #define FOR_EACH_WEB_STREAMS_REACTION_HANDLER_CROSS_REALM(V) \
@@ -227,6 +233,7 @@ namespace WebCore {
     FOR_EACH_WEB_STREAMS_REACTION_HANDLER_WS_CONTROLLER(V)         \
     FOR_EACH_WEB_STREAMS_REACTION_HANDLER_TS_OPERATIONS(V)         \
     FOR_EACH_WEB_STREAMS_REACTION_HANDLER_TS_CONTROLLER(V)         \
+    FOR_EACH_WEB_STREAMS_REACTION_HANDLER_CODEC(V)                 \
     FOR_EACH_WEB_STREAMS_REACTION_HANDLER_CROSS_REALM(V)           \
     FOR_EACH_WEB_STREAMS_REACTION_HANDLER_BUN_SOURCE(V)            \
     FOR_EACH_WEB_STREAMS_REACTION_HANDLER_DIRECT_CONTROLLER(V)     \

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,23 +59,29 @@ 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 };
+    // Compression/Decompression only: the chunk behind m_codecPromise started on the thread
+    // pool, so its remaining steps are dispatched there too (never run on this thread).
+    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
     // output straight to `m_nativeSinkPtr` via the Rust SinkHandle dispatcher
     // (Bun__NativeTransformSink__writeBytes) instead of wrapping it in a
     // JSUint8Array and enqueueing on the readable.
-    // `m_nativeSinkReadyPromise` is the transform-algorithm result returned on
-    // sink backpressure; the sink's onReady resolves it.
+    // `m_nativeSinkReadyPromise` is whatever promise the arm parked on sink
+    // backpressure (the transform-algorithm result itself, or a codec chunk's
+    // resume gate); the sink's onReady (or detaching from the sink) 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: the transform-algorithm promise of a chunk (or
+    // flush) whose codec steps span turns, because a step ran off-thread or stopped at
+    // the coder's output cap and is parked until the consumer has room. Set means the
+    // coder still holds that chunk's state, so ClearAlgorithms defers the coder release
+    // to the chunk's terminal (JSCompressionStreamShared.cpp).
+    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,26 @@ 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 is the one shared terminal (post-flush, error, cancel), but it can reach
+// a coder that is still busy: a re-entrant reader.cancel() from user JS inside a native arm
+// (m_nativeStateInUse), an off-thread codec step (m_asyncCodecInFlight), or a codec chunk
+// whose steps are still pending across turns (m_codecPromise; the close algorithm clears
+// algorithms as soon as the flush arm returns, while a large flush may still be parked
+// mid-way). Defer; whoever finishes that work runs nativeTransformReleaseStateIfIdle.
 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/WebStreamsInternals.h

@@ -463,6 +463,10 @@ 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 the release ClearAlgorithms deferred (m_nativeStateReleasePending), provided nothing
+// holds the native state any more: no arm on the stack, no off-thread codec step, no codec
+// chunk pending across turns. No-op otherwise.
+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 +486,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