s3: fail a streaming upload at VM teardown instead of leaking it

src/jsc/bindings/NativePromiseContext.h

@@ -49,6 +49,7 @@ class NativePromiseContext final : public JSC::JSCell {
         HTMLRewriterSuspension,
         // Task-only tag on the Rust side; never stored in a context cell.
         HTMLRewriterPipeFree,
+        S3UploadStreamWrapper,
     };
 
     // `held` is visited, so the reaction keeps it alive for as long as the

src/runtime/api/NativePromiseContext.rs

@@ -28,6 +28,7 @@ use bun_jsc::{JSGlobalObject, JSValue};
 
 use crate::api::html_rewriter;
 use crate::api::server;
+use crate::webcore::s3::client::S3UploadStreamWrapper;
 
 // Request contexts are a single generic
 // `NewRequestContext<ThisServer, SSL, DEBUG, MUX>`; alias the eight
@@ -66,10 +67,12 @@ pub enum Tag {
     /// Task-only tag (never a context cell): drops the last ref of a
     /// `RewriterPipe` on behalf of `RewriterPipe::deref_outside_caller`.
     HTMLRewriterPipeFree,
+    /// The stream pump's claim on an `S3UploadStreamWrapper` (`pump_cell`).
+    S3UploadStreamWrapper,
 }
 
 impl Tag {
-    pub const COUNT: usize = 10;
+    pub const COUNT: usize = 11;
 
     #[inline]
     const fn from_raw(n: u8) -> Tag {
@@ -84,6 +87,7 @@ impl Tag {
             7 => Tag::DebugHTTPSServerMuxRequestContext,
             8 => Tag::HTMLRewriterSuspension,
             9 => Tag::HTMLRewriterPipeFree,
+            10 => Tag::S3UploadStreamWrapper,
             _ => unreachable!(),
         }
     }
@@ -118,6 +122,9 @@ impl<ThisServer, const SSL: bool, const DBG: bool, const MUX: bool> NativePromis
 impl NativePromiseContextType for html_rewriter::RewriterPipe {
     const TAG: Tag = Tag::HTMLRewriterSuspension;
 }
+impl NativePromiseContextType for S3UploadStreamWrapper {
+    const TAG: Tag = Tag::S3UploadStreamWrapper;
+}
 
 // `&JSGlobalObject` is ABI-identical to a non-null pointer. `ctx` is stored
 // opaquely (never dereferenced by the C++ side), so the FFI itself has no
@@ -202,6 +209,9 @@ fn clear_remembered_cell(ctx: *mut c_void, tag: Tag) {
             Tag::DebugHTTPSServerMuxRequestContext => {
                 (*ctx.cast::<DebugHTTPSServerMuxRequestContext>()).promise_cell_collected()
             }
+            Tag::S3UploadStreamWrapper => {
+                (*ctx.cast::<S3UploadStreamWrapper>()).pump_cell_collected()
+            }
             Tag::HTMLRewriterSuspension | Tag::HTMLRewriterPipeFree => {}
         }
     }
