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/streams.rs

@@ -31,6 +31,8 @@
 // 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 bun_s3_signing::error::S3Error;
 }
 
 /// `Blob.SizeType` is `u64` (see `webcore::blob::SizeType`).
@@ -2089,7 +2091,7 @@
     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 +2381,14 @@
         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 +2412,53 @@
         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);
+        // Reject with the caller's error before `fail()` reaches the wrapper
+        // callback; `reject()` swaps the Strong to empty so the callback's
+        // Failure branch is a no-op and only `finalize()` runs.
+        let _ = self.end_promise.reject(global_this, Ok(err));
+        if self
+            .task_mut()
+            .is_some_and(|t| t.state != bun_s3::MultiPartUploadState::Finished)
+        {
+            let _ = self.task_mut().unwrap().fail(bun_s3::S3Error {
+                code: b"UnknownError",
+                message: b"The upload was aborted by the writer",
+            });
+        }
+        bun_sys::Result::Ok(value)
+    }
+
     pub fn to_js(&mut self, global_this: &JSGlobalObject) -> JSValue {
         NetworkSinkJSSink::create_object(global_this, self, 0)
     }
@@ -2447,8 +2503,8 @@
     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)

test/js/bun/s3/s3-writer-end-error.test.ts

@@ -0,0 +1,191 @@
+import { expect, test } from "bun:test";
+import { bunEnv, bunExe } from "harness";
+
+// S3File.writer().end(new Error(...)) must abort the multipart upload and
+// reject. Previously the error argument was ignored: the buffered tail was
+// uploaded, CompleteMultipartUpload was sent, and the promise resolved,
+// silently publishing a truncated object.
+
+const fixture = `
+import * as net from "node:net";
+
+const reqs: string[] = [];
+const committed = new Set<string>();
+let nextId = 0;
+
+const server = net.createServer(sock => {
+  let buf = Buffer.alloc(0);
+  sock.on("error", () => {});
+  sock.on("data", chunk => {
+    buf = Buffer.concat([buf, chunk]);
+    for (;;) {
+      const headerEnd = buf.indexOf("\\r\\n\\r\\n");
+      if (headerEnd < 0) return;
+      const head = buf.toString("latin1", 0, headerEnd);
+      const len = Number(/^content-length: *(\\d+)/im.exec(head)?.[1] ?? 0);
+      if (buf.length < headerEnd + 4 + len) return;
+      const body = Buffer.from(buf.subarray(headerEnd + 4, headerEnd + 4 + len));
+      buf = buf.subarray(headerEnd + 4 + len);
+      const [method, target] = head.split("\\r\\n")[0].split(" ");
+      const key = decodeURIComponent(target.split("?")[0]).replace(/^\\/bucket\\//, "");
+      const q = new URLSearchParams(target.split("?")[1] ?? "");
+      let status = 200, out = "", extra = "";
+      if (method === "POST" && q.has("uploads")) {
+        const id = "up" + ++nextId;
+        reqs.push("INIT " + key);
+        out = '<?xml version="1.0"?><InitiateMultipartUploadResult><Bucket>bucket</Bucket><Key>k</Key><UploadId>' + id + '</UploadId></InitiateMultipartUploadResult>';
+      } else if (method === "PUT" && q.has("partNumber")) {
+        reqs.push("PART " + key + " " + q.get("partNumber") + " " + body.length);
+        extra = 'ETag: "p' + q.get("partNumber") + '"\\r\\n';
+      } else if (method === "POST" && q.has("uploadId")) {
+        reqs.push("COMMIT " + key);
+        committed.add(key);
+        out = '<?xml version="1.0"?><CompleteMultipartUploadResult><Key>k</Key><ETag>"e"</ETag></CompleteMultipartUploadResult>';
+      } else if (method === "DELETE" && q.has("uploadId")) {
+        reqs.push("ABORT " + key);
+        status = 204;
+      } else if (method === "PUT") {
+        reqs.push("PUT " + key + " " + body.length);
+        committed.add(key);
+      }
+      const b = Buffer.from(out);
+      sock.write("HTTP/1.1 " + status + " X\\r\\n" + extra + "Connection: keep-alive\\r\\nContent-Length: " + (status === 204 ? 0 : b.length) + "\\r\\n\\r\\n");
+      if (status !== 204 && b.length) sock.write(b);
+    }
+  });
+});
+await new Promise<void>(r => server.listen(0, "127.0.0.1", () => r()));
+const port = (server.address() as net.AddressInfo).port;
+
+const s3 = new Bun.S3Client({
+  endpoint: "http://127.0.0.1:" + port,
+  bucket: "bucket",
+  accessKeyId: "AK",
+  secretAccessKey: "SK",
+  region: "us-east-1",
+});
+const PART = 5 * 1024 * 1024;
+
+async function settle(p: Promise<unknown>) {
+  try {
+    await p;
+    return "resolved";
+  } catch (e: any) {
+    return "rejected:" + e.message;
+  }
+}
+
+function summary(key: string) {
+  return {
+    committed: committed.has(key),
+    commits: reqs.filter(r => r.startsWith("COMMIT ")).length,
+    puts: reqs.filter(r => r.startsWith("PUT ")).length,
+    aborts: reqs.filter(r => r.startsWith("ABORT ")).length,
+    inits: reqs.filter(r => r.startsWith("INIT ")).length,
+  };
+}
+
+async function waitFor(predicate: () => boolean, limit = 500) {
+  for (let i = 0; i < limit && !predicate(); i++) await Bun.sleep(10);
+}
+
+const results: Record<string, unknown> = {};
+
+{
+  // Multipart already initiated and a part uploaded; then the source fails.
+  reqs.length = 0;
+  const w = s3.file("multi.bin").writer({ partSize: PART, queueSize: 1, retry: 0 });
+  w.write(new Uint8Array(PART));
+  await w.flush();
+  w.write(new Uint8Array(100));
+  const outcome = await settle(w.end(new Error("source failed mid-stream")));
+  await waitFor(() => reqs.some(r => r.startsWith("ABORT ") || r.startsWith("COMMIT ")));
+  results.multipart = { outcome, ...summary("multi.bin") };
+}
+
+{
+  // Buffered data below partSize; multipart never started.
+  reqs.length = 0;
+  const w = s3.file("single.bin").writer({ partSize: PART, queueSize: 1, retry: 0 });
+  w.write(new Uint8Array(100));
+  const outcome = await settle(w.end(new Error("source failed mid-stream")));
+  // A buggy build uploads the buffered bytes as a single-file PUT and only
+  // then settles, so reqs already reflects it here. Give any late request a
+  // bounded window to appear; the fixed build leaves reqs empty.
+  await waitFor(() => reqs.length > 0, 50);
+  results.single = { outcome, ...summary("single.bin") };
+}
+
+{
+  // Control: end() with no error still commits.
+  reqs.length = 0;
+  const w = s3.file("ok.bin").writer({ partSize: PART, queueSize: 1, retry: 0 });
+  w.write(new Uint8Array(PART));
+  w.write(new Uint8Array(100));
+  const outcome = await settle(w.end());
+  await waitFor(() => reqs.some(r => r.startsWith("COMMIT ")));
+  results.ok = { outcome, ...summary("ok.bin") };
+}
+
+console.log(JSON.stringify(results));
+server.close();
+process.exit(0);
+`;
+
+test("S3File.writer().end(error) aborts the upload and rejects", async () => {
+  await using proc = Bun.spawn({
+    cmd: [bunExe(), "-e", fixture],
+    env: {
+      ...bunEnv,
+      HTTP_PROXY: undefined,
+      HTTPS_PROXY: undefined,
+      http_proxy: undefined,
+      https_proxy: undefined,
+    },
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+
+  let result;
+  try {
+    result = JSON.parse(stdout.trim());
+  } catch {
+    throw new Error(`fixture did not emit JSON\nstdout: ${stdout}\nstderr: ${stderr}`);
+  }
+
+  // After a part has been uploaded, end(error) must reject with the caller's
+  // error, send AbortMultipartUpload, and never send CompleteMultipartUpload.
+  expect(result.multipart).toEqual({
+    outcome: "rejected:source failed mid-stream",
+    committed: false,
+    commits: 0,
+    puts: 0,
+    aborts: 1,
+    inits: 1,
+  });
+
+  // Before anything is sent, end(error) must reject without uploading the
+  // buffered bytes as a single-file PUT.
+  expect(result.single).toEqual({
+    outcome: "rejected:source failed mid-stream",
+    committed: false,
+    commits: 0,
+    puts: 0,
+    aborts: 0,
+    inits: 0,
+  });
+
+  // end() with no error still commits normally.
+  expect(result.ok).toEqual({
+    outcome: "resolved",
+    committed: true,
+    commits: 1,
+    puts: 0,
+    aborts: 0,
+    inits: 1,
+  });
+
+  expect(exitCode).toBe(0);
+});