s3: throw the upload error from writer() calls after a failed upload

src/runtime/webcore/Sink.rs

@@ -729,18 +729,8 @@ impl<T: JsSinkType> JSSink<T> {
         use bun_sys_jsc::ErrorJsc;
         bun_core::mark_binding!();
 
-        // SAFETY: caller contract; the borrow ends before `close_with_error`,
-        // which may re-enter or free the sink.
-        if let Some(err) = unsafe { (*this).get_pending_error() } {
-            // `throw_error` sets the pending JS exception and returns the
-            // `JsError` for `?`-propagation; this host fn returns bare
-            // `JSValue`, so report and return ZERO (caller checks exception).
-            let _ = global.vm().throw_error(global, err);
-            return JSValue::ZERO;
-        }
-
         let result = if reason.is_empty() {
-            // SAFETY: as above; `end` does not free the sink.
+            // SAFETY: caller contract; `end` does not free the sink.
             unsafe { (*this).end(None) }
         } else {
             // SAFETY: caller contract.
@@ -889,9 +879,7 @@ pub(crate) unsafe fn sink_handle_from_id(
 ///
 /// Return shape matches [`streams::result::Writable::to_js`] so
 /// `nativeSinkWriteIsBackpressure` reads a negative number / pending promise
-/// exactly as the previous `js_write_bytes` path produced. No
-/// [`JsSinkType::get_pending_error`] guard: every sink uses the trait-default
-/// `None`, so omitting it is behavior-preserving.
+/// exactly as the previous `js_write_bytes` path produced.
 #[unsafe(no_mangle)]
 #[allow(clippy::not_unsafe_ptr_arg_deref)]
 pub extern "C" fn Bun__NativeTransformSink__writeBytes(

src/runtime/webcore/s3/client.rs

@@ -431,34 +431,36 @@ pub(crate) fn writable_stream(
             .global_this
             .expect("NetworkSink.global_this set at construction");
         let global = global.get();
-        if sink.end_promise.has_value() || sink.flush_promise.has_value() {
+        let has_pending_promise = sink.end_promise.has_value() || sink.flush_promise.has_value();
+        let _exit_guard = has_pending_promise.then(|| {
             // SAFETY: `bun_vm()` returns the live per-thread VM pointer.
             let event_loop = global.bun_vm().as_mut().event_loop();
             // SAFETY: event_loop is initialised for the lifetime of the VM.
             // RAII: `enter()` now, `exit()` on drop.
-            let _exit_guard = unsafe { bun_jsc::event_loop::EventLoop::enter_scope(event_loop) };
-            match result {
-                S3UploadResult::Success => {
-                    if sink.flush_promise.has_value() {
-                        sink.flush_promise
-                            .resolve(global, JSValue::js_number(0.0))?;
-                    }
-                    if sink.end_promise.has_value() {
-                        sink.end_promise
-                            .resolve(global, JSValue::js_number(uploaded as f64))?;
-                    }
+            unsafe { bun_jsc::event_loop::EventLoop::enter_scope(event_loop) }
+        });
+        match result {
+            S3UploadResult::Success => {
+                if sink.flush_promise.has_value() {
+                    sink.flush_promise
+                        .resolve(global, JSValue::js_number(0.0))?;
                 }
-                S3UploadResult::Failure(err) => {
-                    let js_err = s3_error_to_js(&err, global, sink.path());
-                    if sink.flush_promise.has_value() {
-                        sink.flush_promise.reject(global, Ok(js_err))?;
-                    }
-                    if sink.end_promise.has_value() {
-                        sink.end_promise.reject(global, Ok(js_err))?;
-                    }
-                    if !sink.done {
-                        sink.abort();
-                    }
+                if sink.end_promise.has_value() {
+                    sink.end_promise
+                        .resolve(global, JSValue::js_number(uploaded as f64))?;
+                }
+            }
+            S3UploadResult::Failure(err) => {
+                let js_err = s3_error_to_js(&err, global, sink.path());
+                sink.pending_error.set(global, js_err);
+                if sink.flush_promise.has_value() {
+                    sink.flush_promise.reject(global, Ok(js_err))?;
+                }
+                if sink.end_promise.has_value() {
+                    sink.end_promise.reject(global, Ok(js_err))?;
+                }
+                if !sink.done {
+                    sink.abort();
                 }
             }
         }

src/runtime/webcore/streams.rs

@@ -2283,6 +2283,8 @@ pub struct NetworkSink {
     /// failure callback can reject with the original JS error (e.g. S3
     /// `NoSuchKey`) instead of the generic `UnknownError` passed to `fail()`.
     pub(crate) upstream_error: jsc::strong::Optional,
+    /// Error of a failed `writer()` upload, returned by `get_pending_error`.
+    pub(crate) pending_error: jsc::strong::Optional,
     pub(crate) ended: bool,
     pub(crate) done: bool,
     /// `s3file.writer()`: the box is referenced by the JS wrapper (`finalize`) and by the upload's
@@ -2301,6 +2303,7 @@ impl Default for NetworkSink {
             pending: WritablePending::default(),
             end_promise: JSPromiseStrong::default(),
             upstream_error: jsc::strong::Optional::empty(),
+            pending_error: jsc::strong::Optional::empty(),
             ended: false,
             done: false,
             writer_holders: core::cell::Cell::new(0),
@@ -2648,8 +2651,11 @@ impl NetworkSink {
         unsafe { crate::webcore::s3::client::S3UploadStreamWrapper::deref(wrapper) };
     }
 
-    pub(crate) fn end_from_js(&mut self, _cx: &bun_jsc::JsThread<'_>) -> bun_sys::Result<JSValue> {
+    pub(crate) fn end_from_js(&mut self, cx: &bun_jsc::JsThread<'_>) -> bun_sys::Result<JSValue> {
         let _ = self.end(None);
+        if let Some(err) = self.pending_error.get() {
+            return bun_sys::Result::Ok(JSPromise::rejected_promise(cx.global(), err).to_js());
+        }
         if self.end_promise.has_value() {
             // we are already waiting for the end
             return bun_sys::Result::Ok(self.end_promise.value());
@@ -2706,6 +2712,9 @@ impl crate::webcore::sink::JsSinkType for NetworkSink {
     fn end_from_js(&mut self, cx: &bun_jsc::JsThread<'_>) -> bun_sys::Result<JSValue> {
         Self::end_from_js(self, cx)
     }
+    fn get_pending_error(&mut self) -> Option<JSValue> {
+        self.pending_error.get()
+    }
     fn source(&mut self) -> Option<&mut SourceHandle> {
         Some(&mut self.source)
     }

test/js/bun/s3/s3.test.ts

@@ -2133,6 +2133,81 @@ describe("s3 upload stream body error", () => {
   });
 });
 
+describe.concurrent("s3 writer upload failure", () => {
+  // A mock S3 that answers 403 to every UploadPart. `abortSeen` settles when
+  // AbortMultipartUpload arrives, which the upload sends after it reported the
+  // failure to the sink.
+  function failingS3(abortSeen: PromiseWithResolvers<void>) {
+    return Bun.serve({
+      port: 0,
+      async fetch(req) {
+        const q = new URL(req.url).searchParams;
+        if (req.method === "POST" && q.has("uploads")) {
+          return new Response(
+            "<InitiateMultipartUploadResult><Bucket>b</Bucket><Key>k</Key><UploadId>UP1</UploadId></InitiateMultipartUploadResult>",
+          );
+        }
+        if (req.method === "PUT" && q.has("partNumber")) {
+          await req.arrayBuffer();
+          return new Response("<Error><Code>AccessDenied</Code><Message>Access Denied</Message></Error>", {
+            status: 403,
+          });
+        }
+        if (req.method === "DELETE") {
+          abortSeen.resolve();
+          return new Response(null, { status: 204 });
+        }
+        return new Response("", { headers: { ETag: '"e"' } });
+      },
+    });
+  }
+  const accessDenied = expect.objectContaining({ code: "AccessDenied" });
+
+  it("throws the S3 error from write(), flush() and end() when no promise was pending", async () => {
+    const abortSeen = Promise.withResolvers<void>();
+    await using server = failingS3(abortSeen);
+    const client = new S3Client({ endpoint: server.url.href, accessKeyId: "a", secretAccessKey: "b", bucket: "bkt" });
+    const writer = client.file("obj.bin").writer({ partSize: 5 * 1024 * 1024, retry: 0 });
+    // A full part starts the multipart upload. No promise is kept.
+    writer.write(new Uint8Array(5 * 1024 * 1024));
+    await abortSeen.promise;
+
+    expect(() => writer.write(new Uint8Array(1024))).toThrow(accessDenied);
+    expect(() => writer.flush()).toThrow(accessDenied);
+    expect(() => writer.end()).toThrow(accessDenied);
+    // close() is cleanup. It does not throw the error again.
+    expect((writer as any).close()).toBeUndefined();
+  });
+
+  it("rejects the pending flush() and end() and throws from later write() and end()", async () => {
+    const abortSeen = Promise.withResolvers<void>();
+    await using server = failingS3(abortSeen);
+    const client = new S3Client({ endpoint: server.url.href, accessKeyId: "a", secretAccessKey: "b", bucket: "bkt" });
+    const writer = client.file("obj.bin").writer({ partSize: 5 * 1024 * 1024, retry: 0 });
+    writer.write(new Uint8Array(5 * 1024 * 1024));
+    const pendingFlush = writer.flush().then(
+      () => "resolved",
+      e => e,
+    );
+    await expect(writer.end()).rejects.toEqual(accessDenied);
+    expect(await pendingFlush).toEqual(accessDenied);
+    await abortSeen.promise;
+
+    expect(() => writer.write(new Uint8Array(1024))).toThrow(accessDenied);
+    expect(() => writer.end()).toThrow(accessDenied);
+    expect((writer as any).close()).toBeUndefined();
+  });
+
+  it("rejects end() when the upload fails inside end() itself", async () => {
+    // Under partSize, end() signs and sends the single PUT. Without credentials
+    // the signing fails before any request, so the failure is synchronous.
+    const writer = new S3Client({ bucket: "bkt" }).file("obj.bin").writer();
+    writer.write("hello");
+    await expect(writer.end()).rejects.toEqual(expect.objectContaining({ code: "ERR_S3_MISSING_CREDENTIALS" }));
+    expect((writer as any).close()).toBeUndefined();
+  });
+});
+
 describe("presigned url signature", () => {
   function verifyPresignedUrl(presigned: string, credentials: { secretAccessKey: string; region: string }) {
     const url = new URL(presigned);