webstreams: propagate exceptions through helpers, convert only at boundaries

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

@@ -136,31 +136,29 @@ struct CodecStepResult {
     // The coder stopped at its output bound; the chunk needs another step.
     bool more { false };
     bool sinkBackpressure { false };
-    // Codec error or failed delivery; empty on success (and on VM termination).
-    JSValue thrown;
 };
 
 // One step on this thread, delivered straight into the native JSSink when one is attached
-// (no JSUint8Array), otherwise enqueued. `input` is only passed on a chunk's first step.
+// (no JSUint8Array), otherwise enqueued. `input` is only passed on a chunk's first step. A codec
+// error or failed delivery throws.
 static CodecStepResult runStepHere(JSGlobalObject* globalObject, JSTransformStream* stream, void* coder, const uint8_t* input, size_t inputLen, bool finish)
 {
     auto& vm = getVM(globalObject);
-    auto catchScope = DECLARE_TOP_EXCEPTION_SCOPE(vm);
+    auto scope = DECLARE_THROW_SCOPE(vm);
     CodecStepResult step;
     if (void* sinkPtr = stream->m_nativeSinkPtr) {
         JSValue wrote = JSValue::decode(CompressionStreamCoder__transformInto(coder, globalObject, input, inputLen, finish, stream->m_nativeSinkId, sinkPtr, &step.more));
-        if (!catchScope.exception())
-            step.sinkBackpressure = nativeSinkWriteIsBackpressure(vm, wrote);
-    } else {
-        JSValue out = JSValue::decode(CompressionStreamCoder__transform(coder, globalObject, input, inputLen, finish, &step.more));
-        if (!catchScope.exception()) {
-            auto* view = dynamicDowncast<JSArrayBufferView>(out);
-            if (view && view->length())
-                transformStreamDefaultControllerEnqueue(globalObject, stream->m_controller.get(), out);
-        }
+        RETURN_IF_EXCEPTION(scope, step);
+        step.sinkBackpressure = nativeSinkWriteIsBackpressure(vm, wrote);
+        return step;
+    }
+    JSValue out = JSValue::decode(CompressionStreamCoder__transform(coder, globalObject, input, inputLen, finish, &step.more));
+    RETURN_IF_EXCEPTION(scope, step);
+    auto* view = dynamicDowncast<JSArrayBufferView>(out);
+    if (view && view->length()) {
+        transformStreamDefaultControllerEnqueue(globalObject, stream->m_controller.get(), out);
+        RETURN_IF_EXCEPTION(scope, step);
     }
-    if (catchScope.exception()) [[unlikely]]
-        step.thrown = takeAbruptCompletion(globalObject, catchScope);
     return step;
 }
 
@@ -169,7 +167,6 @@ enum class CodecOutcome : uint8_t {
     // Done, but its last write left the sink full: the transform promise becomes
     // m_nativeSinkReadyPromise and settles when the sink drains, as any backpressured write.
     DoneSinkFull,
-    Failed,
     // Nothing to settle: output is still pending and the consumer is full (it will continue
     // the chunk), or a terminal reached from inside a delivery already abandoned it.
     Pending,
@@ -180,24 +177,20 @@ static bool consumerFull(JSTransformStream* stream, const CodecStepResult& step)
     return stream->m_nativeSinkPtr ? step.sinkBackpressure : stream->m_backpressure;
 }
 
-// Steps on this thread until the chunk completes, fails, or its consumer is full. `input` is
-// only fed on the first step. The caller holds m_nativeStateInUse, so a terminal reached from
+// Steps on this thread until the chunk completes, fails (throws), or its consumer is full. `input`
+// is only fed on the first step. The caller holds m_nativeStateInUse, so a terminal reached from
 // inside a delivery defers the coder release instead of freeing it under the loop.
