s3: do not complete the upload when writer().end() gets an error

docs/runtime/s3.mdx

@@ -177,6 +177,25 @@ for (let i = 0; i < 10; i++) {
 await writer.end();
 ```
 
+`end()` stores the bytes written so far under the key, also when you call it from a `catch` block. If the source of the data fails before the data is complete, pass the error to `end()`. The argument must be `instanceof Error`. With any other value, `end()` completes the upload as it does without an argument.
+
+With an error, Bun does not complete the upload, so it does not create or replace the object under the key. An object that is already under the key stays as it is. Bun cancels the parts that wait in the queue. If a multipart upload has started, Bun sends `AbortMultipartUpload` for it. If S3 has not yet returned the upload id, Bun sends the abort when the id arrives. The promise that `end(error)` returns and a pending `flush()` reject with `error`. Bun marks these rejections as handled, so you do not have to await them.
+
+```ts s3.ts icon="/icons/typescript.svg"
+const writer = s3file.writer();
+try {
+  for await (const chunk of source) {
+    writer.write(chunk);
+    await writer.flush();
+  }
+} catch (error) {
+  // Discard the incomplete upload
+  writer.end(error instanceof Error ? error : new Error("The source failed", { cause: error }));
+  throw error;
+}
+await writer.end();
+```
+
 ---
 
 ## Presigning URLs

packages/bun-types/s3.d.ts

@@ -85,8 +85,38 @@ declare module "bun" {
     /**
      * Finish the upload. This also flushes the internal buffer.
      *
-     * @param error Optional error to associate with the end operation
+     * Without an argument, `end()` stores the bytes written so far under the
+     * key, also when you call it from a `catch` block.
+     *
+     * @param error Pass the error if the data is incomplete because its source
+     * failed. The value must be `instanceof Error`. With any other value,
+     * `end()` completes the upload as it does without an argument. With an
+     * error, the upload is not completed, so the object under the key is not
+     * created or replaced. An object that is already under the key stays as it
+     * is. The parts that wait in the queue are cancelled. If a multipart upload
+     * has started, `AbortMultipartUpload` is sent for it. If S3 has not yet
+     * returned the upload id, the abort is sent when the id arrives. The
+     * returned Promise and a pending `flush()` reject with `error`. These
+     * rejections count as handled, so you do not have to await them. After an
+     * earlier `end()`, or after the upload has failed, the argument has no
+     * effect.
      * @returns Number of bytes written or a Promise resolving to the number of bytes
+     *
+     * @example
+     * ```ts
+     * const writer = s3.file("backup.tar").writer();
+     * try {
+     *   for await (const chunk of source) {
+     *     writer.write(chunk);
+     *     await writer.flush();
+     *   }
+     * } catch (error) {
+     *   // Discard the incomplete upload: "backup.tar" is not created or replaced
+     *   writer.end(error instanceof Error ? error : new Error("The source failed", { cause: error }));
+     *   throw error;
+     * }
+     * await writer.end();
+     * ```
      */
     end(error?: Error): number | Promise<number>;
 

src/codegen/generate-jssink.ts

@@ -301,8 +301,10 @@ async function implementation() {
 #include "JSDOMWrapperCache.h"
 #include "ScriptExecutionContext.h"
 #include "WebCoreJSClientData.h"
+#include <JavaScriptCore/ErrorPrototype.h>
 #include <JavaScriptCore/FunctionPrototype.h>
 #include <JavaScriptCore/HeapAnalyzer.h>
+#include <JavaScriptCore/ProxyObject.h>
 
 #include <JavaScriptCore/JSDestructibleObjectHeapCellType.h>
 #include <JavaScriptCore/SlotVisitorMacros.h>
@@ -1017,6 +1019,28 @@ ${classes
     RELEASE_ASSERT_NOT_REACHED();
 }
 
+// Whether the argument of a sink's end() is an error: \`value instanceof Error\`,
+// for the Error of any realm. Only stored prototypes are read, and a Proxy is
+// replaced by its target, so no script runs. A DOMException and an ES5-style
+// error (\`util.inherits(MyError, Error)\`) are not ErrorInstance cells, but
+// their prototype chain has an Error.prototype.
+extern "C" bool JSSink__isErrorValue(JSC::EncodedJSValue encodedValue)
+{
+    JSC::JSValue value = JSC::JSValue::decode(encodedValue);
+    if (!value.isObject())
+        return false;
+
+    JSC::JSObject* object = asObject(value);
+    while (object->type() == JSC::ProxyObjectType)
+        object = uncheckedDowncast<JSC::ProxyObject>(object)->target();
+
+    for (; object; object = object->getPrototypeDirect().getObject()) {
+        if (object->type() == JSC::ErrorInstanceType || object->inherits<JSC::ErrorPrototype>())
+            return true;
+    }
+    return false;
+}
+
 extern "C" void JSSinkController__onReady(JSC::EncodedJSValue controllerValue, JSC::EncodedJSValue amt, JSC::EncodedJSValue offset)
 {
     (void)amt;

src/runtime/webcore/Sink.rs

@@ -330,6 +330,10 @@ fn start_pump(global: &JSGlobalObject, stream: JSValue, controller: JSValue) ->
     result
 }
 
+/// [`JsSinkType::END_WITH_ERROR_FROM_JS`]: the sink, the caller, the error.
+pub(crate) type EndWithErrorFromJs<T> =
+    unsafe fn(*mut T, &bun_jsc::JsThread<'_>, JSValue) -> bun_jsc::JsResult<JSValue>;
+
 /// Trait collecting every method `JSSink` may call on the wrapped `SinkType`.
 /// Most of these are optional, modeled with default method bodies and
 /// associated `const` gates.
@@ -339,6 +343,13 @@ pub(crate) trait JsSinkType: Sized + JsSinkAbi {
     const HAS_CONSTRUCT: bool = false;
     /// Mirrors `@hasDecl(SinkType, "flushFromJS")`.
     const HAS_FLUSH_FROM_JS: bool = false;
+    /// `end(error)` from JS with a value that is `instanceof Error`: the
+    /// caller's source failed, so the bytes written so far are a truncated
+    /// body. `None` keeps the clean end of `end_from_js` for any argument.
+    ///
+    /// The pointer is the cell's live sink. It is raw because failing can
+    /// re-enter the sink through its owner.
+    const END_WITH_ERROR_FROM_JS: Option<EndWithErrorFromJs<Self>> = None;
     /// Mirrors `@hasDecl(SinkType, "protectJSWrapper")`.
     const HAS_PROTECT_JS_WRAPPER: bool = false;
     /// Mirrors `@hasDecl(SinkType, "updateRef")`.
@@ -459,6 +470,13 @@ pub(crate) trait JsSinkType: Sized + JsSinkAbi {
 // no lut entry and no C++ caller.
 // ──────────────────────────────────────────────────────────────────────────
 
+unsafe extern "C" {
+    /// `value instanceof Error` for the Error of any realm, also through a
+    /// Proxy. Runs no script. Defined in `generate-jssink.ts`.
+    #[link_name = "JSSink__isErrorValue"]
+    safe fn is_error_value(value: JSValue) -> bool;
+}
+
 impl<T: JsSinkType> JSSink<T> {
     /// `JSSink.getThis` — recover `&mut JSSink<T>` from `callframe.this()` or
     /// throw the appropriate detached/cast-failed error.
@@ -682,6 +700,17 @@ impl<T: JsSinkType> JSSink<T> {
             return Err(global.throw_value(err));
         }
 
+        if let Some(end_with_error) = T::END_WITH_ERROR_FROM_JS {
+            // Only an error value: `pipeTo` of `node:stream/iter` ends its
+            // writer with an options object, and that is a clean end.
+            let err = frame.argument(0);
+            if is_error_value(err) {
+                let sink: *mut T = &raw mut this.sink;
+                // SAFETY: `sink` is the cell's live sink; `this` is not used again.
+                return unsafe { end_with_error(sink, &cx, err) };
+            }
+        }
+
         let result = match this.sink.end_from_js(&cx) {
             sys::Result::Ok(value) => Ok(value),
             sys::Result::Err(err) => Err(global.throw_value(err.to_js(global)?)),

src/runtime/webcore/streams.rs

@@ -2582,6 +2582,44 @@ impl NetworkSink {
         });
     }
 
+    /// `writer().end(error)`: the caller's source failed, so the upload is
+    /// aborted as in `fail_from_js_pump`. The returned promise and a pending
+    /// `flush()` reject with `error`. The caller already has that error, so
+    /// both rejections are marked handled: an `end(error)` or a `flush()`
+    /// that nobody awaits must not end the process.
+    ///
+    /// `error` is not stored on the sink. It can reference the writer, and a
+    /// `Strong` to it that nothing clears would keep both alive.
+    ///
+    /// Raw `*mut Self` for the same reason as `fail_from_js_pump`.
+    pub(crate) fn end_with_error_from_js(
+        this: *mut Self,
+        cx: &bun_jsc::JsThread<'_>,
+        err: JSValue,
+    ) -> JsResult<JSValue> {
+        let global = cx.global();
+        // SAFETY: `this` is the live sink the JS wrapper holds.
+        if unsafe { (*this).ended } {
+            // The outcome was decided by the earlier `end()` or failure.
+            // SAFETY: as above; `end_from_js` does not re-enter the sink.
+            return match unsafe { (*this).end_from_js(cx) } {
+                bun_sys::Result::Ok(value) => Ok(value),
+                bun_sys::Result::Err(e) => Err(e.throw(global)),
+            };
+        }
+        // Taken first, so that the failure callback of the upload does not
+        // reject it with the generic error that `fail()` is given.
+        // SAFETY: as above; the borrow ends before `fail_from_js_pump` re-enters.
+        let mut flush_promise = unsafe { (*this).flush_promise.take() };
+        Self::fail_from_js_pump(this, global, JSValue::ZERO);
+        if flush_promise.has_value() {
+            flush_promise.swap().reject_as_handled(global, err)?;
+        }
+        let promise = JSPromise::create(global);
+        promise.reject_as_handled(global, err)?;
+        Ok(promise.to_js())
+    }
+
     /// Native-path terminator called from `SinkHandle::end`. Unlike `end()`
     /// (clean EOF / commit), an upstream error on the ByteStream fast-path must
     /// abort the upload and surface the original JS error to the caller.
@@ -2683,6 +2721,8 @@ crate::impl_js_sink_abi!(NetworkSink, "NetworkSink");
 impl crate::webcore::sink::JsSinkType for NetworkSink {
     const NAME: &'static str = Self::NAME;
     const HAS_FLUSH_FROM_JS: bool = true;
+    const END_WITH_ERROR_FROM_JS: Option<crate::webcore::sink::EndWithErrorFromJs<Self>> =
+        Some(Self::end_with_error_from_js);
     const START_TAG: Option<StartTag> = Some(StartTag::NetworkSink);
 
     crate::impl_js_sink_forwarders!();