s3: abort an upload that can never finish (collected writer, Create after fail, 204 abort)

src/event_loop/ConcurrentTask.rs

@@ -112,6 +112,7 @@ pub mod task_tag {
         RuntimeTranspilerStore,
         S3HttpDownloadStreamingTask,
         S3HttpSimpleTask,
+        S3UploadWriterCollected,  // MultiPartUpload (its `writer()` was collected before `end()`)
         SendQueueDeferred,        // bun_runtime::ipc::SendQueue (close / after-close hop)
         ServerAllConnectionsClosedTask,
         HTTPServerDeinit,

src/runtime/dispatch.rs

@@ -84,6 +84,7 @@ use crate::webcore::file_sink::FlushPendingTask as FlushPendingFileSinkTask;
 #[cfg(not(windows))]
 use crate::webcore::file_sink::Poll as FileSinkPoll;
 use crate::webcore::s3::download_stream::S3HttpDownloadStreamingTask;
+use crate::webcore::s3::multipart::WriterCollected as S3UploadWriterCollected;
 use crate::webcore::s3::simple_request::S3HttpSimpleTask;
 use crate::webcore::streams::Pending as StreamPending;
 
@@ -485,6 +486,9 @@ pub(crate) fn run_task(
         task_tag::S3HttpDownloadStreamingTask => {
             S3HttpDownloadStreamingTask::on_response(cast_ptr!(S3HttpDownloadStreamingTask));
         }
+        task_tag::S3UploadWriterCollected => {
+            S3UploadWriterCollected::run(cast_ptr!(S3UploadWriterCollected))?;
+        }
 
         // ── napi ─────────────────────────────────────────────────────────
         task_tag::NapiAsyncWork => {
@@ -722,7 +726,7 @@ fn run_task_cold(task: Task) {
 /// `release_task_unrun` track `bun_event_loop::task_tag::COUNT`. Bump when
 /// adding a variant — and give it an arm in both.
 const _: () = assert!(
-    task_tag::COUNT == 82,
+    task_tag::COUNT == 83,
     "dispatch::run_task / release_task_unrun arm count out of sync with bun_event_loop::task_tag",
 );
 
@@ -1454,6 +1458,7 @@ fn __bun_release_task_unrun(task: bun_event_loop::Task) {
         task_tag::RuntimeTranspilerStore => release!(RuntimeTranspilerStore),
         task_tag::S3HttpDownloadStreamingTask => release!(S3HttpDownloadStreamingTask),
         task_tag::S3HttpSimpleTask => release!(S3HttpSimpleTask),
+        task_tag::S3UploadWriterCollected => release!(S3UploadWriterCollected),
         task_tag::SendQueueDeferred => release!(crate::ipc::SendQueue),
         task_tag::ServerAllConnectionsClosedTask => release!(ServerAllConnectionsClosedTask),
         task_tag::ShellAsync => release!(crate::shell::dispatch_tasks::ShellAsyncTask),

src/runtime/webcore/s3/client.rs

@@ -418,10 +418,11 @@ pub(crate) fn writable_stream(
     storage_class: Option<StorageClass>,
     request_payer: bool,
 ) -> JsResult<JSValue> {
-    // Local callback wrapper. `uploaded` is read off the upload (see `MultiPartUpload::callback`).
+    // Local callback wrapper. `uploaded` and `path` are read off the upload (see `MultiPartUpload::callback`).
     fn wrapper_callback(
         result: S3UploadResult,
         uploaded: u64,
+        path: &[u8],
         sink: &mut NetworkSink,
     ) -> JsResult<()> {
         // `global_this` is a `BackRef` set at construction; copy it so the
@@ -448,7 +449,7 @@ pub(crate) fn writable_stream(
                     }
                 }
                 S3UploadResult::Failure(err) => {
-                    let js_err = s3_error_to_js(&err, global, sink.path());
+                    let js_err = s3_error_to_js(&err, global, Some(path));
                     if sink.flush_promise.has_value() {
                         sink.flush_promise.reject(global, Ok(js_err))?;
                     }
@@ -475,7 +476,9 @@ pub(crate) fn writable_stream(
         let sink = ctx.cast::<NetworkSink>();
         // SAFETY: ctx was set to `response_stream: *mut NetworkSink` below; the box is live
         // while the upload holds it.
-        let r = wrapper_callback(result, task.uploaded_bytes.get(), unsafe { &mut *sink });
+        let r = wrapper_callback(result, task.uploaded_bytes.get(), &task.path, unsafe {
+            &mut *sink
+        });
         // SAFETY: the upload's hold on the box ends here; `sink` is not used afterwards.
         unsafe { NetworkSink::release_writer_holder(sink) };
         r

src/runtime/webcore/s3/multipart.rs

@@ -113,8 +113,8 @@ use bun_s3_signing::storage_class::StorageClass;
 // re-export hub instead.
 use crate::webcore::s3::multipart_options::MultiPartUploadOptions;
 use crate::webcore::s3::simple_request::{
-    self as s3_simple_request, S3CommitResult, S3DownloadResult, S3PartResult, S3UploadResult,
-    execute_simple_s3_request,
+    self as s3_simple_request, S3CommitResult, S3DeleteResult, S3DownloadResult, S3PartResult,
+    S3UploadResult, execute_simple_s3_request,
 };
 use crate::webcore::s3::xml_response;
 use bun_collections::index_sort;
@@ -184,6 +184,32 @@ bun_jsc::impl_abort_handle_owner!(MultiPartUpload, abort_handle, |this, _cause|
     }
 });
 
+/// [`MultiPartUpload::fail_writer_collected`]'s task: same pointer as the upload, its own tag.
+#[repr(transparent)]
+pub(crate) struct WriterCollected(MultiPartUpload);
+impl bun_event_loop::Taskable for WriterCollected {
+    const TAG: bun_event_loop::TaskTag = bun_event_loop::task_tag::S3UploadWriterCollected;
+    /// Drops the task's +1. The context that stopped fails the upload through `abort_handle`.
+    unsafe fn release_unrun(this: *mut Self) {
+        MultiPartUpload::deref_(this.cast::<MultiPartUpload>());
+    }
+    /// The context of the script that made the writer.
+    unsafe fn context(this: *const Self) -> bun_event_loop::ContextId {
+        // SAFETY: fn contract — the live upload `fail_writer_collected` queued.
+        unsafe { (*this.cast::<MultiPartUpload>()).context }
+    }
+}
+impl WriterCollected {
+    pub(crate) fn run(this: *mut Self) -> bun_jsc::JsResult<()> {
+        // SAFETY: adopts the +1 `fail_writer_collected` took; released after `fail`.
+        let upload = unsafe { RefPtr::from_raw(this.cast::<MultiPartUpload>()) };
+        upload.fail(S3Error {
+            code: b"UnknownError",
+            message: b"S3 writer was garbage collected before end() was called",
+        })
+    }
+}
+
 #[repr(u8)]
 #[derive(Clone, Copy, PartialEq, Eq)]
 pub(crate) enum State {
@@ -623,6 +649,16 @@ impl MultiPartUpload {
         Ok(())
     }
 
+    /// `fail` from a task: the caller is a finalizer inside a GC sweep, and `fail` settles promises.
+    pub(crate) fn fail_writer_collected(&self) {
+        self.ref_();
+        self.vm
+            .event_loop_ref()
+            .enqueue_task(bun_event_loop::Task::init(
+                self.root_ptr().cast::<WriterCollected>(),
+            ));
+    }
+
     fn done(&self) -> bun_jsc::JsResult<()> {
         let state = self.state.get();
         if state == State::MultipartCompleted && self.is_queue_empty() {
@@ -674,11 +710,13 @@ impl MultiPartUpload {
         let _guard = unsafe { RefPtr::from_raw(this) };
         // SAFETY: `this` is live for at least the guard's duration.
         let self_ = unsafe { &*this };
-        if self_.state.get() == State::Finished {
-            return Ok(());
-        }
+        // `fail` ran while this request was in flight: an upload the server created is still open.
+        let failed = self_.state.get() == State::Finished;
         match result {
             S3DownloadResult::Failure(err) => {
+                if failed {
+                    return Ok(());
+                }
                 scoped_log!(
                     S3MultiPartUpload,
                     "startMultiPartRequestResult {} failed {}: {}",
@@ -711,6 +749,20 @@ impl MultiPartUpload {
                 if let Some(upload_id) = upload_id {
                     self_.upload_id.set(upload_id);
                 }
+                if failed {
+                    if valid {
+                        scoped_log!(
+                            S3MultiPartUpload,
+                            "startMultiPartRequestResult {} aborting after fail id: {}",
+                            BStr::new(&self_.path),
+                            BStr::new(self_.upload_id.get())
+                        );
+                        // The rollback callback releases this ref.
+                        self_.ref_();
+                        self_.rollback_multi_part_request()?;
+                    }
+                    return Ok(());
+                }
                 if !valid {
                     // Unknown type of response error from AWS
                     scoped_log!(
@@ -735,10 +787,15 @@ impl MultiPartUpload {
                 self_.drain_enqueued_parts(0)
             }
             // this is "unreachable" but we cover in case AWS returns 404
-            S3DownloadResult::NotFound(_) => self_.fail(S3Error {
-                code: b"UnknownError",
-                message: b"Failed to initiate multipart upload",
-            }),
+            S3DownloadResult::NotFound(_) => {
+                if failed {
+                    return Ok(());
+                }
+                self_.fail(S3Error {
+                    code: b"UnknownError",
+                    message: b"Failed to initiate multipart upload",
+                })
+            }
         }
     }
 
@@ -792,7 +849,7 @@ impl MultiPartUpload {
 
     /// We do a best effort to rollback the multipart upload, if it fails we will retry, if it still we just deinit the upload
     pub(crate) fn on_rollback_multi_part_request(
-        result: S3UploadResult,
+        result: S3DeleteResult,
         this: *mut c_void,
     ) -> bun_jsc::JsResult<()> {
         let this = this.cast::<Self>();
@@ -805,7 +862,7 @@ impl MultiPartUpload {
             BStr::new(self_.upload_id.get())
         );
         match result {
-            S3UploadResult::Failure(_err) => {
+            S3DeleteResult::Failure(_err) => {
                 let mut options = self_.options.get();
                 if options.retry > 0 {
                     options.retry -= 1;
@@ -817,7 +874,8 @@ impl MultiPartUpload {
                 MultiPartUpload::deref_(this);
                 Ok(())
             }
-            S3UploadResult::Success => {
+            // 404: the store no longer has the upload, which is what a rollback is for.
+            S3DeleteResult::Success | S3DeleteResult::NotFound(_) => {
                 MultiPartUpload::deref_(this);
                 Ok(())
             }
@@ -883,7 +941,7 @@ impl MultiPartUpload {
                 request_payer: self.request_payer,
                 ..Default::default()
             },
-            s3_simple_request::S3Callback::Upload(Self::on_rollback_multi_part_request),
+            s3_simple_request::S3Callback::Delete(Self::on_rollback_multi_part_request),
             self.as_ctx_ptr(),
         )
     }

src/runtime/webcore/streams.rs

@@ -2331,13 +2331,6 @@ impl NetworkSink {
         Box::new(init)
     }
 
-    pub fn path(&self) -> Option<&[u8]> {
-        if let Some(task) = self.task_ref() {
-            return Some(&task.path);
-        }
-        None
-    }
-
     pub(crate) fn start(&mut self, _stream_start: &Start) -> bun_sys::Result<()> {
         if self.ended {
             return bun_sys::Result::Ok(());
@@ -2373,6 +2366,18 @@ impl NetworkSink {
         self.task = None;
     }
 
+    /// The `writer()` wrapper was collected before `end()`: nothing can finish the upload, so fail it.
+    fn abort_on_collect(&mut self) {
+        if self.ended || self.writer_holders.get() == 0 {
+            return;
+        }
+        self.ended = true;
+        self.done = true;
+        if let Some(task) = self.task_ref() {
+            task.fail_writer_collected();
+        }
+    }
+
     /// The S3 upload drained: settle the flush/write promises (terminal, like
     /// `flush_promise`) and wake the source.
     pub(crate) fn on_writable(
@@ -2691,6 +2696,7 @@ impl crate::webcore::sink::JsSinkType for NetworkSink {
     unsafe fn finalize(this: *mut Self) {
         // SAFETY: trait contract — `this` is live and not used after this call.
         unsafe {
+            (*this).abort_on_collect();
             (*this).finalize();
             Self::release_writer_holder(this);
         }