worker: survive terminate() while Bun.serve is streaming a type:"direct" body

src/jsc/VM.rs

@@ -118,10 +118,6 @@ impl VM {
         JSC__VM__notifyNeedTermination(self)
     }
 
-    pub(crate) fn clear_has_termination_request(&self) {
-        crate::cpp::JSC__VM__clearHasTerminationRequest(self)
-    }
-
     #[track_caller]
     pub fn throw_error(&self, global_object: &JSGlobalObject, value: JSValue) -> JsError {
         crate::validation_scope!(scope, global_object);

src/jsc/bindings/bindings.cpp

@@ -4923,11 +4923,6 @@ bool JSC__VM__isTerminationException(JSC::VM* vm, JSC::Exception* exception)
     return vm->isTerminationException(exception);
 }
 
-[[ZIG_EXPORT(nothrow)]]
-void JSC__VM__clearHasTerminationRequest(JSC::VM* vm)
-{
-    vm->clearHasTerminationRequest();
-}
 [[ZIG_EXPORT(nothrow)]]
 bool JSC__VM__hasTerminationRequest(JSC::VM* vm)
 {

src/jsc/web_worker.rs

@@ -1243,12 +1243,11 @@ impl WebWorker {
             // SAFETY: vm_ptr valid; unpublished above under vm_lock, so no
             // other thread can dereference it now — `&mut` is exclusive.
             let vm = unsafe { &mut *vm_ptr };
-            // terminate() set the JSC termination flag to interrupt running JS;
-            // clear it so process.on('exit') handlers can run. teardownJSCVM
-            // re-sets it for the JSC VM teardown.
-            vm.jsc_vm().clear_has_termination_request();
             vm.is_shutting_down = true;
+            // on_exit first: dispatchExitInternal skips 'exit' on terminate (node semantics).
             vm.on_exit();
+            // Cleared here so close_all_socket_groups can re-enter JS; teardownJSCVM re-sets it.
+            vm.global().clear_termination_exception();
             if let Some(hooks) = runtime_hooks() {
                 (hooks.cron_clear_all_teardown)(vm);
                 // Drain `TimeoutObject`s from this worker's timer heap before

src/runtime/server/RequestContext.rs

@@ -2129,6 +2129,14 @@ where
                 stream_log!("returned a promise");
                 this.drain_microtasks();
 
+                // terminate() inside the drain leaves a pending TerminationException; every branch below re-enters JS.
+                if global_this.has_exception() {
+                    response_stream.sink.on_first_write = None;
+                    response_stream.sink.ctx = None;
+                    this.flags.set_has_marked_pending(true);
+                    return;
+                }
+
                 // `MarkHandled` matters for the Rejected arm: the promise
                 // settled before any reaction was attached, so without the
                 // flag the VM would report it as an unhandled rejection even
@@ -2679,6 +2687,11 @@ where
         response_value.ensure_still_alive();
         ctx.drain_microtasks();
 
+        // Same as do_render_stream: terminate() inside the drain leaves a pending TerminationException.
+        if this.global_this().has_exception() {
+            return;
+        }
+
         if ctx.is_aborted_or_ended() {
             return;
         }

src/runtime/server/mod.rs

@@ -1313,6 +1313,11 @@ impl<const SSL: bool, const DEBUG: bool> NewServer<SSL, DEBUG> {
                     needs_to_drain = false;
                     // SAFETY: `vm` is the process-static VirtualMachine.
                     unsafe { (*vm).drain_microtasks() };
+                    // Same as RequestContext::on_response: terminate() inside the drain leaves a pending TerminationException.
+                    if global.has_exception() {
+                        strong_promise.deinit();
+                        break 'brk HttpResult::Success;
+                    }
                     status = promise.status();
                 }
 

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

@@ -11,6 +11,198 @@ const rounds = slow ? 4 : 8;
 const perRound = slow ? 12 : 32;
 const timeout = slow ? 60_000 : 20_000;
 
+// Regression: Bun.serve() inside a worker streaming a type:"direct" body,
+// then worker.terminate() mid-stream. drain_microtasks() drives the async
+// pull loop; when the termination trap fires inside it, the outermost
+// VMEntryScope's executeEntryScopeServicesOnExit clears hasTerminationRequest
+// while the TerminationException stays pending. do_render_stream then kept
+// going and (a) lazy-initialized NativePromiseContextStructure inside a
+// DeferTerminationForAWhile scope, asserting hasTerminationRequest() in
+// VMTraps::deferTerminationSlow, and (b) re-entered JS via then_with_value,
+// asserting !exception() in executeCallImpl. Separately, WebWorker::shutdown
+// cleared only the request flag before close_all_socket_groups, leaving the
+// sticky exception in place for the same class of assert. Both asserts are
+// compiled out outside debug/ASAN builds.
+test.skipIf(!(isDebug || isASAN))(
+  'terminate() while Bun.serve is streaming a type:"direct" body does not trip DeferTermination / assertNoException',
+  async () => {
+    await using proc = Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `
+        const { Worker } = require("node:worker_threads");
+        const src = 'const server = Bun.serve({ port: 0, fetch() {' +
+          '  return new Response(new ReadableStream({ type: "direct", async pull(ctrl) {' +
+          '    for (let i = 0; i < 20000; i++) {' +
+          '      ctrl.write(new Uint8Array(8192).fill(i));' +
+          '      await ctrl.flush();' +
+          '    }' +
+          '    ctrl.close();' +
+          '  }, cancel() {} }));' +
+          '} });' +
+          'require("node:worker_threads").parentPort.postMessage(server.port);';
+        for (let i = 0; i < ${rounds}; i++) {
+          const w = new Worker(src, { eval: true });
+          const port = await new Promise(r => w.once("message", r));
+          const res = await fetch("http://127.0.0.1:" + port + "/");
+          const rd = res.body.getReader();
+          await rd.read();
+          const done = new Promise(r => w.once("exit", r));
+          w.terminate();
+          await done;
+          await rd.cancel().catch(() => {});
+        }
+        console.log("survived");
+      `,
+      ],
+      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("survived\n");
+    expect(exitCode).toBe(0);
+  },
+  timeout,
+);
+
+// Sibling of the above: on_response's drain_microtasks() driving an async
+// fetch handler that never yields past `await 1`. Same DeferTermination /
+// assertNoException pair, same fix shape. Also asserts Node's
+// "process.on('exit') is skipped on terminate()" when terminate lands while
+// JS is on the stack (sab[1] stays 0): in that state the NeedTermination trap
+// has already been consumed, so only the sticky TerminationException guards
+// dispatchExitInternal; clearing it before on_exit would let the handler run.
+test.skipIf(!(isDebug || isASAN))(
+  "terminate() while Bun.serve's async fetch is looping in drain_microtasks does not trip DeferTermination",
+  async () => {
+    await using proc = Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `
+        const { Worker } = require("node:worker_threads");
+        const sab = new SharedArrayBuffer(8);
+        const flag = new Int32Array(sab);
+        const src = 'const { workerData } = require("node:worker_threads");' +
+          'const flag = new Int32Array(workerData.sab);' +
+          'process.on("exit", () => Atomics.add(flag, 1, 1));' +
+          'const server = Bun.serve({ port: 0, async fetch() {' +
+          '  for (;;) { Atomics.add(flag, 0, 1); await 1; }' +
+          '} });' +
+          'require("node:worker_threads").parentPort.postMessage(server.port);';
+        for (let i = 0; i < ${rounds}; i++) {
+          Atomics.store(flag, 0, 0);
+          Atomics.store(flag, 1, 0);
+          const w = new Worker(src, { eval: true, workerData: { sab } });
+          const port = await new Promise(r => w.once("message", r));
+          const ac = new AbortController();
+          fetch("http://127.0.0.1:" + port + "/", { signal: ac.signal }).catch(() => {});
+          while (Atomics.load(flag, 0) < 100) await Bun.sleep(0);
+          const done = new Promise(r => w.once("exit", r));
+          w.terminate();
+          await done;
+          ac.abort();
+          if (Atomics.load(flag, 1) !== 0) throw new Error("process.on('exit') fired on terminate");
+        }
+        console.log("survived");
+      `,
+      ],
+      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("survived\n");
+    expect(exitCode).toBe(0);
+  },
+  timeout,
+);
+
+// Same again for the node:http-compat path (on_node_http_request_with_upgrade_ctx).
+test.skipIf(!(isDebug || isASAN))(
+  "terminate() while a node:http server's async handler is looping in drain_microtasks does not trip DeferTermination",
+  async () => {
+    await using proc = Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `
+        const { Worker } = require("node:worker_threads");
+        const sab = new SharedArrayBuffer(4);
+        const flag = new Int32Array(sab);
+        const src = 'const { workerData } = require("node:worker_threads");' +
+          'const flag = new Int32Array(workerData.sab);' +
+          'const http = require("node:http");' +
+          'const server = http.createServer(async (req, res) => {' +
+          '  for (;;) { Atomics.add(flag, 0, 1); await 1; }' +
+          '}).listen(0, () => require("node:worker_threads").parentPort.postMessage(server.address().port));';
+        for (let i = 0; i < ${rounds}; i++) {
+          Atomics.store(flag, 0, 0);
+          const w = new Worker(src, { eval: true, workerData: { sab } });
+          const port = await new Promise(r => w.once("message", r));
+          const ac = new AbortController();
+          fetch("http://127.0.0.1:" + port + "/", { signal: ac.signal }).catch(() => {});
+          while (Atomics.load(flag, 0) < 100) await Bun.sleep(0);
+          const done = new Promise(r => w.once("exit", r));
+          w.terminate();
+          await done;
+          ac.abort();
+        }
+        console.log("survived");
+      `,
+      ],
+      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("survived\n");
+    expect(exitCode).toBe(0);
+  },
+  timeout,
+);
+
+// Node skips the worker's process.on('exit') listeners on worker.terminate()
+// when the worker is idle (the NeedTermination trap bit is still armed here,
+// so this is the simpler half of the invariant; the mid-JS half is checked in
+// the async-fetch test above).
+test("process.on('exit') does not fire on worker.terminate() when idle", async () => {
+  await using proc = Bun.spawn({
+    cmd: [
+      bunExe(),
+      "-e",
+      `
+      const { Worker } = require("node:worker_threads");
+      const sab = new SharedArrayBuffer(4);
+      const counter = new Int32Array(sab);
+      const src = 'const { workerData, parentPort } = require("node:worker_threads");' +
+        'const counter = new Int32Array(workerData.sab);' +
+        'process.on("exit", () => Atomics.add(counter, 0, 1));' +
+        'parentPort.postMessage("ready");' +
+        'setInterval(() => {}, 100000);';
+      const w = new Worker(src, { eval: true, workerData: { sab } });
+      await new Promise(r => w.once("message", r));
+      const done = new Promise(r => w.once("exit", r));
+      w.terminate();
+      await done;
+      console.log("exitCalls=" + Atomics.load(counter, 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("exitCalls=0\n");
+  expect(exitCode).toBe(0);
+});
+
 // Regression: `new Worker(url, { ref: false })` was silently ignored — the
 // Zig-side `user_keep_alive` field was set from it but never read, and the
 // parent keep-alive was taken unconditionally in `create()`. `.unref()` after