diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs index 456faa35776c..64b6a0ac0134 100644 --- a/src/runtime/valkey_jsc/js_valkey.rs +++ b/src/runtime/valkey_jsc/js_valkey.rs @@ -1366,13 +1366,6 @@ impl JSValkeyClient { // Callback for when Valkey client needs to reconnect pub(crate) fn on_valkey_reconnect(&self) { - // SAFETY: adopts connect()'s socket keep-alive ref for the just-closed - // socket (or the one `ValkeyDeferredClose::run` took in its place). - // Reached only from `ValkeyClient::on_close()`'s reconnect branch, - // which never calls `on_valkey_close()`, so this scope is the sole - // releaser. The caller holds its own scoped ref, so count > 0. - let _socket_ref = unsafe { ScopedRef::adopt(self.as_ctx_ptr()) }; - // This timer was bounding the attempt that just ended; left armed it // fires during the retry delay, and `fail()` then has no socket to // close and nothing settles connect(). `reconnect()` arms a new one. @@ -1384,11 +1377,6 @@ impl JSValkeyClient { // Callback for when Valkey client closes pub(crate) fn on_valkey_close(&self) -> JsResult<()> { let global_object = self.global_object; - - // SAFETY: adopts connect()'s socket keep-alive ref (or the one - // `ValkeyDeferredClose::run` took in its place); the caller holds its - // own scoped ref so count stays > 0 until this drops. - let _socket_ref = unsafe { ScopedRef::adopt(self.as_ctx_ptr()) }; let _defer = scopeguard::guard(BackRef::new(self), |p| p.update_poll_ref()); let Some(this_jsvalue) = self.this_value.get().try_get() else { @@ -1537,8 +1525,11 @@ impl JSValkeyClient { // `owner_ptr` opaquely (no overlapping write). let owner_ptr: *mut JSValkeyClient = std::ptr::from_ref::(self).cast_mut(); let client_ptr: *mut valkey::ValkeyClient = self.client.as_ptr(); - // Socket keep-alive ref, released by on_valkey_close/on_valkey_reconnect. - // Forgotten once there is a socket to own it. + // Socket keep-alive ref. Forgotten once there is a socket to own it; + // adopted by the guard at the entry of the socket's close event + // (`SocketHandler::on_close`, `SocketHandler::on_connect_error`, or + // `ValkeyClient::close()` for a half-open socket), which is the one + // event uSockets delivers for every socket this returns. let socket_ref = self.ref_scope(); // SAFETY: `client_ptr` is live; `group` is the lazy-initialised per-VM // `SocketGroup` (stable for the VM's lifetime). `ssl_ctx` is a +1-ref @@ -1859,6 +1850,10 @@ impl SocketHandler { ) -> JsResult<()> { debug!("Socket closed."); let _guard = this.ref_scope(); + // SAFETY: adopts the keep-alive ref `connect()` forgot for this + // socket; this is its one close event. Released after `_defer` runs, + // while `_guard` still holds the client. + let _socket_ref = unsafe { ScopedRef::adopt(this.as_ctx_ptr()) }; // Ensure the socket pointer is updated. this.client_mut().socket = Socket::SocketTcp(uws::SocketTCP::detached()); // Before `on_close()`: it runs `onclose` and settles the connect() @@ -1886,6 +1881,8 @@ impl SocketHandler { // Ensure the socket pointer is updated. this.client_mut().socket = Socket::SocketTcp(uws::SocketTCP::detached()); let _guard = this.ref_scope(); + // SAFETY: as in `on_close`; a dial that fails gets this event instead. + let _socket_ref = unsafe { ScopedRef::adopt(this.as_ctx_ptr()) }; this.client_mut().status = valkey::Status::Disconnected; let _defer = scopeguard::guard(BackRef::new(this), |p| p.update_poll_ref()); @@ -2039,9 +2036,8 @@ impl ValkeyDeferredClose { if !this.client.get().socket.is_closed() { return; } - // `on_close()` ends in `on_valkey_close`/`on_valkey_reconnect`, - // which release the ref the socket would have held. - this.ref_(); + // No socket ref to give back: `connect()` forgets it only once + // it has a socket, and this task exists because it never did. this.client_mut().status = valkey::Status::Disconnected; let closed = this.client_mut().on_close(); this.update_poll_ref(); diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs index 5e23976d16e2..afd8c14935da 100644 --- a/src/runtime/valkey_jsc/valkey.rs +++ b/src/runtime/valkey_jsc/valkey.rs @@ -6,6 +6,7 @@ use bun_collections::VecExt; use bun_collections::OffsetByteList; use bun_jsc::virtual_machine::VirtualMachine; use bun_jsc::{GlobalRef, JSGlobalObject, JSPromise, JSValue, JsResult}; +use bun_ptr::ScopedRef; use bun_uws::{self as uws, AnySocket, SocketGroup, SocketKind, SslCtx}; use bun_valkey::valkey_protocol as protocol; use bun_valkey::valkey_protocol::{RESPValue, RedisError}; @@ -486,19 +487,23 @@ impl ValkeyClient { ) -> JsResult<()> { let mut pending = core::mem::take(pending_ptr); let mut entries = core::mem::take(entries_ptr); - // Note: `defer pending.deinit()` / `defer entries.deinit()` — handled by Drop. - // Reject commands in the command queue + // A rejection fails once the VM's termination is pending; the rest of + // both queues still has to be read out and dropped. + let mut result = Ok(()); while let Some(mut command_pair) = pending.pop_front() { - command_pair.reject_command(global_this, jsvalue)?; + let rejected = command_pair.reject_command(global_this, jsvalue); + if result.is_ok() { + result = rejected; + } } - - // Reject commands in the offline queue while let Some(mut cmd) = entries.pop_front() { - // Note: `defer cmd.deinit(allocator)` — Entry should impl Drop. - cmd.promise.reject(global_this, Ok(jsvalue))?; + let rejected = cmd.promise.reject(global_this, Ok(jsvalue)); + if result.is_ok() { + result = rejected; + } } - Ok(()) + result } fn reject_in_flight_commands(&mut self, message: &[u8], err: RedisError) -> JsResult<()> { @@ -618,25 +623,29 @@ impl ValkeyClient { // hasn't resolved yet (`POLL_TYPE_SEMI_SOCKET` — DNS resolved // synchronously so `connect()` got a real `us_socket_t*` rather than // a `us_connecting_socket_t*`). See `us_internal_socket_close_raw`. - // The valkey client relies on one of those callbacks (via - // `on_valkey_close`/`on_valkey_reconnect`) to release the `+1` - // keep-alive ref `connect()` took, so without one the - // `JSValkeyClient` box leaks. Detect a SEMI_SOCKET before closing - // and run the close path ourselves afterwards. + // The close event is what releases the keep-alive ref `connect()` + // took, so detect a SEMI_SOCKET before closing and run the close + // event by hand afterwards. let is_semi_socket = matches!(socket.socket(), uws::InternalSocket::Connected(_)) && !socket.is_established(); // TODO: make socket.close() return a JsResult. socket.close(code); - if global.has_exception() { - return Err(bun_jsc::JsError::Thrown); - } - if is_semi_socket { - self.status = Status::Disconnected; - // A half-open socket never gets uSockets' close dispatch, so run the - // close event here. - return self.on_close(); + let thrown = if global.has_exception() { + Err(bun_jsc::JsError::Thrown) + } else { + Ok(()) + }; + if !is_semi_socket { + return thrown; } - Ok(()) + // SAFETY: adopts the keep-alive ref `connect()` forgot for this + // socket, as `SocketHandler::on_close` does for one uSockets closes. + // Every caller of `close()` holds a scoped ref of its own, so the + // client outlives this scope. + let _socket_ref = unsafe { ScopedRef::adopt(self.parent_ptr()) }; + self.status = Status::Disconnected; + let closed = self.on_close(); + thrown.and(closed) } /// Handle connection closed event diff --git a/test/js/valkey/valkey-gc.test.ts b/test/js/valkey/valkey-gc.test.ts index 5ceeea2bc115..e4cbd22a8019 100644 --- a/test/js/valkey/valkey-gc.test.ts +++ b/test/js/valkey/valkey-gc.test.ts @@ -1,5 +1,5 @@ -import { expect, test } from "bun:test"; -import { bunEnv, bunExe, bunRun, isASAN } from "harness"; +import { describe, expect, test } from "bun:test"; +import { bunEnv, bunExe, bunRun, isASAN, isMusl, isWindows } from "harness"; import net from "node:net"; import { join } from "node:path"; @@ -43,6 +43,245 @@ test.concurrent("RedisClient survives a failed custom-TLS context without freein expect(exitCode).toBe(0); }); +// The socket's close event (ValkeyClient::on_close) rejects the commands the +// connection still owes replies for, then settles connect()/onclose or arms the +// retry. Rejecting a promise fails once the VM's termination is pending, which +// is the state a terminated worker's teardown closes its sockets in, and +// on_close() then returns before its callees ran. The keep-alive ref the socket +// held on the client used to be released by those callees, so every +// RedisClient terminated with commands owed leaked its Box; now +// the close event's entry releases it whatever on_close() returns. One case per +// branch of on_close() (retry scheduled, autoReconnect off, retries exhausted) +// with the commands in flight, one with commands in the offline queue as well, +// and one whose close event is on_connect_error (a dial that never completes) +// rather than on_close. With the offline queue, the first rejection failing +// used to leave the queue's remaining entries undropped, leaking their +// serialized bytes as well. Only observable via LSan, so ASAN-only. (The main +// thread's teardown under process.exit() closes the same sockets with no +// termination pending, so the rejections succeed there and nothing leaked.) +describe.skipIf(!isASAN)("VM teardown with commands owed to a RedisClient leaks nothing", () => { + const CRLF = "\\r\\n"; + // Answers HELLO, never replies to a command, and resolves `ready` once the + // first INCR has arrived, so the commands it owes are in flight from then on. + const server = ` + const HELLO = "%1${CRLF}$5${CRLF}proto${CRLF}:3${CRLF}"; + const { promise: ready, resolve: onReady } = Promise.withResolvers(); + const server = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + socket: { + open(s) { s.data = { buf: "", hello: false }; }, + data(s, chunk) { + s.data.buf += chunk.toString("latin1"); + if (!s.data.hello && s.data.buf.includes("HELLO")) { + s.data.hello = true; + s.write(HELLO); + } + if (s.data.buf.includes("INCR")) onReady(); + }, + close() {}, + error() {}, + }, + }); + `; + + async function expectCleanExit(src: string, stdout: string) { + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", src], + env: { + ...bunEnv, + BUN_DESTRUCT_VM_ON_EXIT: "1", + ASAN_OPTIONS: [bunEnv.ASAN_OPTIONS, "detect_leaks=1"].filter(Boolean).join(":"), + LSAN_OPTIONS: `print_suppressions=0:suppressions=${join(import.meta.dirname, "../../leaksan.supp")}`, + }, + stdout: "pipe", + stderr: "pipe", + }); + const [out, err, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect({ stdout: out, stderr: err, exitCode }).toEqual({ stdout, stderr: "", exitCode: 0 }); + } + + // The client is created from a macrotask on purpose: allocations made while + // the worker's module body is still on the stack match the module-evaluation + // entries of leaksan.supp, and a leaked client would then go unreported. + const workerSrc = (body: string) => ` + const { parentPort, workerData } = require("node:worker_threads"); + setImmediate(() => { + const client = new Bun.RedisClient(workerData.url, workerData.options); + globalThis.client = client; + ${body} + }); + `; + + // Terminates the worker once the server reports `ready`, with whatever the + // client still owes. + function terminateWorker(options: object, worker: string) { + return expectCleanExit( + ` + const { Worker } = require("node:worker_threads"); + ${server} + const worker = new Worker(${JSON.stringify(workerSrc(worker))}, { + eval: true, + workerData: { url: "redis://127.0.0.1:" + server.port, options: ${JSON.stringify(options)} }, + }); + worker.on("error", (err) => { console.error(err); process.exit(2); }); + worker.on("exit", (code) => { console.error("worker exited on its own with " + code); process.exit(3); }); + await ready; + worker.removeAllListeners("exit"); + console.log("terminated", await worker.terminate()); + server.stop(true); + `, + "terminated 1\n", + ); + } + + const inFlight = `client.connect().then(() => { for (let i = 0; i < 4; i++) client.incr("k").catch(() => {}); });`; + + // Symbolizing a leak report takes LSan several seconds on a debug binary. + const timeout = 60_000; + test.concurrent("worker.terminate(): retry scheduled", () => terminateWorker({}, inFlight), timeout); + test.concurrent( + "worker.terminate(): autoReconnect off", + () => terminateWorker({ autoReconnect: false }, inFlight), + timeout, + ); + test.concurrent("worker.terminate(): retries exhausted", () => terminateWorker({ maxRetries: 0 }, inFlight), timeout); + + // WATCH is not auto-pipelined, so it waits in the offline queue while the + // INCRs are in flight, and the INCRs sent after it queue up behind it. The + // first in-flight rejection failing then leaves the whole queue behind. + test.concurrent( + "worker.terminate(): commands queued behind a non-pipelined command", + () => + terminateWorker( + { autoReconnect: false }, + `client.connect().then(() => { + for (let i = 0; i < 4; i++) client.incr("k").catch(() => {}); + client.send("WATCH", ["k"]).catch(() => {}); + for (let i = 0; i < 4; i++) client.incr("k").catch(() => {}); + });`, + ), + timeout, + ); + + // A listener nobody accepts from, with a backlog one filler connection + // fills, so the kernel drops every later SYN and a dial to `port` sits in + // EINPROGRESS for good. Needs listen(2) with the smallest backlog that + // admits exactly one connection (macOS treats 0 as unlimited), which Bun's + // own listeners do not expose, so the listener is a raw libc socket. + const blackhole = ` + const net = require("node:net"); + const { dlopen, ptr } = require("bun:ffi"); + const darwin = process.platform === "darwin"; + const libc = dlopen(darwin ? "libSystem.B.dylib" : "libc.so.6", { + socket: { args: ["int", "int", "int"], returns: "int" }, + bind: { args: ["int", "ptr", "int"], returns: "int" }, + listen: { args: ["int", "int"], returns: "int" }, + getsockname: { args: ["int", "ptr", "ptr"], returns: "int" }, + }); + const AF_INET = 2, SOCK_STREAM = 1; + const addr = new Uint8Array(16); + if (darwin) { addr[0] = 16; addr[1] = AF_INET; } else new DataView(addr.buffer).setUint16(0, AF_INET, true); + addr.set([127, 0, 0, 1], 4); + const fd = libc.symbols.socket(AF_INET, SOCK_STREAM, 0); + if (fd < 0 || libc.symbols.bind(fd, ptr(addr), 16) !== 0 || libc.symbols.listen(fd, darwin ? 1 : 0) !== 0) throw new Error("listen failed"); + const len = new Uint32Array([16]); + if (libc.symbols.getsockname(fd, ptr(addr), ptr(len)) !== 0) throw new Error("getsockname failed"); + const port = (addr[2] << 8) | addr[3]; + // The error listener outlives the await, so a later error on the filler + // is swallowed rather than thrown. + const filler = net.connect(port, "127.0.0.1"); + await new Promise((resolve, reject) => filler.on("connect", resolve).on("error", reject)); + `; + + // The dial never completes, so teardown delivers on_connect_error for it, + // with the commands in the offline queue. + test.concurrent.skipIf(isWindows || isMusl)( + "worker.terminate(): commands queued behind a dial that stays pending", + () => + expectCleanExit( + ` + const { Worker } = require("node:worker_threads"); + ${blackhole} + const { promise: dialing, resolve: onDialing } = Promise.withResolvers(); + const worker = new Worker(${JSON.stringify( + workerSrc(` + client.connect().catch(() => {}); + for (let i = 0; i < 4; i++) client.incr("k").catch(() => {}); + parentPort.postMessage("dialing"); + `), + )}, { + eval: true, + workerData: { url: "redis://127.0.0.1:" + port, options: { autoReconnect: false } }, + }); + worker.on("message", onDialing); + worker.on("error", (err) => { console.error(err); process.exit(2); }); + worker.on("exit", (code) => { console.error("worker exited on its own with " + code); process.exit(3); }); + await dialing; + worker.removeAllListeners("exit"); + console.log("terminated", await worker.terminate()); + filler.destroy(); + `, + "terminated 1\n", + ), + timeout, + ); + + // A dial to an IP literal gets a real us_socket_t back from uSockets before + // the TCP handshake completes (POLL_TYPE_SEMI_SOCKET), and uSockets delivers + // no close event when the application closes one of those, so + // ValkeyClient::close() releases connect()'s keep-alive ref and runs the + // close event by hand. One case per entry into that branch: close() while + // the dial is pending, and the connection timeout firing during it. The + // command in the offline queue tells the two apart: connect() itself is + // always rejected as connection-closed. The client is created from a + // macrotask for the same reason as the workers'. + function closePendingDial(options: object, body: string, stdout: string) { + return expectCleanExit( + ` + ${blackhole} + const { promise: done, resolve: onDone } = Promise.withResolvers(); + setImmediate(async () => { + const client = new Bun.RedisClient("redis://127.0.0.1:" + port, ${JSON.stringify(options)}); + let closes = 0; + client.onclose = () => { closes++; }; + const connecting = client.connect(); + const queued = client.get("k"); + ${body} + const code = (p) => p.then(() => "resolved", (err) => err.code); + console.log(await code(connecting), await code(queued), closes, client.connected); + Bun.gc(true); + onDone(); + }); + await done; + Bun.gc(true); + filler.destroy(); + `, + stdout + "\n", + ); + } + test.concurrent.skipIf(isWindows || isMusl)( + "close() while a dial to an IP literal is pending", + () => + closePendingDial( + { autoReconnect: false }, + "client.close();", + "ERR_REDIS_CONNECTION_CLOSED ERR_REDIS_CONNECTION_CLOSED 1 false", + ), + timeout, + ); + test.concurrent.skipIf(isWindows || isMusl)( + "connection timeout while a dial to an IP literal is pending", + () => + closePendingDial( + { connectionTimeout: 1 }, + "", + "ERR_REDIS_CONNECTION_CLOSED ERR_REDIS_CONNECTION_TIMEOUT 1 false", + ), + timeout, + ); +}); + // Fuzzer found a heap-use-after-free that survived the ScopedRef refactor: // on_connection_timeout's unconditional `ScopedRef::adopt` released a ref the // timer no longer held, so the ScopedRef drop at scope end brought the