fetch/S3: replace ResumableSink with proper JSSinks

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

@@ -270,8 +270,9 @@ static NextStep asyncIterHandleNextResult(JSGlobalObject* globalObject, JSAsyncI
         JSValue wrote = invokeOptionalMethod(globalObject, controller, WebCore::builtinNames(vm).writePublicName(), writeArgs);
         if (scope.exception()) [[unlikely]]
             goto abrupt;
-        if (wrote && wrote.isNumber() && wrote.asNumber() < 0) {
-            // The HTTP sink reports backpressure with a negative return: wait for the drain.
+        if ((wrote && wrote.isNumber() && wrote.asNumber() < 0) || wrote.isFalse()) {
+            // Backpressure: a JSSink returns a negative number, a ResumableSink returns false.
+            // Either way `flush(true)` returns the drain promise.
             MarkedArgumentBuffer flushArgs;
             flushArgs.append(jsBoolean(true));
             JSValue flushed = invokeOptionalMethod(globalObject, controller, builtinNames(vm).flushPublicName(), flushArgs);

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

@@ -1381,7 +1381,7 @@
         if (hasError)
             args.append(error);
         ASSERT(!args.hasOverflowed());
-        invokeMethod(vm, globalObject, sink, builtinNames(vm).endPublicName(), args);
+        invokeMethod(vm, globalObject, sink, hasError ? vm.propertyNames->error : builtinNames(vm).endPublicName(), args);
         if (catchScope.exception()) [[unlikely]] {
             if (takeAbruptCompletion(globalObject, catchScope).isEmpty())
                 return;
@@ -1527,12 +1527,87 @@
     RELEASE_AND_RETURN(scope, resumableDrain(vm, globalObject, op));
 }
 
+// DirectPending -> ResumableSink without an ArrayBufferSink intermediary: the underlying
+// source's pull() writes straight into sink.write (bool backpressure, flush(true) drain
+// promise) exactly as readDirectStream hands a JSSink to the same pull.
+static void resumableSetupDirect(JSC::VM& vm, JSGlobalObject* globalObject, JSReadableStream* stream, JSObject* sink, JSObject* underlyingSource)
+{
+    auto scope = DECLARE_THROW_SCOPE(vm);
+    auto* runtime = WebCore::JSStreamsRuntime::from(globalObject);
+
+    stream->m_directUnderlyingSource.clear();
+    stream->m_bunMode = BunStreamMode::Default;
+
+    JSValue pull = underlyingSource->get(globalObject, builtinNames(vm).pullPublicName());
+    RETURN_IF_EXCEPTION(scope, );
+
+    double rawHighWaterMark = stream->m_bunHighWaterMark;
+    auto* startOptions = constructEmptyObject(globalObject);
+    startOptions->putDirect(vm, builtinNames(vm).highWaterMarkPublicName(), jsNumber(std::isnan(rawHighWaterMark) ? 0 : rawHighWaterMark));
+    MarkedArgumentBuffer startArgs;
+    startArgs.append(startOptions);
+    ASSERT(!startArgs.hasOverflowed());
+    invokeMethod(vm, globalObject, sink, builtinNames(vm).startPublicName(), startArgs);
+    RETURN_IF_EXCEPTION(scope, );
+
+    auto* cancelBound = createBoundHandler(globalObject, runtime->boundResumableSinkDirectCancel(), underlyingSource);
+    RETURN_IF_EXCEPTION(scope, );
+    MarkedArgumentBuffer handlerArgs;
+    handlerArgs.append(jsUndefined());
+    handlerArgs.append(cancelBound);
+    ASSERT(!handlerArgs.hasOverflowed());
+    invokeMethod(vm, globalObject, sink, builtinNames(vm).setHandlersPublicName(), handlerArgs);
+    RETURN_IF_EXCEPTION(scope, );
+
+    stream->m_lockedWithoutReader = true;
+
+    if (!pull.isCallable()) {
+        MarkedArgumentBuffer noArgs;
+        scope.release();
+        invokeMethod(vm, globalObject, sink, builtinNames(vm).endPublicName(), noArgs);
+        return;
+    }
+
+    MarkedArgumentBuffer pullArgs;
+    pullArgs.append(sink);
+    ASSERT(!pullArgs.hasOverflowed());
+    JSValue maybePromise = JSC::call(globalObject, pull, getCallData(pull), underlyingSource, pullArgs);
+    RETURN_IF_EXCEPTION(scope, );
+
+    if (auto* pullPromise = dynamicDowncast<JSPromise>(maybePromise)) {
+        markPromiseAsHandled(vm, pullPromise);
+        pullPromise->performPromiseThenWithContext(vm, globalObject, runtime->onReturnUndefined(), runtime->onResumableSinkDirectPullRejected(), jsUndefined(), sink);
+    }
+}
+
 JSValue assignStreamIntoResumableSink(JSGlobalObject* globalObject, JSReadableStream* stream, JSObject* resumableSink)
 {
     auto& vm = getVM(globalObject);
     auto scope = DECLARE_THROW_SCOPE(vm);
     auto* domGlobalObject = defaultGlobalObject(globalObject);
     auto* runtime = WebCore::JSStreamsRuntime::from(globalObject);
+
+    if (JSObject* underlyingSource = stream->m_directUnderlyingSource.get(); stream->m_bunMode == BunStreamMode::DirectPending && underlyingSource) {
+        JSValue thrown;
+        {
+            auto catchScope = DECLARE_TOP_EXCEPTION_SCOPE(vm);
+            resumableSetupDirect(vm, globalObject, stream, resumableSink, underlyingSource);
+            if (catchScope.exception()) [[unlikely]] {
+                thrown = takeAbruptCompletion(globalObject, catchScope);
+                if (thrown.isEmpty())
+                    return {};
+            }
+        }
+        if (!thrown.isEmpty()) {
+            MarkedArgumentBuffer args;
+            args.append(thrown);
+            ASSERT(!args.hasOverflowed());
+            invokeMethod(vm, globalObject, resumableSink, vm.propertyNames->error, args);
+        }
+        RETURN_IF_EXCEPTION(scope, {});
+        return jsUndefined();
+    }
+
     auto* op = JSResumableSinkPumpOperation::create(vm, runtime->resumableSinkPumpOperationStructure(domGlobalObject));
     op->m_stream.set(vm, op, stream);
     op->m_sink.set(vm, op, resumableSink);
@@ -1803,4 +1878,41 @@
     return JSValue::encode(jsUndefined());
 }
 
+// Native cancel on the direct path: (underlyingSource, unused, reason) -> source.cancel(reason).
+JSC_DEFINE_HOST_FUNCTION(jsWebStreamsHandler_boundResumableSinkDirectCancel, (JSGlobalObject * globalObject, CallFrame* callFrame))
+{
+    auto& vm = getVM(globalObject);
+    auto scope = DECLARE_THROW_SCOPE(vm);
+    JSObject* underlyingSource = callFrame->argument(0).getObject();
+    if (!underlyingSource)
+        return JSValue::encode(jsUndefined());
+    MarkedArgumentBuffer args;
+    args.append(callFrame->argument(2));
+    ASSERT(!args.hasOverflowed());
+    JSValue result = invokeOptionalMethod(globalObject, underlyingSource, builtinNames(vm).cancelPublicName(), args);
+    RETURN_IF_EXCEPTION(scope, {});
+    if (result && result.isCell()) {
+        if (auto* promise = dynamicDowncast<JSPromise>(result))
+            markPromiseAsHandled(vm, promise);
+    }
+    return JSValue::encode(jsUndefined());
+}
+
+// pull(sink) rejected on the direct path: end the sink with the error so the native side
+// aborts the upload; the pump already ran iterator.throw().
+JSC_DEFINE_HOST_FUNCTION(jsWebStreamsHandler_onResumableSinkDirectPullRejected, (JSGlobalObject * globalObject, CallFrame* callFrame))
+{
+    auto& vm = getVM(globalObject);
+    auto scope = DECLARE_THROW_SCOPE(vm);
+    JSObject* sink = callFrame->argument(1).getObject();
+    if (!sink)
+        return JSValue::encode(jsUndefined());
+    MarkedArgumentBuffer args;
+    args.append(callFrame->argument(0));
+    ASSERT(!args.hasOverflowed());
+    invokeOptionalMethod(globalObject, sink, vm.propertyNames->error, args);
+    RETURN_IF_EXCEPTION(scope, {});
+    return JSValue::encode(jsUndefined());
+}
+
 } // namespace WebCore

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

@@ -172,7 +172,8 @@ class JSDirectStreamController;
     V(onResumableSinkChunk)                                 \
     V(onResumableSinkClose)                                 \
     V(onResumableSinkReadRejected)                          \
-    V(onResumableSinkEndMicrotask)
+    V(onResumableSinkEndMicrotask)                          \
+    V(onResumableSinkDirectPullRejected)
 
 // owner: JSDirectStreamController.cpp. context = the JSDirectStreamController.
 #define FOR_EACH_WEB_STREAMS_REACTION_HANDLER_DIRECT_CONTROLLER(V) \
@@ -248,13 +249,16 @@ class JSDirectStreamController;
 //   boundReadStreamIntoSinkOnClose(op, stream, reason): readStreamIntoSink's JSSink onClose.
 //   boundResumableSinkDrain(op) / boundResumableSinkCancel(op, unused, reason): stored on
 //     the native ResumableSink via setHandlers.
+//   boundResumableSinkDirectCancel(underlyingSource, unused, reason): the direct-path
+//     oncancel for a DirectPending stream handed to a ResumableSink.
 #define FOR_EACH_WEB_STREAMS_BOUND_HANDLER_TARGET_BUN_SOURCE(V) \
     V(boundOnNativeSourceClose)                                 \
     V(boundOnNativeSourceDrain)                                 \
     V(boundReadDirectStreamOnClose)                             \
     V(boundReadStreamIntoSinkOnClose)                           \
     V(boundResumableSinkDrain)                                  \
-    V(boundResumableSinkCancel)
+    V(boundResumableSinkCancel)                                 \
+    V(boundResumableSinkDirectCancel)
 
 // owner: JSDirectStreamController.cpp — the FIVE detachable own methods of the direct
 // controller: `end` and `close` are two bound cells over the ONE boundDirectClose target.

src/jsc/generated.rs

@@ -1133,9 +1133,9 @@ js_class_module!(JSImmediate = "Immediate" { callback, arguments });
 js_class_module!(JSBlob      = "Blob"      as crate::webcore_types::Blob { name, stream });
 js_class_module!(JSResponse  = "Response"  { body, headers, url, statusText, stream });
 js_class_module!(JSRequest   = "Request"   { body, headers, url, signal, stream });
-// `values: ["ondrain", "oncancel", "stream"]` in src/runtime/api/ResumableSink.classes.ts.
-js_class_module!(JSResumableFetchSink    = "ResumableFetchSink"    { ondrain, oncancel, stream });
-js_class_module!(JSResumableS3UploadSink = "ResumableS3UploadSink" { ondrain, oncancel, stream });
+// `values: ["ondrain", "oncancel", "stream", "flushPromise"]` in src/runtime/api/ResumableSink.classes.ts.
+js_class_module!(JSResumableFetchSink    = "ResumableFetchSink"    { ondrain, oncancel, stream, flushPromise });
+js_class_module!(JSResumableS3UploadSink = "ResumableS3UploadSink" { ondrain, oncancel, stream, flushPromise });
 // `values: ["resolve", "reject"]` in src/runtime/api/Shell.classes.ts.
 js_class_module!(JSShellInterpreter      = "ShellInterpreter"      { resolve, reject });
 // `src/runtime/crypto/crypto.classes.ts` — one entry per `StaticCryptoHasher`

src/runtime/api/ResumableSink.classes.ts

@@ -19,17 +19,33 @@ function generate(name) {
         fn: "jsWrite",
         length: 1,
       },
+      flush: {
+        fn: "jsFlush",
+        length: 1,
+        passThis: true,
+      },
       end: {
+        fn: "jsClose",
+        length: 0,
+        passThis: true,
+      },
+      close: {
+        fn: "jsClose",
+        length: 0,
+        passThis: true,
+      },
+      error: {
         fn: "jsEnd",
         length: 1,
+        passThis: true,
       },
       setHandlers: {
         fn: "jsSetHandlers",
         length: 2,
         passThis: true,
       },
     },
-    values: ["ondrain", "oncancel", "stream"],
+    values: ["ondrain", "oncancel", "stream", "flushPromise"],
   });
 }
 export default [generate("ResumableFetchSink"), generate("ResumableS3UploadSink")];

