Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion src/runtime/node/types.rs
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ use bun_paths::{MAX_PATH_BYTES, OSPathBuffer, OSPathSliceZ, PathBuffer};
use bun_sys::{self, Fd, Mode, O};

use crate::node::util::validators;
use crate::webcore::{Blob, Request, Response};
use crate::webcore::{Blob, BlobExt as _, Request, Response};

pub(crate) use jsc::MarkedArrayBuffer as Buffer;
use jsc::PinnedArrayBuffer;
Expand Down Expand Up @@ -126,6 +126,14 @@ impl BlobOrStringOrBuffer {
}
}
Comment thread
robobun marked this conversation as resolved.

/// [`Self::slice`], or `None` for a file- or S3-backed Blob, which has to be read first.
pub(crate) fn in_memory_view(&self) -> Option<&[u8]> {
match self {
Self::Blob(blob) => blob.in_memory_view(),
Self::StringOrBuffer(str) => Some(str.slice()),
}
}
Comment thread
robobun marked this conversation as resolved.

pub(crate) fn byte_length(&self) -> usize {
self.slice().len()
}
Expand Down
11 changes: 6 additions & 5 deletions src/runtime/socket/socket_body.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2891,12 +2891,13 @@ impl<const SSL: bool> NewSocket<SSL> {
Err(_) => return WriteResult::Fail,
};
// `buffer` Drop frees.
if matches!(&buffer, BlobOrStringOrBuffer::Blob(b) if b.needs_to_read_file()) {
let Some(data) = buffer.in_memory_view() else {
let _ = global.throw(format_args!(
"File blob not supported yet in this function."
"{} cannot send a file- or S3-backed Blob synchronously; await blob.bytes() first",
Comment thread
robobun marked this conversation as resolved.
if IS_END { "Socket.end" } else { "Socket.write" },
));
return WriteResult::Fail;
}
};

