diff --git a/src/js/internal/streams/iter/pull.ts b/src/js/internal/streams/iter/pull.ts index 01e253b2503a..28c084361f50 100644 --- a/src/js/internal/streams/iter/pull.ts +++ b/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) { diff --git a/src/jsc/JSValue.rs b/src/jsc/JSValue.rs index d2e90883f262..6471c5f8ec41 100644 --- a/src/jsc/JSValue.rs +++ b/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() { diff --git a/src/jsc/bindings/bindings.cpp b/src/jsc/bindings/bindings.cpp index 7ea92a6e4e69..7864b47e3690 100644 --- a/src/jsc/bindings/bindings.cpp +++ b/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(); } - return type == JSC::ErrorInstanceType; + // DOMException uses JSC::ObjectType but chains its prototype through + // Error.prototype, so `instanceof Error` is true. + return cell->inherits(); } // This implementation closely mimics the one in JSC::JSPromise::reject diff --git a/src/runtime/webcore/ArrayBufferSink.rs b/src/runtime/webcore/ArrayBufferSink.rs index 4f0bd90b1a49..7aa52bc0ef0a 100644 --- a/src/runtime/webcore/ArrayBufferSink.rs +++ b/src/runtime/webcore/ArrayBufferSink.rs @@ -283,7 +283,7 @@ impl crate::webcore::sink::JsSinkType for ArrayBufferSink { fn end(&mut self, err: Option) -> bun_sys::Result<()> { Self::end(self, err) } - fn end_from_js(&mut self, global: &JSGlobalObject) -> bun_sys::Result { + fn end_from_js(&mut self, global: &JSGlobalObject, _err: JSValue) -> bun_sys::Result { 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, diff --git a/src/runtime/webcore/FileSink.rs b/src/runtime/webcore/FileSink.rs index 4a17e701fc50..2322cc6db0a9 100644 --- a/src/runtime/webcore/FileSink.rs +++ b/src/runtime/webcore/FileSink.rs @@ -1190,7 +1190,7 @@ impl crate::webcore::sink::JsSinkType for FileSink { fn end(&mut self, err: Option) -> sys::Result<()> { Self::end(self, err) } - fn end_from_js(&mut self, global: &JSGlobalObject) -> sys::Result { + fn end_from_js(&mut self, global: &JSGlobalObject, _err: JSValue) -> sys::Result { Self::end_from_js(self, global) } fn flush(&mut self) -> sys::Result<()> { diff --git a/src/runtime/webcore/Sink.rs b/src/runtime/webcore/Sink.rs index a6edf0c55f03..50b38a5020e6 100644 --- a/src/runtime/webcore/Sink.rs +++ b/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) -> sys::Result<()>; - fn end_from_js(&mut self, global: &JSGlobalObject) -> sys::Result; + fn end_from_js(&mut self, global: &JSGlobalObject, err: JSValue) -> sys::Result; fn flush(&mut self) -> sys::Result<()>; fn start(&mut self, config: streams::Start) -> sys::Result<()>; @@ -1010,7 +1010,10 @@ impl JSSink { 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 JSSink { } // 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) => { diff --git a/src/runtime/webcore/s3/multipart.rs b/src/runtime/webcore/s3/multipart.rs index d8b4d16bd1ec..8ea8bdd01d52 100644 --- a/src/runtime/webcore/s3/multipart.rs +++ b/src/runtime/webcore/s3/multipart.rs @@ -657,6 +657,23 @@ impl MultiPartUpload { } } + /// Extract and validate `` from an InitiateMultipartUpload body. + fn parse_upload_id(body: &[u8]) -> Option> { + let start = strings::index_of(body, b"")? + b"".len(); + let end = strings::index_of(body, b"")?; + 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 - 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"") { - let value_start = start + b"".len(); - if let Some(end) = strings::index_of(slice, b"") { - 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: {}", diff --git a/src/runtime/webcore/streams.rs b/src/runtime/webcore/streams.rs index 0d86d23081bf..2eb6afd04b8d 100644 --- a/src/runtime/webcore/streams.rs +++ b/src/runtime/webcore/streams.rs @@ -31,6 +31,9 @@ pub type ByteListPoolNode = bun_collections::pool::Node>; // 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 crate::webcore::sink::JsSinkType fn end(&mut self, err: Option) -> bun_sys::Result<()> { Self::end(self, err) } - fn end_from_js(&mut self, global: &JSGlobalObject) -> bun_sys::Result { + fn end_from_js(&mut self, global: &JSGlobalObject, _err: JSValue) -> bun_sys::Result { 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 { + pub fn end_from_js( + &mut self, + global_this: &JSGlobalObject, + err: JSValue, + ) -> bun_sys::Result { + // 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 { + 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) -> bun_sys::Result<()> { Self::end(self, err) } - fn end_from_js(&mut self, global: &JSGlobalObject) -> bun_sys::Result { - Self::end_from_js(self, global) + fn end_from_js(&mut self, global: &JSGlobalObject, err: JSValue) -> bun_sys::Result { + Self::end_from_js(self, global, err) } fn flush(&mut self) -> bun_sys::Result<()> { Self::flush(self) diff --git a/test/js/bun/s3/s3-writer-end-error.test.ts b/test/js/bun/s3/s3-writer-end-error.test.ts new file mode 100644 index 000000000000..bd12d4a19f47 --- /dev/null +++ b/test/js/bun/s3/s3-writer-end-error.test.ts @@ -0,0 +1,263 @@ +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(); +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 = 'bucketk' + id + ''; + } 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 = 'k"e"'; + } 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(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) { + 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 = {}; + +{ + // 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") }; +} + +{ + // end(error) in the same JS turn as the write that triggered the init + // request; the UploadId arrives after fail() and must still be rolled back. + reqs.length = 0; + const w = s3.file("race.bin").writer({ partSize: PART, queueSize: 1, retry: 0 }); + w.write(new Uint8Array(PART)); + 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.race = { outcome, ...summary("race.bin") }; +} + +{ + // DOMException (the default AbortSignal.reason type) must abort like any Error. + reqs.length = 0; + const w = s3.file("domex.bin").writer({ partSize: PART, queueSize: 1, retry: 0 }); + w.write(new Uint8Array(PART)); + await w.flush(); + w.write(new Uint8Array(100)); + const ac = new AbortController(); + ac.abort(new DOMException("source failed mid-stream", "AbortError")); + const outcome = await settle(w.end(ac.signal.reason)); + await waitFor(() => reqs.some(r => r.startsWith("ABORT ") || r.startsWith("COMMIT "))); + results.domex = { outcome, ...summary("domex.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") }; +} + +{ + // Control: a non-Error argument (e.g. an options bag) is not treated as an + // abort request and the upload commits. + reqs.length = 0; + const w = s3.file("opts.bin").writer({ partSize: PART, queueSize: 1, retry: 0 }); + w.write(new Uint8Array(PART)); + w.write(new Uint8Array(100)); + const outcome = await settle((w.end as any)({ signal: new AbortController().signal })); + await waitFor(() => reqs.some(r => r.startsWith("COMMIT ") || r.startsWith("ABORT "))); + results.opts = { outcome, ...summary("opts.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, + // LSan flags pre-existing transpiler/sourcemap leaks from the -e fixture + // itself; unrelated to S3, so don't let it abort the child. + ASAN_OPTIONS: [bunEnv.ASAN_OPTIONS, "symbolize=0", "detect_leaks=0"].filter(Boolean).join(":"), + }, + 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, + }); + + // end(error) in the same turn as the write that dispatched the init request + // must still roll back the UploadId once it arrives. + expect(result.race).toEqual({ + outcome: "rejected:source failed mid-stream", + committed: false, + commits: 0, + puts: 0, + aborts: 1, + inits: 1, + }); + + // DOMException (AbortSignal.reason) must abort like any Error. + expect(result.domex).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, + }); + + // end() with a non-Error argument commits normally. + expect(result.opts).toEqual({ + outcome: "resolved", + committed: true, + commits: 1, + puts: 0, + aborts: 0, + inits: 1, + }); + + expect(exitCode).toBe(0); +});