src/runtime/webcore/ResumableSink.rs

@@ -34,6 +34,8 @@
     fn ondrain_get_cached(this_value: JSValue) -> Option<JSValue>;
     fn stream_set_cached(this_value: JSValue, global: &JSGlobalObject, value: JSValue);
     fn stream_get_cached(this_value: JSValue) -> Option<JSValue>;
+    fn flushpromise_set_cached(this_value: JSValue, global: &JSGlobalObject, value: JSValue);
+    fn flushpromise_get_cached(this_value: JSValue) -> Option<JSValue>;
 }
 
 /// Trait capturing the per-`Context` callbacks the sink invokes.
@@ -118,6 +120,19 @@
     fn set_stream(this_value: JSValue, global: &JSGlobalObject, value: JSValue) {
         Js::stream_set_cached(this_value, global, value);
     }
+    #[inline]
+    fn set_flush_promise(this_value: JSValue, global: &JSGlobalObject, value: JSValue) {
+        Js::flushpromise_set_cached(this_value, global, value);
+    }
+    fn take_flush_promise(
+        this_value: JSValue,
+        global: &JSGlobalObject,
+    ) -> Option<&'static mut bun_jsc::JSPromise> {
+        let value = Js::flushpromise_get_cached(this_value)?;
+        let promise = value.as_promise()?;
+        Js::flushpromise_set_cached(this_value, global, JSValue::ZERO);
+        Some(bun_jsc::JSPromise::opaque_mut(promise))
+    }
 
     #[inline]
     fn on_write(ctx: *mut Context, bytes: &[u8]) -> ResumableSinkBackpressure {
@@ -333,26 +348,81 @@
         Ok(JSValue::from(this.status != Status::Paused))
     }
 
