diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs index 5a592cd7ad0e..026190d9578d 100644 --- a/src/runtime/valkey_jsc/js_valkey.rs +++ b/src/runtime/valkey_jsc/js_valkey.rs @@ -1382,6 +1382,13 @@ impl JSValkeyClient { let _socket_ref = unsafe { ScopedRef::adopt(self.as_ctx_ptr()) }; let _defer = scopeguard::guard(BackRef::new(self), |p| p.update_poll_ref()); + // Reached from `ValkeyClient::on_close()`'s stopped-VM branch during + // worker shutdown; the ref adoption above is the only thing that must + // still run. + if self.vm().script_execution_status() != jsc::ScriptExecutionStatus::Running { + return Ok(()); + } + let Some(this_jsvalue) = self.this_value.get().try_get() else { return Ok(()); }; diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs index cd183884c1bb..5d7d1ad88ed1 100644 --- a/src/runtime/valkey_jsc/valkey.rs +++ b/src/runtime/valkey_jsc/valkey.rs @@ -651,6 +651,26 @@ impl ValkeyClient { self.unregister_auto_flusher(); self.write_buffer.clear_and_free(); + // Worker shutdown's socket-group drain reaches here with the VM + // stopped and a TerminationException still pending; every branch below + // materialises a coded JS error, which lazily initialises + // `nodeErrorCache` under a DeferTermination scope and trips + // `ASSERT(vm.hasTerminationRequest())`. Drop the queues without + // building errors and release the socket ref. + if self.vm.script_execution_status() != bun_jsc::ScriptExecutionStatus::Running { + let mut pending = + core::mem::replace(&mut self.in_flight, command::promise_pair::Queue::init()); + let mut entries = core::mem::replace(&mut self.queue, command::entry::Queue::init()); + while let Some(pair) = pending.read_item() { + drop(pair); + } + while let Some(cmd) = entries.read_item() { + drop(cmd); + } + self.on_valkey_close()?; + return Ok(()); + } + // If manually closing, don't attempt to reconnect if self.flags.is_manually_closed { debug!("skip reconnecting since the connection is manually closed"); diff --git a/test/js/web/workers/worker-terminate-lifetime.test.ts b/test/js/web/workers/worker-terminate-lifetime.test.ts index 6b9f7c1b3366..b63d13df0857 100644 --- a/test/js/web/workers/worker-terminate-lifetime.test.ts +++ b/test/js/web/workers/worker-terminate-lifetime.test.ts @@ -402,6 +402,92 @@ test.skipIf(!isDebug)( 120_000, ); +// Regression: worker shutdown's socket-group drain fires valkey's on_close, +// which built a coded JS Error for every in-flight command. With a +// TerminationException still pending and hasTerminationRequest() already +// cleared for process.on('exit'), the first coded error in that worker +// lazily initialised nodeErrorCache under a DeferTermination scope and +// tripped ASSERT(vm.hasTerminationRequest()) in VMTraps::deferTerminationSlow, +// SIGABRTing the whole process. Release WebKit compiles that ASSERT out. +test.skipIf(!isDebug)( + "terminate() while a worker's Bun.RedisClient has commands in flight does not trip deferTerminationSlow's ASSERT", + async () => { + // Each worker connects to an inline RESP3 responder in the parent, keeps + // 2000 INCRs in flight, and is terminated once it reports hot. The + // socket-group drain at shutdown then hits on_close with the in-flight + // queue full. Unpatched debug builds abort within the first few iterations. + const ITER = 20; + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const { Worker } = require("node:worker_threads"); + const HELLO = "%3\\r\\n+server\\r\\n+fake\\r\\n+version\\r\\n+7.4.0\\r\\n+proto\\r\\n:3\\r\\n"; + const server = Bun.listen({ + hostname: "127.0.0.1", port: 0, + socket: { + open(s) { s.helloDone = false; }, + data(s, d) { + let out = ""; + for (const line of d.toString("latin1").split("\\r\\n")) { + if (line[0] !== "*") continue; + out += s.helloDone ? ":1\\r\\n" : HELLO; + s.helloDone = true; + } + if (out) s.write(out); + }, + close() {}, error() {}, + }, + }); + const url = "redis://127.0.0.1:" + server.port; + const src = + "const { parentPort, workerData: d } = require('node:worker_threads');" + + "const c = new Bun.RedisClient(d.url, { autoReconnect: d.reconnect });" + + "await c.connect();" + + "parentPort.postMessage('hot');" + + "for (;;) {" + + " const ps = [];" + + " for (let i = 0; i < 2000; i++) ps.push(c.incr('k').catch(() => {}));" + + " await Promise.all(ps);" + + "}"; + function ready(w) { + return new Promise((res, rej) => { + w.once("message", res); + w.once("error", rej); + w.once("exit", (c) => rej(new Error("worker exited " + c + " before ready"))); + }); + } + let done = 0; + for (let i = 0; i < ${ITER}; i++) { + const w = new Worker(src, { + eval: true, + workerData: { url, reconnect: i % 2 === 0 }, + }); + await ready(w); + w.on("error", () => {}); + await Bun.sleep((i * 7) % 60); + await w.terminate(); + done++; + } + server.stop(true); + if (done !== ${ITER}) throw new Error("only " + done + "/${ITER} terminated"); + console.log("PASS"); + `, + ], + 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); + }, + 120_000, +); + // Regression: Bun__handleUncaughtException probed process._fatalException (a // JS get(), where the worker's termination trap fires) and then called // wrapped.emit("uncaughtException") with the sticky TerminationException