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
81 changes: 57 additions & 24 deletions src/runtime/server/FileResponseStream.rs
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@

use core::cell::Cell;
use core::ffi::c_void;
#[cfg(any(target_os = "linux", target_os = "android"))]
use core::sync::atomic::{AtomicBool, Ordering};

use bun_io::Closer;
#[cfg(windows)]
Expand Down Expand Up @@ -223,60 +225,70 @@ impl FileResponseStream {
return;
}

// BufferedReader path
this_ref.reader.with_mut(|reader| {
this_ref.start_reader(
opts.file_type,
opts.pollable,
(opts.offset > 0).then_some(opts.offset),
opts.length,
);
}

/// `pread` from `offset` when given, else `read` from the fd's position.
fn start_reader(
&self,
file_type: FileType,
pollable: bool,
offset: Option<u64>,
length: Option<u64>,
) {
let fd = self.fd.get();
self.reader.with_mut(|reader| {
reader.flags.remove(ReaderFlags::CLOSE_HANDLE); // we own fd via auto_close
reader.flags.set(ReaderFlags::POLLABLE, opts.pollable);
reader.flags.set(ReaderFlags::POLLABLE, pollable);
reader
.flags
.set(ReaderFlags::NONBLOCKING, opts.file_type != FileType::File);
.set(ReaderFlags::NONBLOCKING, file_type != FileType::File);
#[cfg(unix)]
if opts.file_type == FileType::Socket {
if file_type == FileType::Socket {
reader.flags.insert(ReaderFlags::SOCKET);
}
// The reader reports the end of the body as EOF, so `on_read_chunk` ends the response there like at a real EOF.
reader.set_limit(opts.length.map(|len| len as usize));
reader.set_parent(this.cast::<c_void>());
reader.set_limit(length.map(|len| len as usize));
reader.set_parent(self.as_ptr().cast::<c_void>());
});

// SAFETY: `start()`/`start_file_offset()` re-enter this object through
// the parent pointer (`loop_`/`event_loop`), so no cell borrow spans them.
let reader = this_ref.reader_mut();
let start_result = if opts.offset > 0 {
reader.start_file_offset(opts.fd, opts.pollable, opts.offset as usize)
} else {
reader.start(opts.fd, opts.pollable)
let reader = self.reader_mut();
let start_result = match offset {
Some(offset) => reader.start_file_offset(fd, pollable, offset as usize),
None => reader.start(fd, pollable),
};
if let Err(err) = start_result {
this_ref.fail_with(err);
self.fail_with(err);
return;
}

// SAFETY: as above — `update_ref` re-enters `event_loop` through the parent pointer.
this_ref.reader_mut().update_ref(true);
self.reader_mut().update_ref(true);

#[cfg(unix)]
if let Some(poll) = this_ref.reader.get().handle.get_poll() {
if this_ref
.reader
.get()
.flags
.contains(ReaderFlags::NONBLOCKING)
{
if let Some(poll) = self.reader.get().handle.get_poll() {
if self.reader.get().flags.contains(ReaderFlags::NONBLOCKING) {
poll.set_flag(FilePollFlag::Nonblocking);
}
match opts.file_type {
match file_type {
FileType::Socket => poll.set_flag(FilePollFlag::Socket),
FileType::NonblockingPipe | FileType::Pipe => poll.set_flag(FilePollFlag::Fifo),
FileType::File => {}
}
}

// hold a ref for the in-flight read; released in on_reader_done/on_reader_error
this_ref.hold_read_ref();
self.hold_read_ref();
// SAFETY: `reader` is live for the stream's lifetime; `read` is the
// raw re-entrancy-safe entry (its dispatch runs user JS).
unsafe { BufferedReader::read(this_ref.reader.as_ptr()) };
unsafe { BufferedReader::read(self.reader.as_ptr()) };
}

#[inline]
Expand Down Expand Up @@ -449,6 +461,20 @@ impl FileResponseStream {
}
sys::E::EINTR => continue,
sys::E::EAGAIN => return self.arm_sendfile_writable(),
sys::E::EINVAL | sys::E::ENOSYS | sys::E::ENOTSUP | sys::E::EPERM => {
bun_output::scoped_log!(
FileResponseStream,
"sendfile refused ({}), falling back to reader",
errno
);
if errno == sys::E::ENOSYS {
SENDFILE_UNAVAILABLE.store(true, Ordering::Relaxed);
}
self.mode.set(Mode::Reader);
let sf = self.sendfile.get();
self.start_reader(FileType::File, false, Some(sf.offset), Some(sf.remain));
return true;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}
_ => {
self.fail_with(
sys::Error::from_code(errno, sys::Tag::sendfile).with_fd(self.fd.get()),
Expand Down Expand Up @@ -660,7 +686,14 @@ fn can_sendfile(resp: AnyResponse, file_type: FileType, length: Option<u64>) ->
return false;
}
let Some(len) = length else { return false };
if SENDFILE_UNAVAILABLE.load(Ordering::Relaxed) {
return false;
}
// Below ~1MB the syscall + dual-readiness overhead doesn't pay off.
len >= (1 << 20)
}
}

/// Set once `sendfile(2)` answers `ENOSYS`.
#[cfg(any(target_os = "linux", target_os = "android"))]
static SENDFILE_UNAVAILABLE: AtomicBool = AtomicBool::new(false);
2 changes: 2 additions & 0 deletions src/uws_sys/libuwsockets.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -1568,6 +1568,7 @@ __attribute__((callback (corker, ctx)))
char *ptr = pair.first;
ptr[0] = '\r';
ptr[1] = '\n';
uwsRes->getHttpResponseData()->state |= uWS::HttpResponseData<true>::HTTP_WRITE_CALLED;
uwsRes->uncork();
}
else
Expand All @@ -1578,6 +1579,7 @@ __attribute__((callback (corker, ctx)))
char *ptr = pair.first;
ptr[0] = '\r';
ptr[1] = '\n';
uwsRes->getHttpResponseData()->state |= uWS::HttpResponseData<false>::HTTP_WRITE_CALLED;
uwsRes->uncork();
}
}
Expand Down
206 changes: 205 additions & 1 deletion test/js/bun/http/bun-serve-file.test.ts
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
import type { Server } from "bun";
import { afterAll, beforeAll, describe, expect, it, mock, test } from "bun:test";
import { bunEnv, bunExe, isASAN, isWindows, rmScope, rss, tempDir, tempDirWithFiles } from "harness";
import { bunEnv, bunExe, isASAN, isLinux, isWindows, rmScope, rss, tempDir, tempDirWithFiles } from "harness";
import { mkfifo } from "mkfifo";
import { closeSync, openSync, unlinkSync, writeSync } from "node:fs";
import net from "node:net";
import { join } from "node:path";

const LARGE_SIZE = 1024 * 1024 * 8;
Expand Down Expand Up @@ -1514,3 +1515,206 @@ test("file route serves a burst of concurrent requests after reloads", async ()
const a = await fetch(`${server.url}a`).then(r => r.text());
expect(a).toBe("a-new");
});

// A regular file of 1 MiB or more over plain TCP is sent with sendfile(2). A
// kernel, seccomp policy or filesystem that refuses sendfile for the source fd
// answers EINVAL, ENOSYS, EOPNOTSUPP or EPERM, after the head with its
// Content-Length is already on the wire. The stream has to serve the declared
// bytes with read+write instead. The broken build closed the socket, so the
// client saw a 200 head and then a reset with no body.
//
// A seccomp filter installed in the server process makes sendfile(2) fail with
// the errno under test. Linux only: the filter is a Linux API. With
// SENDFILE_ALLOW_COUNT set, a call whose byte count equals it (the first call,
// which asks for the whole file) is allowed and only the later calls fail, so
// the fallback starts after partial progress.
const seccompFixture = `
const { dlopen, FFIType, ptr } = require("bun:ffi");

function openLibc() {
for (const name of ["libc.so.6", "libc.musl-x86_64.so.1", "libc.musl-aarch64.so.1", "libc.so"]) {
try {
return dlopen(name, {
prctl: { args: [FFIType.i32, FFIType.u64, FFIType.u64, FFIType.u64, FFIType.u64], returns: FFIType.i32 },
});
} catch {}
}
return null;
}

// Returns false when this environment does not allow a seccomp filter.
function failSendfile(errno, allowCount) {
const libc = openLibc();
if (libc === null) return false;
const NR_SENDFILE = process.arch === "x64" ? 40 : 71;
const BPF_LD_W_ABS = 0x20, BPF_JMP_JEQ_K = 0x15, BPF_RET_K = 0x06;
const SECCOMP_RET_ALLOW = 0x7fff0000, SECCOMP_RET_ERRNO = 0x00050000;
const insns = [
[BPF_LD_W_ABS, 0, 0, 0], // A = seccomp_data.nr
[BPF_JMP_JEQ_K, 0, 3, NR_SENDFILE], // not sendfile: allow
[BPF_LD_W_ABS, 0, 0, 40], // A = low 32 bits of seccomp_data.args[3] (count)
[BPF_JMP_JEQ_K, 1, 0, allowCount], // count == allowCount: allow
[BPF_RET_K, 0, 0, SECCOMP_RET_ERRNO | errno],
[BPF_RET_K, 0, 0, SECCOMP_RET_ALLOW],
];
const prog = new Uint8Array(insns.length * 8);
const view = new DataView(prog.buffer);
insns.forEach(([code, jt, jf, k], i) => {
view.setUint16(i * 8, code, true);
view.setUint8(i * 8 + 2, jt);
view.setUint8(i * 8 + 3, jf);
view.setUint32(i * 8 + 4, k, true);
});
const fprog = new Uint8Array(16);
const fview = new DataView(fprog.buffer);
fview.setUint16(0, insns.length, true);
fview.setBigUint64(8, BigInt(ptr(prog)), true);
const PR_SET_NO_NEW_PRIVS = 38, PR_SET_SECCOMP = 22, SECCOMP_MODE_FILTER = 2;
if (libc.symbols.prctl(PR_SET_NO_NEW_PRIVS, 1n, 0n, 0n, 0n) !== 0) return false;
return libc.symbols.prctl(PR_SET_SECCOMP, BigInt(SECCOMP_MODE_FILTER), BigInt(ptr(fprog)), 0n, 0n) === 0;
}

// A count of 0 is never asked for, so the default filter rejects every call.
if (!failSendfile(Number(process.env.SENDFILE_ERRNO), Number(process.env.SENDFILE_ALLOW_COUNT ?? 0))) {
process.exit(3);
}
if (process.env.SENDFILE_PROBE) process.exit(0);
const server = Bun.serve({
port: 0,
hostname: "127.0.0.1",
fetch(req) {
if (new URL(req.url).pathname === "/probe") return new Response("ok");
return new Response(Bun.file(process.env.FILE));
},
error(e) {
console.log("error-handler " + e.code);
return new Response("error", { status: 500 });
},
});
console.log("port " + server.port);
`;

const seccompAvailable =
isLinux &&
Bun.spawnSync({
cmd: [bunExe(), "-e", seccompFixture],
env: { ...bunEnv, SENDFILE_ERRNO: "22", SENDFILE_PROBE: "1" },
}).exitCode === 0;

describe.skipIf(!seccompAvailable)("Bun.serve falls back to read+write when sendfile(2) is refused", () => {
function patternFile(size: number) {
const content = Buffer.alloc(size);
for (let i = 0; i < size; i += 4) content.writeUInt32LE((i * 2654435761) >>> 0, i);
return content;
}

async function startServer(dir: string, env: Record<string, string>) {
const proc = Bun.spawn({
cmd: [bunExe(), "-e", seccompFixture],
cwd: dir,
env: { ...bunEnv, ...env },
stdout: "pipe",
stderr: "inherit",
});
const reader = proc.stdout.getReader();
let stdout = "";
let port: RegExpMatchArray | null = null;
while ((port = stdout.match(/^port (\d+)$/m)) === null) {
const { value, done } = await reader.read();
if (done) throw new Error(`server exited before it printed its port: ${stdout}`);
stdout += Buffer.from(value).toString();
}
return { proc, port: Number(port[1]) };
}

for (const [name, errno] of [
["EINVAL", 22],
["ENOSYS", 38],
["EOPNOTSUPP", 95],
["EPERM", 1],
] as const) {
test.concurrent(name, async () => {
Comment thread
robobun marked this conversation as resolved.
const SIZE = 2 * 1024 * 1024;
using dir = tempDir("serve-sendfile-fallback", {});
const filePath = join(String(dir), "large.bin");
const content = patternFile(SIZE);
await Bun.write(filePath, content);

const { proc, port } = await startServer(String(dir), { FILE: filePath, SENDFILE_ERRNO: String(errno) });
await using _proc = proc;

// Two requests on one keep-alive connection: the second only parses if
// the first body was framed by the declared Content-Length.
const results = [];
for (let i = 0; i < 2; i++) {
const res = await fetch(`http://127.0.0.1:${port}/`);
const body = Buffer.from(await res.arrayBuffer());
results.push({
status: res.status,
contentLength: res.headers.get("content-length"),
bodyLength: body.length,
bodyMatches: body.equals(content),
});
}
expect(results).toEqual(
Array(2).fill({ status: 200, contentLength: String(SIZE), bodyLength: SIZE, bodyMatches: true }),
);
});
}

// The first sendfile(2) call moves as much as the socket buffers take and
// then reports EAGAIN. The client does not read until then, so the call
// stops well short of the file. Every later call is refused, and the
// fallback has to continue from the bytes already sent.
test.concurrent("after partial progress", async () => {
const SIZE = 16 * 1024 * 1024;
using dir = tempDir("serve-sendfile-fallback-partial", {});
const filePath = join(String(dir), "large.bin");
const content = patternFile(SIZE);
await Bun.write(filePath, content);

const { proc, port } = await startServer(String(dir), {
FILE: filePath,
SENDFILE_ERRNO: "22",
SENDFILE_ALLOW_COUNT: String(SIZE),
});
await using _proc = proc;

const { promise: done, resolve, reject } = Promise.withResolvers<Buffer>();
const chunks: Buffer[] = [];
let received = 0;
let headEnd = -1;
let bodyStart = 0;
const client = net.connect(port, "127.0.0.1");
client.pause();
client.on("connect", () => client.write("GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n"));
client.on("data", chunk => {
chunks.push(chunk);
received += chunk.length;
if (headEnd === -1) {
const wire = Buffer.concat(chunks);
headEnd = wire.indexOf("\r\n\r\n");
if (headEnd !== -1) {
bodyStart = headEnd + 4;
const head = wire.subarray(0, headEnd).toString("latin1");
if (!/^content-length:\s*16777216$/im.test(head)) reject(new Error(`unexpected head: ${head}`));
}
}
if (headEnd !== -1 && received >= bodyStart + SIZE) client.end();
});
client.on("close", () => resolve(Buffer.concat(chunks)));
client.on("error", reject);

// The probe is handled after the file request on the server's event loop,
// so once it answers, the first sendfile(2) call has returned.
expect(await fetch(`http://127.0.0.1:${port}/probe`).then(r => r.text())).toBe("ok");
client.resume();

const wire = await done;
const body = wire.subarray(bodyStart);
expect({ bodyLength: body.length, bodyMatches: body.equals(content) }).toEqual({
bodyLength: SIZE,
bodyMatches: true,
});
});
});
Loading