+    /// `flush(true)` while paused returns a promise that `drain()` fulfills (or `cancel()`
+    /// rejects); the direct-stream pump awaits it when `write()` returns `false`.
+    #[bun_jsc::host_fn(method)]
+    pub fn js_flush(
+        this: &mut Self,
+        global_this: &JSGlobalObject,
+        callframe: &CallFrame,
+        this_value: JSValue,
+    ) -> JsResult<JSValue> {
+        bun_jsc::mark_binding!();
+        if this.is_detached() {
+            return Ok(JSValue::UNDEFINED);
+        }
+        let args = callframe.arguments();
+        let wait = args.first().is_some_and(|v| v.to_boolean());
+        if !wait || this.status != Status::Paused {
+            return Ok(JSValue::UNDEFINED);
+        }
+        if let Some(pending) = Js::flushpromise_get_cached(this_value) {
+            if pending.as_promise().is_some() {
+                return Ok(pending);
+            }
+        }
+        let promise = bun_jsc::JSPromise::create(global_this);
+        let promise_value = promise.to_js();
+        Self::set_flush_promise(this_value, global_this, promise_value);
+        Ok(promise_value)
+    }
+
+    /// `error(e)` on the controller surface: abort the upload with `e`.
     #[bun_jsc::host_fn(method)]
     pub fn js_end(
         this: &mut Self,
-        _global_this: &JSGlobalObject,
+        global_this: &JSGlobalObject,
         callframe: &CallFrame,
+        this_value: JSValue,
     ) -> JsResult<JSValue> {
         bun_jsc::mark_binding!();
         let args = callframe.arguments();
         // ignore any call if detached
         if this.is_detached() {
             return Ok(JSValue::UNDEFINED);
         }
+        let err = if args.len() > 0 { Some(args[0]) } else { None };
+        if let Some(promise) = Self::take_flush_promise(this_value, global_this) {
+            let _ = promise.reject_as_handled(global_this, err.unwrap_or(JSValue::UNDEFINED));
+        }
         this.detach_js();
         scoped_log!(ResumableSink, "jsEnd {}", args.len());
         this.status = Status::Done;
 
-        Self::on_end(
-            this.context,
-            if args.len() > 0 { Some(args[0]) } else { None },
-        );
+        Self::on_end(this.context, err);
+        Ok(JSValue::UNDEFINED)
+    }
+
+    /// `end()` / `close()` on the controller surface: a clean end regardless of the reason
+    /// argument (the direct controller's `end`/`close` are the same no-error target).
+    #[bun_jsc::host_fn(method)]
+    pub fn js_close(
+        this: &mut Self,
+        global_this: &JSGlobalObject,
+        _callframe: &CallFrame,
+        this_value: JSValue,
+    ) -> JsResult<JSValue> {
+        bun_jsc::mark_binding!();
+        if this.is_detached() {
+            return Ok(JSValue::UNDEFINED);
+        }
+        if let Some(promise) = Self::take_flush_promise(this_value, global_this) {
+            let _ = promise.resolve(global_this, JSValue::UNDEFINED);
+        }
+        this.detach_js();
+        scoped_log!(ResumableSink, "jsClose");
+        this.status = Status::Done;
+        Self::on_end(this.context, None);
         Ok(JSValue::UNDEFINED)
     }
 
