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

src/js/internal/streams/iter/pull.ts

@@ -954,7 +954,7 @@ async function pipeTo(source, ...args) {
 
     if (!options?.preventClose) {
       if (!hasEndSync || writer.endSync() < 0) {
-        await writer.end?.(signal ? { __proto__: null, signal } : undefined);
+        await writer.end?.();
       }
     }
   } catch (error) {

src/jsc/JSValue.rs

@@ -354,7 +354,7 @@ impl JSValue {
     pub fn is_function(self) -> bool {
         self.is_cell() && self.js_type().is_function()
     }
-    /// `JSValue.isAnyError()` — Error, Exception, or has `[Symbol.error]`.
+    /// `JSValue.isAnyError()` — Error, Exception, or DOMException.
     #[inline]
     pub fn is_any_error(self) -> bool {
         if !self.is_cell() {

src/jsc/bindings/bindings.cpp

@@ -115,6 +115,7 @@
 #include "JSDOMConvertSequences.h"
 #include "JSDOMConvertStrings.h"
 #include "JSDOMConvertUnion.h"
+#include "JSDOMException.h"
 #include "JSDOMExceptionHandling.h"
 #include "JSDOMGlobalObjectInlines.h"
 #include "JSDOMIterator.h"
@@ -3626,11 +3627,17 @@ bool JSC__JSValue__isAnyError(JSC::EncodedJSValue JSValue0)
     JSC::JSCell* cell = value.asCell();
     JSC::JSType type = cell->type();
 
+    if (type == JSC::ErrorInstanceType) {
+        return true;
+    }
+
     if (type == JSC::CellType) {
         return cell->inherits<JSC::Exception>();
     }
 
-    return type == JSC::ErrorInstanceType;
+    // DOMException uses JSC::ObjectType but chains its prototype through
+    // Error.prototype, so `instanceof Error` is true.
+    return cell->inherits<WebCore::JSDOMException>();
 }
 
 // This implementation closely mimics the one in JSC::JSPromise::reject

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

@@ -657,6 +657,23 @@ impl MultiPartUpload {
         }
     }
 
+    /// Extract and validate `<UploadId>` from an InitiateMultipartUpload body.
+    fn parse_upload_id(body: &[u8]) -> Option<Box<[u8]>> {
+        let start = strings::index_of(body, b"<UploadId>")? + b"<UploadId>".len();
+        let end = strings::index_of(body, b"</UploadId>")?;
+        if end < start {
+            return None;
+        }
+        let id = &body[start..end];
+        if id.is_empty()
+            || id.len() > Self::MAX_UPLOAD_ID_LEN
+            || id.iter().any(|b| !b.is_ascii() || b.is_ascii_control())
+        {
+            return None;
+        }
+        Some(Box::from(id))
+    }
+
     /// Result of the Multipart request, after this we can start draining the parts
     pub fn start_multi_part_request_result(
         result: S3DownloadResult,
@@ -669,6 +686,16 @@ impl MultiPartUpload {
         // 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 {
+                if let Some(id) = Self::parse_upload_id(response.body.list.as_slice()) {
+                    this.upload_id = id;
+                    // rollback consumes one ref via its completion callback
+                    this.ref_();
+                    return this.rollback_multi_part_request();
+                }
+            }
             return Ok(());
         }
         match result {
@@ -684,24 +711,9 @@ impl MultiPartUpload {
             }
             S3DownloadResult::Success(response) => {
                 // response.body is bun.MutableString — `list` is a Vec<u8>
-                let slice = response.body.list.as_slice();
-                // PERF: upload_id is duped out of the body instead of slicing into it
-                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]);
-                        }
-                    }
-                }
+                let upload_id = Self::parse_upload_id(response.body.list.as_slice());
                 this.uploadid_buffer = response.body;
-                if this.upload_id.is_empty()
-                    || this.upload_id.len() > Self::MAX_UPLOAD_ID_LEN
-                    || this
-                        .upload_id
-                        .iter()
-                        .any(|b| !b.is_ascii() || b.is_ascii_control())
-                {
+                let Some(upload_id) = upload_id else {
                     // Unknown type of response error from AWS
                     scoped_log!(
                         S3MultiPartUpload,
@@ -713,7 +725,8 @@ impl MultiPartUpload {
                         message: b"Failed to initiate multipart upload",
                     })?;
                     return Ok(());
-                }
+                };
+                this.upload_id = upload_id;
                 scoped_log!(
                     S3MultiPartUpload,
                     "startMultiPartRequestResult {} success id: {}",

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,16 @@ 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> {
+        // Only an Error-like value aborts; other argument shapes (e.g. option
+        // bags) are ignored so callers that pass them still commit.
+        if err.is_any_error() {
+            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 +2415,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 +2520,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)