@@ -240,15 +250,16 @@ impl Taskable for DeferredDerefTask {
 impl DeferredDerefTask {
     const TAG_MASK: usize = 0b1111;
 
+    /// From a cell's destructor (and `RewriterPipe::deref_outside_caller`).
     pub(crate) fn schedule(ctx: *mut c_void, tag: Tag) {
         // SAFETY: called from the JS thread (GC sweep → C++ destructor); the
         // thread-local VM is alive for the duration of this call.
-        let vm = VirtualMachine::get();
-        if vm.event_loop_ref().is_closed_for_tasks() {
+        if VirtualMachine::get().event_loop_ref().is_closed_for_tasks() {
             // Teardown has forbidden script and released the queue; from here on only GC destructors
             // (possibly mid-sweep in `~VM`) reach this, and theirs is the last use of `ctx` in that
             // frame. A worker's HTMLRewriter pipe would outlive it, so those refs are released now,
-            // sweep-safe; a RequestContext's deref is not sweep-safe and dies with the VM instead.
+            // sweep-safe; a RequestContext's or an S3 wrapper's release is not sweep-safe and dies
+            // with the VM instead.
             match tag {
                 // SAFETY: the destroyed context held the suspension's ref on this live pipe.
                 Tag::HTMLRewriterSuspension => unsafe {
@@ -267,7 +278,17 @@ impl DeferredDerefTask {
             }
             return;
         }
+        Self::enqueue(ctx, tag);
+    }
 
+    /// For JS-thread code that owns a ref on `ctx` but must not release it in the
+    /// current frame (a frame of `ctx`'s own may be below it): the release runs on the
+    /// next tick, or at once if teardown has closed the queue, when no such frame exists.
+    pub(crate) fn schedule_outside_caller(ctx: *mut c_void, tag: Tag) {
+        Self::enqueue(ctx, tag);
+    }
+
+    fn enqueue(ctx: *mut c_void, tag: Tag) {
         let addr = ctx as usize;
         debug_assert!(addr & Self::TAG_MASK == 0);
 
@@ -277,9 +298,10 @@ impl DeferredDerefTask {
             <DeferredDerefTask as Taskable>::TAG,
             (addr | (tag as usize)) as *mut (),
         );
-        // SAFETY: event_loop() returns the VM's owned EventLoop; we are the
-        // sole mutator on the JS thread here.
-        vm.event_loop_ref().enqueue_task(task);
+        // A closed queue releases the task on arrival (`release_unrun` above runs it).
+        // SAFETY: JS thread; event_loop() returns the VM's owned EventLoop; we are the
+        // sole mutator here.
+        VirtualMachine::get().event_loop_ref().enqueue_task(task);
     }
 
     pub(crate) fn run_from_js_thread(packed_ptr: usize) {
@@ -289,7 +311,8 @@ impl DeferredDerefTask {
         // of the type indicated by `tag`, and this task owns one ref on it,
         // released below (for the HTMLRewriter tags: the ref taken in
         // `begin_suspension`, or the last ref handed over by
-        // `deref_outside_caller`). We are on the JS thread.
+        // `deref_outside_caller`; for the S3 tag: the pump's claim from
+        // `upload_stream`). We are on the JS thread.
         unsafe {
             match tag {
                 Tag::HTTPServerRequestContext => (*ctx.cast::<HTTPServerRequestContext>()).deref(),
@@ -325,6 +348,9 @@ impl DeferredDerefTask {
                         NonNull::new_unchecked(ctx.cast::<html_rewriter::RewriterPipe>()),
                     );
                 }
+                Tag::S3UploadStreamWrapper => {
+                    S3UploadStreamWrapper::release_pump_claim(ctx.cast::<S3UploadStreamWrapper>());
+                }
             }
         }
     }
@@ -353,3 +379,4 @@ const _: () = assert!(
 );
 const _: () =
     assert!(core::mem::align_of::<html_rewriter::RewriterPipe>() > DeferredDerefTask::TAG_MASK);
+const _: () = assert!(core::mem::align_of::<S3UploadStreamWrapper>() > DeferredDerefTask::TAG_MASK);

src/runtime/jsc_hooks.rs

@@ -129,6 +129,8 @@ pub(crate) enum ActiveHandle {
     /// An S3 request / streaming download out on the HTTP thread; same.
     S3Request(ptr::NonNull<crate::webcore::s3::simple_request::S3HttpSimpleTask>),
     S3Download(ptr::NonNull<crate::webcore::s3::download_stream::S3HttpDownloadStreamingTask>),
+    /// An S3 upload; between parts nothing of it is out on the HTTP thread.
+    S3Upload(ptr::NonNull<crate::webcore::s3::MultiPartUpload>),
     /// A `Bun.build` running on the bundle thread with this VM's plugins/env.
     Bundle(ptr::NonNull<crate::api::js_bundle_completion_task::JSBundleCompletionTask>),
     /// A `dns.Resolver` (or the VM-global one) with a live c-ares channel.
@@ -231,6 +233,21 @@ impl ActiveHandle {
             handles.swap_remove(self);
         }
     }
+
+    /// Whether a `bun test --isolate` swap leaves this running; its completion
+    /// lands on the next file's loop.
+    fn outlives_test_isolation(&self) -> bool {
+        match *self {
+            // A live VM cannot cancel a build: hop tasks it already queued would
+            // still be dispatched against the finished pass.
+            ActiveHandle::Bundle(_) => true,
+            // SAFETY: registered ⇒ live (it unregisters in `on_response`). A field
+            // read through the raw place: the HTTP thread may be writing the task's
+            // other fields, so no `&` to the whole task is formed.
+            ActiveHandle::S3Request(t) => unsafe { (*t.as_ptr()).outlives_test_isolation },
+            _ => false,
+        }
+    }
 }
 
 /// Per-VM lazy DNS resolver storage. Shared borrow only — c-ares callbacks
@@ -1713,11 +1730,19 @@ fn stop_dns_for_vm_teardown() -> SweepResult {
 /// `--isolate` swap: a microtask still pending at end-of-file (queued by
 /// `tick_immediate_tasks` or `handle_rejected_promises`) can register new
 /// handles when it runs, so drain first so they land in the registry before it
-/// empties, then stop. (VM teardown must *not* drain here — its
-/// prepareForDestruction discards the pre-exit queues.)
+/// empties, then stop. A stop can queue such microtasks too (a failed upload's
+/// `.catch`), and the swap closes every socket blind right after this, so
+/// repeat while a round stopped something. Bounded: a `.catch` that restarts
+/// its upload would otherwise never let this end. (VM teardown must *not*
+/// drain here — its prepareForDestruction discards the pre-exit queues.)
 pub(crate) fn stop_active_handles_for_test_isolation(vm: &mut VirtualMachine) {
-    let _ = vm.event_loop_mut().drain_microtasks();
-    let _ = stop_active_handles(vm, StopReason::TestIsolation);
+    const MAX_ROUNDS: usize = 8;
+    for _ in 0..MAX_ROUNDS {
+        let _ = vm.event_loop_mut().drain_microtasks();
+        if stop_active_handles(vm, StopReason::TestIsolation) == SweepResult::Idle {
+            break;
+        }
+    }
 }
 
 pub(crate) fn stop_active_handles_for_vm_teardown(vm: &mut VirtualMachine) -> SweepResult {
@@ -1760,14 +1785,20 @@ fn stop_active_handles(vm: &mut VirtualMachine, reason: StopReason) -> SweepResu
             unsafe { (*all).event_loop_delay.disable() };
         }
     }
-    // Entries that stay registered across a test-isolation swap.
+    // Entries that stay registered across a test-isolation swap (including what
+    // a stop below registers: the rollback of a failed upload). Not counted as
+    // stopped, so a sweep that only meets these again reports `Idle`.
     let mut kept: Vec<ActiveHandle> = Vec::new();
     loop {
         // SAFETY: live boxed per-thread `RuntimeState`; the borrow ends before
         // the close below re-enters JS.
         let Some(kv) = (unsafe { &mut (*state).active_handles }).pop() else {
             break;
         };
+        if reason == StopReason::TestIsolation && kv.key.outlives_test_isolation() {
+            kept.push(kv.key);
+            continue;
+        }
         result = SweepResult::Stopped;
         match kv.key {
             // SAFETY: live until it unregisters in `detach`.
@@ -1809,10 +1840,11 @@ fn stop_active_handles(vm: &mut VirtualMachine, reason: StopReason) -> SweepResu
             ActiveHandle::S3Download(t) => unsafe {
                 crate::webcore::s3::download_stream::S3HttpDownloadStreamingTask::stop_for_vm_teardown(t.as_ptr())
             },
-            // A live VM cannot cancel a build: hop tasks it already queued here
-            // would still be dispatched against the finished pass. The build
-            // runs on; its completion lands on the next file's global.
-            ActiveHandle::Bundle(_) if reason == StopReason::TestIsolation => kept.push(kv.key),
+            // SAFETY: live until it unregisters in `Drop`; may free itself inside,
+            // not touched after.
+            ActiveHandle::S3Upload(u) => unsafe {
+                crate::webcore::s3::MultiPartUpload::stop_for_vm_teardown(u.as_ptr())
+            },
             // SAFETY: live until it unregisters in `on_complete_anytask`.
             ActiveHandle::Bundle(c) => unsafe {
                 crate::api::js_bundle_completion_task::JSBundleCompletionTask::stop_for_vm_teardown(