diff --git a/patches/zstd/free-workers-before-dictionaries.patch b/patches/zstd/free-workers-before-dictionaries.patch new file mode 100644 index 000000000000..838c01d8cef7 --- /dev/null +++ b/patches/zstd/free-workers-before-dictionaries.patch @@ -0,0 +1,37 @@ +Free the worker threads of a context before the dictionaries they read. + +ZSTD_freeCCtxContent() calls ZSTD_clearAllDicts() and then ZSTDMT_freeCCtx(). +With nbWorkers >= 1, ZSTD_compressStream2() posts a job to a worker thread and +returns. The first job of a frame reads the dictionary of the context through +a pointer that it does not own (job->cdict, the localDict of the CCtx). A job +stops only in ZSTDMT_freeCCtx(), which joins the threads (POOL_free). + +So ZSTD_freeCCtx() of a context that has a dictionary and a running job frees +the dictionary under that job: heap-use-after-free in ZSTDMT_compressionJob(). +destroy() and close() of a node:zlib zstd stream make that call. + +ZSTDMT_freeCCtx() has this order for its own dictionary. It joins the threads +and then frees mtctx->cdictLocal. + +Not covered: ZSTD_CCtx_reset() with parameters, ZSTD_CCtx_loadDictionary(), +ZSTD_CCtx_refCDict() and ZSTD_CCtx_refPrefix() on a context whose jobs still +run. They free the dictionaries too. Bun does not call them on such a context. + +facebook/zstd has the same order on its dev branch (01b7154f11), and no issue +there reports it (checked 2026-10-01). + +--- a/lib/compress/zstd_compress.c ++++ b/lib/compress/zstd_compress.c +@@ -171,10 +171,11 @@ + { + assert(cctx != NULL); + assert(cctx->staticSize == 0); +- ZSTD_clearAllDicts(cctx); + #ifdef ZSTD_MULTITHREAD + ZSTDMT_freeCCtx(cctx->mtctx); cctx->mtctx = NULL; + #endif ++ /* Bun: after the join above. A job that still runs reads these dictionaries. */ ++ ZSTD_clearAllDicts(cctx); + ZSTD_cwksp_free(&cctx->workspace, cctx->customMem); + } + diff --git a/scripts/build/deps/zstd.ts b/scripts/build/deps/zstd.ts index 0e12a93386f2..2458518fdc80 100644 --- a/scripts/build/deps/zstd.ts +++ b/scripts/build/deps/zstd.ts @@ -41,10 +41,16 @@ export const zstd: Dependency = { commit: ZSTD_COMMIT, }), - // x64 targets nehalem, so zstd picks its BMI2 kernels at run time and - // probes CPUID in every CCtx/DCtx init. CPUID is a VM exit under a - // hypervisor (about 2 us each, two per init). Probe once instead. - patches: ["patches/zstd/bmi2-probe-once.patch"], + patches: [ + // x64 targets nehalem, so zstd picks its BMI2 kernels at run time and + // probes CPUID in every CCtx/DCtx init. CPUID is a VM exit under a + // hypervisor (about 2 us each, two per init). Probe once instead. + "patches/zstd/bmi2-probe-once.patch", + // ZSTD_freeCCtx frees the dictionaries of a context before it joins the + // worker threads, and a job that still runs reads them (#44201). Not + // reported upstream yet. + "patches/zstd/free-workers-before-dictionaries.patch", + ], build: cfg => { const sources = [...SOURCES]; diff --git a/scripts/build/workarounds.ts b/scripts/build/workarounds.ts index 97314c535842..7904402634e5 100644 --- a/scripts/build/workarounds.ts +++ b/scripts/build/workarounds.ts @@ -32,6 +32,7 @@ import { readFileSync } from "node:fs"; import { join } from "node:path"; import type { Config } from "./config.ts"; +import { zstd } from "./deps/zstd.ts"; import { BuildError } from "./error.ts"; import { satisfiesRange, toolchainOverride } from "./tools.ts"; @@ -116,6 +117,28 @@ export const workarounds: Workaround[] = [ `(options.detached block), replace the local 0x80 with libc::POSIX_SPAWN_SETSID, ` + `drop the explanatory comments, and delete this entry.`, }, + { + id: "zstd-free-workers-before-dictionaries", + issue: "https://github.com/oven-sh/bun/issues/44201 (no facebook/zstd issue yet)", + description: + "ZSTD_freeCCtx() frees the dictionaries of a context before it joins the worker threads, " + + "and a job that still runs reads them. patches/zstd/free-workers-before-dictionaries.patch " + + "joins first.", + applies: () => true, + expectedToBeFixed: cfg => { + // No fixed release is known: zstd's dev branch had the same order at + // 01b7154f11 (2026-10-01). So every bump of the pin asks for a look. + const PATCH_WRITTEN_FOR = "f8745da6ff1ad1e7bab384bd1f9d742439278e99"; + const source = zstd.source(cfg); + return source.kind === "github-archive" && source.commit !== PATCH_WRITTEN_FOR; + }, + cleanup: + `Read ZSTD_freeCCtxContent() in lib/compress/zstd_compress.c at the new zstd commit. If it ` + + `calls ZSTDMT_freeCCtx() before ZSTD_clearAllDicts(), delete ` + + `patches/zstd/free-workers-before-dictionaries.patch, its line in ` + + `scripts/build/deps/zstd.ts and this entry. If it does not, set PATCH_WRITTEN_FOR ` + + `in this entry to the new commit.`, + }, ]; /** diff --git a/src/runtime/node/zlib/NativeZstd.rs b/src/runtime/node/zlib/NativeZstd.rs index 8a56248734c2..5942abd3e483 100644 --- a/src/runtime/node/zlib/NativeZstd.rs +++ b/src/runtime/node/zlib/NativeZstd.rs @@ -644,29 +644,10 @@ mod _impl { pub(crate) fn close(&mut self) { // Idempotent: a handle that was never (successfully) initialized, - // or that was already closed, has no CCtx/DCtx to reset or free. - if self.state.is_none() { - self.mode = NodeMode::NONE; - return; + // or that was already closed, has no CCtx/DCtx to free. + if self.state.is_some() { + self.deinit_state(); } - let _ = match self.mode { - // SAFETY: state is a valid CCtx/DCtx for this mode. - NodeMode::ZSTD_COMPRESS => unsafe { - c::ZSTD_CCtx_reset( - self.state_ptr().cast(), - c::ZSTD_reset_session_and_parameters, - ) - }, - // SAFETY: state is a valid DCtx set by init() for this mode. - NodeMode::ZSTD_DECOMPRESS => unsafe { - c::ZSTD_DCtx_reset( - self.state_ptr().cast(), - c::ZSTD_reset_session_and_parameters, - ) - }, - _ => unreachable!(), - }; - self.deinit_state(); self.mode = NodeMode::NONE; } diff --git a/src/zstd/lib.rs b/src/zstd/lib.rs index 974a545f386d..2b58c94bc6c7 100644 --- a/src/zstd/lib.rs +++ b/src/zstd/lib.rs @@ -168,7 +168,6 @@ pub mod c { param: ZSTD_dParameter, value: c_int, ) -> usize; - pub fn ZSTD_CCtx_reset(cctx: *mut ZSTD_CCtx, reset: ZSTD_ResetDirective) -> usize; pub fn ZSTD_DCtx_reset(dctx: *mut ZSTD_DCtx, reset: ZSTD_ResetDirective) -> usize; pub fn ZSTD_compressStream2( cctx: *mut ZSTD_CCtx, diff --git a/test/js/node/zlib/zlib-handle-bounds-check.test.ts b/test/js/node/zlib/zlib-handle-bounds-check.test.ts index 8e345664be0f..aefa5ca72e0a 100644 --- a/test/js/node/zlib/zlib-handle-bounds-check.test.ts +++ b/test/js/node/zlib/zlib-handle-bounds-check.test.ts @@ -531,3 +531,118 @@ describe.concurrent("zlib native handle argument validation", () => { ).toEqual({ stdout: "ok hello world", exitCode: 0 }); }); }); + +// A zstd compressor with ZSTD_c_nbWorkers >= 1 compresses on zstd's own threads: a write posts a +// job and returns, so the job still runs after the write callback. The first job of a frame reads +// the dictionary of the stream, and zstd freed that dictionary before it joined its workers +// (https://github.com/oven-sh/bun/issues/44201). Each case leaves that job running and then takes +// one way out of the stream, in a subprocess. ASAN reports the read of freed memory in every run. +// Not concurrent: on a busy machine five cases at once passed the 5 s limit on a debug build +// (a case runs zstd at level 19 for most of its time). In sequence a case took 1.5 to 3.5 s. +describe("zstd: a compressor with workers and a dictionary", () => { + const prelude = /* js */ ` + const zlib = require("node:zlib"); + const { pbkdf2Sync } = require("node:crypto"); + const { finished } = require("node:stream/promises"); + const { ZSTD_c_compressionLevel, ZSTD_c_jobSize, ZSTD_c_nbWorkers, ZSTD_c_overlapLog, ZSTD_e_continue, ZSTD_e_end } = + zlib.constants; + + // The hex digits of a fixed pseudo-random sequence. + const hex = (seed, length) => Buffer.from(pbkdf2Sync(seed, "", 1, length / 2, "sha256").toString("hex")); + // A small dictionary makes only a sanitizer see the wrong free. With 192 KiB a release build + // of the unfixed code usually crashes too: the allocator gives the freed tables of the + // dictionary back to the operating system while the job still searches them. + const dictionary = hex("dictionary", 192 * 1024); + // One job of the smallest size (512 KiB). At level 19 a worker needs far longer for it than + // the calls after the write need. + const input = hex("input", 512 * 1024); + // With its default overlap, level 19 makes a job as long as its window (8 MiB). + const params = { + [ZSTD_c_nbWorkers]: 2, + [ZSTD_c_jobSize]: input.length, + [ZSTD_c_overlapLog]: 1, + [ZSTD_c_compressionLevel]: 19, + }; + + const open = () => zlib.createZstdCompress({ dictionary, params }).on("data", () => {}); + const write = (stream, chunk) => new Promise(done => stream.write(chunk, done)); + // A stream whose first job runs on a worker. + async function busy() { + const stream = open(); + await write(stream, input); + return stream; + } + `; + + async function run(body: string) { + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", prelude + body], + // The finalizer of every handle runs at exit, as on the ASAN CI lanes. + env: { ...bunEnv, BUN_DESTRUCT_VM_ON_EXIT: "1" }, + stderr: "pipe", + }); + // stderr is drained and not compared: a debug build writes to it. + const [stdout, , exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + return { stdout: stdout.trim(), exitCode }; + } + + const cases: [name: string, body: string, expected: string][] = [ + ["destroy()", `(await busy()).destroy(); console.log("destroyed");`, "destroyed"], + ["close()", `(await busy()).close(() => console.log("closed"));`, "closed"], + // The native close waits for the write and runs when it completes. + [ + "destroy() while the write is in flight", + `const stream = open(); + stream.on("close", () => console.log("closed")); + stream.write(input); + stream.destroy();`, + "closed", + ], + [ + "a second _handle.init()", + `const stream = await busy(); + stream._handle.init(new Uint32Array(0), undefined, stream._writeState, () => {}, dictionary); + console.log("initialized");`, + "initialized", + ], + [ + "reset() and the next frame", + `const stream = await busy(); + stream.reset(); + stream.end("x"); + await finished(stream); + console.log("finished");`, + "finished", + ], + // The second call posts a job of zero bytes. zstd's own wait for the jobs of a frame skips it. + // Only a sanitizer sees this one. + [ + "close() while a job of zero bytes runs", + `const handle = open()._handle; + const out = Buffer.alloc(64); + handle.writeSync(ZSTD_e_continue, null, 0, 0, out, 0, out.length); + handle.writeSync(ZSTD_e_end, null, 0, 0, out, 0, 0); + handle.close(); + console.log("closed");`, + "closed", + ], + // The stream still works to its end with the frees in the new order. + [ + "a frame that ends decompresses to the input", + `const stream = zlib.createZstdCompress({ dictionary, params }); + const chunks = []; + stream.on("data", chunk => chunks.push(chunk)); + await write(stream, input); + stream.end(); + await finished(stream); + console.log(zlib.zstdDecompressSync(Buffer.concat(chunks), { dictionary }).equals(input));`, + "true", + ], + ]; + + for (const [name, body, expected] of cases) { + test(name, async () => { + expect(await run(body)).toEqual({ stdout: expected, exitCode: 0 }); + }); + } +});