@@ -365,8 +435,11 @@
             let global_object = self.global_this;
             let global_object = global_object.get();
 
+            self.status = Status::Started;
+            if let Some(promise) = Self::take_flush_promise(js_this, global_object) {
+                let _ = promise.resolve(global_object, JSValue::UNDEFINED);
+            }
             if let Some(ondrain) = Self::get_drain(js_this) {
-                self.status = Status::Started;
                 // SAFETY: `bun_vm()` returns a live `*mut VirtualMachine` owned by
                 // the global; `event_loop()` returns its self-referential
                 // `*mut EventLoop`. Both outlive this call.
@@ -402,6 +475,9 @@
             let global_object = self.global_this;
             let global_object = global_object.get();
 
+            if let Some(promise) = Self::take_flush_promise(js_this, global_object) {
+                let _ = promise.reject_as_handled(global_object, reason);
+            }
             // detach first so if cancel calls end will be a no-op
             self.detach_js();
 
@@ -443,6 +519,7 @@
             Self::set_drain(js_this, global, JSValue::ZERO);
             Self::set_cancel(js_this, global, JSValue::ZERO);
             Self::set_stream(js_this, global, JSValue::ZERO);
+            Self::set_flush_promise(js_this, global, JSValue::ZERO);
             self.js_this.downgrade();
         }
     }
@@ -617,6 +694,14 @@
             fn stream_get_cached(this: JSValue) -> Option<JSValue> {
                 bun_jsc::generated::$name::stream_get_cached(this)
             }
+            #[inline]
+            fn flushpromise_set_cached(this: JSValue, global: &JSGlobalObject, v: JSValue) {
+                bun_jsc::generated::$name::flush_promise_set_cached(this, global, v)
+            }
+            #[inline]
+            fn flushpromise_get_cached(this: JSValue) -> Option<JSValue> {
+                bun_jsc::generated::$name::flush_promise_get_cached(this)
+            }
         }
     )*};
 }