streams: settle a native sink's owner when a direct stream closes, not when pull() returns

src/codegen/generate-jssink.ts

@@ -232,8 +232,8 @@ public:
     SinkID m_sinkId;
     SinkSource m_sourceKind { SinkSource::None };
     mutable WriteBarrier<JSC::JSCell> m_source;
-    mutable WriteBarrier<JSC::JSPromise> m_closePromise; // DirectStream: readDirectStream's result while pull() is sync and open
-    mutable WriteBarrier<JSC::Unknown> m_failReason; // close(error)'s error, so the owner's promise rejects even though pull() itself resolved
+    mutable WriteBarrier<JSC::JSPromise> m_closePromise; // DirectStream: readDirectStream's result, taken by whatever closes the controller
+    mutable WriteBarrier<JSC::Unknown> m_failReason; // close(error)'s error, so the owner's promise rejects with it even when detach() or readDirectStream only see it afterwards
     // Strong, and cleared as soon as the pipe is over, so a controller the user still holds does not keep the stream alive.
     mutable WriteBarrier<JSC::JSObject> m_readableStream;
     // While a native sink pipes a stream in, it roots only this cell; this and m_readableStream hold the rest (streams.rs PipeCell).

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

@@ -679,8 +679,9 @@ static void directStreamOnClose(JSC::VM& vm, JSGlobalObject* globalObject, WebCo
         }
     }
     if (closePromise) {
+        // Only the sink's owner sees this promise, and it can close the sink before it subscribes: never an unhandled rejection.
         if (reason.toBoolean(globalObject))
-            rejectPromise(globalObject, closePromise, reason);
+            closePromise->rejectAsHandled(vm, reason);
         else
             resolvePromise(globalObject, closePromise, jsUndefined());
         RETURN_IF_EXCEPTION(scope, );
@@ -755,13 +756,10 @@ JSValue readDirectStream(JSGlobalObject* globalObject, JSReadableStream* stream,
     JSValue maybePromise = call(globalObject, pull, getCallData(pull), source->thisValue(), pullArgs);
     RETURN_IF_EXCEPTION(scope, {});
 
-    // Resolving without close()/end() ends the sink controller first; a sync return waits for close()/end().
-    if (auto* pullPromise = dynamicDowncast<JSPromise>(maybePromise)) {
-        auto* result = JSPromise::create(vm, globalObject->promiseStructure());
-        pullPromise->performPromiseThenWithContext(vm, globalObject, runtime->onReadDirectStreamPullFulfilled(), runtime->onReadDirectStreamPullRejected(), result, sinkController);
-        return result;
-    }
-    // A sync pull() that called close(error): the owner hears it the way it hears an async one.
+    // The owner hears the controller close, not pull() return: a pull() reaction only closes a controller that an async pull() left open.
+    if (auto* pullPromise = dynamicDowncast<JSPromise>(maybePromise))
+        pullPromise->performPromiseThenWithContext(vm, globalObject, runtime->onReadDirectStreamPullFulfilled(), runtime->onReadDirectStreamPullRejected(), jsUndefined(), sinkController);
+    // pull() already called close(error).
     if (JSValue failed = sinkController->m_failReason.get()) {
         auto* rejected = promiseRejectedWith(globalObject, failed);
         RETURN_IF_EXCEPTION(scope, {});
@@ -1354,37 +1352,50 @@ extern "C" void Bun__NativeStreamSourceAdapter__onClose(JSGlobalObject* globalOb
     Bun::WebStreams::queueStreamsMicrotask(globalObject, WebCore::JSStreamsRuntime::from(globalObject)->onNativeSourceHandleClosedMicrotask(), jsUndefined(), JSValue::decode(adapter));
 }
 
+// A pull() reaction that closes the controller settles the owner's promise in place of that close: with what end() throws, or pull()'s own reason. Null: it closed earlier.
+static JSPromise* takeClosePromise(WebCore::JSReadableSinkControllerBase* sinkController)
+{
+    auto* closePromise = sinkController->m_closePromise.get();
+    sinkController->m_closePromise.clear();
+    return closePromise;
+}
+
 // [reaction-convention] pull() runs once for a native sink: its promise resolving without close()/end() ends the sink (a no-op once detached).
 JSC_DEFINE_HOST_FUNCTION(jsWebStreamsHandler_onReadDirectStreamPullFulfilled, (JSGlobalObject * globalObject, CallFrame* callFrame))
 {
-    auto scope = DECLARE_THROW_SCOPE(getVM(globalObject));
+    auto& vm = getVM(globalObject);
     auto* sinkController = uncheckedDowncast<WebCore::JSReadableSinkControllerBase>(callFrame->argument(1));
-    // pull() called close(error) before resolving: the owner's promise rejects with that error.
-    if (JSValue failed = sinkController->m_failReason.get()) {
-        throwException(globalObject, scope, failed);
-        return {};
-    }
-    sinkController->end(globalObject);
-    RETURN_IF_EXCEPTION(scope, {});
-    return JSValue::encode(jsUndefined());
+    auto* closePromise = takeClosePromise(sinkController);
+    JSC::EnsureStillAliveScope keepAlive(closePromise);
+    return enterStreams(globalObject, [&] {
+        auto scope = DECLARE_THROW_SCOPE(vm);
+        sinkController->end(globalObject);
+        RETURN_IF_EXCEPTION(scope, );
+        if (closePromise)
+            resolvePromise(globalObject, closePromise, jsUndefined()); }, [&](JSValue error) {
+        if (closePromise)
+            closePromise->rejectAsHandled(vm, error); });
 }
 
 // ... and its promise rejecting is close(reason): the source failed, so fail the sink from this side (no cancel()) and pass the rejection on to the sink's owner.
 JSC_DEFINE_HOST_FUNCTION(jsWebStreamsHandler_onReadDirectStreamPullRejected, (JSGlobalObject * globalObject, CallFrame* callFrame))
 {
-    auto scope = DECLARE_THROW_SCOPE(getVM(globalObject));
+    auto& vm = getVM(globalObject);
     JSValue reason = callFrame->argument(0);
     auto* sinkController = uncheckedDowncast<WebCore::JSReadableSinkControllerBase>(callFrame->argument(1));
-    if (!sinkController->wrapped()) {
-        // close(error) already failed the sink: the owner hears that error. A clean close()/end() or a peer abort already settled it: nothing left to fail.
-        if (JSValue failed = sinkController->m_failReason.get())
-            throwException(globalObject, scope, failed);
-        return scope.exception() ? EncodedJSValue() : JSValue::encode(jsUndefined());
-    }
-    sinkController->close(globalObject, reason.toBoolean(globalObject) ? reason : JSValue(createError(globalObject, "pull() rejected"_s)));
-    RETURN_IF_EXCEPTION(scope, {});
-    throwException(globalObject, scope, reason);
-    return {};
+    // close()/end(), close(error) or a peer abort closed it first and the owner heard that: nothing left to fail.
+    if (!sinkController->wrapped())
+        return JSValue::encode(jsUndefined());
+    auto* closePromise = takeClosePromise(sinkController);
+    JSC::EnsureStillAliveScope keepAlive(closePromise);
+    return enterStreams(globalObject, [&] {
+        auto scope = DECLARE_THROW_SCOPE(vm);
+        sinkController->close(globalObject, reason.toBoolean(globalObject) ? reason : JSValue(createError(globalObject, "pull() rejected"_s)));
+        RETURN_IF_EXCEPTION(scope, );
+        if (closePromise)
+            closePromise->rejectAsHandled(vm, reason); }, [&](JSValue error) {
+        if (closePromise)
+            closePromise->rejectAsHandled(vm, error); });
 }
 
 } // namespace WebCore

src/runtime/server/RequestContext.rs

@@ -958,7 +958,7 @@ where
         ctx_log!("deinit<d> ({:p})<r>", self);
         debug_assert!(self.flags.has_finalized());
 
-        // A response body stream suspended inside its `pull()` never settles the promise
+        // A client abort while the body stream is in flight reclaims the claim of the promise
         // whose reactions consume the sink (`handleResolveStream` / `handleRejectStream`),
         // so a client abort in that state reaches deinit with the sink still owned here.
         // This is the owner's last exit: release it exactly like the settle paths do.
@@ -2913,10 +2913,7 @@ where
         stream_log!("onResolve({})", wrote_anything);
         // HTTP/1 only: the sink already fully ended the response, so `resp`
         // can no longer be dereferenced (see `end_already_responded_stream`).
-        // This resolution can run arbitrarily later than the end: e.g. a
-        // direct stream whose `pull()` calls `controller.end()` and then
-        // awaits a promise the user only settles after the client has
-        // disconnected. H2/H3 keep the end_stream() path: their `resp` is still
+        // H2/H3 keep the end_stream() path: their `resp` is still
         // alive here and its still-armed onAborted must be disarmed.
         if !MUX && ended_response {
             self.end_already_responded_stream();