From 86df1a468809ec07e75f1d728c65969fab68e4af Mon Sep 17 00:00:00 2001 From: Jarred Sumner Date: Tue, 16 Jun 2026 20:28:00 +0000 Subject: [PATCH 01/14] fetch: add automatic request body compression via the `compress` option Accepts `boolean | "gzip" | "deflate" | "br" | "zstd" | { encoding, level? }`. Compresses buffered bodies (string, ArrayBuffer/TypedArray, Blob) on the JS thread using a thread-local libdeflate compressor + 512 KiB scratch buffer (mirroring the response-decompression fast path), and injects the `Content-Encoding` request header. ReadableStream and sendfile bodies are left untouched, as are requests that already set `Content-Encoding`. --- packages/bun-types/globals.d.ts | 31 ++ src/runtime/webcore/fetch.rs | 61 ++++ src/runtime/webcore/fetch/compress_body.rs | 355 +++++++++++++++++++++ test/js/web/fetch/fetch-compress.test.ts | 204 ++++++++++++ 4 files changed, 651 insertions(+) create mode 100644 src/runtime/webcore/fetch/compress_body.rs create mode 100644 test/js/web/fetch/fetch-compress.test.ts diff --git a/packages/bun-types/globals.d.ts b/packages/bun-types/globals.d.ts index 4dd3dea89e07..fd93c014af22 100644 --- a/packages/bun-types/globals.d.ts +++ b/packages/bun-types/globals.d.ts @@ -2034,6 +2034,37 @@ interface BunFetchRequestInit extends RequestInit { */ decompress?: boolean; + /** + * Automatically compress the request body before sending and set the + * `Content-Encoding` request header accordingly. + * + * - `true` is equivalent to `"gzip"`. + * - A string selects the encoding with its default level. + * - An object selects the encoding and an explicit compression `level`. + * + * Only buffered bodies (string, `ArrayBuffer`/`TypedArray`, `Blob`) are + * compressed; `ReadableStream` bodies are sent as-is. If the request + * already has a `Content-Encoding` header, the body is left unchanged. + * This is a custom property that is not part of the Fetch API specification. + * + * @default false + * @example + * ```js + * await fetch("https://example.com/upload", { + * method: "POST", + * body: JSON.stringify(bigPayload), + * compress: "gzip", + * }); + * ``` + */ + compress?: + | boolean + | "gzip" + | "deflate" + | "br" + | "zstd" + | { encoding: "gzip" | "deflate" | "br" | "zstd"; level?: number }; + /** * The maximum number of redirects to follow when `redirect` is `"follow"`. * If the response chain redirects more than this many times, the request diff --git a/src/runtime/webcore/fetch.rs b/src/runtime/webcore/fetch.rs index e1e4441351dd..504abd4cd02c 100644 --- a/src/runtime/webcore/fetch.rs +++ b/src/runtime/webcore/fetch.rs @@ -40,6 +40,9 @@ pub(crate) const FETCH_TYPE_ERROR_STRINGS: [&str; 8] = FETCH_TYPE_ERROR_STRING_V #[path = "fetch/FetchTasklet.rs"] pub mod fetch_tasklet; +#[path = "fetch/compress_body.rs"] +pub mod compress_body; + // ────────────────────────────────────────────────────────────────────────── // fetch() implementation // ────────────────────────────────────────────────────────────────────────── @@ -420,6 +423,7 @@ fn fetch_impl( let mut disable_timeout = false; let mut disable_keepalive = false; let mut disable_decompression = false; + let mut compress: Option = None; let mut max_redirects: Option = None; let mut verbose: http::HTTPVerboseLevel = if vm .log_ref() @@ -664,6 +668,34 @@ fn fetch_impl( return Ok(JSValue::ZERO); } + // "compress: boolean | string | { encoding, level? }" + 'extract_compress: { + let objects_to_try = [ + options_object.unwrap_or(JSValue::ZERO), + request_init_object.unwrap_or(JSValue::ZERO), + ]; + + for obj in objects_to_try { + if !obj.is_empty() { + if let Some(compress_value) = obj.get(global_this, "compress")? { + if !compress_value.is_undefined() { + compress = + compress_body::CompressOption::from_js(global_this, compress_value)?; + break 'extract_compress; + } + } + + if global_this.has_exception() { + return Ok(JSValue::ZERO); + } + } + } + } + + if global_this.has_exception() { + return Ok(JSValue::ZERO); + } + // "maxRedirects: number" 'extract_max_redirects: { let objects_to_try = [ @@ -1731,6 +1763,35 @@ fn fetch_impl( } } + // Automatic request-body compression. Only buffered bodies (Blob bytes, + // ArrayBuffer/TypedArray, string) are handled; ReadableStream and sendfile + // are skipped. S3 destinations replace the header set with a signed one, + // so compression is skipped there too. + if let Some(compress_opt) = compress + && let HTTPRequestBody::AnyBlob(_) = &body + && !url.is_s3() + { + let already_has_encoding = headers + .as_ref() + .and_then(|h| h.get_content_encoding()) + .is_some(); + if !already_has_encoding { + let input = body.slice(); + if !input.is_empty() { + let compressed = + compress_body::compress_request_body(global_this, input, &compress_opt)?; + let mut old = core::mem::replace( + &mut body, + HTTPRequestBody::AnyBlob(blob::Any::from_owned_slice(compressed)), + ); + old.detach(); + headers + .get_or_insert_default() + .append(b"Content-Encoding", compress_opt.encoding.header_value()); + } + } + } + if url.is_s3() { // get ENV config — `Transpiler::env_mut` is the safe accessor for the // process-singleton dotenv loader (set during init). diff --git a/src/runtime/webcore/fetch/compress_body.rs b/src/runtime/webcore/fetch/compress_body.rs new file mode 100644 index 000000000000..ddb44c8e5211 --- /dev/null +++ b/src/runtime/webcore/fetch/compress_body.rs @@ -0,0 +1,355 @@ +//! Automatic request-body compression for `fetch()`. +//! +//! Mirrors the response-decompression fast path in `bun_http::InternalState`: +//! a thread-local libdeflate compressor + fixed 512 KiB scratch buffer reused +//! across calls. gzip/deflate go through libdeflate; brotli/zstd use their +//! one-shot encoders. Only buffered bodies are handled here — streams and +//! sendfile are skipped by the caller. + +use core::cell::UnsafeCell; + +use crate::webcore::jsc::{self, JSGlobalObject, JSValue, JsResult}; + +#[derive(Copy, Clone, Eq, PartialEq, Debug)] +pub enum CompressEncoding { + Gzip, + Deflate, + Brotli, + Zstd, +} + +impl CompressEncoding { + pub fn header_value(self) -> &'static [u8] { + match self { + CompressEncoding::Gzip => b"gzip", + CompressEncoding::Deflate => b"deflate", + CompressEncoding::Brotli => b"br", + CompressEncoding::Zstd => b"zstd", + } + } + + fn from_str(s: &[u8]) -> Option { + match s { + b"gzip" => Some(CompressEncoding::Gzip), + b"deflate" => Some(CompressEncoding::Deflate), + b"br" => Some(CompressEncoding::Brotli), + b"zstd" => Some(CompressEncoding::Zstd), + _ => None, + } + } +} + +#[derive(Copy, Clone, Debug)] +pub struct CompressOption { + pub encoding: CompressEncoding, + pub level: Option, +} + +impl CompressOption { + /// Parses `compress?: boolean | "gzip" | "deflate" | "br" | "zstd" | { encoding, level? }`. + /// Returns `Ok(None)` for `false` / `undefined` / `null`. + pub fn from_js(global: &JSGlobalObject, value: JSValue) -> JsResult> { + if value.is_undefined_or_null() { + return Ok(None); + } + if value.is_boolean() { + return Ok(if value.as_boolean() { + Some(CompressOption { + encoding: CompressEncoding::Gzip, + level: None, + }) + } else { + None + }); + } + if value.is_string() { + let s = bun_core::OwnedString::new(value.to_bun_string(global)?); + let bytes = s.to_utf8(); + return match CompressEncoding::from_str(bytes.slice()) { + Some(encoding) => Ok(Some(CompressOption { + encoding, + level: None, + })), + None => Err(global.throw_invalid_arguments(format_args!( + "fetch: 'compress' must be \"gzip\", \"deflate\", \"br\", or \"zstd\"" + ))), + }; + } + if value.is_object() { + let encoding = match value.get(global, "encoding")? { + Some(enc) if enc.is_string() => { + let s = bun_core::OwnedString::new(enc.to_bun_string(global)?); + let bytes = s.to_utf8(); + match CompressEncoding::from_str(bytes.slice()) { + Some(e) => e, + None => { + return Err(global.throw_invalid_arguments(format_args!( + "fetch: 'compress.encoding' must be \"gzip\", \"deflate\", \"br\", or \"zstd\"" + ))); + } + } + } + _ => { + return Err(global.throw_invalid_argument_type_value( + b"compress.encoding", + b"string", + value, + )); + } + }; + let level = match value.get(global, "level")? { + Some(lvl) if !lvl.is_undefined_or_null() => { + if !lvl.is_number() { + return Err(global.throw_invalid_argument_type_value( + b"compress.level", + b"number", + lvl, + )); + } + let n = lvl.to_int32(); + let (min, max) = match encoding { + CompressEncoding::Gzip | CompressEncoding::Deflate => (0, 12), + CompressEncoding::Brotli => ( + bun_brotli::c::BROTLI_MIN_QUALITY, + bun_brotli::c::BROTLI_MAX_QUALITY, + ), + CompressEncoding::Zstd => (1, 22), + }; + if n < min || n > max { + return Err(global.throw_invalid_arguments(format_args!( + "fetch: 'compress.level' for \"{}\" must be between {} and {}", + bstr::BStr::new(encoding.header_value()), + min, + max, + ))); + } + Some(n) + } + _ => None, + }; + return Ok(Some(CompressOption { encoding, level })); + } + Err(global.throw_invalid_argument_type_value( + b"compress", + b"boolean, string, or object", + value, + )) + } +} + +// ────────────────────────────────────────────────────────────────────────── +// Thread-local compressor state (mirrors `LibdeflateState` on the HTTP thread) +// ────────────────────────────────────────────────────────────────────────── + +/// libdeflate's default level. Reused for the cached compressor so the common +/// `compress: true` / `compress: "gzip"` path never allocates a fresh handle. +const DEFAULT_DEFLATE_LEVEL: i32 = 6; +const DEFAULT_BROTLI_QUALITY: i32 = 6; + +const SHARED_BUFFER_SIZE: usize = 512 * 1024; + +struct CompressorState { + compressor: *mut bun_libdeflate_sys::libdeflate::Compressor, + shared_buffer: [u8; SHARED_BUFFER_SIZE], +} + +// SAFETY: `*mut T` (null) and `[u8; N]` are both valid at the all-zero bit pattern. +unsafe impl bun_core::Zeroable for CompressorState {} + +impl CompressorState { + #[inline] + fn compressor_mut<'a>(&self) -> &'a mut bun_libdeflate_sys::libdeflate::Compressor { + // SAFETY: `compressor` is set once in `with_state` from + // `libdeflate_alloc_compressor` (panics on null) and never freed for + // the thread's lifetime. The handle is a separate C heap allocation + // disjoint from `self`, so the returned `&mut` does not alias + // `shared_buffer`. Thread-local — sole live borrow. + unsafe { &mut *self.compressor } + } +} + +thread_local! { + static LAZY_COMPRESSOR: UnsafeCell>> = + const { UnsafeCell::new(None) }; +} + +fn with_state(f: impl FnOnce(&mut CompressorState) -> R) -> R { + LAZY_COMPRESSOR.with(|cell| { + // SAFETY: thread-local; sole accessor; no re-entrance from `f` back into + // this function. + let slot = unsafe { &mut *cell.get() }; + if slot.is_none() { + let compressor = + bun_libdeflate_sys::libdeflate::Compressor::alloc(DEFAULT_DEFLATE_LEVEL); + if compressor.is_null() { + bun_core::out_of_memory(); + } + let mut state: Box = bun_core::boxed_zeroed(); + state.compressor = compressor; + *slot = Some(state); + } + f(slot.as_deref_mut().unwrap()) + }) +} + +// ────────────────────────────────────────────────────────────────────────── +// One-shot body compression +// ────────────────────────────────────────────────────────────────────────── + +pub fn compress_request_body( + global: &JSGlobalObject, + input: &[u8], + opt: &CompressOption, +) -> JsResult> { + match opt.encoding { + CompressEncoding::Gzip | CompressEncoding::Deflate => { + let enc = if opt.encoding == CompressEncoding::Gzip { + bun_libdeflate_sys::libdeflate::Encoding::Gzip + } else { + // HTTP "deflate" is the zlib-wrapped DEFLATE stream (RFC 9110 + // §8.4.1.2); libdeflate's `Deflate` is the raw stream. + bun_libdeflate_sys::libdeflate::Encoding::Zlib + }; + Ok(compress_libdeflate(input, enc, opt.level)) + } + CompressEncoding::Brotli => compress_brotli(global, input, opt.level), + CompressEncoding::Zstd => compress_zstd(global, input, opt.level), + } +} + +fn compress_libdeflate( + input: &[u8], + enc: bun_libdeflate_sys::libdeflate::Encoding, + level: Option, +) -> Vec { + use bun_libdeflate_sys::libdeflate::Compressor; + + with_state(|state| { + // Custom level → allocate a temporary compressor; the cached handle is + // pinned to DEFAULT_DEFLATE_LEVEL. + let mut tmp: *mut Compressor = core::ptr::null_mut(); + let compressor: &mut Compressor = match level { + Some(l) if l != DEFAULT_DEFLATE_LEVEL => { + tmp = Compressor::alloc(l); + if tmp.is_null() { + bun_core::out_of_memory(); + } + // SAFETY: just allocated, non-null, exclusive. + unsafe { &mut *tmp } + } + _ => state.compressor_mut(), + }; + let _guard = scopeguard::guard(tmp, |tmp| { + if !tmp.is_null() { + // SAFETY: `tmp` was returned by `Compressor::alloc` above and is + // not used after this call. + unsafe { Compressor::destroy(tmp) }; + } + }); + + // Fast path: compress into the shared buffer, then copy out. + let bound = compressor.max_bytes_needed(input, enc); + if bound <= state.shared_buffer.len() { + let result = compressor.compress(input, &mut state.shared_buffer, enc); + return state.shared_buffer[..result.written].to_vec(); + } + + // Slow path: body is large; allocate the bound up front and compress + // directly into the Vec's spare capacity. + let mut out = Vec::with_capacity(bound); + compressor.compress_to_vec(input, &mut out, enc); + out + }) +} + +fn compress_brotli(global: &JSGlobalObject, input: &[u8], level: Option) -> JsResult> { + use bun_brotli::c; + let quality = level.unwrap_or(DEFAULT_BROTLI_QUALITY); + + with_state(|state| { + let bound = c::BrotliEncoderMaxCompressedSize(input.len()); + // BrotliEncoderMaxCompressedSize returns 0 when the bound would + // overflow size_t — fall back to a heap buffer in that case. + if bound != 0 && bound <= state.shared_buffer.len() { + let mut out_len = state.shared_buffer.len(); + // SAFETY: input/output slices are valid for their lengths; + // BrotliEncoderCompress only reads `input` and writes `out_len` + // bytes to `shared_buffer`, updating `out_len` to bytes written. + let ok = unsafe { + c::BrotliEncoderCompress( + quality, + c::BROTLI_DEFAULT_WINDOW, + c::BrotliEncoderMode::generic, + input.len(), + input.as_ptr(), + &raw mut out_len, + state.shared_buffer.as_mut_ptr(), + ) + }; + if ok != 0 { + return Ok(state.shared_buffer[..out_len].to_vec()); + } + } + + let cap = if bound != 0 { bound } else { input.len() + 1024 }; + let mut out = vec![0u8; cap]; + let mut out_len = out.len(); + // SAFETY: see above. + let ok = unsafe { + c::BrotliEncoderCompress( + quality, + c::BROTLI_DEFAULT_WINDOW, + c::BrotliEncoderMode::generic, + input.len(), + input.as_ptr(), + &raw mut out_len, + out.as_mut_ptr(), + ) + }; + if ok == 0 { + return Err(global.err( + jsc::ErrorCode::ZLIB_INITIALIZATION_FAILED, + format_args!("brotli compression failed"), + ).throw()); + } + out.truncate(out_len); + Ok(out) + }) +} + +fn compress_zstd(global: &JSGlobalObject, input: &[u8], level: Option) -> JsResult> { + with_state(|state| { + let bound = bun_zstd::compress_bound(input.len()); + if bun_zstd::is_error(bound) { + return Err(global.err( + jsc::ErrorCode::ZLIB_INITIALIZATION_FAILED, + format_args!("zstd compression failed: input too large"), + ).throw()); + } + if bound <= state.shared_buffer.len() { + match bun_zstd::compress(&mut state.shared_buffer, input, level) { + bun_zstd::Result::Success(n) => { + return Ok(state.shared_buffer[..n].to_vec()); + } + bun_zstd::Result::Err(msg) => { + return Err(global.err( + jsc::ErrorCode::ZLIB_INITIALIZATION_FAILED, + format_args!("zstd compression failed: {}", bstr::BStr::new(msg.as_bytes())), + ).throw()); + } + } + } + + let mut out = vec![0u8; bound]; + match bun_zstd::compress(&mut out, input, level) { + bun_zstd::Result::Success(n) => { + out.truncate(n); + Ok(out) + } + bun_zstd::Result::Err(msg) => Err(global.err( + jsc::ErrorCode::ZLIB_INITIALIZATION_FAILED, + format_args!("zstd compression failed: {}", bstr::BStr::new(msg.as_bytes())), + ).throw()), + } + }) +} diff --git a/test/js/web/fetch/fetch-compress.test.ts b/test/js/web/fetch/fetch-compress.test.ts new file mode 100644 index 000000000000..55ba49a2444b --- /dev/null +++ b/test/js/web/fetch/fetch-compress.test.ts @@ -0,0 +1,204 @@ +import { describe, expect, test } from "bun:test"; +import { brotliDecompressSync, gunzipSync, inflateSync, zstdDecompressSync } from "node:zlib"; + +const payload = JSON.stringify({ + msg: Buffer.alloc(1024, "abcdefghij").toString(), + n: 42, +}); + +function makeServer() { + return Bun.serve({ + port: 0, + async fetch(req) { + const raw = Buffer.from(await req.arrayBuffer()); + const encoding = req.headers.get("content-encoding") ?? ""; + let decoded: string; + switch (encoding) { + case "gzip": + decoded = gunzipSync(raw).toString(); + break; + case "deflate": + decoded = inflateSync(raw).toString(); + break; + case "br": + decoded = brotliDecompressSync(raw).toString(); + break; + case "zstd": + decoded = zstdDecompressSync(raw).toString(); + break; + default: + decoded = raw.toString(); + } + return Response.json({ + encoding, + contentLength: req.headers.get("content-length"), + rawLength: raw.length, + decoded, + }); + }, + }); +} + +describe("fetch compress option", () => { + describe.each([ + ["gzip", "gzip" as const], + ["deflate", "deflate" as const], + ["br", "br" as const], + ["zstd", "zstd" as const], + ])("encoding %s", (name, encoding) => { + test.concurrent("string body", async () => { + using server = makeServer(); + const res = await fetch(server.url, { + method: "POST", + body: payload, + compress: encoding, + }); + const json = await res.json(); + expect(json.encoding).toBe(encoding); + expect(json.decoded).toBe(payload); + expect(json.rawLength).toBeLessThan(Buffer.byteLength(payload)); + expect(Number(json.contentLength)).toBe(json.rawLength); + }); + + test.concurrent("Uint8Array body", async () => { + using server = makeServer(); + const res = await fetch(server.url, { + method: "POST", + body: new TextEncoder().encode(payload), + compress: encoding, + }); + const json = await res.json(); + expect(json.encoding).toBe(encoding); + expect(json.decoded).toBe(payload); + expect(json.rawLength).toBeLessThan(Buffer.byteLength(payload)); + }); + + test.concurrent("Blob body", async () => { + using server = makeServer(); + const res = await fetch(server.url, { + method: "POST", + body: new Blob([payload]), + compress: encoding, + }); + const json = await res.json(); + expect(json.encoding).toBe(encoding); + expect(json.decoded).toBe(payload); + expect(json.rawLength).toBeLessThan(Buffer.byteLength(payload)); + }); + + test.concurrent("object form with explicit level", async () => { + using server = makeServer(); + const level = encoding === "zstd" ? 3 : 4; + const res = await fetch(server.url, { + method: "POST", + body: payload, + compress: { encoding, level }, + }); + const json = await res.json(); + expect(json.encoding).toBe(encoding); + expect(json.decoded).toBe(payload); + }); + }); + + test.concurrent("compress: true defaults to gzip", async () => { + using server = makeServer(); + const res = await fetch(server.url, { + method: "POST", + body: payload, + compress: true, + }); + const json = await res.json(); + expect(json.encoding).toBe("gzip"); + expect(json.decoded).toBe(payload); + }); + + test.concurrent("compress: false sends uncompressed", async () => { + using server = makeServer(); + const res = await fetch(server.url, { + method: "POST", + body: payload, + compress: false, + }); + const json = await res.json(); + expect(json.encoding).toBe(""); + expect(json.decoded).toBe(payload); + expect(json.rawLength).toBe(Buffer.byteLength(payload)); + }); + + test.concurrent("explicit Content-Encoding header skips compression", async () => { + using server = makeServer(); + const res = await fetch(server.url, { + method: "POST", + body: payload, + headers: { "Content-Encoding": "identity" }, + compress: "gzip", + }); + const json = await res.json(); + expect(json.encoding).toBe("identity"); + expect(json.rawLength).toBe(Buffer.byteLength(payload)); + }); + + test.concurrent("ReadableStream body is not compressed", async () => { + using server = makeServer(); + const res = await fetch(server.url, { + method: "POST", + body: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(payload)); + controller.close(); + }, + }), + compress: "gzip", + }); + const json = await res.json(); + expect(json.encoding).toBe(""); + expect(json.decoded).toBe(payload); + }); + + test.concurrent("empty body is not compressed", async () => { + using server = makeServer(); + const res = await fetch(server.url, { + method: "POST", + body: "", + compress: "gzip", + }); + const json = await res.json(); + expect(json.encoding).toBe(""); + expect(json.rawLength).toBe(0); + }); + + test.concurrent("body larger than the shared buffer", async () => { + using server = makeServer(); + const big = Buffer.alloc(600 * 1024, "abcdefghij").toString(); + const res = await fetch(server.url, { + method: "POST", + body: big, + compress: "gzip", + }); + const json = await res.json(); + expect(json.encoding).toBe("gzip"); + expect(json.decoded).toBe(big); + expect(json.rawLength).toBeLessThan(big.length); + }); + + test("invalid encoding string throws", () => { + expect(() => + fetch("http://example.com", { + method: "POST", + body: "x", + // @ts-expect-error + compress: "snappy", + }), + ).toThrow(/'compress' must be/); + }); + + test("invalid level throws", () => { + expect(() => + fetch("http://example.com", { + method: "POST", + body: "x", + compress: { encoding: "gzip", level: 99 }, + }), + ).toThrow(/'compress.level'/); + }); +}); From ca9261ce675e02e93887a6bb64a58f89dc47fac6 Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Tue, 16 Jun 2026 20:30:23 +0000 Subject: [PATCH 02/14] [autofix.ci] apply automated fixes --- src/runtime/webcore/fetch/compress_body.rs | 52 +++++++++++++++------- 1 file changed, 35 insertions(+), 17 deletions(-) diff --git a/src/runtime/webcore/fetch/compress_body.rs b/src/runtime/webcore/fetch/compress_body.rs index ddb44c8e5211..342f9d144c0b 100644 --- a/src/runtime/webcore/fetch/compress_body.rs +++ b/src/runtime/webcore/fetch/compress_body.rs @@ -291,7 +291,11 @@ fn compress_brotli(global: &JSGlobalObject, input: &[u8], level: Option) -> } } - let cap = if bound != 0 { bound } else { input.len() + 1024 }; + let cap = if bound != 0 { + bound + } else { + input.len() + 1024 + }; let mut out = vec![0u8; cap]; let mut out_len = out.len(); // SAFETY: see above. @@ -307,10 +311,12 @@ fn compress_brotli(global: &JSGlobalObject, input: &[u8], level: Option) -> ) }; if ok == 0 { - return Err(global.err( - jsc::ErrorCode::ZLIB_INITIALIZATION_FAILED, - format_args!("brotli compression failed"), - ).throw()); + return Err(global + .err( + jsc::ErrorCode::ZLIB_INITIALIZATION_FAILED, + format_args!("brotli compression failed"), + ) + .throw()); } out.truncate(out_len); Ok(out) @@ -321,10 +327,12 @@ fn compress_zstd(global: &JSGlobalObject, input: &[u8], level: Option) -> J with_state(|state| { let bound = bun_zstd::compress_bound(input.len()); if bun_zstd::is_error(bound) { - return Err(global.err( - jsc::ErrorCode::ZLIB_INITIALIZATION_FAILED, - format_args!("zstd compression failed: input too large"), - ).throw()); + return Err(global + .err( + jsc::ErrorCode::ZLIB_INITIALIZATION_FAILED, + format_args!("zstd compression failed: input too large"), + ) + .throw()); } if bound <= state.shared_buffer.len() { match bun_zstd::compress(&mut state.shared_buffer, input, level) { @@ -332,10 +340,15 @@ fn compress_zstd(global: &JSGlobalObject, input: &[u8], level: Option) -> J return Ok(state.shared_buffer[..n].to_vec()); } bun_zstd::Result::Err(msg) => { - return Err(global.err( - jsc::ErrorCode::ZLIB_INITIALIZATION_FAILED, - format_args!("zstd compression failed: {}", bstr::BStr::new(msg.as_bytes())), - ).throw()); + return Err(global + .err( + jsc::ErrorCode::ZLIB_INITIALIZATION_FAILED, + format_args!( + "zstd compression failed: {}", + bstr::BStr::new(msg.as_bytes()) + ), + ) + .throw()); } } } @@ -346,10 +359,15 @@ fn compress_zstd(global: &JSGlobalObject, input: &[u8], level: Option) -> J out.truncate(n); Ok(out) } - bun_zstd::Result::Err(msg) => Err(global.err( - jsc::ErrorCode::ZLIB_INITIALIZATION_FAILED, - format_args!("zstd compression failed: {}", bstr::BStr::new(msg.as_bytes())), - ).throw()), + bun_zstd::Result::Err(msg) => Err(global + .err( + jsc::ErrorCode::ZLIB_INITIALIZATION_FAILED, + format_args!( + "zstd compression failed: {}", + bstr::BStr::new(msg.as_bytes()) + ), + ) + .throw()), } }) } From 526e2a7da8fd5fbeeae3a9cd4bf48f5b2519351e Mon Sep 17 00:00:00 2001 From: Jarred Sumner Date: Tue, 16 Jun 2026 20:37:46 +0000 Subject: [PATCH 03/14] fetch: parse compress encoding via ComptimeStringMap, drop to_utf8 --- src/runtime/webcore/fetch/compress_body.rs | 28 ++++++++++------------ 1 file changed, 12 insertions(+), 16 deletions(-) diff --git a/src/runtime/webcore/fetch/compress_body.rs b/src/runtime/webcore/fetch/compress_body.rs index 342f9d144c0b..a145d960f426 100644 --- a/src/runtime/webcore/fetch/compress_body.rs +++ b/src/runtime/webcore/fetch/compress_body.rs @@ -9,6 +9,7 @@ use core::cell::UnsafeCell; use crate::webcore::jsc::{self, JSGlobalObject, JSValue, JsResult}; +use bun_jsc::ComptimeStringMapExt as _; #[derive(Copy, Clone, Eq, PartialEq, Debug)] pub enum CompressEncoding { @@ -18,6 +19,15 @@ pub enum CompressEncoding { Zstd, } +bun_core::comptime_string_map! { + static COMPRESS_ENCODING_MAP: CompressEncoding = { + b"gzip" => CompressEncoding::Gzip, + b"deflate" => CompressEncoding::Deflate, + b"br" => CompressEncoding::Brotli, + b"zstd" => CompressEncoding::Zstd, + }; +} + impl CompressEncoding { pub fn header_value(self) -> &'static [u8] { match self { @@ -27,16 +37,6 @@ impl CompressEncoding { CompressEncoding::Zstd => b"zstd", } } - - fn from_str(s: &[u8]) -> Option { - match s { - b"gzip" => Some(CompressEncoding::Gzip), - b"deflate" => Some(CompressEncoding::Deflate), - b"br" => Some(CompressEncoding::Brotli), - b"zstd" => Some(CompressEncoding::Zstd), - _ => None, - } - } } #[derive(Copy, Clone, Debug)] @@ -63,9 +63,7 @@ impl CompressOption { }); } if value.is_string() { - let s = bun_core::OwnedString::new(value.to_bun_string(global)?); - let bytes = s.to_utf8(); - return match CompressEncoding::from_str(bytes.slice()) { + return match COMPRESS_ENCODING_MAP.from_js(global, value)? { Some(encoding) => Ok(Some(CompressOption { encoding, level: None, @@ -78,9 +76,7 @@ impl CompressOption { if value.is_object() { let encoding = match value.get(global, "encoding")? { Some(enc) if enc.is_string() => { - let s = bun_core::OwnedString::new(enc.to_bun_string(global)?); - let bytes = s.to_utf8(); - match CompressEncoding::from_str(bytes.slice()) { + match COMPRESS_ENCODING_MAP.from_js(global, enc)? { Some(e) => e, None => { return Err(global.throw_invalid_arguments(format_args!( From f2a097f8dbef9b4eb6b657816729ce1021c49640 Mon Sep 17 00:00:00 2001 From: Jarred Sumner Date: Tue, 16 Jun 2026 21:37:48 +0000 Subject: [PATCH 04/14] fetch(compress): add streaming-decompressor leak test; inline compressor_mut - Add fragmented-response leak test: drips gzip/br/zstd over chunked TE to force the per-request boxed Decompressor path (bypasses libdeflate one-shot fast path), paired with a 700 KiB compress: request body for multi-write send. - Drop CompressorState::compressor_mut<'a>(&self) -> &'a mut and inline its single call site with a SAFETY comment; the unbounded-lifetime signature was correct (C heap handle disjoint from shared_buffer) but read as unsound. --- src/runtime/webcore/fetch/compress_body.rs | 23 +-- test/js/web/fetch/fetch-compress.test.ts | 4 +- test/js/web/fetch/fetch-leak.test.ts | 180 +++++++++++++++++++++ 3 files changed, 190 insertions(+), 17 deletions(-) diff --git a/src/runtime/webcore/fetch/compress_body.rs b/src/runtime/webcore/fetch/compress_body.rs index a145d960f426..6411abcb9516 100644 --- a/src/runtime/webcore/fetch/compress_body.rs +++ b/src/runtime/webcore/fetch/compress_body.rs @@ -85,11 +85,11 @@ impl CompressOption { } } } - _ => { + other => { return Err(global.throw_invalid_argument_type_value( b"compress.encoding", b"string", - value, + other.unwrap_or(JSValue::UNDEFINED), )); } }; @@ -152,18 +152,6 @@ struct CompressorState { // SAFETY: `*mut T` (null) and `[u8; N]` are both valid at the all-zero bit pattern. unsafe impl bun_core::Zeroable for CompressorState {} -impl CompressorState { - #[inline] - fn compressor_mut<'a>(&self) -> &'a mut bun_libdeflate_sys::libdeflate::Compressor { - // SAFETY: `compressor` is set once in `with_state` from - // `libdeflate_alloc_compressor` (panics on null) and never freed for - // the thread's lifetime. The handle is a separate C heap allocation - // disjoint from `self`, so the returned `&mut` does not alias - // `shared_buffer`. Thread-local — sole live borrow. - unsafe { &mut *self.compressor } - } -} - thread_local! { static LAZY_COMPRESSOR: UnsafeCell>> = const { UnsafeCell::new(None) }; @@ -233,7 +221,12 @@ fn compress_libdeflate( // SAFETY: just allocated, non-null, exclusive. unsafe { &mut *tmp } } - _ => state.compressor_mut(), + // SAFETY: `state.compressor` is set once in `with_state` from + // `libdeflate_alloc_compressor` (panics on null) and never freed + // for the thread's lifetime. The handle is a separate C heap + // allocation disjoint from `state.shared_buffer`, so this `&mut` + // does not alias the buffer borrow below. + _ => unsafe { &mut *state.compressor }, }; let _guard = scopeguard::guard(tmp, |tmp| { if !tmp.is_null() { diff --git a/test/js/web/fetch/fetch-compress.test.ts b/test/js/web/fetch/fetch-compress.test.ts index 55ba49a2444b..56265c25c19f 100644 --- a/test/js/web/fetch/fetch-compress.test.ts +++ b/test/js/web/fetch/fetch-compress.test.ts @@ -183,7 +183,7 @@ describe("fetch compress option", () => { test("invalid encoding string throws", () => { expect(() => - fetch("http://example.com", { + fetch("http://127.0.0.1:1/", { method: "POST", body: "x", // @ts-expect-error @@ -194,7 +194,7 @@ describe("fetch compress option", () => { test("invalid level throws", () => { expect(() => - fetch("http://example.com", { + fetch("http://127.0.0.1:1/", { method: "POST", body: "x", compress: { encoding: "gzip", level: 99 }, diff --git a/test/js/web/fetch/fetch-leak.test.ts b/test/js/web/fetch/fetch-leak.test.ts index f32c2c7fbd99..7f9a6f39e0fa 100644 --- a/test/js/web/fetch/fetch-leak.test.ts +++ b/test/js/web/fetch/fetch-leak.test.ts @@ -235,6 +235,186 @@ test("fetch(data:) with percent-encoding does not leak", async () => { expect(exitCode).toBe(0); }, 60000); +test("fetch() compress option does not leak bodies or compressor state", async () => { + // Exercises: + // - all four encodings + // - the custom-level path (allocates a temporary libdeflate compressor that + // must be freed each call) + // - a small body (thread-local 512 KiB shared-buffer fast path) + // - a ~700 KiB body (slow-path Vec allocation + multi-write send) + using server = Bun.serve({ + port: 0, + idleTimeout: 0, + async fetch(req) { + // Drain the body so the request completes; don't decompress (server-side + // allocator noise would mask the client-side signal we're measuring). + await req.arrayBuffer(); + return new Response(); + }, + }); + + const script = /* js */ ` + const url = process.env.SERVER; + const small = Buffer.alloc(32 * 1024, "abcdefghij"); + // > 512 KiB shared buffer → slow path; large enough that the compressed + // output also spans multiple socket writes. + const big = Buffer.alloc(700 * 1024, "abcdefghij"); + const opts = [ + { compress: "gzip" }, + { compress: "deflate" }, + { compress: "br" }, + { compress: "zstd" }, + // custom level → temp libdeflate compressor alloc/free each call + { compress: { encoding: "gzip", level: 1 } }, + ]; + + async function round() { + const promises = []; + for (const opt of opts) { + promises.push(fetch(url, { method: "POST", body: small, ...opt }).then(r => r.arrayBuffer())); + promises.push(fetch(url, { method: "POST", body: big, ...opt }).then(r => r.arrayBuffer())); + } + await Promise.all(promises); + } + + // Warm up: thread-local CompressorState is allocated once here and stays. + for (let i = 0; i < 10; i++) await round(); + Bun.gc(true); + const baseline = process.memoryUsage.rss(); + + for (let i = 0; i < 80; i++) await round(); + Bun.gc(true); + const final = process.memoryUsage.rss(); + + const deltaMB = (final - baseline) / 1024 / 1024; + console.log(JSON.stringify({ baselineMB: (baseline / 1024 / 1024) | 0, finalMB: (final / 1024 / 1024) | 0, deltaMB: Math.round(deltaMB) })); + // 80 rounds × 5 encodings × ~700 KiB bodies → a per-request leak of the + // body or compressor state would grow RSS by hundreds of MB. + if (deltaMB > ${isASAN ? 256 : 32}) { + throw new Error("fetch({compress}) leaked " + Math.round(deltaMB) + " MB over 80 rounds"); + } + `; + + await using proc = Bun.spawn({ + cmd: [bunExe(), "--smol", "-e", script], + env: { ...bunEnv, SERVER: server.url.href }, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + console.log(stdout.trim()); + expect({ stderr, exitCode }).toEqual({ stderr: "", exitCode: 0 }); +}, 60000); + +// Response-side: a compressed body that arrives across many packets bypasses +// the libdeflate one-shot fast path (InternalState.rs:256) and allocates a +// boxed streaming Decompressor (zlib/brotli/zstd FFI handle) per request, +// freed via Drop on InternalState::reset(). Dripping the body over chunked +// transfer encoding forces handle_response_body_chunked_encoding_from_multiple_packets. +// Paired with compress: on the request side so the same loop covers the +// multi-write send of a large compressed request body too. +test("fetch() does not leak streaming decompressor state across fragmented compressed responses", async () => { + const script = /* js */ ` + import { createServer } from "node:net"; + import { gzipSync, brotliCompressSync, zstdCompressSync } from "node:zlib"; + + const plain = Buffer.alloc(64 * 1024, "abcdefghij"); + const bodies = { + gzip: gzipSync(plain), + br: brotliCompressSync(plain), + zstd: zstdCompressSync(plain), + }; + // ~700 KiB request body → compressed output exceeds the 512 KiB shared + // buffer and spans multiple socket writes. + const reqBody = Buffer.alloc(700 * 1024, "abcdefghij"); + + const server = createServer(sock => { + let buf = ""; + sock.on("data", async chunk => { + buf += chunk.toString("latin1"); + // Drain the request: headers + (chunked) body terminator. + if (!buf.includes("\\r\\n\\r\\n")) return; + const isChunked = /transfer-encoding:\\s*chunked/i.test(buf); + if (isChunked && !buf.includes("\\r\\n0\\r\\n\\r\\n")) return; + if (!isChunked) { + const m = buf.match(/content-length:\\s*(\\d+)/i); + const need = m ? Number(m[1]) : 0; + const bodyStart = buf.indexOf("\\r\\n\\r\\n") + 4; + if (buf.length - bodyStart < need) return; + } + const enc = buf.match(/x-want:\\s*(\\w+)/i)[1]; + const body = bodies[enc]; + sock.write( + "HTTP/1.1 200 OK\\r\\n" + + "Content-Encoding: " + enc + "\\r\\n" + + "Transfer-Encoding: chunked\\r\\n" + + "Connection: close\\r\\n\\r\\n", + ); + // Drip in small chunks so the client's Decompressor sees many + // update_buffers()/read_all() cycles before the final flush. + for (let i = 0; i < body.length; i += 256) { + const piece = body.subarray(i, i + 256); + sock.write(piece.length.toString(16) + "\\r\\n"); + sock.write(piece); + sock.write("\\r\\n"); + // Yield so chunks land in separate packets. + await new Promise(r => setImmediate(r)); + } + sock.end("0\\r\\n\\r\\n"); + }); + sock.on("error", () => {}); + }).listen(0, "127.0.0.1"); + await new Promise(r => server.once("listening", r)); + const url = "http://127.0.0.1:" + server.address().port + "/"; + + async function round() { + const promises = []; + for (const enc of ["gzip", "br", "zstd"]) { + promises.push( + fetch(url, { + method: "POST", + body: reqBody, + compress: enc, + headers: { "x-want": enc }, + }).then(async r => { + const got = Buffer.from(await r.arrayBuffer()); + if (!got.equals(plain)) throw new Error(enc + " round-trip mismatch (" + got.length + ")"); + }), + ); + } + await Promise.all(promises); + } + + for (let i = 0; i < 10; i++) await round(); + Bun.gc(true); + const baseline = process.memoryUsage.rss(); + + for (let i = 0; i < 60; i++) await round(); + Bun.gc(true); + const final = process.memoryUsage.rss(); + + server.close(); + const deltaMB = (final - baseline) / 1024 / 1024; + console.log(JSON.stringify({ baselineMB: (baseline / 1024 / 1024) | 0, finalMB: (final / 1024 / 1024) | 0, deltaMB: Math.round(deltaMB) })); + // 60 rounds × 3 encodings = 180 streaming Decompressor handles + 180 + // ~700 KiB compressed request bodies. A leaked boxed zlib/brotli/zstd + // reader or an un-freed compressed body Vec would grow RSS by >100 MB. + if (deltaMB > ${isASAN ? 256 : 32}) { + throw new Error("fragmented compressed fetch leaked " + Math.round(deltaMB) + " MB over 60 rounds"); + } + `; + + await using proc = Bun.spawn({ + cmd: [bunExe(), "--smol", "-e", script], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + console.log(stdout.trim()); + expect({ stderr, exitCode }).toEqual({ stderr: "", exitCode: 0 }); +}, 60000); + // Regression for src/runtime/webcore/fetch/FetchTasklet.zig:601,614 — // Holder.resolve/reject use `self.promise.swap()` which *consumes* (clears) // the jsc.Strong handle on the fetch() promise before calling resolve()/reject(). From 018a3db6f3f38dba3a6f5fd4ef4f32f5763a58c4 Mon Sep 17 00:00:00 2001 From: Jarred Sumner Date: Tue, 16 Jun 2026 22:16:50 +0000 Subject: [PATCH 05/14] fetch(compress): disable sendfile when compress is set; reject NaN/non-integer level - Bun.file() bodies that qualified for the sendfile fast path (plain http, no proxy, >=32 KiB, non-Windows) silently skipped explicit compression while the same call over https/proxy/small-file/Windows compressed. Gate sendfile on compress.is_none() so an explicit request always wins. - compress.level: NaN/5.5/Infinity passed is_number() and to_int32() mapped them to in-range values (NaN->0). Validate via as_number() + is_nan()/fract() like maxRedirects does. --- src/runtime/webcore/fetch.rs | 5 ++++- src/runtime/webcore/fetch/compress_body.rs | 10 +++++++-- test/js/web/fetch/fetch-compress.test.ts | 24 ++++++++++++++++++++-- 3 files changed, 34 insertions(+), 5 deletions(-) diff --git a/src/runtime/webcore/fetch.rs b/src/runtime/webcore/fetch.rs index 504abd4cd02c..6cc7392d8950 100644 --- a/src/runtime/webcore/fetch.rs +++ b/src/runtime/webcore/fetch.rs @@ -1656,7 +1656,10 @@ fn fetch_impl( Ok(fd) => fd, }; - if proxy.is_none() && http::SendFile::is_eligible(&url) { + // An explicit `compress` request always wins over the sendfile + // heuristic — otherwise the same `Bun.file()` body would compress + // over https/proxy/<32 KiB/Windows but silently not over plain http. + if proxy.is_none() && compress.is_none() && http::SendFile::is_eligible(&url) { 'use_sendfile: { let stat: bun_sys::Stat = match bun_sys::fstat(opened_fd) { Ok(result) => result, diff --git a/src/runtime/webcore/fetch/compress_body.rs b/src/runtime/webcore/fetch/compress_body.rs index 6411abcb9516..c51d8087f794 100644 --- a/src/runtime/webcore/fetch/compress_body.rs +++ b/src/runtime/webcore/fetch/compress_body.rs @@ -98,11 +98,17 @@ impl CompressOption { if !lvl.is_number() { return Err(global.throw_invalid_argument_type_value( b"compress.level", - b"number", + b"integer", lvl, )); } - let n = lvl.to_int32(); + let raw = lvl.as_number(); + if raw.is_nan() || raw.fract() != 0.0 { + return Err(global.throw_invalid_arguments(format_args!( + "fetch: 'compress.level' must be an integer" + ))); + } + let n = raw as i32; let (min, max) = match encoding { CompressEncoding::Gzip | CompressEncoding::Deflate => (0, 12), CompressEncoding::Brotli => ( diff --git a/test/js/web/fetch/fetch-compress.test.ts b/test/js/web/fetch/fetch-compress.test.ts index 56265c25c19f..fe0da1436067 100644 --- a/test/js/web/fetch/fetch-compress.test.ts +++ b/test/js/web/fetch/fetch-compress.test.ts @@ -1,4 +1,6 @@ import { describe, expect, test } from "bun:test"; +import { isWindows, tempDir } from "harness"; +import { join } from "node:path"; import { brotliDecompressSync, gunzipSync, inflateSync, zstdDecompressSync } from "node:zlib"; const payload = JSON.stringify({ @@ -192,13 +194,31 @@ describe("fetch compress option", () => { ).toThrow(/'compress' must be/); }); - test("invalid level throws", () => { + test.each([99, NaN, 5.5, Infinity])("invalid level %p throws", level => { expect(() => fetch("http://127.0.0.1:1/", { method: "POST", body: "x", - compress: { encoding: "gzip", level: 99 }, + compress: { encoding: "gzip", level }, }), ).toThrow(/'compress.level'/); }); + + // A Bun.file() body large enough to qualify for the sendfile fast path + // (≥32 KiB, plain http, no proxy, non-Windows) must still be compressed + // when compress is explicitly set — the sendfile heuristic must not win. + test.skipIf(isWindows)("Bun.file() body large enough for sendfile is still compressed", async () => { + using server = makeServer(); + const big = Buffer.alloc(64 * 1024, "abcdefghij").toString(); + using dir = tempDir("fetch-compress-sendfile", { "body.txt": big }); + const res = await fetch(server.url, { + method: "POST", + body: Bun.file(join(String(dir), "body.txt")), + compress: "gzip", + }); + const json = await res.json(); + expect(json.encoding).toBe("gzip"); + expect(json.decoded).toBe(big); + expect(json.rawLength).toBeLessThan(big.length); + }); }); From 2794226d1f087f82f927a7179fdb2f70abfd02ef Mon Sep 17 00:00:00 2001 From: Jarred Sumner Date: Tue, 16 Jun 2026 22:18:55 +0000 Subject: [PATCH 06/14] fetch(compress): validate compress.level via validate_integer_range --- src/runtime/webcore/fetch/compress_body.rs | 43 +++++++++------------- test/js/web/fetch/fetch-compress.test.ts | 5 ++- 2 files changed, 20 insertions(+), 28 deletions(-) diff --git a/src/runtime/webcore/fetch/compress_body.rs b/src/runtime/webcore/fetch/compress_body.rs index c51d8087f794..f1dd769c0704 100644 --- a/src/runtime/webcore/fetch/compress_body.rs +++ b/src/runtime/webcore/fetch/compress_body.rs @@ -95,37 +95,27 @@ impl CompressOption { }; let level = match value.get(global, "level")? { Some(lvl) if !lvl.is_undefined_or_null() => { - if !lvl.is_number() { - return Err(global.throw_invalid_argument_type_value( - b"compress.level", - b"integer", - lvl, - )); - } - let raw = lvl.as_number(); - if raw.is_nan() || raw.fract() != 0.0 { - return Err(global.throw_invalid_arguments(format_args!( - "fetch: 'compress.level' must be an integer" - ))); - } - let n = raw as i32; - let (min, max) = match encoding { - CompressEncoding::Gzip | CompressEncoding::Deflate => (0, 12), + let (min, max, default) = match encoding { + CompressEncoding::Gzip | CompressEncoding::Deflate => { + (0, 12, DEFAULT_DEFLATE_LEVEL) + } CompressEncoding::Brotli => ( bun_brotli::c::BROTLI_MIN_QUALITY, bun_brotli::c::BROTLI_MAX_QUALITY, + DEFAULT_BROTLI_QUALITY, ), - CompressEncoding::Zstd => (1, 22), + CompressEncoding::Zstd => (1, 22, DEFAULT_ZSTD_LEVEL), }; - if n < min || n > max { - return Err(global.throw_invalid_arguments(format_args!( - "fetch: 'compress.level' for \"{}\" must be between {} and {}", - bstr::BStr::new(encoding.header_value()), - min, - max, - ))); - } - Some(n) + Some(global.validate_integer_range::( + lvl, + default, + bun_jsc::IntegerRange { + min: i128::from(min), + max: i128::from(max), + field_name: b"compress.level", + always_allow_zero: false, + }, + )?) } _ => None, }; @@ -147,6 +137,7 @@ impl CompressOption { /// `compress: true` / `compress: "gzip"` path never allocates a fresh handle. const DEFAULT_DEFLATE_LEVEL: i32 = 6; const DEFAULT_BROTLI_QUALITY: i32 = 6; +const DEFAULT_ZSTD_LEVEL: i32 = 3; const SHARED_BUFFER_SIZE: usize = 512 * 1024; diff --git a/test/js/web/fetch/fetch-compress.test.ts b/test/js/web/fetch/fetch-compress.test.ts index fe0da1436067..938508871f54 100644 --- a/test/js/web/fetch/fetch-compress.test.ts +++ b/test/js/web/fetch/fetch-compress.test.ts @@ -194,14 +194,15 @@ describe("fetch compress option", () => { ).toThrow(/'compress' must be/); }); - test.each([99, NaN, 5.5, Infinity])("invalid level %p throws", level => { + test.each([99, -1, 5.5, Infinity, "6"])("invalid level %p throws", level => { expect(() => fetch("http://127.0.0.1:1/", { method: "POST", body: "x", + // @ts-expect-error compress: { encoding: "gzip", level }, }), - ).toThrow(/'compress.level'/); + ).toThrow(/compress\.level/); }); // A Bun.file() body large enough to qualify for the sendfile fast path From aada0306ed2001f9674c2b9548951e8bd427a479 Mon Sep 17 00:00:00 2001 From: Jarred Sumner Date: Tue, 16 Jun 2026 22:47:57 +0000 Subject: [PATCH 07/14] fetch(compress): move request-body compression to the HTTP thread Compression now runs in HTTPClient::start() on the HTTP thread, reusing HttpThread.lazy_libdeflater (LibdeflateState gains a lazy compressor handle and the existing 512 KiB shared_buffer is reused for the fast path). - bun_http::compress_body: CompressEncoding/CompressOption + compress_into() taking &mut LibdeflateState; returns bun_core::Error on encoder failure. - HTTPClient: new compress + compressed_request_body fields. start() compresses Bytes bodies before InternalState::init; compress is .take()n so retry/h2-retry re-entries don't double-compress. The compressed Vec is freed in on_async_http_callback_raw alongside redirect/prev_redirect (the threadlocal clone is dealloc'd without Drop, so clone-owned state is torn down explicitly). - async_http::Options.compress threaded through AsyncHTTP::init. - runtime/webcore/fetch/compress_body.rs is now just from_js() re-exporting the bun_http types; the JS-thread thread_local! CompressorState is gone. - fetch_impl no longer compresses; it appends Content-Encoding and forwards the option via FetchOptions. Content-Length was already computed on the HTTP thread in build_request from original_request_body.len(). --- src/http/AsyncHTTP.rs | 5 + src/http/HTTPThread.rs | 23 ++ src/http/compress_body.rs | 205 ++++++++++ src/http/lib.rs | 39 +- src/runtime/webcore/fetch.rs | 33 +- src/runtime/webcore/fetch/FetchTasklet.rs | 3 + src/runtime/webcore/fetch/compress_body.rs | 415 ++++----------------- 7 files changed, 367 insertions(+), 356 deletions(-) create mode 100644 src/http/compress_body.rs diff --git a/src/http/AsyncHTTP.rs b/src/http/AsyncHTTP.rs index a127c5a3d5b8..a307eb3db4ec 100644 --- a/src/http/AsyncHTTP.rs +++ b/src/http/AsyncHTTP.rs @@ -213,6 +213,8 @@ fn make_client<'a>( async_http_id, hostname, unix_socket_path: ZigStringSlice::EMPTY, + compress: None, + compressed_request_body: Vec::new(), } } @@ -270,6 +272,7 @@ pub struct Options<'a> { pub max_redirects: Option, pub reject_unauthorized: Option, pub tls_props: Option, + pub compress: Option, } // ────────────────────────────────────────────────────────────────────────── @@ -530,6 +533,7 @@ impl<'a> AsyncHTTP<'a> { if let Some(val) = options.tls_props { this.client.tls_props = Some(val); } + this.client.compress = options.compress; if let Some(proxy) = &this.http_proxy { if let Some(auth) = build_proxy_authorization(proxy) { @@ -768,6 +772,7 @@ impl<'a> AsyncHTTP<'a> { // Clone-owned (allocated after `ptr::read`). drop(core::mem::take(&mut client.redirect)); drop(core::mem::take(&mut client.prev_redirect)); + drop(core::mem::take(&mut client.compressed_request_body)); if let Some(tunnel) = client.proxy_tunnel.take() { // SAFETY: tunnel was created by ProxyTunnel::start // (heap::alloc) and is refcounted; detach the socket diff --git a/src/http/HTTPThread.rs b/src/http/HTTPThread.rs index 3d4fc21ee488..e7a752acf406 100644 --- a/src/http/HTTPThread.rs +++ b/src/http/HTTPThread.rs @@ -283,6 +283,7 @@ pub struct CertCheckResumeMessage { pub struct LibdeflateState { pub decompressor: *mut bun_libdeflate_sys::libdeflate::Decompressor, + pub compressor: *mut bun_libdeflate_sys::libdeflate::Compressor, pub shared_buffer: [u8; 512 * 1024], } @@ -306,6 +307,28 @@ impl LibdeflateState { // SAFETY: see INVARIANT above. unsafe { &mut *self.decompressor } } + + /// Lazy libdeflate compressor at [`DEFAULT_DEFLATE_LEVEL`]; same lifetime + /// and aliasing invariants as [`decompressor_mut`]. + /// + /// [`DEFAULT_DEFLATE_LEVEL`]: crate::compress_body::DEFAULT_DEFLATE_LEVEL + /// [`decompressor_mut`]: Self::decompressor_mut + #[inline] + pub(crate) fn compressor_mut<'a>( + &mut self, + ) -> &'a mut bun_libdeflate_sys::libdeflate::Compressor { + if self.compressor.is_null() { + self.compressor = bun_libdeflate_sys::libdeflate::Compressor::alloc( + crate::compress_body::DEFAULT_DEFLATE_LEVEL, + ); + if self.compressor.is_null() { + bun_core::out_of_memory(); + } + } + // SAFETY: just ensured non-null; HTTP-thread-only; separate C heap + // allocation disjoint from `shared_buffer`. + unsafe { &mut *self.compressor } + } } pub const REQUEST_BODY_SEND_STACK_BUFFER_SIZE: usize = 32 * 1024; diff --git a/src/http/compress_body.rs b/src/http/compress_body.rs new file mode 100644 index 000000000000..55b825a57f51 --- /dev/null +++ b/src/http/compress_body.rs @@ -0,0 +1,205 @@ +//! Automatic request-body compression for `fetch({ compress })`. +//! +//! Runs on the HTTP thread inside `HTTPClient::start()` so it can reuse +//! [`LibdeflateState`]'s 512 KiB scratch buffer and cached libdeflate handle +//! (the same struct that backs the response-decompression fast path). +//! gzip/deflate go through libdeflate; brotli/zstd use their one-shot +//! encoders. Only buffered bodies reach this — streams and sendfile are +//! filtered out on the JS thread. + +use crate::http_thread::LibdeflateState; + +#[derive(Copy, Clone, Eq, PartialEq, Debug)] +pub enum CompressEncoding { + Gzip, + Deflate, + Brotli, + Zstd, +} + +impl CompressEncoding { + pub fn header_value(self) -> &'static [u8] { + match self { + CompressEncoding::Gzip => b"gzip", + CompressEncoding::Deflate => b"deflate", + CompressEncoding::Brotli => b"br", + CompressEncoding::Zstd => b"zstd", + } + } +} + +#[derive(Copy, Clone, Debug)] +pub struct CompressOption { + pub encoding: CompressEncoding, + pub level: Option, +} + +/// libdeflate's default level. Reused for the cached compressor so the common +/// `compress: true` / `compress: "gzip"` path never allocates a fresh handle. +pub const DEFAULT_DEFLATE_LEVEL: i32 = 6; +pub const DEFAULT_BROTLI_QUALITY: i32 = 6; +pub const DEFAULT_ZSTD_LEVEL: i32 = 3; + +/// One-shot body compression into `out` (cleared first). HTTP-thread-only. +pub(crate) fn compress_into( + state: &mut LibdeflateState, + input: &[u8], + opt: &CompressOption, + out: &mut Vec, +) -> Result<(), bun_core::Error> { + out.clear(); + match opt.encoding { + CompressEncoding::Gzip | CompressEncoding::Deflate => { + let enc = if opt.encoding == CompressEncoding::Gzip { + bun_libdeflate_sys::libdeflate::Encoding::Gzip + } else { + // HTTP "deflate" is the zlib-wrapped DEFLATE stream (RFC 9110 + // §8.4.1.2); libdeflate's `Deflate` is the raw stream. + bun_libdeflate_sys::libdeflate::Encoding::Zlib + }; + compress_libdeflate(state, input, enc, opt.level, out); + Ok(()) + } + CompressEncoding::Brotli => compress_brotli(state, input, opt.level, out), + CompressEncoding::Zstd => compress_zstd(state, input, opt.level, out), + } +} + +fn compress_libdeflate( + state: &mut LibdeflateState, + input: &[u8], + enc: bun_libdeflate_sys::libdeflate::Encoding, + level: Option, + out: &mut Vec, +) { + use bun_libdeflate_sys::libdeflate::Compressor; + + // Custom level → allocate a temporary compressor; the cached handle is + // pinned to DEFAULT_DEFLATE_LEVEL. + let mut tmp: *mut Compressor = core::ptr::null_mut(); + let compressor: &mut Compressor = match level { + Some(l) if l != DEFAULT_DEFLATE_LEVEL => { + tmp = Compressor::alloc(l); + if tmp.is_null() { + bun_core::out_of_memory(); + } + // SAFETY: just allocated, non-null, exclusive. + unsafe { &mut *tmp } + } + _ => state.compressor_mut(), + }; + let _guard = scopeguard::guard(tmp, |tmp| { + if !tmp.is_null() { + // SAFETY: `tmp` was returned by `Compressor::alloc` above and is + // not used after this call. + unsafe { Compressor::destroy(tmp) }; + } + }); + + // Fast path: compress into the shared buffer, then copy out. + let bound = compressor.max_bytes_needed(input, enc); + if bound <= state.shared_buffer.len() { + let result = compressor.compress(input, &mut state.shared_buffer, enc); + out.extend_from_slice(&state.shared_buffer[..result.written]); + return; + } + + // Slow path: body is large; allocate the bound up front and compress + // directly into the Vec's spare capacity. + out.reserve(bound); + compressor.compress_to_vec(input, out, enc); +} + +fn compress_brotli( + state: &mut LibdeflateState, + input: &[u8], + level: Option, + out: &mut Vec, +) -> Result<(), bun_core::Error> { + use bun_brotli::c; + let quality = level.unwrap_or(DEFAULT_BROTLI_QUALITY); + + let bound = c::BrotliEncoderMaxCompressedSize(input.len()); + // BrotliEncoderMaxCompressedSize returns 0 when the bound would overflow + // size_t — fall back to a heap buffer in that case. + if bound != 0 && bound <= state.shared_buffer.len() { + let mut out_len = state.shared_buffer.len(); + // SAFETY: input/output slices are valid for their lengths; + // BrotliEncoderCompress only reads `input` and writes `out_len` + // bytes to `shared_buffer`, updating `out_len` to bytes written. + let ok = unsafe { + c::BrotliEncoderCompress( + quality, + c::BROTLI_DEFAULT_WINDOW, + c::BrotliEncoderMode::generic, + input.len(), + input.as_ptr(), + &raw mut out_len, + state.shared_buffer.as_mut_ptr(), + ) + }; + if ok != 0 { + out.extend_from_slice(&state.shared_buffer[..out_len]); + return Ok(()); + } + } + + let cap = if bound != 0 { + bound + } else { + input.len() + 1024 + }; + out.resize(cap, 0); + let mut out_len = out.len(); + // SAFETY: see above. + let ok = unsafe { + c::BrotliEncoderCompress( + quality, + c::BROTLI_DEFAULT_WINDOW, + c::BrotliEncoderMode::generic, + input.len(), + input.as_ptr(), + &raw mut out_len, + out.as_mut_ptr(), + ) + }; + if ok == 0 { + out.clear(); + return Err(bun_core::err!(CompressionFailed)); + } + out.truncate(out_len); + Ok(()) +} + +fn compress_zstd( + state: &mut LibdeflateState, + input: &[u8], + level: Option, + out: &mut Vec, +) -> Result<(), bun_core::Error> { + let bound = bun_zstd::compress_bound(input.len()); + if bun_zstd::is_error(bound) { + return Err(bun_core::err!(CompressionFailed)); + } + if bound <= state.shared_buffer.len() { + return match bun_zstd::compress(&mut state.shared_buffer, input, level) { + bun_zstd::Result::Success(n) => { + out.extend_from_slice(&state.shared_buffer[..n]); + Ok(()) + } + bun_zstd::Result::Err(_) => Err(bun_core::err!(CompressionFailed)), + }; + } + + out.resize(bound, 0); + match bun_zstd::compress(out, input, level) { + bun_zstd::Result::Success(n) => { + out.truncate(n); + Ok(()) + } + bun_zstd::Result::Err(_) => { + out.clear(); + Err(bun_core::err!(CompressionFailed)) + } + } +} diff --git a/src/http/lib.rs b/src/http/lib.rs index 82ba943c9e12..0a44be00776d 100644 --- a/src/http/lib.rs +++ b/src/http/lib.rs @@ -5,6 +5,7 @@ pub mod async_http; #[path = "CertificateInfo.rs"] pub mod certificate_info; +pub mod compress_body; #[path = "Decompressor.rs"] pub mod decompressor; #[path = "H2Client.rs"] @@ -650,6 +651,14 @@ pub struct HTTPClient<'a> { pub async_http_id: u32, pub hostname: Option<&'a [u8]>, pub unix_socket_path: ZigStringSlice, + /// `fetch({ compress })` — when set, [`Self::start`] compresses the + /// `Bytes` body on the HTTP thread before connecting. Cleared after the + /// first compression so retry/h2-retry paths that re-enter `start()` with + /// the already-compressed slice don't double-compress. + pub compress: Option, + /// Backing storage for the compressed body; `state.original_request_body` + /// borrows from this for the lifetime of the request. + pub compressed_request_body: Vec, } impl<'a> HTTPClient<'a> { @@ -2407,9 +2416,37 @@ impl<'a> HTTPClient<'a> { self.url.is_https() } - pub fn start(&mut self, body: HTTPRequestBody<'a>, body_out_str: &mut MutableString) { + pub fn start(&mut self, mut body: HTTPRequestBody<'a>, body_out_str: &mut MutableString) { body_out_str.reset(); + if let Some(opt) = self.compress.take() + && let HTTPRequestBody::Bytes(input) = body + && !input.is_empty() + { + match compress_body::compress_into( + http_thread().deflater(), + input, + &opt, + &mut self.compressed_request_body, + ) { + Ok(()) => { + // SAFETY: `compressed_request_body` lives on `self` for the + // request's lifetime; never reallocated after this point + // (`compress` was just `.take()`n). `state.original_request_body` + // / `state.request_body` borrow it until the terminal + // result-callback drops the client. + body = HTTPRequestBody::Bytes(unsafe { + &*core::ptr::from_ref(self.compressed_request_body.as_slice()) + }); + } + Err(e) => { + self.state = InternalState::init(body, body_out_str); + self.fail(e); + return; + } + } + } + debug_assert!(self.state.response_message_buffer.list.capacity() == 0); self.state = InternalState::init(body, body_out_str); diff --git a/src/runtime/webcore/fetch.rs b/src/runtime/webcore/fetch.rs index 6cc7392d8950..59d2cb4981db 100644 --- a/src/runtime/webcore/fetch.rs +++ b/src/runtime/webcore/fetch.rs @@ -679,8 +679,7 @@ fn fetch_impl( if !obj.is_empty() { if let Some(compress_value) = obj.get(global_this, "compress")? { if !compress_value.is_undefined() { - compress = - compress_body::CompressOption::from_js(global_this, compress_value)?; + compress = compress_body::from_js(global_this, compress_value)?; break 'extract_compress; } } @@ -1769,8 +1768,11 @@ fn fetch_impl( // Automatic request-body compression. Only buffered bodies (Blob bytes, // ArrayBuffer/TypedArray, string) are handled; ReadableStream and sendfile // are skipped. S3 destinations replace the header set with a signed one, - // so compression is skipped there too. - if let Some(compress_opt) = compress + // so compression is skipped there too. The actual compression runs on the + // HTTP thread (HTTPClient::start) so it can reuse LibdeflateState's shared + // scratch buffer; here we only commit to it by appending Content-Encoding + // and forwarding the option. + if let Some(compress_opt) = &compress && let HTTPRequestBody::AnyBlob(_) = &body && !url.is_s3() { @@ -1778,21 +1780,15 @@ fn fetch_impl( .as_ref() .and_then(|h| h.get_content_encoding()) .is_some(); - if !already_has_encoding { - let input = body.slice(); - if !input.is_empty() { - let compressed = - compress_body::compress_request_body(global_this, input, &compress_opt)?; - let mut old = core::mem::replace( - &mut body, - HTTPRequestBody::AnyBlob(blob::Any::from_owned_slice(compressed)), - ); - old.detach(); - headers - .get_or_insert_default() - .append(b"Content-Encoding", compress_opt.encoding.header_value()); - } + if !already_has_encoding && !body.slice().is_empty() { + headers + .get_or_insert_default() + .append(b"Content-Encoding", compress_opt.encoding.header_value()); + } else { + compress = None; } + } else { + compress = None; } if url.is_s3() { @@ -2030,6 +2026,7 @@ fn fetch_impl( force_http3, force_http1, is_node_http_client: ALLOW_GET_BODY, + compress, check_server_identity: if check_server_identity.is_empty_or_undefined_or_null() { jsc::strong::Optional::empty() } else { diff --git a/src/runtime/webcore/fetch/FetchTasklet.rs b/src/runtime/webcore/fetch/FetchTasklet.rs index 24ea04f47fc6..8eadb2adbfef 100644 --- a/src/runtime/webcore/fetch/FetchTasklet.rs +++ b/src/runtime/webcore/fetch/FetchTasklet.rs @@ -1895,6 +1895,7 @@ impl FetchTasklet { reject_unauthorized: Some(fetch_options.reject_unauthorized), verbose: Some(fetch_options.verbose), tls_props: fetch_options.ssl_config, + compress: fetch_options.compress, }, ))); // enable streaming the write side @@ -2408,6 +2409,7 @@ pub struct FetchOptions { pub force_http3: bool, pub force_http1: bool, pub is_node_http_client: bool, + pub compress: Option, } impl Default for FetchOptions { @@ -2442,6 +2444,7 @@ impl Default for FetchOptions { force_http3: false, force_http1: false, is_node_http_client: false, + compress: None, } } } diff --git a/src/runtime/webcore/fetch/compress_body.rs b/src/runtime/webcore/fetch/compress_body.rs index f1dd769c0704..10bdde43329b 100644 --- a/src/runtime/webcore/fetch/compress_body.rs +++ b/src/runtime/webcore/fetch/compress_body.rs @@ -1,23 +1,14 @@ -//! Automatic request-body compression for `fetch()`. -//! -//! Mirrors the response-decompression fast path in `bun_http::InternalState`: -//! a thread-local libdeflate compressor + fixed 512 KiB scratch buffer reused -//! across calls. gzip/deflate go through libdeflate; brotli/zstd use their -//! one-shot encoders. Only buffered bodies are handled here — streams and -//! sendfile are skipped by the caller. +//! JS-side parsing of `fetch({ compress })`. The actual compression runs on +//! the HTTP thread (`bun_http::compress_body`) so it can reuse +//! `LibdeflateState`'s shared scratch buffer. -use core::cell::UnsafeCell; - -use crate::webcore::jsc::{self, JSGlobalObject, JSValue, JsResult}; +use crate::webcore::jsc::{JSGlobalObject, JSValue, JsResult}; use bun_jsc::ComptimeStringMapExt as _; -#[derive(Copy, Clone, Eq, PartialEq, Debug)] -pub enum CompressEncoding { - Gzip, - Deflate, - Brotli, - Zstd, -} +pub use bun_http::compress_body::{ + CompressEncoding, CompressOption, DEFAULT_BROTLI_QUALITY, DEFAULT_DEFLATE_LEVEL, + DEFAULT_ZSTD_LEVEL, +}; bun_core::comptime_string_map! { static COMPRESS_ENCODING_MAP: CompressEncoding = { @@ -28,332 +19,82 @@ bun_core::comptime_string_map! { }; } -impl CompressEncoding { - pub fn header_value(self) -> &'static [u8] { - match self { - CompressEncoding::Gzip => b"gzip", - CompressEncoding::Deflate => b"deflate", - CompressEncoding::Brotli => b"br", - CompressEncoding::Zstd => b"zstd", - } +/// Parses `compress?: boolean | "gzip" | "deflate" | "br" | "zstd" | { encoding, level? }`. +/// Returns `Ok(None)` for `false` / `undefined` / `null`. +pub fn from_js(global: &JSGlobalObject, value: JSValue) -> JsResult> { + if value.is_undefined_or_null() { + return Ok(None); } -} - -#[derive(Copy, Clone, Debug)] -pub struct CompressOption { - pub encoding: CompressEncoding, - pub level: Option, -} - -impl CompressOption { - /// Parses `compress?: boolean | "gzip" | "deflate" | "br" | "zstd" | { encoding, level? }`. - /// Returns `Ok(None)` for `false` / `undefined` / `null`. - pub fn from_js(global: &JSGlobalObject, value: JSValue) -> JsResult> { - if value.is_undefined_or_null() { - return Ok(None); - } - if value.is_boolean() { - return Ok(if value.as_boolean() { - Some(CompressOption { - encoding: CompressEncoding::Gzip, - level: None, - }) - } else { - None - }); - } - if value.is_string() { - return match COMPRESS_ENCODING_MAP.from_js(global, value)? { - Some(encoding) => Ok(Some(CompressOption { - encoding, - level: None, - })), - None => Err(global.throw_invalid_arguments(format_args!( - "fetch: 'compress' must be \"gzip\", \"deflate\", \"br\", or \"zstd\"" - ))), - }; - } - if value.is_object() { - let encoding = match value.get(global, "encoding")? { - Some(enc) if enc.is_string() => { - match COMPRESS_ENCODING_MAP.from_js(global, enc)? { - Some(e) => e, - None => { - return Err(global.throw_invalid_arguments(format_args!( - "fetch: 'compress.encoding' must be \"gzip\", \"deflate\", \"br\", or \"zstd\"" - ))); - } - } - } - other => { - return Err(global.throw_invalid_argument_type_value( - b"compress.encoding", - b"string", - other.unwrap_or(JSValue::UNDEFINED), - )); - } - }; - let level = match value.get(global, "level")? { - Some(lvl) if !lvl.is_undefined_or_null() => { - let (min, max, default) = match encoding { - CompressEncoding::Gzip | CompressEncoding::Deflate => { - (0, 12, DEFAULT_DEFLATE_LEVEL) - } - CompressEncoding::Brotli => ( - bun_brotli::c::BROTLI_MIN_QUALITY, - bun_brotli::c::BROTLI_MAX_QUALITY, - DEFAULT_BROTLI_QUALITY, - ), - CompressEncoding::Zstd => (1, 22, DEFAULT_ZSTD_LEVEL), - }; - Some(global.validate_integer_range::( - lvl, - default, - bun_jsc::IntegerRange { - min: i128::from(min), - max: i128::from(max), - field_name: b"compress.level", - always_allow_zero: false, - }, - )?) - } - _ => None, - }; - return Ok(Some(CompressOption { encoding, level })); - } - Err(global.throw_invalid_argument_type_value( - b"compress", - b"boolean, string, or object", - value, - )) + if value.is_boolean() { + return Ok(if value.as_boolean() { + Some(CompressOption { + encoding: CompressEncoding::Gzip, + level: None, + }) + } else { + None + }); } -} - -// ────────────────────────────────────────────────────────────────────────── -// Thread-local compressor state (mirrors `LibdeflateState` on the HTTP thread) -// ────────────────────────────────────────────────────────────────────────── - -/// libdeflate's default level. Reused for the cached compressor so the common -/// `compress: true` / `compress: "gzip"` path never allocates a fresh handle. -const DEFAULT_DEFLATE_LEVEL: i32 = 6; -const DEFAULT_BROTLI_QUALITY: i32 = 6; -const DEFAULT_ZSTD_LEVEL: i32 = 3; - -const SHARED_BUFFER_SIZE: usize = 512 * 1024; - -struct CompressorState { - compressor: *mut bun_libdeflate_sys::libdeflate::Compressor, - shared_buffer: [u8; SHARED_BUFFER_SIZE], -} - -// SAFETY: `*mut T` (null) and `[u8; N]` are both valid at the all-zero bit pattern. -unsafe impl bun_core::Zeroable for CompressorState {} - -thread_local! { - static LAZY_COMPRESSOR: UnsafeCell>> = - const { UnsafeCell::new(None) }; -} - -fn with_state(f: impl FnOnce(&mut CompressorState) -> R) -> R { - LAZY_COMPRESSOR.with(|cell| { - // SAFETY: thread-local; sole accessor; no re-entrance from `f` back into - // this function. - let slot = unsafe { &mut *cell.get() }; - if slot.is_none() { - let compressor = - bun_libdeflate_sys::libdeflate::Compressor::alloc(DEFAULT_DEFLATE_LEVEL); - if compressor.is_null() { - bun_core::out_of_memory(); - } - let mut state: Box = bun_core::boxed_zeroed(); - state.compressor = compressor; - *slot = Some(state); - } - f(slot.as_deref_mut().unwrap()) - }) -} - -// ────────────────────────────────────────────────────────────────────────── -// One-shot body compression -// ────────────────────────────────────────────────────────────────────────── - -pub fn compress_request_body( - global: &JSGlobalObject, - input: &[u8], - opt: &CompressOption, -) -> JsResult> { - match opt.encoding { - CompressEncoding::Gzip | CompressEncoding::Deflate => { - let enc = if opt.encoding == CompressEncoding::Gzip { - bun_libdeflate_sys::libdeflate::Encoding::Gzip - } else { - // HTTP "deflate" is the zlib-wrapped DEFLATE stream (RFC 9110 - // §8.4.1.2); libdeflate's `Deflate` is the raw stream. - bun_libdeflate_sys::libdeflate::Encoding::Zlib - }; - Ok(compress_libdeflate(input, enc, opt.level)) - } - CompressEncoding::Brotli => compress_brotli(global, input, opt.level), - CompressEncoding::Zstd => compress_zstd(global, input, opt.level), + if value.is_string() { + return match COMPRESS_ENCODING_MAP.from_js(global, value)? { + Some(encoding) => Ok(Some(CompressOption { + encoding, + level: None, + })), + None => Err(global.throw_invalid_arguments(format_args!( + "fetch: 'compress' must be \"gzip\", \"deflate\", \"br\", or \"zstd\"" + ))), + }; } -} - -fn compress_libdeflate( - input: &[u8], - enc: bun_libdeflate_sys::libdeflate::Encoding, - level: Option, -) -> Vec { - use bun_libdeflate_sys::libdeflate::Compressor; - - with_state(|state| { - // Custom level → allocate a temporary compressor; the cached handle is - // pinned to DEFAULT_DEFLATE_LEVEL. - let mut tmp: *mut Compressor = core::ptr::null_mut(); - let compressor: &mut Compressor = match level { - Some(l) if l != DEFAULT_DEFLATE_LEVEL => { - tmp = Compressor::alloc(l); - if tmp.is_null() { - bun_core::out_of_memory(); + if value.is_object() { + let encoding = match value.get(global, "encoding")? { + Some(enc) if enc.is_string() => match COMPRESS_ENCODING_MAP.from_js(global, enc)? { + Some(e) => e, + None => { + return Err(global.throw_invalid_arguments(format_args!( + "fetch: 'compress.encoding' must be \"gzip\", \"deflate\", \"br\", or \"zstd\"" + ))); } - // SAFETY: just allocated, non-null, exclusive. - unsafe { &mut *tmp } + }, + other => { + return Err(global.throw_invalid_argument_type_value( + b"compress.encoding", + b"string", + other.unwrap_or(JSValue::UNDEFINED), + )); } - // SAFETY: `state.compressor` is set once in `with_state` from - // `libdeflate_alloc_compressor` (panics on null) and never freed - // for the thread's lifetime. The handle is a separate C heap - // allocation disjoint from `state.shared_buffer`, so this `&mut` - // does not alias the buffer borrow below. - _ => unsafe { &mut *state.compressor }, }; - let _guard = scopeguard::guard(tmp, |tmp| { - if !tmp.is_null() { - // SAFETY: `tmp` was returned by `Compressor::alloc` above and is - // not used after this call. - unsafe { Compressor::destroy(tmp) }; - } - }); - - // Fast path: compress into the shared buffer, then copy out. - let bound = compressor.max_bytes_needed(input, enc); - if bound <= state.shared_buffer.len() { - let result = compressor.compress(input, &mut state.shared_buffer, enc); - return state.shared_buffer[..result.written].to_vec(); - } - - // Slow path: body is large; allocate the bound up front and compress - // directly into the Vec's spare capacity. - let mut out = Vec::with_capacity(bound); - compressor.compress_to_vec(input, &mut out, enc); - out - }) -} - -fn compress_brotli(global: &JSGlobalObject, input: &[u8], level: Option) -> JsResult> { - use bun_brotli::c; - let quality = level.unwrap_or(DEFAULT_BROTLI_QUALITY); - - with_state(|state| { - let bound = c::BrotliEncoderMaxCompressedSize(input.len()); - // BrotliEncoderMaxCompressedSize returns 0 when the bound would - // overflow size_t — fall back to a heap buffer in that case. - if bound != 0 && bound <= state.shared_buffer.len() { - let mut out_len = state.shared_buffer.len(); - // SAFETY: input/output slices are valid for their lengths; - // BrotliEncoderCompress only reads `input` and writes `out_len` - // bytes to `shared_buffer`, updating `out_len` to bytes written. - let ok = unsafe { - c::BrotliEncoderCompress( - quality, - c::BROTLI_DEFAULT_WINDOW, - c::BrotliEncoderMode::generic, - input.len(), - input.as_ptr(), - &raw mut out_len, - state.shared_buffer.as_mut_ptr(), - ) - }; - if ok != 0 { - return Ok(state.shared_buffer[..out_len].to_vec()); + let level = match value.get(global, "level")? { + Some(lvl) if !lvl.is_undefined_or_null() => { + let (min, max, default) = match encoding { + CompressEncoding::Gzip | CompressEncoding::Deflate => { + (0, 12, DEFAULT_DEFLATE_LEVEL) + } + CompressEncoding::Brotli => ( + bun_brotli::c::BROTLI_MIN_QUALITY, + bun_brotli::c::BROTLI_MAX_QUALITY, + DEFAULT_BROTLI_QUALITY, + ), + CompressEncoding::Zstd => (1, 22, DEFAULT_ZSTD_LEVEL), + }; + Some(global.validate_integer_range::( + lvl, + default, + bun_jsc::IntegerRange { + min: i128::from(min), + max: i128::from(max), + field_name: b"compress.level", + always_allow_zero: false, + }, + )?) } - } - - let cap = if bound != 0 { - bound - } else { - input.len() + 1024 - }; - let mut out = vec![0u8; cap]; - let mut out_len = out.len(); - // SAFETY: see above. - let ok = unsafe { - c::BrotliEncoderCompress( - quality, - c::BROTLI_DEFAULT_WINDOW, - c::BrotliEncoderMode::generic, - input.len(), - input.as_ptr(), - &raw mut out_len, - out.as_mut_ptr(), - ) + _ => None, }; - if ok == 0 { - return Err(global - .err( - jsc::ErrorCode::ZLIB_INITIALIZATION_FAILED, - format_args!("brotli compression failed"), - ) - .throw()); - } - out.truncate(out_len); - Ok(out) - }) -} - -fn compress_zstd(global: &JSGlobalObject, input: &[u8], level: Option) -> JsResult> { - with_state(|state| { - let bound = bun_zstd::compress_bound(input.len()); - if bun_zstd::is_error(bound) { - return Err(global - .err( - jsc::ErrorCode::ZLIB_INITIALIZATION_FAILED, - format_args!("zstd compression failed: input too large"), - ) - .throw()); - } - if bound <= state.shared_buffer.len() { - match bun_zstd::compress(&mut state.shared_buffer, input, level) { - bun_zstd::Result::Success(n) => { - return Ok(state.shared_buffer[..n].to_vec()); - } - bun_zstd::Result::Err(msg) => { - return Err(global - .err( - jsc::ErrorCode::ZLIB_INITIALIZATION_FAILED, - format_args!( - "zstd compression failed: {}", - bstr::BStr::new(msg.as_bytes()) - ), - ) - .throw()); - } - } - } - - let mut out = vec![0u8; bound]; - match bun_zstd::compress(&mut out, input, level) { - bun_zstd::Result::Success(n) => { - out.truncate(n); - Ok(out) - } - bun_zstd::Result::Err(msg) => Err(global - .err( - jsc::ErrorCode::ZLIB_INITIALIZATION_FAILED, - format_args!( - "zstd compression failed: {}", - bstr::BStr::new(msg.as_bytes()) - ), - ) - .throw()), - } - }) + return Ok(Some(CompressOption { encoding, level })); + } + Err(global.throw_invalid_argument_type_value( + b"compress", + b"boolean, string, or object", + value, + )) } From 19d861c05f46681dbbd16b469e0e830b06500c12 Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Tue, 16 Jun 2026 22:50:18 +0000 Subject: [PATCH 08/14] [autofix.ci] apply automated fixes --- src/runtime/webcore/fetch/compress_body.rs | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/src/runtime/webcore/fetch/compress_body.rs b/src/runtime/webcore/fetch/compress_body.rs index 10bdde43329b..7eeeeb99e4f1 100644 --- a/src/runtime/webcore/fetch/compress_body.rs +++ b/src/runtime/webcore/fetch/compress_body.rs @@ -92,9 +92,5 @@ pub fn from_js(global: &JSGlobalObject, value: JSValue) -> JsResult Date: Tue, 16 Jun 2026 23:11:03 +0000 Subject: [PATCH 09/14] fetch(compress): compress at write time; spill to Vec only on partial send Common case (h1, compressed bound <= 512 KiB, body fits in one socket write) now allocates no per-request Vec: - Compression moves from HTTPClient::start() to write time so the output can borrow LibdeflateState::shared_buffer for the synchronous send. - state.original_request_body stays as the original uncompressed slice; only state.request_body (the cursor) is re-seated to compressed bytes. Redirects (307/308) and h2/idempotent retries re-read the original and re-compress on the next hop. New state.flags.body_compressed gates per-attempt re-entry. - Content-Length comes from HTTPClient::body_len_for_send() (compressed_body_len when set, else original_request_body.len()); h1/h2/h3/proxy build_request call sites updated. - h1: send_initial_request_payload compresses into shared_buffer, writes, then spill_compressed_body() copies any unsent tail into compressed_request_body before yielding to the event loop. Covers both the amount==0 early return and the normal return. - h2/h3/proxy-tunnel: compress_body_for_send(false) writes straight into the Vec (their body sends span event-loop ticks). - gzip/deflate slow path (bound > 512 KiB) now uses streaming zlib deflate(Z_FINISH) into a Vec growing in 64 KiB steps, instead of libdeflate one-shot which would prealloc the worst-case bound. zlib level clamped to 9. - Tests: incompressible 600 KiB body (forces zlib streaming + spill); 307 redirect with compressed body (re-compress from original on second hop). --- src/http/AsyncHTTP.rs | 1 + src/http/InternalState.rs | 6 + src/http/compress_body.rs | 171 +++++++++++++++++------ src/http/h2_client/ClientSession.rs | 11 +- src/http/h3_client/encode.rs | 5 +- src/http/lib.rs | 146 ++++++++++++++----- test/js/web/fetch/fetch-compress.test.ts | 53 +++++++ 7 files changed, 311 insertions(+), 82 deletions(-) diff --git a/src/http/AsyncHTTP.rs b/src/http/AsyncHTTP.rs index a307eb3db4ec..b7a1a75ee0b2 100644 --- a/src/http/AsyncHTTP.rs +++ b/src/http/AsyncHTTP.rs @@ -215,6 +215,7 @@ fn make_client<'a>( unix_socket_path: ZigStringSlice::EMPTY, compress: None, compressed_request_body: Vec::new(), + compressed_body_len: 0, } } diff --git a/src/http/InternalState.rs b/src/http/InternalState.rs index 4daf7442d721..81d82df71b7f 100644 --- a/src/http/InternalState.rs +++ b/src/http/InternalState.rs @@ -74,6 +74,11 @@ pub struct InternalStateFlags { /// check passed (and implicitly by `InternalState::reset()` on every /// redirect hop / failure, so each hop re-parks independently). pub is_waiting_for_cert_check: bool, + /// Set once `HTTPClient::compress_body_for_send` has run for this attempt. + /// Guards header-retry re-entries from compressing again. Cleared by + /// `reset()`/`init()` so each redirect/retry hop re-compresses from the + /// original uncompressed `original_request_body`. + pub body_compressed: bool, } impl InternalStateFlags { @@ -88,6 +93,7 @@ impl InternalStateFlags { resend_request_body_on_redirect: false, clear_hostname_on_redirect: false, is_waiting_for_cert_check: false, + body_compressed: false, } } } diff --git a/src/http/compress_body.rs b/src/http/compress_body.rs index 55b825a57f51..a1b402ebed4f 100644 --- a/src/http/compress_body.rs +++ b/src/http/compress_body.rs @@ -40,40 +40,66 @@ pub const DEFAULT_DEFLATE_LEVEL: i32 = 6; pub const DEFAULT_BROTLI_QUALITY: i32 = 6; pub const DEFAULT_ZSTD_LEVEL: i32 = 3; -/// One-shot body compression into `out` (cleared first). HTTP-thread-only. +/// Where [`compress_into`] put its output. +pub(crate) enum CompressOutput { + /// Output is in `LibdeflateState::shared_buffer[..len]`. Valid only until + /// the next use of `shared_buffer` (i.e. the current synchronous callback). + Shared(usize), + /// Output is in the caller's `spill` Vec. + Spilled, +} + +/// One-shot body compression. HTTP-thread-only. Writes into +/// `state.shared_buffer` when the bound fits (returning [`CompressOutput::Shared`]); +/// otherwise allocates into `spill` (cleared first). Callers that need the +/// output to outlive the current callback must copy `Shared` into owned storage. pub(crate) fn compress_into( state: &mut LibdeflateState, input: &[u8], opt: &CompressOption, - out: &mut Vec, -) -> Result<(), bun_core::Error> { - out.clear(); + spill: &mut Vec, +) -> Result { + spill.clear(); match opt.encoding { CompressEncoding::Gzip | CompressEncoding::Deflate => { - let enc = if opt.encoding == CompressEncoding::Gzip { + let gzip = opt.encoding == CompressEncoding::Gzip; + let enc = if gzip { bun_libdeflate_sys::libdeflate::Encoding::Gzip } else { // HTTP "deflate" is the zlib-wrapped DEFLATE stream (RFC 9110 // §8.4.1.2); libdeflate's `Deflate` is the raw stream. bun_libdeflate_sys::libdeflate::Encoding::Zlib }; - compress_libdeflate(state, input, enc, opt.level, out); - Ok(()) + match compress_libdeflate_fast(state, input, enc, opt.level) { + Some(n) => Ok(CompressOutput::Shared(n)), + None => { + compress_zlib_streaming(input, gzip, opt.level, spill)?; + Ok(CompressOutput::Spilled) + } + } } - CompressEncoding::Brotli => compress_brotli(state, input, opt.level, out), - CompressEncoding::Zstd => compress_zstd(state, input, opt.level, out), + CompressEncoding::Brotli => compress_brotli(state, input, opt.level, spill), + CompressEncoding::Zstd => compress_zstd(state, input, opt.level, spill), } } -fn compress_libdeflate( +/// libdeflate one-shot fast path into `state.shared_buffer`. Returns `None` +/// when the worst-case bound exceeds the shared buffer — caller falls back to +/// [`compress_zlib_streaming`]. +fn compress_libdeflate_fast( state: &mut LibdeflateState, input: &[u8], enc: bun_libdeflate_sys::libdeflate::Encoding, level: Option, - out: &mut Vec, -) { +) -> Option { use bun_libdeflate_sys::libdeflate::Compressor; + // Bound is level-independent — use the cached handle so the slow-path + // bail-out doesn't pay for a temp compressor it never uses. + if state.compressor_mut().max_bytes_needed(input, enc) > state.shared_buffer.len() { + return None; + } + // Custom level → allocate a temporary compressor; the cached handle is // pinned to DEFAULT_DEFLATE_LEVEL. let mut tmp: *mut Compressor = core::ptr::null_mut(); @@ -96,26 +122,89 @@ fn compress_libdeflate( } }); - // Fast path: compress into the shared buffer, then copy out. - let bound = compressor.max_bytes_needed(input, enc); - if bound <= state.shared_buffer.len() { - let result = compressor.compress(input, &mut state.shared_buffer, enc); - out.extend_from_slice(&state.shared_buffer[..result.written]); - return; + Some( + compressor + .compress(input, &mut state.shared_buffer, enc) + .written, + ) +} + +/// Slow path for gzip/deflate when the libdeflate one-shot bound exceeds the +/// shared buffer: streaming zlib `deflate(Z_FINISH)` into a Vec that grows in +/// 64 KiB steps so the allocation tracks the actual compressed size, not the +/// worst-case bound. +fn compress_zlib_streaming( + input: &[u8], + gzip: bool, + level: Option, + out: &mut Vec, +) -> Result<(), bun_core::Error> { + use bun_zlib::{FlushValue, ReturnCode, deflate, deflateEnd, deflateInit2_, zlibVersion}; + + let mut strm: bun_zlib::z_stream = bun_core::ffi::zeroed(); + strm.next_in = input.as_ptr(); + strm.avail_in = input.len() as _; + // gzip wrapper: +16; HTTP "deflate" is the zlib-wrapped stream + // (RFC 9110 §8.4.1.2): plain 15. + let window_bits = if gzip { 15 + 16 } else { 15 }; + // libdeflate accepts 0..=12; zlib only 0..=9. + let level = level.unwrap_or(DEFAULT_DEFLATE_LEVEL).min(9); + // SAFETY: `strm` is zeroed; version/size match the linked zlib. + let rc = unsafe { + deflateInit2_( + &raw mut strm, + level, + 8, // Z_DEFLATED + window_bits, + 8, // default memLevel + 0, // Z_DEFAULT_STRATEGY + zlibVersion().cast::(), + size_of::() as _, + ) + }; + if rc != ReturnCode::Ok { + return Err(bun_core::err!(CompressionFailed)); } + // SAFETY: `strm` is stack-pinned for the function's lifetime; the guard + // runs `deflateEnd` on it at scope exit (before `strm` is dropped). Raw + // pointer avoids the borrow conflict with the loop body. + let strm_p: *mut bun_zlib::z_stream = &raw mut strm; + let _guard = scopeguard::guard(strm_p, |p| unsafe { + deflateEnd(p); + }); - // Slow path: body is large; allocate the bound up front and compress - // directly into the Vec's spare capacity. - out.reserve(bound); - compressor.compress_to_vec(input, out, enc); + loop { + if out.capacity() == out.len() { + out.reserve(64 * 1024); + } + let spare = out.spare_capacity_mut(); + strm.next_out = spare.as_mut_ptr().cast::(); + strm.avail_out = spare.len() as _; + // SAFETY: `strm` initialized; next_in/avail_in/next_out/avail_out are + // valid for their lengths; `deflate` only reads input and writes the + // tail of `out`'s spare capacity. + let rc = unsafe { deflate(&raw mut strm, FlushValue::Finish) }; + let produced = spare.len() - strm.avail_out as usize; + // SAFETY: zlib has initialized `produced` bytes at the start of + // `spare`; new len is within capacity. + unsafe { out.set_len(out.len() + produced) }; + match rc { + ReturnCode::StreamEnd => return Ok(()), + ReturnCode::Ok => continue, + _ => { + out.clear(); + return Err(bun_core::err!(CompressionFailed)); + } + } + } } fn compress_brotli( state: &mut LibdeflateState, input: &[u8], level: Option, - out: &mut Vec, -) -> Result<(), bun_core::Error> { + spill: &mut Vec, +) -> Result { use bun_brotli::c; let quality = level.unwrap_or(DEFAULT_BROTLI_QUALITY); @@ -139,8 +228,7 @@ fn compress_brotli( ) }; if ok != 0 { - out.extend_from_slice(&state.shared_buffer[..out_len]); - return Ok(()); + return Ok(CompressOutput::Shared(out_len)); } } @@ -149,8 +237,8 @@ fn compress_brotli( } else { input.len() + 1024 }; - out.resize(cap, 0); - let mut out_len = out.len(); + spill.resize(cap, 0); + let mut out_len = spill.len(); // SAFETY: see above. let ok = unsafe { c::BrotliEncoderCompress( @@ -160,45 +248,42 @@ fn compress_brotli( input.len(), input.as_ptr(), &raw mut out_len, - out.as_mut_ptr(), + spill.as_mut_ptr(), ) }; if ok == 0 { - out.clear(); + spill.clear(); return Err(bun_core::err!(CompressionFailed)); } - out.truncate(out_len); - Ok(()) + spill.truncate(out_len); + Ok(CompressOutput::Spilled) } fn compress_zstd( state: &mut LibdeflateState, input: &[u8], level: Option, - out: &mut Vec, -) -> Result<(), bun_core::Error> { + spill: &mut Vec, +) -> Result { let bound = bun_zstd::compress_bound(input.len()); if bun_zstd::is_error(bound) { return Err(bun_core::err!(CompressionFailed)); } if bound <= state.shared_buffer.len() { return match bun_zstd::compress(&mut state.shared_buffer, input, level) { - bun_zstd::Result::Success(n) => { - out.extend_from_slice(&state.shared_buffer[..n]); - Ok(()) - } + bun_zstd::Result::Success(n) => Ok(CompressOutput::Shared(n)), bun_zstd::Result::Err(_) => Err(bun_core::err!(CompressionFailed)), }; } - out.resize(bound, 0); - match bun_zstd::compress(out, input, level) { + spill.resize(bound, 0); + match bun_zstd::compress(spill, input, level) { bun_zstd::Result::Success(n) => { - out.truncate(n); - Ok(()) + spill.truncate(n); + Ok(CompressOutput::Spilled) } bun_zstd::Result::Err(_) => { - out.clear(); + spill.clear(); Err(bun_core::err!(CompressionFailed)) } } diff --git a/src/http/h2_client/ClientSession.rs b/src/http/h2_client/ClientSession.rs index d798f4276a6a..f2f622a3b3b3 100644 --- a/src/http/h2_client/ClientSession.rs +++ b/src/http/h2_client/ClientSession.rs @@ -419,7 +419,16 @@ impl ClientSession { } self.rearm_timeout(); - let request = client.h2_build_request(client.state.original_request_body.len()); + // DATA-frame encoding may yield mid-body — compress into the Vec so the + // cursor stays valid across event-loop ticks. + if let Err(e) = client.compress_body_for_send(false) { + self.streams.swap_remove(&stream_ref.id); + drop_stream(stream); + client.h2 = None; + client.fail(e); + return; + } + let request = client.h2_build_request(client.body_len_for_send()); if let Err(err) = encode::write_request(self, client, stream_ref, &request) { // encodeHeader pushes into the HPACK encoder's dynamic table per // call, so a mid-encode failure leaves entries the server will diff --git a/src/http/h3_client/encode.rs b/src/http/h3_client/encode.rs index bba6f0979aee..871442aea6e0 100644 --- a/src/http/h3_client/encode.rs +++ b/src/http/h3_client/encode.rs @@ -33,8 +33,11 @@ pub fn write_request( let href: &[u8] = client.url.href; let host: &[u8] = client.url.host; let reject_unauthorized = client.flags.reject_unauthorized; + // h3 body bytes flow into lsquic's send buffer asynchronously — compress + // into the Vec so the cursor stays valid across event-loop ticks. + client.compress_body_for_send(false)?; let req_body: bun_ptr::RawSlice = client.state.request_body; - let body_len = client.state.original_request_body.len(); + let body_len = client.body_len_for_send(); let is_streaming = client.state.original_request_body.is_stream(); let is_bytes = matches!( client.state.original_request_body, diff --git a/src/http/lib.rs b/src/http/lib.rs index 0a44be00776d..65399cee4705 100644 --- a/src/http/lib.rs +++ b/src/http/lib.rs @@ -651,14 +651,19 @@ pub struct HTTPClient<'a> { pub async_http_id: u32, pub hostname: Option<&'a [u8]>, pub unix_socket_path: ZigStringSlice, - /// `fetch({ compress })` — when set, [`Self::start`] compresses the - /// `Bytes` body on the HTTP thread before connecting. Cleared after the - /// first compression so retry/h2-retry paths that re-enter `start()` with - /// the already-compressed slice don't double-compress. + /// `fetch({ compress })` — when set, the body is compressed lazily at + /// write time (h1: `send_initial_request_payload`; h2/h3: at attach) so + /// the output can borrow `LibdeflateState::shared_buffer`. Persists across + /// redirects/retries so each hop re-compresses from the original + /// `state.original_request_body`. pub compress: Option, - /// Backing storage for the compressed body; `state.original_request_body` - /// borrows from this for the lifetime of the request. + /// Backing storage for the compressed body when it must outlive a single + /// synchronous write (output > shared buffer, partial h1 write, or h2/h3 + /// frame encoding). Empty in the common one-write h1 case. pub compressed_request_body: Vec, + /// Compressed length for `Content-Length`; 0 when `compress` is None or + /// the body hasn't been compressed yet. + pub compressed_body_len: usize, } impl<'a> HTTPClient<'a> { @@ -2416,37 +2421,9 @@ impl<'a> HTTPClient<'a> { self.url.is_https() } - pub fn start(&mut self, mut body: HTTPRequestBody<'a>, body_out_str: &mut MutableString) { + pub fn start(&mut self, body: HTTPRequestBody<'a>, body_out_str: &mut MutableString) { body_out_str.reset(); - if let Some(opt) = self.compress.take() - && let HTTPRequestBody::Bytes(input) = body - && !input.is_empty() - { - match compress_body::compress_into( - http_thread().deflater(), - input, - &opt, - &mut self.compressed_request_body, - ) { - Ok(()) => { - // SAFETY: `compressed_request_body` lives on `self` for the - // request's lifetime; never reallocated after this point - // (`compress` was just `.take()`n). `state.original_request_body` - // / `state.request_body` borrow it until the terminal - // result-callback drops the client. - body = HTTPRequestBody::Bytes(unsafe { - &*core::ptr::from_ref(self.compressed_request_body.as_slice()) - }); - } - Err(e) => { - self.state = InternalState::init(body, body_out_str); - self.fail(e); - return; - } - } - } - debug_assert!(self.state.response_message_buffer.list.capacity() == 0); self.state = InternalState::init(body, body_out_str); @@ -2610,6 +2587,90 @@ impl<'a> HTTPClient<'a> { self.complete_connecting_process(); } + /// Body length for `Content-Length` — the compressed length once + /// [`compress_body_for_send`] has run, otherwise the original. + #[inline] + pub fn body_len_for_send(&self) -> usize { + if self.state.flags.body_compressed { + self.compressed_body_len + } else { + self.state.original_request_body.len() + } + } + + /// Lazy one-shot request-body compression at write time. Re-seats + /// `state.request_body` (the send cursor) to the compressed bytes; + /// `state.original_request_body` stays as the original uncompressed slice + /// so redirects/retries can re-compress from it. When `into_shared` and + /// the bound fits, the cursor borrows `LibdeflateState::shared_buffer` — + /// callers must [`spill_compressed_body`] before returning to the event + /// loop with bytes left to send. Idempotent per attempt via + /// `state.flags.body_compressed`. + /// + /// [`spill_compressed_body`]: Self::spill_compressed_body + pub fn compress_body_for_send(&mut self, into_shared: bool) -> Result<(), bun_core::Error> { + let Some(opt) = self.compress else { + return Ok(()); + }; + if self.state.flags.body_compressed { + return Ok(()); + } + let HTTPRequestBody::Bytes(input) = self.state.original_request_body else { + return Ok(()); + }; + if input.is_empty() { + return Ok(()); + } + + let deflater = http_thread().deflater(); + let out = compress_body::compress_into( + deflater, + input, + &opt, + &mut self.compressed_request_body, + )?; + let slice: &[u8] = match out { + compress_body::CompressOutput::Shared(n) if into_shared => &deflater.shared_buffer[..n], + compress_body::CompressOutput::Shared(n) => { + self.compressed_request_body + .extend_from_slice(&deflater.shared_buffer[..n]); + self.compressed_request_body.as_slice() + } + compress_body::CompressOutput::Spilled => self.compressed_request_body.as_slice(), + }; + self.compressed_body_len = slice.len(); + // SAFETY: `slice` borrows either `LibdeflateState::shared_buffer` + // (HTTP-thread singleton, valid for the current synchronous callback — + // caller spills before yielding) or `self.compressed_request_body` + // (lives on `self`, only mutated by this function via `clear()` on the + // next attempt after `state.reset()`). `state.request_body` is a + // `RawSlice` cursor; this is the same erasure pattern + // `InternalState::init` uses for `original_request_body`. + self.state.request_body = + bun_ptr::RawSlice::new(unsafe { &*core::ptr::from_ref::<[u8]>(slice) }); + self.state.flags.body_compressed = true; + Ok(()) + } + + /// Copy any unsent compressed bytes still borrowing `shared_buffer` into + /// `compressed_request_body` and re-seat the cursor. No-op when the cursor + /// already points at the Vec (or is empty). + fn spill_compressed_body(&mut self) { + if !self.state.flags.body_compressed + || !self.compressed_request_body.is_empty() + || self.state.request_body.is_empty() + { + return; + } + self.compressed_request_body + .extend_from_slice(self.state.request_body.slice()); + // SAFETY: `compressed_request_body` lives on `self`; same erasure as + // `compress_body_for_send`. + self.state.request_body = bun_ptr::RawSlice::new(unsafe { + &*core::ptr::from_ref::<[u8]>(self.compressed_request_body.as_slice()) + }); + } + fn estimated_request_header_byte_length(&self) -> usize { let sliced = self.header_entries.slice(); let mut count: usize = 0; @@ -2628,6 +2689,8 @@ impl<'a> HTTPClient<'a> { &mut self, socket: HttpSocket, ) -> Result { + self.compress_body_for_send(true)?; + let mut request_body_buffer = self.get_request_body_send_buffer(); // request_body_buffer drops at scope exit (was `defer .deinit()`) let mut temporary_send_buffer = request_body_buffer.to_array_list(); @@ -2635,7 +2698,7 @@ impl<'a> HTTPClient<'a> { let writer = &mut temporary_send_buffer; // Vec impls bun_io::Write - let request = self.build_request(self.state.original_request_body.len()); + let request = self.build_request(self.body_len_for_send()); if self.http_proxy.is_some() { if self.url.is_https() { @@ -2673,6 +2736,7 @@ impl<'a> HTTPClient<'a> { if IS_FIRST_CALL { if amount == 0 { // don't worry about it + self.spill_compressed_body(); return Ok(InitialRequestPayloadResult { has_sent_headers: self.state.request_sent_len >= headers_len, has_sent_body: false, @@ -2708,6 +2772,8 @@ impl<'a> HTTPClient<'a> { false }; + self.spill_compressed_body(); + Ok(InitialRequestPayloadResult { has_sent_headers, has_sent_body, @@ -3033,10 +3099,16 @@ impl<'a> HTTPClient<'a> { // `proxy_tunnel::raw_as_mut` INVARIANT). let proxy = proxy_tunnel::raw_as_mut(proxy_ptr); self.set_timeout(&socket); + // Proxy-tunnel writes can be partial across event-loop ticks + // — compress straight into the Vec. + if let Err(e) = self.compress_body_for_send(false) { + self.close_and_fail::(e, socket); + return; + } let mut temporary_send_buffer: Vec = Vec::with_capacity(16 * 1024); let writer = &mut temporary_send_buffer; - let request = self.build_request(self.request_body().len()); + let request = self.build_request(self.body_len_for_send()); if write_request(writer, &request).is_err() { self.close_and_fail::(err!(OutOfMemory), socket); return; diff --git a/test/js/web/fetch/fetch-compress.test.ts b/test/js/web/fetch/fetch-compress.test.ts index 938508871f54..725526fb4245 100644 --- a/test/js/web/fetch/fetch-compress.test.ts +++ b/test/js/web/fetch/fetch-compress.test.ts @@ -183,6 +183,59 @@ describe("fetch compress option", () => { expect(json.rawLength).toBeLessThan(big.length); }); + // gzip bound on 600 KiB of random bytes is > 512 KiB so compress_into spills + // straight to the per-request Vec instead of borrowing the shared buffer. + test.concurrent("incompressible body larger than the shared buffer (spill path)", async () => { + using server = Bun.serve({ + port: 0, + async fetch(req) { + const raw = Buffer.from(await req.arrayBuffer()); + return Response.json({ + encoding: req.headers.get("content-encoding"), + contentLength: req.headers.get("content-length"), + rawLength: raw.length, + sha: new Bun.CryptoHasher("sha1").update(gunzipSync(raw)).digest("hex"), + }); + }, + }); + const big = crypto.getRandomValues(Buffer.alloc(600 * 1024)); + const res = await fetch(server.url, { + method: "POST", + body: big, + compress: "gzip", + }); + const json = await res.json(); + expect(json.encoding).toBe("gzip"); + expect(Number(json.contentLength)).toBe(json.rawLength); + expect(json.sha).toBe(new Bun.CryptoHasher("sha1").update(big).digest("hex")); + }); + + // 307 preserves method+body; the HTTP-thread compression must re-run on the + // second hop from the original uncompressed slice (state.original_request_body + // is never re-seated to the compressed bytes). + test.concurrent("307 redirect re-sends compressed body", async () => { + let target: URL; + using dest = Bun.serve({ + port: 0, + async fetch(req) { + const raw = Buffer.from(await req.arrayBuffer()); + return Response.json({ + encoding: req.headers.get("content-encoding"), + decoded: gunzipSync(raw).toString(), + }); + }, + }); + target = dest.url; + using src = Bun.serve({ + port: 0, + fetch: () => new Response(null, { status: 307, headers: { Location: String(target) } }), + }); + const res = await fetch(src.url, { method: "POST", body: payload, compress: "gzip" }); + const json = await res.json(); + expect(json.encoding).toBe("gzip"); + expect(json.decoded).toBe(payload); + }); + test("invalid encoding string throws", () => { expect(() => fetch("http://127.0.0.1:1/", { From 9c058825fc53cca3293fc755bcfe88bc90e4d708 Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Tue, 16 Jun 2026 23:13:23 +0000 Subject: [PATCH 10/14] [autofix.ci] apply automated fixes --- src/http/lib.rs | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/src/http/lib.rs b/src/http/lib.rs index 65399cee4705..096f929cd926 100644 --- a/src/http/lib.rs +++ b/src/http/lib.rs @@ -2623,12 +2623,8 @@ impl<'a> HTTPClient<'a> { } let deflater = http_thread().deflater(); - let out = compress_body::compress_into( - deflater, - input, - &opt, - &mut self.compressed_request_body, - )?; + let out = + compress_body::compress_into(deflater, input, &opt, &mut self.compressed_request_body)?; let slice: &[u8] = match out { compress_body::CompressOutput::Shared(n) if into_shared => &deflater.shared_buffer[..n], compress_body::CompressOutput::Shared(n) => { From f3b0b2469f8f72f300c155366271b1d61bcbc821 Mon Sep 17 00:00:00 2001 From: Jarred Sumner Date: Tue, 16 Jun 2026 23:18:38 +0000 Subject: [PATCH 11/14] fetch(compress): add h2/h3 compress tests and large-deflate zlib-wrapped test - h2: withH2Server echo that gunzips the received body; small (32 KiB, shared-buffer fast path) and large (600 KiB, zlib-streaming spill) variants; asserts content-encoding, content-length == compressed length, round-trip. - h3: new /raw-echo route; same small/large variants. - h1: 600 KiB compress:"deflate" to verify the zlib-streaming slow path emits a zlib-wrapped (RFC 1950) stream that inflateSync decodes. --- test/js/web/fetch/fetch-compress.test.ts | 11 ++++ test/js/web/fetch/fetch-http2-client.test.ts | 56 ++++++++++++++++++++ test/js/web/fetch/fetch-http3-client.test.ts | 29 ++++++++++ 3 files changed, 96 insertions(+) diff --git a/test/js/web/fetch/fetch-compress.test.ts b/test/js/web/fetch/fetch-compress.test.ts index 725526fb4245..abb86e633fd9 100644 --- a/test/js/web/fetch/fetch-compress.test.ts +++ b/test/js/web/fetch/fetch-compress.test.ts @@ -183,6 +183,17 @@ describe("fetch compress option", () => { expect(json.rawLength).toBeLessThan(big.length); }); + // >512 KiB takes the streaming-zlib slow path; verify it produces a + // zlib-wrapped (RFC 1950) stream, not raw deflate. + test.concurrent("deflate body larger than the shared buffer (zlib streaming)", async () => { + using server = makeServer(); + const big = Buffer.alloc(600 * 1024, "abcdefghij").toString(); + const res = await fetch(server.url, { method: "POST", body: big, compress: "deflate" }); + const json = await res.json(); + expect(json.encoding).toBe("deflate"); + expect(json.decoded).toBe(big); + }); + // gzip bound on 600 KiB of random bytes is > 512 KiB so compress_into spills // straight to the per-request Vec instead of borrowing the shared buffer. test.concurrent("incompressible body larger than the shared buffer (spill path)", async () => { diff --git a/test/js/web/fetch/fetch-http2-client.test.ts b/test/js/web/fetch/fetch-http2-client.test.ts index d4b2facf2c83..a4122c9f99e5 100644 --- a/test/js/web/fetch/fetch-http2-client.test.ts +++ b/test/js/web/fetch/fetch-http2-client.test.ts @@ -1534,6 +1534,62 @@ describe.concurrent("fetch() over HTTP/2 (BUN_FEATURE_FLAG_EXPERIMENTAL_HTTP2_CL ); }); + test.each([ + ["small (shared-buffer fast path)", 32 * 1024], + ["large (zlib-streaming spill path)", 600 * 1024], + ])("compress: gzip request body over h2 — %s", async (_, size) => { + await withH2Server( + (req, res) => { + const chunks: Buffer[] = []; + req.on("data", c => chunks.push(c)); + req.on("end", () => { + const raw = Buffer.concat(chunks); + res.writeHead(200, { + "x-recv-len": String(raw.length), + "x-recv-encoding": req.headers["content-encoding"] ?? "", + "x-recv-content-length": req.headers["content-length"] ?? "", + }); + res.end(zlib.gunzipSync(raw)); + }); + }, + async url => { + await using proc = await spawnCapped({ + cmd: [ + bunExe(), + "--no-warnings", + "-e", + `const payload = Buffer.alloc(${size}, "abcdefghij"); + const r = await fetch("${url}", { + method: "POST", + body: payload, + compress: "gzip", + protocol: "http2", + tls: { rejectUnauthorized: false }, + }); + const decoded = Buffer.from(await r.arrayBuffer()); + console.log(JSON.stringify({ + recvLen: Number(r.headers.get("x-recv-len")), + encoding: r.headers.get("x-recv-encoding"), + contentLength: r.headers.get("x-recv-content-length"), + match: decoded.equals(payload), + }));`, + ], + env: { ...bunEnv, NODE_TLS_REJECT_UNAUTHORIZED: "0" }, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + const out = JSON.parse(stdout.trim()); + expect(out.encoding).toBe("gzip"); + expect(out.recvLen).toBeLessThan(size); + expect(out.contentLength).toBe(String(out.recvLen)); + expect(out.match).toBe(true); + expect(exitCode).toBe(0); + }, + ); + }); + test("protocol:'http2' against an h1-only server fails with HTTP2Unsupported", async () => { const server = https.createServer({ ...tls }, (_req, res) => res.end("h1")); server.listen(0); diff --git a/test/js/web/fetch/fetch-http3-client.test.ts b/test/js/web/fetch/fetch-http3-client.test.ts index cc7559efc3c8..c98692015448 100644 --- a/test/js/web/fetch/fetch-http3-client.test.ts +++ b/test/js/web/fetch/fetch-http3-client.test.ts @@ -42,6 +42,16 @@ beforeAll(async () => { "/head": () => new Response("should not appear", { headers: { "x-head": "1" } }), "/route/:id": req => new Response("id=" + req.params.id, { headers: { "x-route": "param" } }), "/headers-echo": req => Response.json(Object.fromEntries(req.headers)), + "/raw-echo": async req => { + const raw = Buffer.from(await req.arrayBuffer()); + return new Response(raw, { + headers: { + "x-recv-len": String(raw.length), + "x-recv-encoding": req.headers.get("content-encoding") ?? "", + "x-recv-content-length": req.headers.get("content-length") ?? "", + }, + }); + }, // Response body driven by pull — one chunk per consumer read. "/pull": () => { @@ -228,6 +238,25 @@ describe("fetch protocol: http3", () => { expect(Buffer.from(await res.bytes()).equals(payload)).toBe(true); }); + test.each([ + ["small (shared-buffer fast path)", 32 * 1024], + ["large (zlib-streaming spill path)", 600 * 1024], + ])("compress: gzip request body — %s", async (_, size) => { + const { gunzipSync } = await import("node:zlib"); + const payload = Buffer.alloc(size, "abcdefghij"); + const res = await fetch(`${base}/raw-echo`, { + ...h3, + method: "POST", + body: payload, + compress: "gzip", + }); + const raw = Buffer.from(await res.bytes()); + expect(res.headers.get("x-recv-encoding")).toBe("gzip"); + expect(Number(res.headers.get("x-recv-len"))).toBeLessThan(size); + expect(res.headers.get("x-recv-content-length")).toBe(String(raw.length)); + expect(gunzipSync(raw).equals(payload)).toBe(true); + }); + test.each([200, 204, 404, 500])("status %d", async code => { const res = await fetch(`${base}/status?code=${code}&body=${code === 204 ? "" : "x"}`, h3); expect(res.status).toBe(code); From 81cf964f780f48589eccf432d0c6313cdf574035 Mon Sep 17 00:00:00 2001 From: Jarred Sumner Date: Tue, 16 Jun 2026 23:20:11 +0000 Subject: [PATCH 12/14] fetch(compress): drop compressed_request_body in dealloc_in_flight_for_exit Mirrors the clone-only teardown in on_async_http_callback_raw so an in-flight fetch({compress}) at shutdown_for_exit() doesn't leak its compressed Vec under LSan. --- src/http/HTTPThread.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/http/HTTPThread.rs b/src/http/HTTPThread.rs index e7a752acf406..61865db45f44 100644 --- a/src/http/HTTPThread.rs +++ b/src/http/HTTPThread.rs @@ -1072,6 +1072,7 @@ impl HttpThread { let client = &mut (*nn.as_ptr()).async_http.client; drop(core::mem::take(&mut client.redirect)); drop(core::mem::take(&mut client.prev_redirect)); + drop(core::mem::take(&mut client.compressed_request_body)); if let Some(tunnel) = client.proxy_tunnel.take() { (*tunnel.as_ptr()).detach_socket(); tunnel.deref(); From c0945b798edc7152c960357b12bf45817307efac Mon Sep 17 00:00:00 2001 From: Jarred Sumner Date: Tue, 16 Jun 2026 23:40:32 +0000 Subject: [PATCH 13/14] fetch(compress): chunk zlib avail_in for >=4 GiB bodies; fix clippy SAFETY - compress_zlib_streaming: avail_in is c_uint; feed input in <=u32::MAX chunks with NoFlush until the tail, then Finish, so a >=4 GiB body isn't silently truncated to its low 32 bits. avail_out also clamped via try_from. - Move the SAFETY comment inside the deflateEnd scopeguard closure (-D clippy::undocumented-unsafe-blocks). --- src/http/compress_body.rs | 41 +++++++++++++++++++++++++++++---------- 1 file changed, 31 insertions(+), 10 deletions(-) diff --git a/src/http/compress_body.rs b/src/http/compress_body.rs index a1b402ebed4f..8d2cddfc32f2 100644 --- a/src/http/compress_body.rs +++ b/src/http/compress_body.rs @@ -141,9 +141,19 @@ fn compress_zlib_streaming( ) -> Result<(), bun_core::Error> { use bun_zlib::{FlushValue, ReturnCode, deflate, deflateEnd, deflateInit2_, zlibVersion}; + // `avail_in` is `c_uint`; feed the input in ≤u32::MAX-sized chunks so a + // ≥4 GiB body isn't silently truncated. + fn take<'a>(rem: &mut &'a [u8]) -> &'a [u8] { + let n = rem.len().min(u32::MAX as usize); + let (head, tail) = rem.split_at(n); + *rem = tail; + head + } + let mut remaining = input; + let first = take(&mut remaining); let mut strm: bun_zlib::z_stream = bun_core::ffi::zeroed(); - strm.next_in = input.as_ptr(); - strm.avail_in = input.len() as _; + strm.next_in = first.as_ptr(); + strm.avail_in = first.len() as _; // gzip wrapper: +16; HTTP "deflate" is the zlib-wrapped stream // (RFC 9110 §8.4.1.2): plain 15. let window_bits = if gzip { 15 + 16 } else { 15 }; @@ -165,26 +175,37 @@ fn compress_zlib_streaming( if rc != ReturnCode::Ok { return Err(bun_core::err!(CompressionFailed)); } - // SAFETY: `strm` is stack-pinned for the function's lifetime; the guard - // runs `deflateEnd` on it at scope exit (before `strm` is dropped). Raw - // pointer avoids the borrow conflict with the loop body. let strm_p: *mut bun_zlib::z_stream = &raw mut strm; - let _guard = scopeguard::guard(strm_p, |p| unsafe { - deflateEnd(p); + let _guard = scopeguard::guard(strm_p, |p| { + // SAFETY: `strm` is stack-pinned for the function's lifetime; the + // guard runs `deflateEnd` on it at scope exit (before `strm` drops). + // The raw pointer avoids a borrow conflict with the loop body. + unsafe { deflateEnd(p) }; }); loop { + if strm.avail_in == 0 && !remaining.is_empty() { + let next = take(&mut remaining); + strm.next_in = next.as_ptr(); + strm.avail_in = next.len() as _; + } + let flush = if remaining.is_empty() { + FlushValue::Finish + } else { + FlushValue::NoFlush + }; if out.capacity() == out.len() { out.reserve(64 * 1024); } let spare = out.spare_capacity_mut(); strm.next_out = spare.as_mut_ptr().cast::(); - strm.avail_out = spare.len() as _; + strm.avail_out = u32::try_from(spare.len()).unwrap_or(u32::MAX); + let avail_out_before = strm.avail_out; // SAFETY: `strm` initialized; next_in/avail_in/next_out/avail_out are // valid for their lengths; `deflate` only reads input and writes the // tail of `out`'s spare capacity. - let rc = unsafe { deflate(&raw mut strm, FlushValue::Finish) }; - let produced = spare.len() - strm.avail_out as usize; + let rc = unsafe { deflate(&raw mut strm, flush) }; + let produced = (avail_out_before - strm.avail_out) as usize; // SAFETY: zlib has initialized `produced` bytes at the start of // `spare`; new len is within capacity. unsafe { out.set_len(out.len() + produced) }; From 6aec4e2ec6b9f2dcc9360c8c76bb29628fe4e903 Mon Sep 17 00:00:00 2001 From: Jarred Sumner Date: Tue, 16 Jun 2026 23:52:22 +0000 Subject: [PATCH 14/14] fetch(compress): update stale leak-test comments; hoist gunzipSync import --- test/js/web/fetch/fetch-http3-client.test.ts | 5 ++--- test/js/web/fetch/fetch-leak.test.ts | 8 +++++--- 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/test/js/web/fetch/fetch-http3-client.test.ts b/test/js/web/fetch/fetch-http3-client.test.ts index c98692015448..2bdba2ef6dff 100644 --- a/test/js/web/fetch/fetch-http3-client.test.ts +++ b/test/js/web/fetch/fetch-http3-client.test.ts @@ -1,4 +1,4 @@ -import { gzipSync, type Server } from "bun"; +import { gunzipSync, gzipSync, type Server } from "bun"; import { afterAll, beforeAll, describe, expect, test } from "bun:test"; import { bunEnv, bunExe, tempDir, tls } from "harness"; @@ -242,7 +242,6 @@ describe("fetch protocol: http3", () => { ["small (shared-buffer fast path)", 32 * 1024], ["large (zlib-streaming spill path)", 600 * 1024], ])("compress: gzip request body — %s", async (_, size) => { - const { gunzipSync } = await import("node:zlib"); const payload = Buffer.alloc(size, "abcdefghij"); const res = await fetch(`${base}/raw-echo`, { ...h3, @@ -254,7 +253,7 @@ describe("fetch protocol: http3", () => { expect(res.headers.get("x-recv-encoding")).toBe("gzip"); expect(Number(res.headers.get("x-recv-len"))).toBeLessThan(size); expect(res.headers.get("x-recv-content-length")).toBe(String(raw.length)); - expect(gunzipSync(raw).equals(payload)).toBe(true); + expect(Buffer.from(gunzipSync(raw)).equals(payload)).toBe(true); }); test.each([200, 204, 404, 500])("status %d", async code => { diff --git a/test/js/web/fetch/fetch-leak.test.ts b/test/js/web/fetch/fetch-leak.test.ts index 7f9a6f39e0fa..937c775d86c9 100644 --- a/test/js/web/fetch/fetch-leak.test.ts +++ b/test/js/web/fetch/fetch-leak.test.ts @@ -240,8 +240,9 @@ test("fetch() compress option does not leak bodies or compressor state", async ( // - all four encodings // - the custom-level path (allocates a temporary libdeflate compressor that // must be freed each call) - // - a small body (thread-local 512 KiB shared-buffer fast path) - // - a ~700 KiB body (slow-path Vec allocation + multi-write send) + // - a small body (HTTP-thread LibdeflateState shared_buffer fast path) + // - a ~700 KiB body (zlib-streaming slow path → per-request Vec, freed in + // on_async_http_callback_raw) using server = Bun.serve({ port: 0, idleTimeout: 0, @@ -277,7 +278,8 @@ test("fetch() compress option does not leak bodies or compressor state", async ( await Promise.all(promises); } - // Warm up: thread-local CompressorState is allocated once here and stays. + // Warm up: HTTP-thread LibdeflateState (lazy compressor + 512 KiB + // shared_buffer) is allocated once here and stays for the process. for (let i = 0; i < 10; i++) await round(); Bun.gc(true); const baseline = process.memoryUsage.rss();