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

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

@@ -5,12 +5,14 @@
 #include "root.h"
 #include "StreamsForward.h"
 
-// CompressionStreamCoder.rs
+// 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);
-// 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, 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 {

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

@@ -147,6 +147,11 @@ namespace WebCore {
 #define FOR_EACH_WEB_STREAMS_REACTION_HANDLER_TS_CONTROLLER(V) \
     V(onTSPerformTransformRejected)
 
+// owner: JSCompressionStreamShared.cpp. context = the JSCompressionStream / JSDecompressionStream
+// whose codec chunk is parked.
+#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 +232,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,25 @@ 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
     // 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 the promise an arm parked on sink backpressure;
+    // 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: transform-algorithm promise of a chunk whose codec
+    // steps span turns (off-thread, or parked at the output cap). While set, the coder holds
+    // that chunk's state and ClearAlgorithms defers the coder release to the chunk's terminal.
+    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/TransformStreamOperations.cpp

@@ -277,8 +277,15 @@ JSPromise* transformStreamDefaultSourceCancelAlgorithm(JSGlobalObject* globalObj
     auto& vm = getVM(globalObject);
     auto scope = DECLARE_THROW_SCOPE(vm);
     auto* controller = stream->m_controller.get();
-    if (auto* finishPromise = controller->m_finishPromise.get())
+    if (auto* finishPromise = controller->m_finishPromise.get()) {
+        // The spec skips the reaction below (and its unblock-write step) while a close is in
+        // flight; a native codec flush parked on [[backpressureChangePromise]] still needs it,
+        // as the readable this cancel just closed will never pull again. Nothing else waits on
+        // that promise once a close has started.
+        transformStreamUnblockWrite(globalObject, stream);
+        RETURN_IF_EXCEPTION(scope, nullptr);
         return finishPromise;
+    }
     auto* finishPromise = JSPromise::create(vm, globalObject->promiseStructure());
     controller->m_finishPromise.set(vm, controller, finishPromise);
 

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

@@ -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