worker: re-throw TerminationException from Bun__reportUnhandledError so terminate() breaks a microtask-bound ReadableStream loop

src/jsc/bindings/bindings.cpp

@@ -4899,6 +4899,21 @@ bool JSC__VM__hasTerminationRequest(JSC::VM* vm)
     return vm->hasTerminationRequest();
 }
 
+// InternalMicrotask::BunPerformMicrotaskJob catches any exception from the
+// job callback with an unconditional clearException() and then calls
+// Bun__reportUnhandledError with the Exception*. When the caught exception
+// is the TerminationException that clear spends the one shot the
+// NeedTermination trap fired, so the microtask drain never observes
+// termination and a worker in a microtask-bound loop can spin forever.
+// Re-establishing the TerminationException here lets MicrotaskQueue's
+// runMicrotask see it on return and break out of the drain.
+[[ZIG_EXPORT(nothrow)]]
+void JSC__VM__rethrowTerminationException(JSC::VM* vm)
+{
+    if (vm->hasTerminationRequest() && !vm->hasPendingTerminationException())
+        vm->throwTerminationException();
+}
+
 void JSC__VM__setExecutionForbidden(JSC::VM* arg0, bool arg1)
 {
     (*arg0).setExecutionForbidden();

src/jsc/bindings/webcore/EventEmitter.cpp

@@ -252,6 +252,13 @@ bool EventEmitter::innerInvokeEventListeners(const Identifier& eventType, Simple
         auto* exception = exceptionPtr.get();
 
         if (exception) [[unlikely]] {
+            // The NakedPtr<Exception>& call() overload has already cleared the
+            // exception. Running more listeners (or error listeners) after
+            // termination would re-enter JS with the TerminationException
+            // re-thrown by Bun__reportUnhandledError and trip
+            // executeCallImpl's assertNoException().
+            if (vm.isTerminationException(exception)) [[unlikely]]
+                break;
             auto errorIdentifier = vm.propertyNames->error;
             auto hasErrorListener = this->hasActiveEventListeners(errorIdentifier);
             if (!hasErrorListener || eventType == errorIdentifier) {

src/jsc/virtual_machine_exports.rs

@@ -117,12 +117,14 @@ pub fn queue_task(global: &JSGlobalObject, task: *mut crate::cpp_task::CppTask)
 pub fn report_unhandled_error(global: &JSGlobalObject, value: JSValue) -> JSValue {
     crate::mark_binding!();
 
-    if !value.is_termination_exception() {
-        let _ = global
-            .bun_vm()
-            .as_mut()
-            .uncaught_exception(global, value, false);
+    if value.is_termination_exception() {
+        crate::cpp::JSC__VM__rethrowTerminationException(global.vm());
+        return JSValue::UNDEFINED;
     }
+    let _ = global
+        .bun_vm()
+        .as_mut()
+        .uncaught_exception(global, value, false);
     JSValue::UNDEFINED
 }
 

src/runtime/webcore/Body.rs

@@ -1688,12 +1688,19 @@ pub(crate) trait BodyMixin: BodyOwnerJs + Sized {
     /// Migrate any `Locked.readable` strong ref
     /// into the GC-traced `js.gc.stream` slot to break the cycle (the JS
     /// wrapper owns the stream; native side must not hold it strongly).
+    ///
+    /// Called from constructors after the native object has been heap-allocated,
+    /// so it must not reach a VMTraps safepoint (a TerminationException thrown
+    /// here would leave the generated `construct` holding a non-null ptr with
+    /// an exception pending, tripping its leak assertion). `Strong::value()`
+    /// is a pure slot read; `Strong::get()` would re-tag through `from_js`,
+    /// which is a safepoint.
     fn check_body_stream_ref(&self, global_object: &JSGlobalObject) {
         if let Some(js_value) = self.js_ref() {
             if let Value::Locked(locked) = self.get_body_value() {
-                if let Some(stream) = locked.readable.get(global_object) {
-                    stream.value.ensure_still_alive();
-                    Self::stream_set_cached(js_value, global_object, stream.value);
+                if let Some(stream_value) = locked.readable.value() {
+                    stream_value.ensure_still_alive();
+                    Self::stream_set_cached(js_value, global_object, stream_value);
                     locked.readable.downgrade();
                 }
             }

src/runtime/webcore/ReadableStream.rs

@@ -44,7 +44,7 @@ impl Default for Strong {
 }
 
 impl Strong {
-    fn value(&self) -> Option<JSValue> {
+    pub(crate) fn value(&self) -> Option<JSValue> {
         self.held.get().or_else(|| {
             if self.weak.is_empty() {
                 None

test/js/web/workers/worker-terminate-lifetime.test.ts

@@ -177,6 +177,74 @@ test.skipIf(!isASAN)(
   timeout,
 );
 
+// Regression: InternalMicrotask::BunPerformMicrotaskJob (used by
+// queueMicrotask and by the C++ stream start/pull reaction jobs) catches any
+// exception from the job callback with an unconditional clearException(), so
+// when the caught exception is the TerminationException the microtask drain
+// never observes termination. A worker in a microtask-bound loop that
+// constructs a JS-source ReadableStream (or calls queueMicrotask) each turn
+// would spin forever with terminate() never resolving.
+// The Response / Request variants additionally cover a trap safepoint inside
+// check_body_stream_ref that fired the TerminationException after the native
+// Response/Request had been heap-allocated, tripping the generated
+// constructor's "Memory leak detected: new Response()" assertion.
+describe("terminate() resolves for a worker in a microtask-bound ReadableStream loop", () => {
+  const variants: Record<string, string> = {
+    "new ReadableStream({pull})": `new ReadableStream({ pull(c) { c.enqueue(1); c.close(); } })`,
+    "new ReadableStream({start})": `new ReadableStream({ start(c) { c.close(); } })`,
+    "new Response(new ReadableStream)": `new Response(new ReadableStream({ async pull() {} }))`,
+    "new Request(new ReadableStream)": `new Request("http://x", { method: "POST", body: new ReadableStream({ pull(c) { c.close(); } }), duplex: "half" })`,
+    "queueMicrotask": `queueMicrotask(() => {})`,
+  };
+  // A single hang is deterministic on an unfixed build (the loop is purely
+  // microtask-bound), so a small sweep of offsets is plenty.
+  const localRounds = slow ? 3 : 6;
+  const deadline = slow ? 10_000 : 4_000;
+
+  for (const [name, expr] of Object.entries(variants)) {
+    test.concurrent(
+      name,
+      async () => {
+        await using proc = Bun.spawn({
+          cmd: [
+            bunExe(),
+            "-e",
+            `
+            const { Worker } = require("node:worker_threads");
+            const src = 'require("node:worker_threads").parentPort.postMessage("up");' +
+              '(async () => { for (;;) { ' + ${JSON.stringify(expr)} + '; await 0; } })();';
+            for (let r = 0; r < ${localRounds}; r++) {
+              const w = new Worker(src, { eval: true });
+              w.on("error", () => {});
+              await new Promise((res) => w.once("message", res));
+              await Bun.sleep((r * 23) % 80);
+              const winner = await Promise.race([
+                w.terminate().then(() => "ok"),
+                Bun.sleep(${deadline}).then(() => "hung"),
+              ]);
+              if (winner !== "ok") {
+                console.log("HUNG round " + r);
+                process.exit(1);
+              }
+            }
+            console.log("PASS");
+            process.exit(0);
+          `,
+          ],
+          env: bunEnv,
+          stdout: "pipe",
+          stderr: "pipe",
+        });
+        const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+        expect(stderr).toBe("");
+        expect(stdout).toBe("PASS\n");
+        expect(exitCode).toBe(0);
+      },
+      timeout,
+    );
+  }
+});
+
 // Regression: Bun.serve() inside a worker, streaming a JS ReadableStream body,
 // then worker.terminate() mid-stream. Worker shutdown stops the server which
 // tears down the in-flight HTTP(S)ResponseSink and fires its JS onClose hook