diff --git a/src/jsc/bindings/bindings.cpp b/src/jsc/bindings/bindings.cpp index d99045a75ad4..3ff5910afb31 100644 --- a/src/jsc/bindings/bindings.cpp +++ b/src/jsc/bindings/bindings.cpp @@ -4899,6 +4899,13 @@ bool JSC__VM__hasTerminationRequest(JSC::VM* vm) return vm->hasTerminationRequest(); } +[[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(); diff --git a/src/jsc/bindings/webcore/EventEmitter.cpp b/src/jsc/bindings/webcore/EventEmitter.cpp index 48148ff4e3c6..2c234cb891cd 100644 --- a/src/jsc/bindings/webcore/EventEmitter.cpp +++ b/src/jsc/bindings/webcore/EventEmitter.cpp @@ -252,6 +252,8 @@ bool EventEmitter::innerInvokeEventListeners(const Identifier& eventType, Simple auto* exception = exceptionPtr.get(); if (exception) [[unlikely]] { + if (vm.isTerminationException(exception)) [[unlikely]] + break; auto errorIdentifier = vm.propertyNames->error; auto hasErrorListener = this->hasActiveEventListeners(errorIdentifier); if (!hasErrorListener || eventType == errorIdentifier) { diff --git a/src/jsc/virtual_machine_exports.rs b/src/jsc/virtual_machine_exports.rs index 035dee7f05bc..14c7812bdc82 100644 --- a/src/jsc/virtual_machine_exports.rs +++ b/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 } diff --git a/src/runtime/webcore/Body.rs b/src/runtime/webcore/Body.rs index 201e73b7c338..88a40c6fc348 100644 --- a/src/runtime/webcore/Body.rs +++ b/src/runtime/webcore/Body.rs @@ -1691,9 +1691,10 @@ pub(crate) trait BodyMixin: BodyOwnerJs + Sized { 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); + // `Strong::get()` is a VMTraps safepoint; this runs post-alloc. + 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(); } } diff --git a/src/runtime/webcore/ReadableStream.rs b/src/runtime/webcore/ReadableStream.rs index 6ef9d5771301..4754ab75bcae 100644 --- a/src/runtime/webcore/ReadableStream.rs +++ b/src/runtime/webcore/ReadableStream.rs @@ -44,7 +44,7 @@ impl Default for Strong { } impl Strong { - fn value(&self) -> Option { + pub(crate) fn value(&self) -> Option { self.held.get().or_else(|| { if self.weak.is_empty() { None diff --git a/test/js/web/workers/worker-terminate-lifetime.test.ts b/test/js/web/workers/worker-terminate-lifetime.test.ts index 6b9f7c1b3366..be28663b68e8 100644 --- a/test/js/web/workers/worker-terminate-lifetime.test.ts +++ b/test/js/web/workers/worker-terminate-lifetime.test.ts @@ -177,6 +177,78 @@ 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 = { + "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 }); + await new Promise((res, rej) => { + w.once("message", res); + w.once("error", rej); + w.once("exit", (c) => rej(new Error("worker exited " + c + " before ready"))); + }); + w.on("error", () => {}); + 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