-static CodecOutcome stepChunkHere(JSGlobalObject* globalObject, JSTransformStream* stream, void* coder, const uint8_t* input, size_t inputLen, bool finish, JSValue& thrown)
+static CodecOutcome stepChunkHere(JSGlobalObject* globalObject, JSTransformStream* stream, void* coder, const uint8_t* input, size_t inputLen, bool finish)
 {
     auto& vm = getVM(globalObject);
     auto scope = DECLARE_THROW_SCOPE(vm);
     ASSERT(stream->m_nativeStateInUse);
     bool continuation = !!stream->m_codecPromise;
     for (;;) {
         CodecStepResult step = runStepHere(globalObject, stream, coder, input, inputLen, finish);
-        RETURN_IF_EXCEPTION(scope, CodecOutcome::Failed);
+        RETURN_IF_EXCEPTION(scope, CodecOutcome::Pending);
         if (continuation && !stream->m_codecPromise)
             return CodecOutcome::Pending;
-        if (!step.thrown.isEmpty()) {
-            thrown = step.thrown;
-            return CodecOutcome::Failed;
-        }
         bool full = consumerFull(stream, step);
         if (!step.more)
             return full && stream->m_nativeSinkPtr ? CodecOutcome::DoneSinkFull : CodecOutcome::Done;
@@ -238,16 +231,13 @@ static void settleCodecChunk(JSGlobalObject* globalObject, JSTransformStream* st
 }
 
 // Applies the outcome of a step that ran from a later turn to the pending chunk.
-static void settlePendingChunk(JSGlobalObject* globalObject, JSTransformStream* stream, CodecOutcome outcome, JSValue thrown)
+static void settlePendingChunk(JSGlobalObject* globalObject, JSTransformStream* stream, CodecOutcome outcome)
 {
     auto& vm = getVM(globalObject);
     switch (outcome) {
     case CodecOutcome::Done:
         settleCodecChunk(globalObject, stream, JSValue());
         return;
-    case CodecOutcome::Failed:
-        settleCodecChunk(globalObject, stream, thrown);
-        return;
     case CodecOutcome::DoneSinkFull:
         if (auto* promise = takeCodecPromise(stream))
             stream->m_nativeSinkReadyPromise.set(vm, stream, promise);
@@ -257,8 +247,8 @@ static void settlePendingChunk(JSGlobalObject* globalObject, JSTransformStream*
     }
 }
 
-// The transform / flush arm (under runNativeArm). Abrupt completions become a rejected
-// promise: an arm must never throw synchronously into ProcessWrite/ProcessClose.
+// The transform / flush arm (under runNativeArm). It returns a promise, so a failed step is its
+// rejection: an arm must never throw synchronously into ProcessWrite/ProcessClose.
 static JSPromise* transformChunk(JSGlobalObject* globalObject, JSTransformStream* stream, void* coder, JSValue chunk, const uint8_t* input, size_t inputLen, bool finish)
 {
     auto& vm = getVM(globalObject);
@@ -275,14 +265,12 @@ static JSPromise* transformChunk(JSGlobalObject* globalObject, JSTransformStream
         return promise;
     }
 
-    JSValue thrown;
-    CodecOutcome outcome = stepChunkHere(globalObject, stream, coder, input, inputLen, finish, thrown);
-    RETURN_IF_EXCEPTION(scope, nullptr);
+    CodecOutcome outcome = stepChunkHere(globalObject, stream, coder, input, inputLen, finish);
+    if (scope.exception()) [[unlikely]]
+        RELEASE_AND_RETURN(scope, promiseRejectedWithPendingException(globalObject, scope));
     switch (outcome) {
     case CodecOutcome::Done:
         RELEASE_AND_RETURN(scope, promiseFulfilledWith(globalObject, jsUndefined()));
-    case CodecOutcome::Failed:
-        RELEASE_AND_RETURN(scope, promiseRejectedWith(globalObject, thrown));
     case CodecOutcome::DoneSinkFull: {
         auto* ready = JSPromise::create(vm, globalObject->promiseStructure());
         stream->m_nativeSinkReadyPromise.set(vm, stream, ready);
@@ -312,12 +300,19 @@ void nativeCodecContinue(JSGlobalObject* globalObject, JSTransformStream* stream
             return;
         RELEASE_AND_RETURN(scope, dispatchStepOffThread(globalObject, stream, coder, jsUndefined(), nullptr, 0, false));
     }
-    JSValue thrown;
     stream->m_nativeStateInUse = true;
-    CodecOutcome outcome = stepChunkHere(globalObject, stream, coder, nullptr, 0, false, thrown);
+    CodecOutcome outcome = stepChunkHere(globalObject, stream, coder, nullptr, 0, false);
     stream->m_nativeStateInUse = false;
-    RETURN_IF_EXCEPTION(scope, void());
-    settlePendingChunk(globalObject, stream, outcome, thrown);
+    if (scope.exception()) [[unlikely]] {
+        // BOUNDARY: this continues the pending chunk on its behalf, so a failed step is that
+        // chunk's rejection. If a terminal reached inside the step already abandoned the chunk,
+        // that terminal carries the error and there is nothing left to settle.
+        JSValue thrown = takeException(scope);
+        RETURN_IF_EXCEPTION(scope, );
+        if (stream->m_codecPromise)
+            settleCodecChunk(globalObject, stream, thrown);
+    } else
+        settlePendingChunk(globalObject, stream, outcome);
     // An abandon from inside the loop deferred its release to here.
     RELEASE_AND_RETURN(scope, nativeTransformReleaseStateIfIdle(stream));
 }
@@ -336,18 +331,10 @@ static JSPromise* compressionStreamTransformImpl(JSGlobalObject* globalObject, J
 {
     auto& vm = getVM(globalObject);
     auto scope = DECLARE_THROW_SCOPE(vm);
-    JSValue thrown;
     WTF::CString scratch;
-    std::optional<std::span<const uint8_t>> bytes;
-    {
-        auto catchScope = DECLARE_TOP_EXCEPTION_SCOPE(vm);
-        bytes = bufferSourceBytes(globalObject, chunk, scratch);
-        if (catchScope.exception()) [[unlikely]]
-            thrown = takeAbruptCompletion(globalObject, catchScope);
-    }
-    RETURN_IF_EXCEPTION(scope, nullptr);
-    if (!thrown.isEmpty())
-        RELEASE_AND_RETURN(scope, promiseRejectedWith(globalObject, thrown));
+    std::optional<std::span<const uint8_t>> bytes = bufferSourceBytes(globalObject, chunk, scratch);
+    if (scope.exception()) [[unlikely]]
+        RELEASE_AND_RETURN(scope, promiseRejectedWithPendingException(globalObject, scope));
     if (!bytes) [[unlikely]]
         return nullptr;
     RELEASE_AND_RETURN(scope, transformChunk(globalObject, stream, stream->m_coder, chunk, bytes->data(), bytes->size(), false));
@@ -373,9 +360,36 @@ JSPromise* decompressionStreamFlush(JSGlobalObject* globalObject, JSDecompressio
     return transformChunk(globalObject, stream, stream->m_coder, jsUndefined(), nullptr, 0, true);
 }
 
+// Hands an off-thread step's output to the consumer (native sink or readable). Throws on a failed
+// delivery.
+static CodecStepResult deliverAsyncOutput(JSGlobalObject* globalObject, JSTransformStream* stream, const uint8_t* out, size_t outLen, bool more)
+{
+    auto& vm = getVM(globalObject);
+    auto scope = DECLARE_THROW_SCOPE(vm);
+    CodecStepResult step;
+    step.more = more;
+    if (void* sinkPtr = stream->m_nativeSinkPtr) {
+        JSValue wrote = JSValue::decode(Bun__NativeTransformSink__writeBytes(stream->m_nativeSinkId, sinkPtr, globalObject, out, outLen));
+        RETURN_IF_EXCEPTION(scope, step);
+        step.sinkBackpressure = nativeSinkWriteIsBackpressure(vm, wrote);
+        return step;
+    }
+    auto copied = JSC::ArrayBuffer::tryCreate(std::span<const uint8_t>(out, outLen));
+    if (!copied) [[unlikely]] {
+        throwOutOfMemoryError(globalObject, scope);
+        return step;
+    }
+    auto* view = JSC::JSUint8Array::create(globalObject, globalObject->typedArrayStructure(JSC::TypeUint8, false), WTF::move(copied), 0, outLen);
+    RETURN_IF_EXCEPTION(scope, step);
+    transformStreamDefaultControllerEnqueue(globalObject, stream->m_controller.get(), JSValue(view));
+    RETURN_IF_EXCEPTION(scope, step);
+    return step;
+}
+
 // JS-thread completion of an off-thread step. `out` borrows the coder's buffer, so it is
 // consumed BEFORE m_asyncCodecInFlight is cleared and the coder may be released or
-// re-dispatched. Returns into Rust, so this is an exception boundary rather than a throw scope.
+// re-dispatched. BOUNDARY: entered from the event loop with no JS above it; a failed delivery is
+// the pending chunk's rejection.
 extern "C" void Bun__CompressionStream__deliverAsync(JSC::JSGlobalObject* globalObject, JSC::EncodedJSValue streamCell, const uint8_t* out, size_t outLen, bool more, JSC::EncodedJSValue error)
 {
     auto& vm = getVM(globalObject);
@@ -388,27 +402,14 @@ extern "C" void Bun__CompressionStream__deliverAsync(JSC::JSGlobalObject* global
 
     CodecStepResult step;
     step.more = more;
-    bool abandoned = !stream->m_codecPromise;
+    JSValue thrown;
     if (error) {
-        step.thrown = JSValue::decode(error);
-    } else if (outLen && !abandoned) {
-        if (void* sinkPtr = stream->m_nativeSinkPtr) {
-            JSValue wrote = JSValue::decode(Bun__NativeTransformSink__writeBytes(stream->m_nativeSinkId, sinkPtr, globalObject, out, outLen));
-            if (!scope.exception())
-                step.sinkBackpressure = nativeSinkWriteIsBackpressure(vm, wrote);
-        } else {
-            auto copied = JSC::ArrayBuffer::tryCreate(std::span<const uint8_t>(out, outLen));
-            if (!copied) [[unlikely]] {
-                step.thrown = JSC::createOutOfMemoryError(globalObject);
-            } else {
-                auto* view = JSC::JSUint8Array::create(globalObject, globalObject->typedArrayStructure(JSC::TypeUint8, false), WTF::move(copied), 0, outLen);
-                if (!scope.exception())
-                    transformStreamDefaultControllerEnqueue(globalObject, stream->m_controller.get(), JSValue(view));
-            }
-        }
+        thrown = JSValue::decode(error);
+    } else if (outLen && stream->m_codecPromise) {
+        step = deliverAsyncOutput(globalObject, stream, out, outLen, more);
         if (scope.exception()) [[unlikely]] {
-            step.thrown = takeAbruptCompletion(globalObject, scope);
-            if (step.thrown.isEmpty()) {
+            thrown = takeException(scope);
+            if (!thrown) {
                 // VM termination: the chunk stays pending; teardown's finalizer releases the coder.
                 Bun__VM__takeTerminationOutsideScript(globalObject);
                 stream->m_asyncCodecInFlight = false;
@@ -423,10 +424,10 @@ extern "C" void Bun__CompressionStream__deliverAsync(JSC::JSGlobalObject* global
         nativeTransformReleaseStateIfIdle(stream);
         return;
     }
-    if (!step.thrown.isEmpty())
-        settleCodecChunk(globalObject, stream, step.thrown);
+    if (thrown)
+        settleCodecChunk(globalObject, stream, thrown);
     else if (!step.more)
-        settlePendingChunk(globalObject, stream, consumerFull(stream, step) && stream->m_nativeSinkPtr ? CodecOutcome::DoneSinkFull : CodecOutcome::Done, JSValue());
+        settlePendingChunk(globalObject, stream, consumerFull(stream, step) && stream->m_nativeSinkPtr ? CodecOutcome::DoneSinkFull : CodecOutcome::Done);
     else if (!consumerFull(stream, step)) {
         if (void* coder = coderOf(stream)) [[likely]]
             dispatchStepOffThread(globalObject, stream, coder, jsUndefined(), nullptr, 0, false);