node:zlib: block worker shutdown on in-flight async compression (UAF)

src/jsc/event_loop.rs

@@ -10,7 +10,7 @@
 //! poll deadline). See PORTING.md §Dispatch.
 
 use core::ptr::NonNull;
-use core::sync::atomic::{AtomicI32, AtomicPtr, Ordering};
+use core::sync::atomic::{AtomicI32, AtomicPtr, AtomicU32, Ordering};
 
 use bun_io::{self as Async, Waker};
 use bun_uws as uws;
@@ -89,6 +89,15 @@ pub struct EventLoop {
 
     pub entered_event_loop_count: isize,
     pub concurrent_ref: AtomicI32,
+    /// Count of `WorkPool` jobs scheduled from this VM's JS thread whose
+    /// pool-thread callback has not yet finished its last access to this
+    /// `EventLoop` (typically `enqueue_task_concurrent`). `WebWorker::shutdown`
+    /// spins on this reaching zero before freeing the JSC heap and the
+    /// `VirtualMachine` box that this `EventLoop` is a field of; without that
+    /// barrier the pool thread's `do_work()` (JSC-heap input/output buffers)
+    /// and completion post are both use-after-free. Bracket with
+    /// [`Self::work_pool_task_ref`] / [`Self::work_pool_task_unref`].
+    pub work_pool_pending: AtomicU32,
     /// Atomic nullable pointer to the next-due `WTFTimer`.
     ///
     /// Note (§Dispatch): payload is `*mut ()` — the real
@@ -129,6 +138,7 @@ impl Default for EventLoop {
             uws_loop: (),
             entered_event_loop_count: 0,
             concurrent_ref: AtomicI32::new(0),
+            work_pool_pending: AtomicU32::new(0),
             imminent_gc_timer: AtomicPtr::new(core::ptr::null_mut()),
             #[cfg(unix)]
             signal_handler: None,
@@ -1005,6 +1015,38 @@ impl EventLoop {
         self.wakeup();
     }
 
+    /// JS-thread: call immediately before `WorkPool::schedule` for a task whose
+    /// pool-thread callback will dereference this `EventLoop` / the owning
+    /// `VirtualMachine` / the JSC heap (e.g. to post a completion via
+    /// [`Self::enqueue_task_concurrent`]). Paired with
+    /// [`Self::work_pool_task_unref`] on the pool thread; see
+    /// [`Self::work_pool_pending`].
+    #[inline]
+    pub fn work_pool_task_ref(&self) {
+        self.work_pool_pending.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Pool-thread: call as the last `EventLoop`/VM access in the `WorkPool`
+    /// callback (after [`Self::enqueue_task_concurrent`]). The `Release` store
+    /// pairs with [`Self::wait_for_pending_work_pool_tasks`]'s `Acquire` load
+    /// so `WebWorker::shutdown` cannot observe zero until every prior access
+    /// is visible, making the subsequent VM dealloc safe.
+    #[inline]
+    pub fn work_pool_task_unref(&self) {
+        self.work_pool_pending.fetch_sub(1, Ordering::Release);
+    }
+
+    /// Worker-thread shutdown barrier. Spins (yielding) until every
+    /// outstanding [`Self::work_pool_task_ref`] has been matched by
+    /// [`Self::work_pool_task_unref`]. Each pending task is one bounded
+    /// compression/IO step, so the wait is bounded; `terminate()` already runs
+    /// user exit handlers here, so this is not a new latency cliff.
+    pub fn wait_for_pending_work_pool_tasks(&self) {
+        while self.work_pool_pending.load(Ordering::Acquire) > 0 {
+            std::thread::yield_now();
+        }
+    }
+
     pub fn ref_concurrently(&self) {
         let _ = self.concurrent_ref.fetch_add(1, Ordering::SeqCst);
         self.wakeup();

src/jsc/web_worker.rs

@@ -1312,6 +1312,14 @@
             // or observes m_isShuttingDown under m_lock and drops. Idempotent;
             // teardownJSCVM sets it again.
             Bun__JSCTaskScheduler__markShuttingDown(vm.global());
+            // Wait for in-flight WorkPool jobs scheduled from this VM
+            // (node:zlib async compression today; see `work_pool_pending`).
+            // The pool-thread callback reads this VM's `EventLoop` and the
+            // JSC-heap-backed input/output buffers; both are freed below
+            // (teardownJSCVM / step-5 dealloc). Runs before the drain so the
+            // completion each job posts is reclaimed by
+            // `release_queued_tasks_for_shutdown`.
+            vm.event_loop_shared().wait_for_pending_work_pool_tasks();
             // Reclaim queued CppTasks (the per-worker stdio/messaging
             // MessagePort drain tasks that can be in self.tasks mid-tick when
             // terminate() lands, and any Worker dispatchExit close task from a

src/runtime/node/node_zlib_binding.rs

@@ -449,6 +449,10 @@ impl<T: CompressionStreamImpl> CompressionStream<T> {
             callback: Self::async_job_run_task,
         });
         this.poll_ref().with_mut(|p| p.ref_(vm));
+        // Hold the worker-shutdown barrier open until the pool thread has
+        // finished `do_work()` and posted the completion; matched by
+        // `work_pool_task_unref()` at the end of `async_job_run`.
+        vm.event_loop_shared().work_pool_task_ref();
         WorkPool::schedule(this.task().as_ptr());
 
         Ok(JSValue::UNDEFINED)
@@ -487,10 +491,18 @@ impl<T: CompressionStreamImpl> CompressionStream<T> {
         // `enqueue_task_concurrent` body only touches the lock-free
         // `concurrent_tasks` queue (thread-safe). `this` is the heap-allocated
         // `m_ctx` payload — the matching `ref()` in `write()` keeps it alive
-        // until `run_from_js_thread` runs and calls `deref()`.
+        // until `run_from_js_thread` runs and calls `deref()`. Liveness of the
+        // VM across a worker terminate is guaranteed by the
+        // `work_pool_task_ref()` taken in `write()`: `WebWorker::shutdown`
+        // blocks on that count reaching zero before freeing the JSC heap or
+        // the VM box, so `global_this`, `vm` and `event_loop` are all still
+        // valid here.
         unsafe {
             (*vm.event_loop()).enqueue_task_concurrent(ConcurrentTask::create(Task::init(this)));
         }
+        // Last VM access; pairs with `work_pool_task_ref()` in `write()` and
+        // releases `WebWorker::shutdown`'s barrier.
+        vm.event_loop_shared().work_pool_task_unref();
     }
 
     /// Dispatched from `dispatch.rs` when the worker-thread `do_work()` posts

test/js/node/zlib/zlib-worker-terminate.test.ts

@@ -0,0 +1,77 @@
+import { test, expect } from "bun:test";
+import { bunEnv, bunExe, isASAN } from "harness";
+
+// Regression test for a heap-use-after-free when a Worker is terminated
+// while async node:zlib operations (gzip/brotliCompress/deflate) are running
+// on the thread pool. The pool-thread completion callback dereferenced the
+// worker's VirtualMachine/EventLoop after WebWorker::shutdown had already
+// freed it:
+//
+//   heap-use-after-free  READ of size 8  thread T16 (Bun Pool 2)
+//     #0 event_loop                       src/jsc/VirtualMachine.rs
+//     #1 async_job_run<NativeBrotli>      src/runtime/node/node_zlib_binding.rs
+//   freed by thread (Worker): WebWorker::shutdown  src/jsc/web_worker.rs
+//
+// Keeping several codecs in flight at once makes the do_work() window wide
+// enough that terminate() reliably lands inside it.
+test(
+  "worker.terminate() during in-flight node:zlib async compression does not UAF",
+  async () => {
+    // ASAN poisons the freed VM immediately; a couple of rounds are enough.
+    // Release builds need the freed page to be reused/unmapped, which takes a
+    // few more.
+    const ROUNDS = isASAN ? 4 : 10;
+
+    const script = /* js */ `
+      const { Worker } = require("node:worker_threads");
+      const src = \`
+        const { parentPort } = require("node:worker_threads");
+        const zlib = require("node:zlib");
+        const { promisify } = require("node:util");
+        const gz = promisify(zlib.gzip);
+        const br = promisify(zlib.brotliCompress);
+        const df = promisify(zlib.deflate);
+        const big = Buffer.alloc(16 << 20, 0x61);
+        const lanes = (n, f) => {
+          for (let i = 0; i < n; i++)
+            (async () => { for (;;) { try { await f(); } catch {} } })();
+        };
+        lanes(2, () => gz(big));
+        lanes(1, () => br(big.subarray(0, 4 << 20)));
+        lanes(2, () => df(big.subarray(0, 10 << 20)));
+        parentPort.postMessage("up");
+      \`;
+      (async () => {
+        for (let r = 0; r < ${ROUNDS}; r++) {
+          const w = new Worker(src, { eval: true });
+          await new Promise(res => w.once("message", res));
+          await Bun.sleep(60 + (r * 41) % 220);
+          await w.terminate();
+        }
+        console.log("ok");
+      })().catch(e => {
+        console.error(e);
+        process.exit(1);
+      });
+    `;
+
+    await using proc = Bun.spawn({
+      cmd: [bunExe(), "-e", script],
+      env: bunEnv,
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+    const [stdout, stderr, exitCode] = await Promise.all([
+      proc.stdout.text(),
+      proc.stderr.text(),
+      proc.exited,
+    ]);
+
+    expect(stderr).not.toContain("heap-use-after-free");
+    expect(stderr).not.toContain("AddressSanitizer");
+    expect({ stdout: stdout.trim(), exitCode }).toEqual({ stdout: "ok", exitCode: 0 });
+  },
+  // Per-test override: each round starts a Worker under ASAN and compresses a
+  // 16 MiB buffer; the default 5s is too short for that even once.
+  60_000,
+);