// Rust can't reference an outer `const` generic in a nested `const` item
// (E0401), so precompute the full label per branch.
Expand Down Expand Up @@ -2928,7 +2929,7 @@ impl<const SSL: bool> NewSocket<SSL> {

let byte_length: usize = 'brk: {
if length_value.is_undefined() {
break 'brk buffer.slice().len();
break 'brk data.len();
}
if !length_value.is_any_int() {
let _ = global.throw_invalid_argument_type(label, "byteLength", "integer");
Expand All @@ -2952,7 +2953,7 @@ impl<const SSL: bool> NewSocket<SSL> {
usize::try_from(l).expect("int cast")
};

let mut bytes = buffer.slice();
let mut bytes = data;

if byte_offset > bytes.len() {
let _ = global.throw_range_error(
Expand Down
142 changes: 142 additions & 0 deletions test/js/bun/net/socket.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1193,6 +1193,148 @@ it("reload() of a stopped listener says that the listener is closed", () => {
expect(() => listener.reload({ socket })).toThrow("Listener is closed");
});

// write() and end() take the bytes of a Blob at once. A file- or S3-backed Blob has none in memory.
describe.concurrent.each([
["tcp", {}, {}],
["tls", { tls }, { tls: { rejectUnauthorized: false } }],
] as const)("a Blob given to write() and end() over %s", (transport, listenOptions, connectOptions) => {
// Runs `use` on a connected client socket, ends the socket, and resolves with what the listener received.
async function received(use: (socket: Socket) => void): Promise<string> {
const { promise, resolve, reject } = Promise.withResolvers<string>();
const chunks: Buffer[] = [];
using listener = Bun.listen({
hostname: "127.0.0.1",
port: 0,
...listenOptions,
socket: {
data: (_, chunk) => void chunks.push(Buffer.from(chunk)),
end: socket => void socket.end(),
close: () => resolve(Buffer.concat(chunks).toString()),
error: (_, error) => reject(error),
},
});
const run = (socket: Socket) => {
try {
use(socket);
} catch (error) {
reject(error);
} finally {
socket.end();
}
};
await connect({
hostname: "127.0.0.1",
port: listener.port,
...connectOptions,
socket: {
data() {},
...(transport === "tls" ? { handshake: run } : { open: run }),
error: (_, error) => reject(error),
connectError: (_, error) => reject(error),
},
});
return await promise;
}
const outcome = (call: () => unknown) => {
try {
return `returned ${call()}`;
} catch (error: any) {
return `${error.name}: ${error.message}`;
}
};

it("refuses a file- or S3-backed Blob, sends nothing and stays open", async () => {
// Stands in for S3 on the loopback, and records each request.
const requests: string[] = [];
using server = Bun.serve({
port: 0,
hostname: "127.0.0.1",
fetch(request) {
requests.push(`${request.method} ${new URL(request.url).pathname}`);
return new Response("from-s3");
},
});
const client = new Bun.S3Client({
endpoint: server.url.href,
accessKeyId: "test",
secretAccessKey: "test",
bucket: "bucket",
});
using dir = tempDir("socket-write-file-blob", { "file.txt": "from-a-file" });
const s3file = () => client.file("key");
const unread: Record<string, () => unknown> = {
"an S3File": s3file,
"a slice of an S3File": () => s3file().slice(0, 4),
"a Blob made of one S3File": () => new Blob([s3file()]),
"a Request with an S3File body": () => new Request("http://example.com/", { method: "POST", body: s3file() }),
"a Response made of one S3File": () => new Response([s3file()] as any),
"a Bun.file()": () => Bun.file(join(String(dir), "file.txt")),
};

const outcomes: Record<string, string[]> = {};
const bytes = await received(socket => {
for (const [name, make] of Object.entries(unread)) {
outcomes[name] = [outcome(() => socket.write(make() as any)), outcome(() => socket.end(make() as any))];
}
outcomes["an S3File with a byteOffset and a byteLength"] = [
outcome(() => socket.write(s3file() as any, 0, 4)),
outcome(() => socket.end(s3file() as any, 0, 4)),
];
// A refused end() that still marked the socket as ended would close it at this flush.
socket.flush();
outcomes["a string, after every refusal"] = [outcome(() => socket.write("still-open"))];
});

const refused = (method: string) =>
`Error: Socket.${method} cannot send a file- or S3-backed Blob synchronously; await blob.bytes() first`;
expect({ outcomes, bytes }).toEqual({
outcomes: {
...Object.fromEntries(Object.keys(unread).map(name => [name, [refused("write"), refused("end")]])),
"an S3File with a byteOffset and a byteLength": [refused("write"), refused("end")],
"a string, after every refusal": ["returned 10"],
},
bytes: "still-open",
});
// A request that a refused call started would reach the stand-in before this one.
expect({ text: await client.file("after").text(), requests }).toEqual({
text: "from-s3",
requests: ["GET /bucket/after"],
});
});

// The control: it fails if the read refuses or misreads a Blob whose bytes are in memory.
it("sends an in-memory Blob and the buffered body of a Request or Response, in order", async () => {
const outcomes: string[] = [];
const bytes = await received(socket => {
outcomes.push(
outcome(() => socket.write("a")),
outcome(() => socket.write(new Blob(["bcd"]) as any)),
outcome(() => socket.write(new Response("ef") as any)),
outcome(() => socket.write(new Request("http://example.com/", { method: "POST", body: "ghij" }) as any)),
outcome(() => socket.write(new Blob(["##lm##"]).slice(2, 4) as any)),
outcome(() => socket.write(new Blob(["#?no##"]).slice(1, 5) as any, 1, 2)),
outcome(() => socket.write(new Blob([]) as any)),
outcome(() => socket.write(new Response(new ReadableStream({ pull() {} })) as any)),
outcome(() => socket.end(new Blob(["k"]) as any)),
);
});
expect({ outcomes, bytes }).toEqual({
outcomes: [
"returned 1",
"returned 3",
"returned 2",
"returned 4",
"returned 2",
"returned 2",
"returned 0",
"TypeError: Only buffered Request/Response bodies are supported for now.",
"returned 1",
],
bytes: "abcdefghijlmnok",
});
});
});

it("getPeerCertificate and exportKeyingMaterial accept undefined for optional arguments", async () => {
const { promise, resolve, reject } = Promise.withResolvers<void>();
using listener = Bun.listen({ hostname: "127.0.0.1", port: 0, tls, socket: { data() {} } });
Expand Down
Loading