Skip to content
13 changes: 10 additions & 3 deletions src/runtime/webcore/Blob.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 => {
Expand All @@ -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),
))
}

Expand Down Expand Up @@ -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)
}

Expand Down
9 changes: 7 additions & 2 deletions src/runtime/webcore/s3/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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) => {
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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();
}
}
Expand Down Expand Up @@ -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,
Expand Down
7 changes: 6 additions & 1 deletion src/runtime/webcore/s3/multipart.rs
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,8 @@ pub struct MultiPartUpload {

pub current_part_number: u16,
pub ref_count: Cell<u32>, // intrusive refcount — see bun_ptr::IntrusiveRc
/// Server-acknowledged payload bytes; resolves `S3File.write()` / `.writer().end()`.
pub uploaded: u64,
pub ended: bool,

pub options: MultiPartUploadOptions,
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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()
}
Expand Down
21 changes: 10 additions & 11 deletions src/runtime/webcore/streams.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2047,6 +2047,8 @@ pub struct NetworkSink {
// JSC_BORROW: process-lifetime VM global; safe `Deref` via `BackRef`.
pub global_this: Option<BackRef<JSGlobalObject>>,
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,
Expand All @@ -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,
Expand Down Expand Up @@ -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))?;
}
Expand Down Expand Up @@ -2274,27 +2279,21 @@ impl NetworkSink {
}

pub fn end_from_js(&mut self, _global_this: &JSGlobalObject) -> bun_sys::Result<JSValue> {
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 {
Expand Down
155 changes: 155 additions & 0 deletions test/js/bun/s3/s3-write-return-bytes.test.ts
Original file line number Diff line number Diff line change
@@ -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(
`<?xml version="1.0"?><InitiateMultipartUploadResult><UploadId>abc123</UploadId></InitiateMultipartUploadResult>`,
{ 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,
});
});
Comment thread
coderabbitai[bot] marked this conversation as resolved.

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(
"<InitiateMultipartUploadResult><UploadId>abc123</UploadId></InitiateMultipartUploadResult>",
{ status: 200 },
);
}
if (req.method === "POST" && req.url.includes("uploadId=")) {
return new Response('<CompleteMultipartUploadResult><ETag>"etag"</ETag></CompleteMultipartUploadResult>', {
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 });
});
});
Loading