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 @@ impl WebWorker {
             // 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,58 @@
+import { expect, test } from "bun:test";
+import { bunEnv, bunExe, isASAN } from "harness";
+
+// worker.terminate() while async node:zlib compression is in flight on the
+// thread pool must not dereference the worker's freed VM/EventLoop from the
+// pool-thread completion. Mixed gzip/brotli/deflate lanes keep do_work() busy
+// so terminate reliably lands mid-compression.
+test("worker.terminate() during in-flight node:zlib async compression does not UAF", async () => {
+  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((resolve, reject) => {
+          w.once("message", resolve);
+          w.once("error", reject);
+          w.once("exit", code => reject(new Error("worker exited " + code + " before ready")));
+        });
+        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 });
+}, 30_000);