diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs index 9f6c186dcb90..d2691258fafd 100644 --- a/src/runtime/valkey_jsc/js_valkey.rs +++ b/src/runtime/valkey_jsc/js_valkey.rs @@ -359,6 +359,9 @@ pub struct JSValkeyClient { pub timer: RefCountedTimer, pub reconnect_timer: RefCountedTimer, + /// Socket keep-alive refs outstanding from [`connect`](Self::connect); + /// consumed by [`take_socket_ref`](Self::take_socket_ref). + socket_refs: Cell, pub ref_count: bun_ptr::RefCount, } @@ -501,6 +504,20 @@ impl JSValkeyClient { // SAFETY: `self` is live; the guard's own ref keeps it alive past Drop. unsafe { ScopedRef::new(self.as_ctx_ptr()) } } + /// Adopt one socket keep-alive ref from [`connect`](Self::connect), or + /// `None` when the re-entrant close/fail path has already taken it. + #[inline] + fn take_socket_ref(&self) -> Option> { + match self.socket_refs.get().checked_sub(1) { + None => None, + Some(n) => { + self.socket_refs.set(n); + // SAFETY: `connect()` took this `+1` via `socket_ref.forget()` + // and recorded it in `socket_refs`; this scope consumes it. + Some(unsafe { ScopedRef::adopt(self.as_ctx_ptr()) }) + } + } + } #[inline] pub fn new(init: JSValkeyClient) -> *mut JSValkeyClient { // bun.TrivialNew(@This()) → heap::alloc(Box::new(init)) @@ -818,6 +835,7 @@ impl JSValkeyClient { _secure: Cell::new(None), timer: RefCountedTimer::new(Timer::Tag::ValkeyConnectionTimeout), reconnect_timer: RefCountedTimer::new(Timer::Tag::ValkeyConnectionReconnect), + socket_refs: Cell::new(0), })) } @@ -938,6 +956,7 @@ impl JSValkeyClient { _secure: Cell::new(None), timer: RefCountedTimer::new(Timer::Tag::ValkeyConnectionTimeout), reconnect_timer: RefCountedTimer::new(Timer::Tag::ValkeyConnectionReconnect), + socket_refs: Cell::new(0), })) } @@ -1379,11 +1398,7 @@ impl JSValkeyClient { // Callback for when Valkey client needs to reconnect pub fn on_valkey_reconnect(&self) { - // SAFETY: adopts connect()'s socket keep-alive ref for the just-closed - // socket. 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()) }; + let _socket_ref = self.take_socket_ref(); self.reconnect_timer .arm(self, self.client.get().get_reconnect_delay()); @@ -1393,9 +1408,7 @@ impl JSValkeyClient { pub fn on_valkey_close(&self) -> JsTerminatedResult<()> { let global_object = self.global_object; - // SAFETY: adopts connect()'s socket keep-alive ref; 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 _socket_ref = self.take_socket_ref(); let _defer = scopeguard::guard(BackRef::new(self), |p| p.update_poll_ref()); let Some(this_jsvalue) = self.this_value.get().try_get() else { @@ -1596,6 +1609,7 @@ impl JSValkeyClient { // `on_valkey_close()` consumes the socket ref; hand it over so it // isn't released twice. socket_ref.forget(); + self.socket_refs.set(self.socket_refs.get() + 1); self.client_mut().on_valkey_close()?; self.client_mut().status = valkey::Status::Disconnected; return Ok(()); @@ -1637,6 +1651,7 @@ impl JSValkeyClient { // Disarm on success: the socket now owns the keep-alive ref. scopeguard::ScopeGuard::into_inner(errdefer_status); socket_ref.forget(); + self.socket_refs.set(self.socket_refs.get() + 1); Ok(()) } @@ -1709,6 +1724,7 @@ impl JSValkeyClient { debug_assert!(this_ref.client.get().socket.is_closed()); debug_assert!(!this_ref.timer.ref_held.get()); debug_assert!(!this_ref.reconnect_timer.ref_held.get()); + debug_assert_eq!(this_ref.socket_refs.get(), 0); if let Some(s) = this_ref._secure.get() { // SAFETY: SSL_CTX is C-refcounted; this releases our ref. unsafe { boringssl::c::SSL_CTX_free(s) }; diff --git a/test/js/valkey/valkey-gc.test.ts b/test/js/valkey/valkey-gc.test.ts index 1a4a8e9e2dfe..b90031305b3c 100644 --- a/test/js/valkey/valkey-gc.test.ts +++ b/test/js/valkey/valkey-gc.test.ts @@ -181,6 +181,54 @@ test.concurrent("RedisClient survives subscribe() + close() against a server tha expect(exitCode).toBe(0); }); +// Drives on_data -> parse fail -> fail_with_js_value -> close() -> synchronous +// on_close -> on_valkey_close, with the connect() rejection drained while the +// socket-handler frames are still on the stack. Path coverage for +// take_socket_ref(); the fuzzer interleaving that over-releases here is not +// reproduced deterministically. +test.concurrent("RedisClient survives a malformed RESP reply that closes the socket from on_data", async () => { + const src = ` + const CRLF = "\\r\\n"; + const server = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + socket: { + open() {}, + data(s) { s.write("!garbage" + CRLF); }, + close() {}, + }, + }); + for (let i = 0; i < 50; i++) { + const c = new Bun.RedisClient("redis://127.0.0.1:" + server.port, { + autoReconnect: false, + connectionTimeout: 5000, + }); + c.onclose = () => {}; + try { await c.connect(); } catch {} + if (typeof c.connected !== "boolean") throw new Error("connected getter broken"); + try { c.close(); } catch {} + Bun.gc(true); + await new Promise(r => setImmediate(r)); + } + server.stop(true); + console.log("OK"); + process.exit(0); + `; + + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", src], + env: bunEnv, + stdout: "pipe", + stderr: "inherit", + }); + + const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]); + + expect(stdout.trim()).toBe("OK"); + expect(proc.signalCode).toBeNull(); + expect(exitCode).toBe(0); +}); + // Fuzzer found a flaky SIGILL when a RedisClient is constructed, a command // throws during argument validation (before any connection attempt), and the // client is then garbage collected. `updatePollRef` could be reached after