diff --git a/src/runtime/webcore/Blob.rs b/src/runtime/webcore/Blob.rs index 48af5644a8bf..e162dd37650e 100644 --- a/src/runtime/webcore/Blob.rs +++ b/src/runtime/webcore/Blob.rs @@ -1663,12 +1663,14 @@ impl BlobExt for Blob { return Ok(promise_value); } jsc::js_promise::Status::Fulfilled => { + // SAFETY: `file_sink` holds our +1 ref; live until deref below. + let written = unsafe { (*file_sink).written.get() }; // SAFETY: release our +1 ref on the sink. unsafe { webcore::FileSink::deref(file_sink) }; readable_stream.done(global_this); return Ok(JSPromise::resolved_promise_value( global_this, - JSValue::js_number(0.0), + JSValue::js_number(written as f64), )); } jsc::js_promise::Status::Rejected => { @@ -1693,12 +1695,14 @@ impl BlobExt for Blob { ); } } + // SAFETY: `file_sink` holds our +1 ref; live until deref below. + let written = unsafe { (*file_sink).written.get() }; // SAFETY: release our +1 ref on the sink. unsafe { webcore::FileSink::deref(file_sink) }; Ok(JSPromise::resolved_promise_value( global_this, - JSValue::js_number(0.0), + JSValue::js_number(written as f64), )) } @@ -5959,7 +5963,10 @@ pub fn on_file_stream_resolve_request_stream( if let Some(stream) = strong.get(global_this) { stream.done(global_this); } - this.promise.resolve(global_this, JSValue::js_number(0.0))?; + // SAFETY: `this.sink` is the +1 ref held until `Drop` below; live here. + let written = unsafe { (*this.sink).written.get() }; + this.promise + .resolve(global_this, JSValue::js_number(written as f64))?; Ok(JSValue::UNDEFINED) } diff --git a/src/runtime/webcore/s3/client.rs b/src/runtime/webcore/s3/client.rs index 80271e2b89eb..710768d6f629 100644 --- a/src/runtime/webcore/s3/client.rs +++ b/src/runtime/webcore/s3/client.rs @@ -445,12 +445,14 @@ pub(crate) fn writable_stream( let _exit_guard = unsafe { bun_jsc::event_loop::EventLoop::enter_scope(event_loop) }; match result { S3UploadResult::Success => { + // flush() keeps its delta contract: 0 = nothing left pending. 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(0.0))?; + sink.end_promise + .resolve(global, JSValue::js_number(sink.wrote as f64))?; } } S3UploadResult::Failure(err) => { @@ -498,6 +500,7 @@ pub(crate) fn writable_stream( available: IntegerBitSet::init_full(), current_part_number: 1, ref_count: core::cell::Cell::new(2), // +1 for the stream + uploaded: 0, ended: false, options, acl: None, @@ -675,9 +678,10 @@ impl S3UploadStreamWrapper { match &result { S3UploadResult::Success => { if self_.end_promise.has_value() { + let uploaded = self_.task_mut().uploaded; self_ .end_promise - .resolve(&self_.global, JSValue::js_number(0.0))?; + .resolve(&self_.global, JSValue::js_number(uploaded as f64))?; self_.end_promise = bun_jsc::JSPromiseStrong::empty(); } } @@ -852,6 +856,7 @@ pub fn upload_stream( available: IntegerBitSet::init_full(), current_part_number: 1, ref_count: core::cell::Cell::new(2), // +1 for the stream ctx (only deinit after task and context ended) + uploaded: 0, ended: false, options, acl, diff --git a/src/runtime/webcore/s3/multipart.rs b/src/runtime/webcore/s3/multipart.rs index 16c37a2af1e0..1dea93636c4e 100644 --- a/src/runtime/webcore/s3/multipart.rs +++ b/src/runtime/webcore/s3/multipart.rs @@ -129,6 +129,8 @@ pub struct MultiPartUpload { pub current_part_number: u16, pub ref_count: Cell, // intrusive refcount — see bun_ptr::IntrusiveRc + /// Server-acknowledged payload bytes; resolves `S3File.write()` / `.writer().end()`. + pub uploaded: u64, pub ended: bool, pub options: MultiPartUploadOptions, @@ -308,6 +310,7 @@ impl UploadPart { this.part_number ); let sent = this.data().len(); + ctx.uploaded += sent as u64; this.free_allocated_slice(); // we will need to order this ctx.multipart_etags.push(UploadPartResult { @@ -459,8 +462,10 @@ impl MultiPartUpload { S3UploadResult::Success => { scoped_log!(S3MultiPartUpload, "singleSendUploadResponse success"); + let sent = this.buffered.size() as u64; + this.uploaded += sent; if let Some(callback) = this.on_writable { - callback(this, this.callback_context, this.buffered.size() as u64); + callback(this, this.callback_context, sent); } this.done() } diff --git a/src/runtime/webcore/streams.rs b/src/runtime/webcore/streams.rs index d4dfdc47f1cf..2fff12c6aac9 100644 --- a/src/runtime/webcore/streams.rs +++ b/src/runtime/webcore/streams.rs @@ -2047,6 +2047,8 @@ pub struct NetworkSink { // JSC_BORROW: process-lifetime VM global; safe `Deref` via `BackRef`. pub global_this: Option>, pub high_water_mark: BlobSizeType, + /// Mirrors `MultiPartUpload.uploaded`; survives `task` detaching before completion. + pub wrote: u64, pub flush_promise: JSPromiseStrong, pub end_promise: JSPromiseStrong, pub ended: bool, @@ -2061,6 +2063,7 @@ impl Default for NetworkSink { signal: Signal::default(), global_this: None, high_water_mark: 2048, + wrote: 0, flush_promise: JSPromiseStrong::default(), end_promise: JSPromiseStrong::default(), ended: false, @@ -2155,8 +2158,10 @@ impl NetworkSink { flushed, task.state as u8 ); + this.wrote = task.uploaded; if this.flush_promise.has_value() { let global = this.global_this.expect("global_this set at construction"); + // flush() resolves with the per-call delta; only end() reports cumulative `wrote`. this.flush_promise .resolve(&global, JSValue::js_number(flushed as f64))?; } @@ -2274,27 +2279,21 @@ impl NetworkSink { } pub fn end_from_js(&mut self, _global_this: &JSGlobalObject) -> bun_sys::Result { - let _ = self.end(None); if self.end_promise.has_value() { // we are already waiting for the end + let _ = self.end(None); return bun_sys::Result::Ok(self.end_promise.value()); } if self.task.is_some() { - // we need to wait for the task to end + // end() may complete synchronously; the promise must exist before it runs. self.end_promise = JSPromiseStrong::init(self.global_this()); let value = self.end_promise.value(); - if !self.ended { - self.ended = true; - // we need to send EOF - if let Some(task) = self.task_mut() { - let _ = task.write_bytes(b"", true); - } - self.signal.close(None); - } + let _ = self.end(None); return bun_sys::Result::Ok(value); } + let _ = self.end(None); // task already detached - bun_sys::Result::Ok(JSValue::js_number(0.0)) + bun_sys::Result::Ok(JSValue::js_number(self.wrote as f64)) } pub fn to_js(&mut self, global_this: &JSGlobalObject) -> JSValue { diff --git a/test/js/bun/s3/s3-write-return-bytes.test.ts b/test/js/bun/s3/s3-write-return-bytes.test.ts new file mode 100644 index 000000000000..c034061d54ff --- /dev/null +++ b/test/js/bun/s3/s3-write-return-bytes.test.ts @@ -0,0 +1,155 @@ +import { S3Client, type S3Options } from "bun"; +import { describe, expect, it } from "bun:test"; +import { isASAN, tempDir } from "harness"; +import path from "node:path"; + +// The S3 client routes through HTTP_PROXY without consulting NO_PROXY, which +// breaks the localhost mock origin on proxied machines. +process.env.HTTP_PROXY = ""; +process.env.HTTPS_PROXY = ""; +process.env.http_proxy = ""; +process.env.https_proxy = ""; + +// writer() leaks its NetworkSink (pre-existing, fix in #34999), which trips +// LeakSanitizer; skip those tests under ASAN until that PR lands. +const itWriter = it.skipIf(isASAN); + +// Every write entry point is typed/documented as "Promise resolving to number of +// bytes written". Buffered sources already returned the true count; streamed +// sources (ReadableStream body, Bun.file, writer().end(), download-to-file) +// resolved a hardcoded 0. + +describe("s3 write() resolves with bytes transferred", () => { + const PAYLOAD = 300_000; + + function mockOrigin() { + let received = 0; + const server = Bun.serve({ + port: 0, + async fetch(req) { + received = (await req.arrayBuffer()).byteLength; + const url = new URL(req.url); + if (url.search === "?uploads=") { + // InitiateMultipartUpload + return new Response( + `abc123`, + { status: 200 }, + ); + } + if (req.method === "GET") { + return new Response(Buffer.alloc(PAYLOAD, "x")); + } + return new Response("", { status: 200, headers: { etag: '"e"' } }); + }, + }); + const options: S3Options = { + endpoint: server.url.href, + accessKeyId: "a", + secretAccessKey: "b", + bucket: "bk", + region: "us-east-1", + }; + return { + server, + options, + client: new S3Client(options), + received: () => received, + [Symbol.dispose]() { + server.stop(true); + }, + }; + } + + it("buffered: Uint8Array source returns byte count", async () => { + using m = mockOrigin(); + const n = await m.client.write("k", new Uint8Array(PAYLOAD)); + expect({ returned: n, received: m.received() }).toEqual({ returned: PAYLOAD, received: PAYLOAD }); + }); + + it("streamed: Response with ReadableStream body returns byte count", async () => { + using m = mockOrigin(); + const stream = new ReadableStream({ + start(c) { + c.enqueue(new Uint8Array(PAYLOAD)); + c.close(); + }, + }); + const n = await m.client.write("k", new Response(stream)); + expect({ returned: n, received: m.received() }).toEqual({ returned: PAYLOAD, received: PAYLOAD }); + }); + + it("streamed: Bun.file source returns byte count", async () => { + using m = mockOrigin(); + using dir = tempDir("s3-write-ret", { + "src.bin": Buffer.alloc(PAYLOAD, "A"), + }); + const n = await Bun.write(m.client.file("k"), Bun.file(path.join(String(dir), "src.bin"))); + expect({ returned: n, received: m.received() }).toEqual({ returned: PAYLOAD, received: PAYLOAD }); + }); + + it("streamed: S3Client.write(key, Bun.file) returns byte count", async () => { + using m = mockOrigin(); + using dir = tempDir("s3-write-ret-direct", { + "src.bin": Buffer.alloc(PAYLOAD, "B"), + }); + const n = await m.client.write("k", Bun.file(path.join(String(dir), "src.bin"))); + expect({ returned: n, received: m.received() }).toEqual({ returned: PAYLOAD, received: PAYLOAD }); + }); + + itWriter("writer(): end() returns total bytes written", async () => { + using m = mockOrigin(); + const w = m.client.file("k").writer(); + w.write(new Uint8Array(PAYLOAD)); + w.write(new Uint8Array(PAYLOAD)); + w.write(new Uint8Array(PAYLOAD)); + const n = await w.end(); + expect({ returned: n, received: m.received() }).toEqual({ + returned: PAYLOAD * 3, + received: PAYLOAD * 3, + }); + }); + + itWriter("writer(): end() returns total bytes for a multipart upload", async () => { + let partsReceived = 0; + using server = Bun.serve({ + port: 0, + async fetch(req) { + const body = await req.arrayBuffer(); + if (req.method === "POST" && req.url.includes("?uploads=")) { + return new Response( + "abc123", + { status: 200 }, + ); + } + if (req.method === "POST" && req.url.includes("uploadId=")) { + return new Response('"etag"', { + status: 200, + }); + } + partsReceived += body.byteLength; + return new Response("", { status: 200, headers: { etag: '"e"' } }); + }, + }); + const client = new S3Client({ + endpoint: server.url.href, + accessKeyId: "a", + secretAccessKey: "b", + bucket: "bk", + region: "us-east-1", + }); + const partSize = 5 * 1024 * 1024; + const total = partSize + 1024 * 1024; + const w = client.file("k").writer({ partSize }); + w.write(Buffer.alloc(total, "a")); + const n = await w.end(); + expect({ returned: n, received: partsReceived }).toEqual({ returned: total, received: total }); + }); + + it("download: Bun.write(path, s3file) returns bytes written to disk", async () => { + using m = mockOrigin(); + using dir = tempDir("s3-dl-ret", {}); + const dest = path.join(String(dir), "out.bin"); + const n = await Bun.write(dest, m.client.file("k")); + expect({ returned: n, onDisk: Bun.file(dest).size }).toEqual({ returned: PAYLOAD, onDisk: PAYLOAD }); + }); +});