redis: gate the socket keep-alive adopt and defer destruction to finalize

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<u32>,
     pub ref_count: bun_ptr::RefCount<JSValkeyClient>,
 }
 
@@ -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<ScopedRef<Self>> {
+        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) };

test/js/valkey/valkey-gc.test.ts

@@ -181,6 +181,56 @@ test.concurrent("RedisClient survives subscribe() + close() against a server tha
   expect(exitCode).toBe(0);
 });
 
+// A malformed reply drives on_data -> parse fail -> fail_with_js_value ->
+// close(), which dispatches SocketHandler::on_close synchronously. Its
+// on_valkey_close() previously adopted the socket keep-alive ref
+// unconditionally; the enter_event_loop_scope drain inside on_valkey_close runs
+// the connect() promise rejection, so user code re-enters close() while the
+// socket handler frames are still on the stack. With take_socket_ref() the
+// second caller sees no outstanding ref and releases nothing.
+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