s3: S3File.writer().end(error) aborts the upload instead of committing

src/runtime/webcore/ArrayBufferSink.rs

@@ -283,7 +283,7 @@ impl crate::webcore::sink::JsSinkType for ArrayBufferSink {
     fn end(&mut self, err: Option<syscall::Error>) -> bun_sys::Result<()> {
         Self::end(self, err)
     }
-    fn end_from_js(&mut self, global: &JSGlobalObject) -> bun_sys::Result<JSValue> {
+    fn end_from_js(&mut self, global: &JSGlobalObject, _err: JSValue) -> bun_sys::Result<JSValue> {
         match Self::end_from_js(self, global) {
             bun_sys::Result::Ok(ab) => bun_sys::Result::Ok(match ab.to_js_unchecked(global) {
                 Ok(v) => v,

src/runtime/webcore/FileSink.rs

@@ -1190,7 +1190,7 @@ impl crate::webcore::sink::JsSinkType for FileSink {
     fn end(&mut self, err: Option<sys::Error>) -> sys::Result<()> {
         Self::end(self, err)
     }
-    fn end_from_js(&mut self, global: &JSGlobalObject) -> sys::Result<JSValue> {
+    fn end_from_js(&mut self, global: &JSGlobalObject, _err: JSValue) -> sys::Result<JSValue> {
         Self::end_from_js(self, global)
     }
     fn flush(&mut self) -> sys::Result<()> {

src/runtime/webcore/Sink.rs

@@ -753,7 +753,7 @@ pub trait JsSinkType: Sized {
     fn write_utf16(&mut self, data: &streams::Result) -> streams::result::Writable;
     fn write_latin1(&mut self, data: &streams::Result) -> streams::result::Writable;
     fn end(&mut self, err: Option<SysError>) -> sys::Result<()>;
-    fn end_from_js(&mut self, global: &JSGlobalObject) -> sys::Result<JSValue>;
+    fn end_from_js(&mut self, global: &JSGlobalObject, err: JSValue) -> sys::Result<JSValue>;
     fn flush(&mut self) -> sys::Result<()>;
     fn start(&mut self, config: streams::Start) -> sys::Result<()>;
 
@@ -1010,7 +1010,10 @@ impl<T: JsSinkType + JsSinkAbi> JSSink<T> {
             return Err(global.throw_value(err));
         }
 
-        let result = match this.sink.end_from_js(global) {
+        let err_arg = frame.argument(0);
+        err_arg.ensure_still_alive();
+
+        let result = match this.sink.end_from_js(global, err_arg) {
             sys::Result::Ok(value) => Ok(value),
             sys::Result::Err(err) => Err(global.throw_value(err.to_js(global)?)),
         };
@@ -1102,7 +1105,7 @@ impl<T: JsSinkType + JsSinkAbi> JSSink<T> {
         }
 
         // TODO: properly propagate exception upwards
-        match this.end_from_js(global) {
+        match this.end_from_js(global, crate::webcore::jsc::JSValue::UNDEFINED) {
             sys::Result::Ok(value) => value,
             sys::Result::Err(err) => match err.to_js(global) {
                 Ok(v) => {

src/runtime/webcore/s3/multipart.rs

@@ -669,6 +669,30 @@
         // SAFETY: callback context — `this` is live (a ref was taken before the request)
         let this = unsafe { &mut *this };
         if this.state == State::Finished {
+            // Already failed while the init request was in flight. If the
+            // server created an upload, roll it back instead of orphaning it.
+            if let S3DownloadResult::Success(response) = result {
+                let slice = response.body.list.as_slice();
+                if let Some(start) = strings::index_of(slice, b"<UploadId>") {
+                    let value_start = start + b"<UploadId>".len();
+                    if let Some(end) = strings::index_of(slice, b"</UploadId>") {
+                        if end >= value_start {
+                            this.upload_id = Box::<[u8]>::from(&slice[value_start..end]);
+                        }
+                    }
+                }
+                if !this.upload_id.is_empty()
+                    && this.upload_id.len() <= Self::MAX_UPLOAD_ID_LEN
+                    && this
+                        .upload_id
+                        .iter()
+                        .all(|b| b.is_ascii() && !b.is_ascii_control())
+                {
+                    // rollback consumes one ref via its completion callback
+                    this.ref_();
+                    return this.rollback_multi_part_request();
+                }
+            }
             return Ok(());
         }
         match result {

src/runtime/webcore/streams.rs

@@ -31,6 +31,9 @@ pub type ByteListPoolNode = bun_collections::pool::Node<Vec<u8>>;
 // for callers that still spell it that way.
 pub mod bun_s3 {
     pub use crate::webcore::s3::MultiPartUpload;
+    pub use crate::webcore::s3::multipart::State as MultiPartUploadState;
+    pub use crate::webcore::s3::simple_request::S3UploadResult;
+    pub use bun_s3_signing::error::S3Error;
 }
 
 /// `Blob.SizeType` is `u64` (see `webcore::blob::SizeType`).
@@ -2089,7 +2092,7 @@ impl<const SSL: bool, const HTTP3: bool> crate::webcore::sink::JsSinkType
     fn end(&mut self, err: Option<SysError>) -> bun_sys::Result<()> {
         Self::end(self, err)
     }
-    fn end_from_js(&mut self, global: &JSGlobalObject) -> bun_sys::Result<JSValue> {
+    fn end_from_js(&mut self, global: &JSGlobalObject, _err: JSValue) -> bun_sys::Result<JSValue> {
         Self::end_from_js(self, global)
     }
     fn flush(&mut self) -> bun_sys::Result<()> {
@@ -2379,7 +2382,14 @@ impl NetworkSink {
         bun_sys::Result::Ok(())
     }
 
-    pub fn end_from_js(&mut self, _global_this: &JSGlobalObject) -> bun_sys::Result<JSValue> {
+    pub fn end_from_js(
+        &mut self,
+        global_this: &JSGlobalObject,
+        err: JSValue,
+    ) -> bun_sys::Result<JSValue> {
+        if !err.is_empty_or_undefined_or_null() {
+            return self.fail_from_js(global_this, err);
+        }
         let _ = self.end(None);
         if self.end_promise.has_value() {
             // we are already waiting for the end
@@ -2403,6 +2413,67 @@ impl NetworkSink {
         bun_sys::Result::Ok(JSValue::js_number(0.0))
     }
 
+    /// Abort the upload with a caller-supplied error: reject any pending
+    /// promises with `err`, then drive the multipart task through `fail()`
+    /// so it cancels in-flight parts and issues AbortMultipartUpload.
+    fn fail_from_js(
+        &mut self,
+        global_this: &JSGlobalObject,
+        err: JSValue,
+    ) -> bun_sys::Result<JSValue> {
+        if self.ended {
+            if self.end_promise.has_value() {
+                return bun_sys::Result::Ok(self.end_promise.value());
+            }
+            return bun_sys::Result::Ok(
+                JSPromise::dangerously_create_rejected_promise_value_without_notifying_vm(
+                    global_this,
+                    err,
+                ),
+            );
+        }
+        self.ended = true;
+        self.done = true;
+        self.cancel = true;
+        self.signal.close(None);
+        if self.flush_promise.has_value() {
+            let _ = self.flush_promise.reject(global_this, Ok(err));
+        }
+        if !self.end_promise.has_value() {
+            self.end_promise = JSPromiseStrong::init(global_this);
+        }
+        let value = self.end_promise.value();
+        let _keep_value = bun_jsc::EnsureStillAlive(value);
+        let _ = self.end_promise.reject(global_this, Ok(err));
+        // Detach the task before driving `fail()`: the wrapper callback would
+        // form a second `&mut NetworkSink` while `&mut self` is live. Swap in
+        // a no-op callback so `fail()` cannot re-enter this sink at all.
+        if let Some(task) = self.task.take() {
+            fn noop(
+                _: bun_s3::S3UploadResult,
+                _: *mut c_void,
+            ) -> core::result::Result<(), jsc::JsTerminated> {
+                core::result::Result::Ok(())
+            }
+            let task_ptr = task.as_ptr();
+            // SAFETY: `task` was the sink's counted ref; it stays live until
+            // `deref_` below. Single JS thread, no other borrow at this site.
+            unsafe {
+                (*task_ptr).callback = noop;
+                (*task_ptr).callback_context = core::ptr::null_mut();
+                (*task_ptr).on_writable = None;
+                if (*task_ptr).state != bun_s3::MultiPartUploadState::Finished {
+                    let _ = (*task_ptr).fail(bun_s3::S3Error {
+                        code: b"UnknownError",
+                        message: b"The upload was aborted by the writer",
+                    });
+                }
+            }
+            bun_s3::MultiPartUpload::deref_(task_ptr);
+        }
+        bun_sys::Result::Ok(value)
+    }
+
     pub fn to_js(&mut self, global_this: &JSGlobalObject) -> JSValue {
         NetworkSinkJSSink::create_object(global_this, self, 0)
     }
@@ -2447,8 +2518,8 @@ impl crate::webcore::sink::JsSinkType for NetworkSink {
     fn end(&mut self, err: Option<SysError>) -> bun_sys::Result<()> {
         Self::end(self, err)
     }
-    fn end_from_js(&mut self, global: &JSGlobalObject) -> bun_sys::Result<JSValue> {
-        Self::end_from_js(self, global)
+    fn end_from_js(&mut self, global: &JSGlobalObject, err: JSValue) -> bun_sys::Result<JSValue> {
+        Self::end_from_js(self, global, err)
     }
     fn flush(&mut self) -> bun_sys::Result<()> {
         Self::flush(self)