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
3 changes: 3 additions & 0 deletions src/http/AsyncHTTP.rs
Original file line number Diff line number Diff line change
Expand Up @@ -196,6 +196,7 @@ fn make_client<'a>(
compressed_request_body: Vec::new(),
compressed_body_len: 0,
pool: crate::PoolOptions::default(),
buffered_sendfile_body: Vec::new(),
}
}

Expand Down Expand Up @@ -756,6 +757,8 @@ impl<'a> AsyncHTTP<'a> {
// populated by the clone (`on_start` → `client.start`); it
// owns the decompressor / compressed_body buffers.
drop(core::mem::take(&mut client.state));
// After `state`: its `original_request_body` borrows this.
drop(core::mem::take(&mut client.buffered_sendfile_body));
}
let elapsed = (*this).elapsed;
bun_core::scoped_log!(AsyncHTTP, "onAsyncHTTPCallback: {:?}", elapsed);
Expand Down
2 changes: 2 additions & 0 deletions src/http/HTTPThread.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1026,6 +1026,8 @@ impl HttpThread {
client.close_proxy_tunnel(false);
drop(core::mem::take(&mut client.custom_ssl_ctx));
drop(core::mem::take(&mut client.state));
// After `state`: its `original_request_body` borrows this.
drop(core::mem::take(&mut client.buffered_sendfile_body));
if let Some(f) = release.release_at_shutdown {
f(release.ctx);
}
Expand Down
12 changes: 11 additions & 1 deletion src/http/InternalState.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,9 @@ use bun_core::MutableString;
use bun_core::Output;

use crate::decompressor::has_zlib_header;
use crate::{CertificateInfo, Decompressor, Encoding, HTTPRequestBody, HTTPResponseMetadata};
use crate::{
CertificateInfo, Decompressor, Encoding, HTTPRequestBody, HTTPResponseMetadata, SendFile,
};

bun_core::define_scoped_log!(log, HTTPInternalState, hidden);

Expand Down Expand Up @@ -51,6 +53,8 @@ pub struct InternalState<'a> {
// outlives-holder invariant (the backing `original_request_body` is a
// sibling field, so it lives exactly as long as this struct).
pub(crate) request_body: bun_ptr::RawSlice<u8>,
/// Send cursor for a `Sendfile` body; `original_request_body` stays untouched for redirects.
pub(crate) sendfile: Option<SendFile>,
pub(crate) original_request_body: HTTPRequestBody<'a>,
pub(crate) request_sent_len: usize,
pub(crate) fail: Option<Error>,
Expand Down Expand Up @@ -136,6 +140,7 @@ impl Default for InternalState<'_> {
content_length: None,
total_body_received: 0,
request_body: bun_ptr::RawSlice::EMPTY,
sendfile: None,
original_request_body: HTTPRequestBody::Bytes(b""),
request_sent_len: 0,
fail: None,
Expand All @@ -152,9 +157,14 @@ impl Default for InternalState<'_> {
impl<'a> InternalState<'a> {
pub(crate) fn init(body: HTTPRequestBody<'a>) -> InternalState<'a> {
let request_body = bun_ptr::RawSlice::new(body.slice());
let sendfile = match &body {
HTTPRequestBody::Sendfile(sendfile) => Some(*sendfile),
_ => None,
};
InternalState {
original_request_body: body,
request_body,
sendfile,
compressed_body: MutableString::init_empty(),
response_message_buffer: MutableString::init_empty(),
decoded_body: MutableString::init_empty(),
Expand Down
15 changes: 15 additions & 0 deletions src/http/SendFile.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ use bun_core::feature_flags;
use bun_sys::{self, Fd};
use bun_url::URL;

/// A body sent with `sendfile(2)`; the JS-side `FetchTasklet` owns the fd for the whole request.
#[derive(Copy, Clone)]
pub struct SendFile {
pub fd: Fd,
Expand All @@ -22,6 +23,20 @@ impl SendFile {
url.is_http() && url.href.len() > 0
}

/// Reads the advertised `content_size` bytes (fewer if the file shrank) for a non-sendfile hop.
pub(crate) fn read_to_vec(&self) -> crate::Result<Vec<u8>> {
let mut bytes: Vec<u8> = Vec::new();
bytes
.try_reserve_exact(self.content_size)
.map_err(|_| bun_alloc::AllocError)?;
bytes.resize(self.content_size, 0);
let read = bun_sys::File::borrow(&self.fd)
.pread_all(&mut bytes, self.offset as u64)
.map_err(bun_errno::SystemErrno::from)?;
bytes.truncate(read);
Ok(bytes)
}

// Takes the resolved fd directly rather than the socket; callers pass
// `socket.fd()`.
pub(crate) fn write(&mut self, socket_fd: Fd) -> Status {
Expand Down
79 changes: 51 additions & 28 deletions src/http/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -832,6 +832,8 @@ pub struct HTTPClient<'a> {
/// the body hasn't been compressed yet.
pub(crate) compressed_body_len: usize,
pub(crate) pool: PoolOptions,
/// A `Sendfile` body read into memory for a TLS hop, borrowed by `state.original_request_body`.
pub(crate) buffered_sendfile_body: Vec<u8>,
}

impl<'a> HTTPClient<'a> {
Expand Down Expand Up @@ -2181,9 +2183,7 @@ impl<'a> HTTPClient<'a> {
if in_progress
&& self.allow_retry
&& self.method.is_idempotent()
// Only a Bytes body can be rebuilt from `original_request_body`.
// Stream/Sendfile bodies are consumed as they are written, so a
// retry would silently replay a truncated request.
// Only an in-memory body is retried; a Stream body is consumed as it is written.
&& matches!(self.state.original_request_body, HTTPRequestBody::Bytes(_))
Comment on lines +2186 to 2187

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟣 pre-existing, not blocking: pre-existing: an idempotent upload of a Bun.file() of 32 KiB or more on a stale keep-alive socket still fails with ConnectionClosed, while a smaller file is retried transparently. The retry predicate at src/http/lib.rs:2187 still admits only HTTPRequestBody::Bytes even though this PR makes a Sendfile body replayable (the write cursor now lives in state.sendfile, so original_request_body stays pristine). Fix: let the keep-alive retry restart Sendfile bodies too, i.e. match Bytes() | Sendfile() there, so the same sibling site as do_redirect uses the pristine body; the PR says this site was left as is on purpose, but the reason given (the body is consumed as it is written) no longer holds after this change.
A small fix can ride a push you are already making; otherwise a short reply is enough.

Why this was flagged

A user calls fetch(url, { method: "PUT", body: Bun.file(path) }) with a file of st_size >= 32 KiB to a plain http:// origin, so src/runtime/webcore/fetch.rs:1629 hands the HTTP thread an HTTPRequestBody::Sendfile. The client picks a pooled keep-alive socket that the server has already closed; on_close runs with in_progress true and response_stage before Body. At src/http/lib.rs:2187 the retry requires matches!(self.state.original_request_body, HTTPRequestBody::Bytes(_)), so the Sendfile request falls through to self.fail(crate::Error::ConnectionClosed) at src/http/lib.rs:2202 and the fetch rejects. A file below 32 KiB is a Bytes body and is silently retried by self.start(body) at src/http/lib.rs:2198. The base branch fails the same way, but on the base the exclusion was justified because SendFile::write mutated original_request_body; after this PR InternalState::init seats a separate cursor in state.sendfile (src/http/InternalState.rs:159-161) and do_redirect already replays the pristine Sendfile through request_body_for_redirect (src/http/lib.rs:2713).

Verification: An idempotent fetch with a Bun.file() body of st_size >= 32 KiB to a plain http:// origin is assigned a pooled keep-alive socket that the server has already closed. on_close only retries if matches!(self.state.original_request_body, HTTPRequestBody::Bytes(_)) (src/http/lib.rs:2187); a Sendfile body falls through to self.fail(crate::Error::ConnectionClosed) at line 2203.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Correct. The old reason does not hold for Sendfile now that the send cursor lives in state.sendfile. The behavior at this site is the same as on main. I am deciding this together with the thread on buffer_sendfile_body_for_tls, and I will reply here with the outcome.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Decision: not in this PR. The head stays at 7e81605, so a merge does not wait on this. A retry for Sendfile bodies at this site is a follow-up. The PR body lists it under Downsides.

&& self.state.response_stage != ResponseStage::Body
&& self.state.response_stage != ResponseStage::BodyChunk
Expand Down Expand Up @@ -2637,17 +2637,7 @@ impl<'a> HTTPClient<'a> {
self.flags.is_streaming_request_body = false;
}

// TODO: what we do with stream body?
let request_body: &[u8] = if self.state.flags.resend_request_body_on_redirect
&& matches!(self.state.original_request_body, HTTPRequestBody::Bytes(_))
{
match &self.state.original_request_body {
HTTPRequestBody::Bytes(b) => b,
_ => unreachable!(),
}
} else {
b""
};
let request_body = self.request_body_for_redirect();

self.state.response_message_buffer = MutableString::default();

Expand Down Expand Up @@ -2710,7 +2700,19 @@ impl<'a> HTTPClient<'a> {
self.flags.protocol = Protocol::Http1_1;
self.reevaluate_proxy_for_redirect();

self.start(HTTPRequestBody::Bytes(request_body));
self.start(request_body);
}

/// The body for the next hop; a `Stream` only reaches this on a 303, already a bodiless GET.
fn request_body_for_redirect(&self) -> HTTPRequestBody<'a> {
if !self.state.flags.resend_request_body_on_redirect {
return HTTPRequestBody::Bytes(b"");
}
match self.state.original_request_body {
HTTPRequestBody::Bytes(bytes) => HTTPRequestBody::Bytes(bytes),
HTTPRequestBody::Sendfile(sendfile) => HTTPRequestBody::Sendfile(sendfile),
HTTPRequestBody::Stream(_) => HTTPRequestBody::Bytes(b""),
}
}

/// Re-resolve `http_proxy` against the post-redirect `self.url`. The
Expand Down Expand Up @@ -2788,6 +2790,12 @@ impl<'a> HTTPClient<'a> {
return;
}

if let Err(err) = self.buffer_sendfile_body_for_tls::<IS_SSL>() {
self.fail(err);
self.complete_connecting_process();
return;
}

// protocol: "http2" is documented as HTTPS-only (h2c is out of scope).
// Every h2 consumer is gated on the SSL const-generic, so without this
// an http:// request would silently fall through to HTTP/1.1.
Expand Down Expand Up @@ -2926,6 +2934,25 @@ impl<'a> HTTPClient<'a> {
self.complete_connecting_process();
}

/// A redirect or env `https://` proxy can put a `Sendfile` body on TLS; send it as `Bytes` there.
fn buffer_sendfile_body_for_tls<const IS_SSL: bool>(&mut self) -> crate::Result<()> {
let HTTPRequestBody::Sendfile(sendfile) = self.state.original_request_body else {
return Ok(());
};
let tunnels_through_proxy = self.http_proxy.is_some() && self.url.is_https();
if !IS_SSL && !tunnels_through_proxy {
return Ok(());
}
debug_assert!(self.buffered_sendfile_body.is_empty());
self.buffered_sendfile_body = sendfile.read_to_vec()?;
Comment on lines +2946 to +2947

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 (optional) Every other in-flight fetch in the process stalls while a large Bun.file() body is read into memory on the HTTP thread when a redirect (or env https proxy) moves it onto TLS. buffer_sendfile_body_for_tls at src/http/lib.rs:2947 calls read_to_vec synchronously on the single shared HTTP thread, allocating and pread-ing the whole content_size (the full file, possibly GBs) before the hop can start. Fix: keep the HTTP thread non-blocking for this fallback, e.g. fail the hop with RequestBodyNotReusable (the spec-conformant option the PR itself names) or bounce the read to the JS/IO side, while still buffering small files so the http->https redirect case keeps working.

Why this was flagged

A user calls fetch("http://...", { method: "PUT", body: Bun.file(largePath) }) where the file is >= 32 KiB, so src/runtime/webcore/fetch.rs:1599-1629 turns the body into HTTPRequestBody::Sendfile. The first hop answers 307 with an https Location (or http_proxy=https://... is set in the environment). do_redirect calls start() (src/http/lib.rs:2703) and start_:: runs buffer_sendfile_body_for_tls (src/http/lib.rs:2793). That function calls SendFile::read_to_vec (src/http/SendFile.rs:27-38), which try_reserve_exact + resize the full content_size and pread_all the whole range synchronously on the HTTP thread. For a multi-hundred-MiB or GiB file on a cold disk this blocks the one HTTP thread for seconds and holds the whole file in memory, so every other fetch() in the process (unrelated requests, timeouts, keep-alive handling) stops progressing until the read finishes. On the base branch this hop sent Content-Length: 0 (or panicked for the env-proxy case), so the stall is new.

Verification: buffer_sendfile_body_for_tls (src/http/lib.rs:2937-2954) does self.buffered_sendfile_body = sendfile.read_to_vec()?; when IS_SSL or http_proxy.is_some() && url.is_https(). SendFile::read_to_vec (src/http/SendFile.rs:27-38) does try_reserve_exact(self.content_size), then pread_all, which loops pread until the whole buffer is filled. Nothing yields to the uws loop, so every other fetch on the process's HTTP thread makes no progress.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This cost is real. The read is synchronous on the HTTP thread and holds content_size bytes, and http to https is a common redirect. I am evaluating a version that keeps the body as Sendfile on a TLS hop and reads the file in bounded chunks in the Body and ProxyBody write arms. That removes this function and the owned buffer. The current head stays unchanged until that is decided. I will post the result here.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Decision: the head stays at 7e81605. A merge does not wait on me.

The cost of this read is about 0.55 ms per MiB with a warm cache (563 ms for 1 GiB), and the file stays in memory until the request ends. A pre-merge check on this head reported 0.1 to 1 s and 1024 MB for 1 GiB. The same file sent to https from the start costs 1 s and 2048 MB on main, because that path reads the whole file on the JS thread. So this fallback is not worse than the https upload path that exists today.

A read in bounded chunks in the write path is still the better end state. It applies to uploads that start on https too, so it is a separate change and not part of this PR. The PR body now lists this cost under Downsides.

// SAFETY: the Vec is assigned only here, once per request (the body is
// `Bytes` from now on), and lives on `self`, which outlives every
// `InternalState` that borrows it.
let bytes: &'a [u8] = unsafe { bun_ptr::detach_lifetime(&self.buffered_sendfile_body) };
self.state = InternalState::init(HTTPRequestBody::Bytes(bytes));
Ok(())
}

/// Body length for `Content-Length` — the compressed length once
/// [`compress_body_for_send`] has run, otherwise the original.
#[inline]
Expand Down Expand Up @@ -3423,7 +3450,7 @@ impl<'a> HTTPClient<'a> {
self.set_timeout(&socket);
}

match &mut self.state.original_request_body {
match self.state.original_request_body {
HTTPRequestBody::Bytes(_) => {
let to_send = self.request_body();
if !to_send.is_empty() {
Expand All @@ -3449,13 +3476,18 @@ impl<'a> HTTPClient<'a> {
// flush without adding any new data
self.flush_stream::<IS_SSL>(socket);
}
HTTPRequestBody::Sendfile(sendfile) => {
HTTPRequestBody::Sendfile(_) => {
if IS_SSL {
panic!(
"sendfile is only supported without SSL. This code should never have been reached!"
);
}

let sendfile = self
.state
.sendfile
.as_mut()
.expect("InternalState::init seats the cursor for a Sendfile body");
// sendfile.write() takes the raw fd, not the socket handle.
match sendfile.write(socket.fd()) {
#[cfg(not(windows))]
Expand Down Expand Up @@ -4425,16 +4457,7 @@ impl<'a> HTTPClient<'a> {
self.flags.is_streaming_request_body = false;
}
self.unix_socket_path = b"";
let request_body: &[u8] = if self.state.flags.resend_request_body_on_redirect
&& matches!(self.state.original_request_body, HTTPRequestBody::Bytes(_))
{
match &self.state.original_request_body {
HTTPRequestBody::Bytes(b) => b,
_ => unreachable!(),
}
} else {
b""
};
let request_body = self.request_body_for_redirect();
self.state.response_message_buffer = MutableString::default();
self.remaining_redirect_count = self.remaining_redirect_count.saturating_sub(1);
self.flags.redirected = true;
Expand All @@ -4450,7 +4473,7 @@ impl<'a> HTTPClient<'a> {
self.flags.proxy_tunneling = false;
self.flags.protocol = Protocol::Http1_1;
self.reevaluate_proxy_for_redirect();
self.start(HTTPRequestBody::Bytes(request_body));
self.start(request_body);
}

pub(crate) fn progress_update_h3(&mut self) {
Expand Down
62 changes: 61 additions & 1 deletion test/js/bun/http/proxy.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,12 @@ import axios from "axios";
import type { Server } from "bun";
import { proxyInternals } from "bun:internal-for-testing";
import { afterAll, beforeAll, describe, expect, test } from "bun:test";
import { bunEnv, bunExe, isASAN, isIPv6, isWindows, tls as tlsCert } from "harness";
import { bunEnv, bunExe, isASAN, isIPv6, isWindows, tempDir, tls as tlsCert } from "harness";
import { HttpsProxyAgent } from "https-proxy-agent";
import { once } from "node:events";
import http from "node:http";
import net from "node:net";
import { join } from "node:path";
import tls from "node:tls";
import { createAdversarialProxy, deadPort, proxyFreeEnv } from "./proxy-stress-helpers";
async function createProxyServer(is_tls: boolean) {
Expand Down Expand Up @@ -231,6 +232,65 @@ for (const proxy_tls of [false, true]) {
}
}

// fetch() uploads a Bun.file() of 32 KiB or more with sendfile(2), which needs a
// plaintext connection, and decides that before a proxy from the environment
// is applied. The HTTP client has to fall back to uploading the file as
// ordinary bytes whenever the connection it ends up on is TLS (an https://
// proxy) or a CONNECT tunnel (a redirect onto an https:// origin through a
// proxy); it used to abort with "sendfile is only supported without SSL" in
// the first case and send an empty body in the second.
describe("Bun.file() body that would use sendfile, with a proxy from the environment", () => {
const SIZE = 64 * 1024;

// Subprocess: the proxy environment is read when the process starts.
async function uploadFromChild(url: string, proxyEnv: Record<string, string>) {
using dir = tempDir("proxy-sendfile-body", { "body.bin": Buffer.alloc(SIZE, "a") });
const env = { ...bunEnv };
for (const key of PROXY_ENV_KEYS) delete env[key];
await using proc = Bun.spawn({
cmd: [
bunExe(),
"-e",
`const res = await fetch(${JSON.stringify(url)}, {
method: "POST",
body: Bun.file(${JSON.stringify(join(String(dir), "body.bin"))}),
tls: { rejectUnauthorized: false },
});
console.log(res.status, (await res.text()).length);`,
],
env: { ...env, ...proxyEnv },
stdout: "pipe",
stderr: "pipe",
});
const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
return { stdout, stderr, exitCode };
}

test.concurrent("is uploaded through a TLS proxy", async () => {
expect(await uploadFromChild(String(httpServer.url), { http_proxy: httpsProxyServer.url })).toEqual({
stdout: `200 ${SIZE}\n`,
stderr: "",
exitCode: 0,
});
});

test.concurrent("is re-uploaded through a CONNECT tunnel when the origin redirects to https", async () => {
using origin = Bun.serve({
port: 0,
async fetch(req) {
await req.arrayBuffer();
return new Response(null, { status: 307, headers: { Location: String(httpsServer.url) } });
},
});
const proxyEnv = { http_proxy: httpProxyServer.url, https_proxy: httpProxyServer.url };
expect(await uploadFromChild(String(origin.url), proxyEnv)).toEqual({
stdout: `200 ${SIZE}\n`,
stderr: "",
exitCode: 0,
});
});
});

for (const server_tls of [false, true]) {
describe.concurrent(`proxy can handle redirects with ${server_tls ? "TLS" : "non-TLS"} server`, () => {
test("with empty body #12007", async () => {
Expand Down
Loading
Loading