From 44c3869e7f02005c64abce6344320c17dc8956c3 Mon Sep 17 00:00:00 2001
From: Alistair Smith <hi@alistair.sh>
Date: Tue, 18 Aug 2026 04:57:17 -0700
Subject: [PATCH 1/9] valkey: reject subscribe on a failed client before
 storing handlers

---
 src/runtime/valkey_jsc/js_valkey.rs           | 33 ++++++++
 src/runtime/valkey_jsc/js_valkey_functions.rs | 33 +++++---
 src/runtime/valkey_jsc/valkey.rs              | 78 +++++++++---------
 .../reliability/connection-failures.test.ts   | 79 +++++++++++++++++++
 4 files changed, 170 insertions(+), 53 deletions(-)

diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs
index 7185779a0b59..eb6b5f252d78 100644
--- a/src/runtime/valkey_jsc/js_valkey.rs
+++ b/src/runtime/valkey_jsc/js_valkey.rs
@@ -178,6 +178,39 @@ impl JSValkeyClient {
         Ok(Some(new_length as usize))
     }
 
+    /// Undo the most recent `upsert_receive_handler` for `channel_name`,
+    /// leaving any handlers registered before it in place.
+    pub(crate) fn remove_last_receive_handler(
+        &self,
+        global_object: &JSGlobalObject,
+        channel_name: JSValue,
+        callback: JSValue,
+    ) -> JsResult<()> {
+        let map = self.subscription_callback_map();
+
+        let existing = map.get(global_object, channel_name)?;
+        if existing.is_undefined_or_null() {
+            return Ok(());
+        }
+        debug_assert!(existing.is_array());
+
+        let length = existing.get_length(global_object)?;
+        if length == 0 || existing.get_index(global_object, (length - 1) as u32)? != callback {
+            return Ok(());
+        }
+        let _ = map.remove(global_object, channel_name)?;
+        if length == 1 {
+            return Ok(());
+        }
+
+        let kept = JSArray::create_empty(global_object, 0)?;
+        for i in 0..(length - 1) as u32 {
+            kept.push(global_object, existing.get_index(global_object, i)?)?;
+        }
+        map.set(global_object, channel_name, kept)?;
+        Ok(())
+    }
+
     /// Add a handler for receiving messages on a specific channel
     pub(crate) fn upsert_receive_handler(
         &self,
diff --git a/src/runtime/valkey_jsc/js_valkey_functions.rs b/src/runtime/valkey_jsc/js_valkey_functions.rs
index a3755ebbe731..4389651e2a11 100644
--- a/src/runtime/valkey_jsc/js_valkey_functions.rs
+++ b/src/runtime/valkey_jsc/js_valkey_functions.rs
@@ -1841,6 +1841,7 @@ impl JSValkeyClient {
 
         let [channel_or_many, handler_callback] = frame.arguments_as_array::<2>();
         let mut redis_channels: Vec<JSArgument> = Vec::with_capacity(1);
+        let mut channel_names: Vec<JSValue> = Vec::with_capacity(1);
 
         if !handler_callback.is_callable() {
             return Err(global.throw_invalid_argument_type("subscribe", "listener", "function"));
@@ -1865,14 +1866,7 @@ impl JSValkeyClient {
                     ));
                 };
                 redis_channels.push(channel);
-
-                // What we do here is add our receive handler. Notice that this doesn't really do anything until the
-                // "SUBSCRIBE" command is sent to redis and we get a response.
-                //
-                // This is less-than-ideal, still, because this assumes a happy path. What happens if
-                // the SUBSCRIBE command fails? We have no way to roll back the addition of the
-                // handler.
-                this.upsert_receive_handler(global, channel_arg, handler_callback)?;
+                channel_names.push(channel_arg);
             }
         } else if channel_or_many.is_string() {
             // It is a single string channel
@@ -1880,8 +1874,7 @@ impl JSValkeyClient {
                 return Err(global.throw_invalid_argument_type("subscribe", "channel", "string"));
             };
             redis_channels.push(channel);
-
-            this.upsert_receive_handler(global, channel_or_many, handler_callback)?;
+            channel_names.push(channel_or_many);
         } else {
             return Err(global.throw_invalid_argument_type(
                 "subscribe",
@@ -1890,6 +1883,19 @@ impl JSValkeyClient {
             ));
         }
 
+        // A client that cannot take the SUBSCRIBE must not keep the handlers
+        // either: an orphaned handler pins the event loop and the client.
+        if let Some(message) = this.client.get().send_rejection() {
+            let error = valkey::ValkeyClient::send_rejection_error(global, message);
+            return Ok(JSPromise::rejected_promise(global, error).to_js());
+        }
+
+        // The handlers only start receiving once redis has answered the
+        // SUBSCRIBE sent below.
+        for &channel_name in &channel_names {
+            this.upsert_receive_handler(global, channel_name, handler_callback)?;
+        }
+
         let command = Command {
             command: b"SUBSCRIBE",
             args: CommandArgs::Args(&redis_channels),
@@ -1898,8 +1904,11 @@ impl JSValkeyClient {
         let promise = match this.send(global, frame.this(), &command) {
             Ok(p) => p,
             Err(err) => {
-                // If we catch an error, we need to clean up any handlers we may have added and fall out of subscription mode
-                this.clear_all_receive_handlers(global)?;
+                // Roll back only the handlers added above.
+                for &channel_name in &channel_names {
+                    this.remove_last_receive_handler(global, channel_name, handler_callback)?;
+                }
+                this.update_poll_ref();
                 return send_err_to_js(global, "Failed to send SUBSCRIBE command", &err);
             }
         };
diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs
index f6e08d03ae71..1ac257a8fb6d 100644
--- a/src/runtime/valkey_jsc/valkey.rs
+++ b/src/runtime/valkey_jsc/valkey.rs
@@ -1437,57 +1437,53 @@ impl ValkeyClient {
         let mut promise = command::Promise::create(global_this, checked_command.meta);
 
         let js_promise: *mut JSPromise = std::ptr::from_mut::<JSPromise>(promise.promise.get());
-        if self.flags.failed {
+        if let Some(message) = self.send_rejection() {
             let _ = promise.reject(
                 global_this,
-                Ok(global_this
-                    .err(
-                        bun_jsc::ErrorCode::REDIS_CONNECTION_CLOSED,
-                        format_args!("Connection has failed"),
-                    )
-                    .to_js()),
+                Ok(Self::send_rejection_error(global_this, message)),
             );
         } else {
-            // Handle disconnected state with offline queue
-            match self.status {
-                Status::Connected => {
-                    self.enqueue(&checked_command, promise)?;
-
-                    // Schedule auto-flushing to process this command if pipelining is enabled
-                    if self.flags.enable_auto_pipelining
-                        && checked_command
-                            .meta
-                            .contains(command::Meta::SUPPORTS_AUTO_PIPELINING)
-                        && self.status == Status::Connected
-                        && !self.queue.is_empty()
-                    {
-                        self.register_auto_flusher(self.vm);
-                    }
-                }
-                Status::NeverConnected | Status::Connecting | Status::Disconnected => {
-                    // Only queue if offline queue is enabled
-                    if self.flags.enable_offline_queue {
-                        self.enqueue(&checked_command, promise)?;
-                    } else {
-                        let _ = promise.reject(
-                            global_this,
-                            Ok(global_this
-                                .err(
-                                    bun_jsc::ErrorCode::REDIS_CONNECTION_CLOSED,
-                                    format_args!(
-                                        "Connection is closed and offline queue is disabled"
-                                    ),
-                                )
-                                .to_js()),
-                        );
-                    }
-                }
+            self.enqueue(&checked_command, promise)?;
+
+            // Schedule auto-flushing to process this command if pipelining is enabled
+            if self.flags.enable_auto_pipelining
+                && checked_command
+                    .meta
+                    .contains(command::Meta::SUPPORTS_AUTO_PIPELINING)
+                && self.status == Status::Connected
+                && !self.queue.is_empty()
+            {
+                self.register_auto_flusher(self.vm);
             }
         }
 
         Ok(js_promise)
     }
 
+    /// Why `send()` would reject a command outright instead of sending or
+    /// queueing it in the current state, or `None` when it would be accepted.
+    pub(crate) fn send_rejection(&self) -> Option<&'static str> {
+        if self.flags.failed {
+            return Some("Connection has failed");
+        }
+        if self.status != Status::Connected && !self.flags.enable_offline_queue {
+            return Some("Connection is closed and offline queue is disabled");
+        }
+        None
+    }
+
+    pub(crate) fn send_rejection_error(
+        global_this: &JSGlobalObject,
+        message: &'static str,
+    ) -> JSValue {
+        global_this
+            .err(
+                bun_jsc::ErrorCode::REDIS_CONNECTION_CLOSED,
+                format_args!("{message}"),
+            )
+            .to_js()
+    }
+
     /// Close the Valkey connection
     pub(crate) fn disconnect(&mut self) -> JsResult<()> {
         self.flags.is_manually_closed = true;
diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts
index ee5e4028fd85..cef575d75751 100644
--- a/test/js/valkey/reliability/connection-failures.test.ts
+++ b/test/js/valkey/reliability/connection-failures.test.ts
@@ -1659,6 +1659,85 @@ describe("Valkey: Recovering After fail()", () => {
       }
     },
   );
+
+  test("subscribe() on a failed client rejects and registers no handler", async () => {
+    const push = (...items: (string | number)[]) =>
+      `>${items.length}\r\n` +
+      items.map(item => (typeof item === "number" ? `:${item}\r\n` : `$${item.length}\r\n${item}\r\n`)).join("");
+    const fake = helloServer();
+    const port = await fake.listen();
+    const client = new RedisClient(`redis://127.0.0.1:${port}`, {
+      connectionTimeout: 0,
+      idleTimeout: 50,
+      autoReconnect: false,
+    });
+    try {
+      const closed = Promise.withResolvers<Error>();
+      client.onclose = err => closed.resolve(err);
+      await client.connect();
+      await closed.promise;
+      const delivered: string[] = [];
+      const listener = (message: string) => delivered.push(message);
+      await expect(client.subscribe("ch", listener)).rejects.toMatchObject({
+        code: "ERR_REDIS_CONNECTION_CLOSED",
+        message: "Connection has failed",
+      });
+      // The rejected subscribe left the client out of subscriber mode.
+      expect(() => client.unsubscribe("ch")).toThrow("can only be called while in subscriber mode");
+
+      // A subscribe on the next connection is the only registration: the
+      // message arrives once, not once per attempt.
+      client.onclose = () => {};
+      await client.connect();
+      const connection2 = fake.sockets[1];
+      connection2.on("data", chunk => {
+        if (chunk.toString("latin1").includes("SUBSCRIBE")) {
+          connection2.write(push("subscribe", "ch", 1) + push("message", "ch", "m0"));
+        }
+      });
+      await client.subscribe("ch", listener);
+      while (delivered.length === 0) await delay(1);
+      await client.ping();
+      expect({ delivered, connections: fake.connections }).toEqual({ delivered: ["m0"], connections: 2 });
+    } finally {
+      client.close();
+      fake.server.close();
+    }
+  });
+
+  test("the process exits after subscribe() is rejected by a failed client", async () => {
+    const fake = helloServer();
+    const port = await fake.listen();
+    try {
+      await using proc = Bun.spawn({
+        cmd: [
+          bunExe(),
+          "-e",
+          `
+          const client = new Bun.RedisClient("redis://127.0.0.1:${port}", { connectionTimeout: 0, idleTimeout: 50, autoReconnect: false });
+          const closed = Promise.withResolvers();
+          client.onclose = err => closed.resolve(err);
+          await client.connect();
+          console.log("onclose", (await closed.promise).code);
+          await client.subscribe("ch", () => {}).catch(err => console.log("subscribe rejected", err.code));
+          `,
+        ],
+        env: bunEnv,
+        stdout: "pipe",
+        stderr: "pipe",
+      });
+      const exitCode = await Promise.race([proc.exited, delay(3000).then(() => "still running")]);
+      if (exitCode === "still running") proc.kill();
+      const [stdout, stderr] = await Promise.all([proc.stdout.text(), proc.stderr.text()]);
+      expect({ stdout, stderr, exitCode }).toEqual({
+        stdout: "onclose ERR_REDIS_CONNECTION_CLOSED\nsubscribe rejected ERR_REDIS_CONNECTION_CLOSED\n",
+        stderr: "",
+        exitCode: 0,
+      });
+    } finally {
+      fake.server.close();
+    }
+  });
 });
 
 describe("Valkey: Offline Queue", () => {

From d0db6e751dfcd50400705b851b2a7b2553dac7d5 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Tue, 18 Aug 2026 18:10:12 +0000
Subject: [PATCH 2/9] test(valkey): share the RESP3 push helper between the two
 tests that use it

---
 .../valkey/reliability/connection-failures.test.ts   | 12 +++++-------
 1 file changed, 5 insertions(+), 7 deletions(-)

diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts
index cef575d75751..a9924633e3f1 100644
--- a/test/js/valkey/reliability/connection-failures.test.ts
+++ b/test/js/valkey/reliability/connection-failures.test.ts
@@ -502,6 +502,11 @@ describe("Valkey: Recovering After fail()", () => {
     return socket.destroyed ? Promise.resolve() : new Promise(resolve => socket.once("close", () => resolve()));
   }
 
+  // RESP3 push frames as the server writes them for SUBSCRIBE and for messages.
+  const push = (...items: (string | number)[]) =>
+    `>${items.length}\r\n` +
+    items.map(item => (typeof item === "number" ? `:${item}\r\n` : `$${item.length}\r\n${item}\r\n`)).join("");
+
   // Calls connect() from the first onclose and reports how that attempt ended.
   function connectFromOnclose(client: RedisClient): Promise<string> {
     const { promise, resolve } = Promise.withResolvers<string>();
@@ -917,10 +922,6 @@ describe("Valkey: Recovering After fail()", () => {
   });
 
   test("a message listener that closes and reconnects is not fed the pushes buffered behind its message", async () => {
-    // RESP3 push frames as the server writes them for SUBSCRIBE and for messages.
-    const push = (...items: (string | number)[]) =>
-      `>${items.length}\r\n` +
-      items.map(item => (typeof item === "number" ? `:${item}\r\n` : `$${item.length}\r\n${item}\r\n`)).join("");
     const fake = helloServer();
     const port = await fake.listen();
     const client = new RedisClient(`redis://127.0.0.1:${port}`);
@@ -1661,9 +1662,6 @@ describe("Valkey: Recovering After fail()", () => {
   );
 
   test("subscribe() on a failed client rejects and registers no handler", async () => {
-    const push = (...items: (string | number)[]) =>
-      `>${items.length}\r\n` +
-      items.map(item => (typeof item === "number" ? `:${item}\r\n` : `$${item.length}\r\n${item}\r\n`)).join("");
     const fake = helloServer();
     const port = await fake.listen();
     const client = new RedisClient(`redis://127.0.0.1:${port}`, {

From 85448904d58d1ff8ede22704a1b1c7f4428aa1c9 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Tue, 18 Aug 2026 18:25:44 +0000
Subject: [PATCH 3/9] test(valkey): await the first delivery and size RESP bulk
 strings in bytes

---
 .../valkey/reliability/connection-failures.test.ts | 14 +++++++++++---
 1 file changed, 11 insertions(+), 3 deletions(-)

diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts
index a9924633e3f1..10b1a63bfc2b 100644
--- a/test/js/valkey/reliability/connection-failures.test.ts
+++ b/test/js/valkey/reliability/connection-failures.test.ts
@@ -505,7 +505,9 @@ describe("Valkey: Recovering After fail()", () => {
   // RESP3 push frames as the server writes them for SUBSCRIBE and for messages.
   const push = (...items: (string | number)[]) =>
     `>${items.length}\r\n` +
-    items.map(item => (typeof item === "number" ? `:${item}\r\n` : `$${item.length}\r\n${item}\r\n`)).join("");
+    items
+      .map(item => (typeof item === "number" ? `:${item}\r\n` : `$${Buffer.byteLength(item)}\r\n${item}\r\n`))
+      .join("");
 
   // Calls connect() from the first onclose and reports how that attempt ended.
   function connectFromOnclose(client: RedisClient): Promise<string> {
@@ -1675,7 +1677,11 @@ describe("Valkey: Recovering After fail()", () => {
       await client.connect();
       await closed.promise;
       const delivered: string[] = [];
-      const listener = (message: string) => delivered.push(message);
+      const firstDelivered = Promise.withResolvers<void>();
+      const listener = (message: string) => {
+        delivered.push(message);
+        firstDelivered.resolve();
+      };
       await expect(client.subscribe("ch", listener)).rejects.toMatchObject({
         code: "ERR_REDIS_CONNECTION_CLOSED",
         message: "Connection has failed",
@@ -1694,7 +1700,9 @@ describe("Valkey: Recovering After fail()", () => {
         }
       });
       await client.subscribe("ch", listener);
-      while (delivered.length === 0) await delay(1);
+      await firstDelivered.promise;
+      // PONG comes back after anything else the stub wrote, so a second
+      // delivery of m0 would be in `delivered` by now.
       await client.ping();
       expect({ delivered, connections: fake.connections }).toEqual({ delivered: ["m0"], connections: 2 });
     } finally {

From 552e216498d4cd9f8f6f47f2ef772a6d09e11ed9 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Tue, 18 Aug 2026 19:02:39 +0000
Subject: [PATCH 4/9] valkey: root the channel names subscribe() collects and
 roll back after a dial that fails inside send()

subscribe() collected the channel names in a Vec<JSValue> between the
argument walk and the handler map. A Vec is not a GC root, so a name
produced by an index getter was collected when the getter for the next
index ran a GC. The names are now appended to a MarkedArgumentBuffer for
the length of the call.

The first dial is made inside send(). When it fails outright (for example
the TLS context cannot be built) the client is failed before the SUBSCRIBE
is looked at, and send() returns an already rejected promise. The
handlers stored for that call were left behind and kept the process
alive. The rollback now also runs when the client rejects the command.
---
 src/runtime/valkey_jsc/js_valkey_functions.rs | 58 +++++++++++----
 .../reliability/connection-failures.test.ts   | 43 +++++++++--
 test/js/valkey/valkey-gc.test.ts              | 71 +++++++++++++++++++
 3 files changed, 155 insertions(+), 17 deletions(-)

diff --git a/src/runtime/valkey_jsc/js_valkey_functions.rs b/src/runtime/valkey_jsc/js_valkey_functions.rs
index 4389651e2a11..e5aef784419b 100644
--- a/src/runtime/valkey_jsc/js_valkey_functions.rs
+++ b/src/runtime/valkey_jsc/js_valkey_functions.rs
@@ -1840,13 +1840,39 @@ impl JSValkeyClient {
         let _guard = this.ref_scope();
 
         let [channel_or_many, handler_callback] = frame.arguments_as_array::<2>();
-        let mut redis_channels: Vec<JSArgument> = Vec::with_capacity(1);
-        let mut channel_names: Vec<JSValue> = Vec::with_capacity(1);
 
         if !handler_callback.is_callable() {
             return Err(global.throw_invalid_argument_type("subscribe", "listener", "function"));
         }
 
+        // The channel names are held from the argument walk until they are in
+        // the handler map. An index getter on the channel array can hand out a
+        // fresh string, and the getter for the next index can run a GC. A `Vec`
+        // is not a GC root, so every name is rooted in the buffer as well.
+        jsc::MarkedArgumentBuffer::new(|rooted| {
+            Self::subscribe_rooted(
+                this,
+                global,
+                frame.this(),
+                channel_or_many,
+                handler_callback,
+                rooted,
+            )
+        })
+    }
+
+    fn subscribe_rooted(
+        this: &Self,
+        global: &JSGlobalObject,
+        this_js: JSValue,
+        channel_or_many: JSValue,
+        handler_callback: JSValue,
+        rooted: &mut jsc::MarkedArgumentBuffer,
+    ) -> JsResult<JSValue> {
+        let mut redis_channels: Vec<JSArgument> = Vec::with_capacity(1);
+        // Mirrors `rooted`, which cannot be read back.
+        let mut channel_names: Vec<JSValue> = Vec::with_capacity(1);
+
         // The first argument given is the channel or may be an array of channels.
         if channel_or_many.is_array() {
             if channel_or_many.get_length(global)? == 0 {
@@ -1858,6 +1884,7 @@ impl JSValkeyClient {
 
             let mut array_iter = channel_or_many.array_iterator(global)?;
             while let Some(channel_arg) = array_iter.next()? {
+                rooted.append(channel_arg);
                 let Some(channel) = from_js(global, channel_arg)? else {
                     return Err(global.throw_invalid_argument_type(
                         "subscribe",
@@ -1870,6 +1897,7 @@ impl JSValkeyClient {
             }
         } else if channel_or_many.is_string() {
             // It is a single string channel
+            rooted.append(channel_or_many);
             let Some(channel) = from_js(global, channel_or_many)? else {
                 return Err(global.throw_invalid_argument_type("subscribe", "channel", "string"));
             };
@@ -1901,19 +1929,23 @@ impl JSValkeyClient {
             args: CommandArgs::Args(&redis_channels),
             meta: CommandMeta::default() | CommandMeta::SUBSCRIPTION_REQUEST,
         };
-        let promise = match this.send(global, frame.this(), &command) {
-            Ok(p) => p,
-            Err(err) => {
-                // Roll back only the handlers added above.
-                for &channel_name in &channel_names {
-                    this.remove_last_receive_handler(global, channel_name, handler_callback)?;
-                }
-                this.update_poll_ref();
-                return send_err_to_js(global, "Failed to send SUBSCRIBE command", &err);
+        let sent = this.send(global, this_js, &command);
+
+        // A first dial is made inside `send()`. When it fails outright (no TLS
+        // context, for example) the client is failed by the time the SUBSCRIBE
+        // is looked at, and the promise comes back already rejected. Roll back
+        // only the handlers added above.
+        if sent.is_err() || this.client.get().send_rejection().is_some() {
+            for &channel_name in &channel_names {
+                this.remove_last_receive_handler(global, channel_name, handler_callback)?;
             }
-        };
+            this.update_poll_ref();
+        }
 
-        Ok(promise_to_js(promise))
+        match sent {
+            Ok(promise) => Ok(promise_to_js(promise)),
+            Err(err) => send_err_to_js(global, "Failed to send SUBSCRIBE command", &err),
+        }
     }
 
     /// Send redis the UNSUBSCRIBE RESP command and clean up anything necessary after the unsubscribe commoand.
diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts
index 10b1a63bfc2b..74d749280191 100644
--- a/test/js/valkey/reliability/connection-failures.test.ts
+++ b/test/js/valkey/reliability/connection-failures.test.ts
@@ -509,6 +509,15 @@ describe("Valkey: Recovering After fail()", () => {
       .map(item => (typeof item === "number" ? `:${item}\r\n` : `$${Buffer.byteLength(item)}\r\n${item}\r\n`))
       .join("");
 
+  // A process that a client keeps alive never exits on its own; report that
+  // as the exit code after 3 s instead of waiting for the test to time out.
+  async function exitOutcome(proc: Bun.Subprocess<"ignore", "pipe", "pipe">) {
+    const exitCode = await Promise.race([proc.exited, delay(3000).then(() => "still running" as const)]);
+    if (exitCode === "still running") proc.kill();
+    const [stdout, stderr] = await Promise.all([proc.stdout.text(), proc.stderr.text()]);
+    return { stdout, stderr, exitCode };
+  }
+
   // Calls connect() from the first onclose and reports how that attempt ended.
   function connectFromOnclose(client: RedisClient): Promise<string> {
     const { promise, resolve } = Promise.withResolvers<string>();
@@ -1732,10 +1741,7 @@ describe("Valkey: Recovering After fail()", () => {
         stdout: "pipe",
         stderr: "pipe",
       });
-      const exitCode = await Promise.race([proc.exited, delay(3000).then(() => "still running")]);
-      if (exitCode === "still running") proc.kill();
-      const [stdout, stderr] = await Promise.all([proc.stdout.text(), proc.stderr.text()]);
-      expect({ stdout, stderr, exitCode }).toEqual({
+      expect(await exitOutcome(proc)).toEqual({
         stdout: "onclose ERR_REDIS_CONNECTION_CLOSED\nsubscribe rejected ERR_REDIS_CONNECTION_CLOSED\n",
         stderr: "",
         exitCode: 0,
@@ -1744,6 +1750,35 @@ describe("Valkey: Recovering After fail()", () => {
       fake.server.close();
     }
   });
+
+  test("the process exits after subscribe() makes a first dial that fails outright", async () => {
+    await using proc = Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `
+        // Neither key nor cert parses, so the dial that subscribe() makes
+        // fails inside the call, before a socket exists.
+        const client = new Bun.RedisClient("rediss://127.0.0.1:1", {
+          tls: { key: "not a key", cert: "not a cert" },
+          autoReconnect: false,
+        });
+        const closed = Promise.withResolvers();
+        client.onclose = err => closed.resolve(err);
+        await client.subscribe("ch", () => {}).catch(err => console.log("subscribe rejected", err.code));
+        console.log("onclose", (await closed.promise).code);
+        `,
+      ],
+      env: bunEnv,
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+    expect(await exitOutcome(proc)).toEqual({
+      stdout: "subscribe rejected ERR_REDIS_CONNECTION_CLOSED\nonclose ERR_REDIS_CONNECTION_CLOSED\n",
+      stderr: "",
+      exitCode: 0,
+    });
+  });
 });
 
 describe("Valkey: Offline Queue", () => {
diff --git a/test/js/valkey/valkey-gc.test.ts b/test/js/valkey/valkey-gc.test.ts
index ec7189eec006..9e0714ddf335 100644
--- a/test/js/valkey/valkey-gc.test.ts
+++ b/test/js/valkey/valkey-gc.test.ts
@@ -905,3 +905,74 @@ test.concurrent("a client closed and collected from within its own reply does no
   // Keep the per-lane count in the CI log so a slide toward zero is visible.
   console.log(`close-from-reply fixture: ${reached![1]} of 10 rounds reached the window`);
 });
+
+// subscribe() walks the channel array before it stores anything in the
+// handler map. A channel produced by an index getter has no other owner while
+// the getters for the later indices run, so the names have to be rooted for
+// that walk; held in a plain Vec, the first ones are collected by the time
+// they are used as map keys.
+test.concurrent(
+  "RedisClient.subscribe() keeps channel names produced by index getters alive across the walk",
+  async () => {
+    const src = `
+    const CRLF = "\\r\\n";
+    const N = 32;
+    const name = i => "ch-" + i;
+    const push = (...items) =>
+      ">" + items.length + CRLF +
+      items.map(item => (typeof item === "number" ? ":" + item + CRLF : "$" + item.length + CRLF + item + CRLF)).join("");
+    const server = Bun.listen({
+      hostname: "127.0.0.1",
+      port: 0,
+      socket: {
+        data(s, d) {
+          const text = d.toString("latin1");
+          if (text.includes("HELLO")) s.write("+OK" + CRLF);
+          if (text.includes("SUBSCRIBE")) {
+            let out = "";
+            for (let i = 0; i < N; i++) out += push("subscribe", name(i), i + 1);
+            for (let i = 0; i < N; i++) out += push("message", name(i), "m");
+            s.write(out);
+          }
+          if (text.includes("PING")) s.write("+PONG" + CRLF);
+        },
+      },
+    });
+    const channels = [];
+    for (let i = 0; i < N; i++) {
+      Object.defineProperty(channels, i, {
+        enumerable: true,
+        get() {
+          // Collects the string handed out for the previous index, then
+          // reuses its memory.
+          Bun.gc(true);
+          const junk = [];
+          for (let j = 0; j < 256; j++) junk.push("junk-" + i + "-" + j);
+          return name(i);
+        },
+      });
+    }
+    const client = new Bun.RedisClient("redis://127.0.0.1:" + server.port, { autoReconnect: false });
+    const delivered = new Set();
+    await client.subscribe(channels, (_message, channel) => delivered.add(channel));
+    // PONG is queued behind the message pushes.
+    await client.ping();
+    if (delivered.size !== N) throw new Error("listener reached " + delivered.size + " of " + N + " channels");
+    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);
+  },
+);

From d114601ff24867c83d85ca8c0e710d7625d899a5 Mon Sep 17 00:00:00 2001
From: Alistair Smith <hi@alistair.sh>
Date: Tue, 18 Aug 2026 12:19:27 -0700
Subject: [PATCH 5/9] valkey: trim subscribe fix to the pre-send state check

---
 src/runtime/valkey_jsc/js_valkey.rs           | 33 -------
 src/runtime/valkey_jsc/js_valkey_functions.rs | 86 ++++++-------------
 .../reliability/connection-failures.test.ts   | 25 +++---
 test/js/valkey/valkey-gc.test.ts              | 71 ---------------
 4 files changed, 39 insertions(+), 176 deletions(-)

diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs
index eb6b5f252d78..7185779a0b59 100644
--- a/src/runtime/valkey_jsc/js_valkey.rs
+++ b/src/runtime/valkey_jsc/js_valkey.rs
@@ -178,39 +178,6 @@ impl JSValkeyClient {
         Ok(Some(new_length as usize))
     }
 
-    /// Undo the most recent `upsert_receive_handler` for `channel_name`,
-    /// leaving any handlers registered before it in place.
-    pub(crate) fn remove_last_receive_handler(
-        &self,
-        global_object: &JSGlobalObject,
-        channel_name: JSValue,
-        callback: JSValue,
-    ) -> JsResult<()> {
-        let map = self.subscription_callback_map();
-
-        let existing = map.get(global_object, channel_name)?;
-        if existing.is_undefined_or_null() {
-            return Ok(());
-        }
-        debug_assert!(existing.is_array());
-
-        let length = existing.get_length(global_object)?;
-        if length == 0 || existing.get_index(global_object, (length - 1) as u32)? != callback {
-            return Ok(());
-        }
-        let _ = map.remove(global_object, channel_name)?;
-        if length == 1 {
-            return Ok(());
-        }
-
-        let kept = JSArray::create_empty(global_object, 0)?;
-        for i in 0..(length - 1) as u32 {
-            kept.push(global_object, existing.get_index(global_object, i)?)?;
-        }
-        map.set(global_object, channel_name, kept)?;
-        Ok(())
-    }
-
     /// Add a handler for receiving messages on a specific channel
     pub(crate) fn upsert_receive_handler(
         &self,
diff --git a/src/runtime/valkey_jsc/js_valkey_functions.rs b/src/runtime/valkey_jsc/js_valkey_functions.rs
index e5aef784419b..522aadfd7c73 100644
--- a/src/runtime/valkey_jsc/js_valkey_functions.rs
+++ b/src/runtime/valkey_jsc/js_valkey_functions.rs
@@ -1840,38 +1840,20 @@ impl JSValkeyClient {
         let _guard = this.ref_scope();
 
         let [channel_or_many, handler_callback] = frame.arguments_as_array::<2>();
+        let mut redis_channels: Vec<JSArgument> = Vec::with_capacity(1);
 
         if !handler_callback.is_callable() {
             return Err(global.throw_invalid_argument_type("subscribe", "listener", "function"));
         }
 
-        // The channel names are held from the argument walk until they are in
-        // the handler map. An index getter on the channel array can hand out a
-        // fresh string, and the getter for the next index can run a GC. A `Vec`
-        // is not a GC root, so every name is rooted in the buffer as well.
-        jsc::MarkedArgumentBuffer::new(|rooted| {
-            Self::subscribe_rooted(
-                this,
-                global,
-                frame.this(),
-                channel_or_many,
-                handler_callback,
-                rooted,
-            )
-        })
-    }
-
-    fn subscribe_rooted(
-        this: &Self,
-        global: &JSGlobalObject,
-        this_js: JSValue,
-        channel_or_many: JSValue,
-        handler_callback: JSValue,
-        rooted: &mut jsc::MarkedArgumentBuffer,
-    ) -> JsResult<JSValue> {
-        let mut redis_channels: Vec<JSArgument> = Vec::with_capacity(1);
-        // Mirrors `rooted`, which cannot be read back.
-        let mut channel_names: Vec<JSValue> = Vec::with_capacity(1);
+        // The walk below stores each listener as it goes. A client that would
+        // reject the SUBSCRIBE outright must not keep the listeners either: a
+        // listener with no subscription behind it pins the event loop and the
+        // client, and cannot be removed with unsubscribe().
+        if let Some(message) = this.client.get().send_rejection() {
+            let error = valkey::ValkeyClient::send_rejection_error(global, message);
+            return Ok(JSPromise::rejected_promise(global, error).to_js());
+        }
 
         // The first argument given is the channel or may be an array of channels.
         if channel_or_many.is_array() {
@@ -1884,7 +1866,6 @@ impl JSValkeyClient {
 
             let mut array_iter = channel_or_many.array_iterator(global)?;
             while let Some(channel_arg) = array_iter.next()? {
-                rooted.append(channel_arg);
                 let Some(channel) = from_js(global, channel_arg)? else {
                     return Err(global.throw_invalid_argument_type(
                         "subscribe",
@@ -1893,16 +1874,23 @@ impl JSValkeyClient {
                     ));
                 };
                 redis_channels.push(channel);
-                channel_names.push(channel_arg);
+
+                // What we do here is add our receive handler. Notice that this doesn't really do anything until the
+                // "SUBSCRIBE" command is sent to redis and we get a response.
+                //
+                // This is less-than-ideal, still, because this assumes a happy path. What happens if
+                // the SUBSCRIBE command fails? We have no way to roll back the addition of the
+                // handler.
+                this.upsert_receive_handler(global, channel_arg, handler_callback)?;
             }
         } else if channel_or_many.is_string() {
             // It is a single string channel
-            rooted.append(channel_or_many);
             let Some(channel) = from_js(global, channel_or_many)? else {
                 return Err(global.throw_invalid_argument_type("subscribe", "channel", "string"));
             };
             redis_channels.push(channel);
-            channel_names.push(channel_or_many);
+
+            this.upsert_receive_handler(global, channel_or_many, handler_callback)?;
         } else {
             return Err(global.throw_invalid_argument_type(
                 "subscribe",
@@ -1911,41 +1899,21 @@ impl JSValkeyClient {
             ));
         }
 
-        // A client that cannot take the SUBSCRIBE must not keep the handlers
-        // either: an orphaned handler pins the event loop and the client.
-        if let Some(message) = this.client.get().send_rejection() {
-            let error = valkey::ValkeyClient::send_rejection_error(global, message);
-            return Ok(JSPromise::rejected_promise(global, error).to_js());
-        }
-
-        // The handlers only start receiving once redis has answered the
-        // SUBSCRIBE sent below.
-        for &channel_name in &channel_names {
-            this.upsert_receive_handler(global, channel_name, handler_callback)?;
-        }
-
         let command = Command {
             command: b"SUBSCRIBE",
             args: CommandArgs::Args(&redis_channels),
             meta: CommandMeta::default() | CommandMeta::SUBSCRIPTION_REQUEST,
         };
-        let sent = this.send(global, this_js, &command);
-
-        // A first dial is made inside `send()`. When it fails outright (no TLS
-        // context, for example) the client is failed by the time the SUBSCRIBE
-        // is looked at, and the promise comes back already rejected. Roll back
-        // only the handlers added above.
-        if sent.is_err() || this.client.get().send_rejection().is_some() {
-            for &channel_name in &channel_names {
-                this.remove_last_receive_handler(global, channel_name, handler_callback)?;
+        let promise = match this.send(global, frame.this(), &command) {
+            Ok(p) => p,
+            Err(err) => {
+                // If we catch an error, we need to clean up any handlers we may have added and fall out of subscription mode
+                this.clear_all_receive_handlers(global)?;
+                return send_err_to_js(global, "Failed to send SUBSCRIBE command", &err);
             }
-            this.update_poll_ref();
-        }
+        };
 
-        match sent {
-            Ok(promise) => Ok(promise_to_js(promise)),
-            Err(err) => send_err_to_js(global, "Failed to send SUBSCRIBE command", &err),
-        }
+        Ok(promise_to_js(promise))
     }
 
     /// Send redis the UNSUBSCRIBE RESP command and clean up anything necessary after the unsubscribe commoand.
diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts
index 74d749280191..81e674e2b201 100644
--- a/test/js/valkey/reliability/connection-failures.test.ts
+++ b/test/js/valkey/reliability/connection-failures.test.ts
@@ -1733,7 +1733,8 @@ describe("Valkey: Recovering After fail()", () => {
           const closed = Promise.withResolvers();
           client.onclose = err => closed.resolve(err);
           await client.connect();
-          console.log("onclose", (await closed.promise).code);
+          await closed.promise;
+          console.log("onclose");
           await client.subscribe("ch", () => {}).catch(err => console.log("subscribe rejected", err.code));
           `,
         ],
@@ -1742,7 +1743,7 @@ describe("Valkey: Recovering After fail()", () => {
         stderr: "pipe",
       });
       expect(await exitOutcome(proc)).toEqual({
-        stdout: "onclose ERR_REDIS_CONNECTION_CLOSED\nsubscribe rejected ERR_REDIS_CONNECTION_CLOSED\n",
+        stdout: "onclose\nsubscribe rejected ERR_REDIS_CONNECTION_CLOSED\n",
         stderr: "",
         exitCode: 0,
       });
@@ -1751,22 +1752,20 @@ describe("Valkey: Recovering After fail()", () => {
     }
   });
 
-  test("the process exits after subscribe() makes a first dial that fails outright", async () => {
+  // subscribe() before connect() with the default offline queue stores the
+  // listener and queues the SUBSCRIBE; when the dial then fails for good the
+  // queued SUBSCRIBE is rejected but the listener stays, so the process is
+  // held alive. #33290 registers the listener on the server's subscribe
+  // confirmation instead, which closes this route; a -ERR reply to SUBSCRIBE
+  // (an ACL NOPERM, say) leaves the same orphan and is closed the same way.
+  test.todo("the process exits after a queued subscribe() is rejected by a dial that fails for good", async () => {
     await using proc = Bun.spawn({
       cmd: [
         bunExe(),
         "-e",
         `
-        // Neither key nor cert parses, so the dial that subscribe() makes
-        // fails inside the call, before a socket exists.
-        const client = new Bun.RedisClient("rediss://127.0.0.1:1", {
-          tls: { key: "not a key", cert: "not a cert" },
-          autoReconnect: false,
-        });
-        const closed = Promise.withResolvers();
-        client.onclose = err => closed.resolve(err);
+        const client = new Bun.RedisClient("redis://127.0.0.1:1", { maxRetries: 0, autoReconnect: false });
         await client.subscribe("ch", () => {}).catch(err => console.log("subscribe rejected", err.code));
-        console.log("onclose", (await closed.promise).code);
         `,
       ],
       env: bunEnv,
@@ -1774,7 +1773,7 @@ describe("Valkey: Recovering After fail()", () => {
       stderr: "pipe",
     });
     expect(await exitOutcome(proc)).toEqual({
-      stdout: "subscribe rejected ERR_REDIS_CONNECTION_CLOSED\nonclose ERR_REDIS_CONNECTION_CLOSED\n",
+      stdout: "subscribe rejected ERR_REDIS_CONNECTION_CLOSED\n",
       stderr: "",
       exitCode: 0,
     });
diff --git a/test/js/valkey/valkey-gc.test.ts b/test/js/valkey/valkey-gc.test.ts
index 9e0714ddf335..ec7189eec006 100644
--- a/test/js/valkey/valkey-gc.test.ts
+++ b/test/js/valkey/valkey-gc.test.ts
@@ -905,74 +905,3 @@ test.concurrent("a client closed and collected from within its own reply does no
   // Keep the per-lane count in the CI log so a slide toward zero is visible.
   console.log(`close-from-reply fixture: ${reached![1]} of 10 rounds reached the window`);
 });
-
-// subscribe() walks the channel array before it stores anything in the
-// handler map. A channel produced by an index getter has no other owner while
-// the getters for the later indices run, so the names have to be rooted for
-// that walk; held in a plain Vec, the first ones are collected by the time
-// they are used as map keys.
-test.concurrent(
-  "RedisClient.subscribe() keeps channel names produced by index getters alive across the walk",
-  async () => {
-    const src = `
-    const CRLF = "\\r\\n";
-    const N = 32;
-    const name = i => "ch-" + i;
-    const push = (...items) =>
-      ">" + items.length + CRLF +
-      items.map(item => (typeof item === "number" ? ":" + item + CRLF : "$" + item.length + CRLF + item + CRLF)).join("");
-    const server = Bun.listen({
-      hostname: "127.0.0.1",
-      port: 0,
-      socket: {
-        data(s, d) {
-          const text = d.toString("latin1");
-          if (text.includes("HELLO")) s.write("+OK" + CRLF);
-          if (text.includes("SUBSCRIBE")) {
-            let out = "";
-            for (let i = 0; i < N; i++) out += push("subscribe", name(i), i + 1);
-            for (let i = 0; i < N; i++) out += push("message", name(i), "m");
-            s.write(out);
-          }
-          if (text.includes("PING")) s.write("+PONG" + CRLF);
-        },
-      },
-    });
-    const channels = [];
-    for (let i = 0; i < N; i++) {
-      Object.defineProperty(channels, i, {
-        enumerable: true,
-        get() {
-          // Collects the string handed out for the previous index, then
-          // reuses its memory.
-          Bun.gc(true);
-          const junk = [];
-          for (let j = 0; j < 256; j++) junk.push("junk-" + i + "-" + j);
-          return name(i);
-        },
-      });
-    }
-    const client = new Bun.RedisClient("redis://127.0.0.1:" + server.port, { autoReconnect: false });
-    const delivered = new Set();
-    await client.subscribe(channels, (_message, channel) => delivered.add(channel));
-    // PONG is queued behind the message pushes.
-    await client.ping();
-    if (delivered.size !== N) throw new Error("listener reached " + delivered.size + " of " + N + " channels");
-    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);
-  },
-);

From 763f1a67590ea993a60f84cb4216a43178124618 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Tue, 18 Aug 2026 19:34:23 +0000
Subject: [PATCH 6/9] ci: retrigger


From 31587f94d6ed464a14b0dd77dd4b75dc784855fa Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Tue, 18 Aug 2026 19:45:15 +0000
Subject: [PATCH 7/9] test(valkey): start reading the child's pipes before
 waiting for it to exit

---
 test/js/valkey/reliability/connection-failures.test.ts | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts
index 81e674e2b201..b681275e5404 100644
--- a/test/js/valkey/reliability/connection-failures.test.ts
+++ b/test/js/valkey/reliability/connection-failures.test.ts
@@ -512,9 +512,10 @@ describe("Valkey: Recovering After fail()", () => {
   // A process that a client keeps alive never exits on its own; report that
   // as the exit code after 3 s instead of waiting for the test to time out.
   async function exitOutcome(proc: Bun.Subprocess<"ignore", "pipe", "pipe">) {
+    const output = Promise.all([proc.stdout.text(), proc.stderr.text()]);
     const exitCode = await Promise.race([proc.exited, delay(3000).then(() => "still running" as const)]);
     if (exitCode === "still running") proc.kill();
-    const [stdout, stderr] = await Promise.all([proc.stdout.text(), proc.stderr.text()]);
+    const [stdout, stderr] = await output;
     return { stdout, stderr, exitCode };
   }
 

From 94bf0f3f466186b945c12f6eb881fa94514c223f Mon Sep 17 00:00:00 2001
From: Alistair Smith <hi@alistair.sh>
Date: Tue, 18 Aug 2026 12:50:07 -0700
Subject: [PATCH 8/9] valkey: dial before the subscribe state check

---
 src/runtime/valkey_jsc/js_valkey.rs           | 43 ++++++++++-------
 src/runtime/valkey_jsc/js_valkey_functions.rs |  5 +-
 .../reliability/connection-failures.test.ts   | 48 +++++++++++++++++--
 3 files changed, 75 insertions(+), 21 deletions(-)

diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs
index 7185779a0b59..a5a593655069 100644
--- a/src/runtime/valkey_jsc/js_valkey.rs
+++ b/src/runtime/valkey_jsc/js_valkey.rs
@@ -1586,29 +1586,38 @@ impl JSValkeyClient {
         // the host-fn shim passes a bare `&self` with no ref of its own.
         let _guard = self.ref_scope();
 
-        if self.client.get().status == valkey::Status::NeverConnected {
-            bun_core::hint::cold();
-
-            match self.connect() {
-                // The command is queued below as for a dial in flight; the
-                // deferred close then rejects it or a retry sends it, like a
-                // refused dial.
-                Err(err) => {
-                    debug!(
-                        "first dial failed before a socket was opened: {}",
-                        err.name()
-                    );
-                    self.close_without_socket_next_tick();
-                }
-                Ok(()) => self.reset_connection_timeout(),
-            }
-        }
+        self.ensure_dialing();
 
         let self_br = BackRef::new(self);
         let _update = scopeguard::guard(self_br, |p| p.update_poll_ref());
         self.client_mut().send(global_this, command)
     }
 
+    /// Start the first dial if the client has never connected. Every command
+    /// entry point runs this before looking at the client's state, so a
+    /// command on a fresh client is queued behind a dial in flight (or
+    /// rejected against a dial that already failed), never against
+    /// `NeverConnected`.
+    pub(crate) fn ensure_dialing(&self) {
+        if self.client.get().status != valkey::Status::NeverConnected {
+            return;
+        }
+        bun_core::hint::cold();
+
+        match self.connect() {
+            // The command is queued as for a dial in flight; the deferred
+            // close then rejects it or a retry sends it, like a refused dial.
+            Err(err) => {
+                debug!(
+                    "first dial failed before a socket was opened: {}",
+                    err.name()
+                );
+                self.close_without_socket_next_tick();
+            }
+            Ok(()) => self.reset_connection_timeout(),
+        }
+    }
+
     // Getter for memory cost - useful for diagnostics
     pub(crate) fn memory_cost(&self) -> usize {
         // TODO(markovejnovic): This is most-likely wrong because I didn't know better.
diff --git a/src/runtime/valkey_jsc/js_valkey_functions.rs b/src/runtime/valkey_jsc/js_valkey_functions.rs
index 522aadfd7c73..2fcea8573e60 100644
--- a/src/runtime/valkey_jsc/js_valkey_functions.rs
+++ b/src/runtime/valkey_jsc/js_valkey_functions.rs
@@ -1849,7 +1849,10 @@ impl JSValkeyClient {
         // The walk below stores each listener as it goes. A client that would
         // reject the SUBSCRIBE outright must not keep the listeners either: a
         // listener with no subscription behind it pins the event loop and the
-        // client, and cannot be removed with unsubscribe().
+        // client, and cannot be removed with unsubscribe(). The dial comes
+        // first, as in `send()`, so a fresh client with the offline queue off
+        // is rejected the way get() is: connecting, not never connected.
+        this.ensure_dialing();
         if let Some(message) = this.client.get().send_rejection() {
             let error = valkey::ValkeyClient::send_rejection_error(global, message);
             return Ok(JSPromise::rejected_promise(global, error).to_js());
diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts
index b681275e5404..181a481f913f 100644
--- a/test/js/valkey/reliability/connection-failures.test.ts
+++ b/test/js/valkey/reliability/connection-failures.test.ts
@@ -2,7 +2,7 @@ import { RedisClient } from "bun";
 import { estimateShallowMemoryUsageOf } from "bun:jsc";
 import { describe, expect, mock, test } from "bun:test";
 import { once } from "events";
-import { bunEnv, bunExe, isWindows, tempDir, tls as tlsCert } from "harness";
+import { bunEnv, bunExe, isASAN, isDebug, isWindows, tempDir, tls as tlsCert } from "harness";
 import net from "net";
 import path from "path";
 import tls from "tls";
@@ -510,10 +510,13 @@ describe("Valkey: Recovering After fail()", () => {
       .join("");
 
   // A process that a client keeps alive never exits on its own; report that
-  // as the exit code after 3 s instead of waiting for the test to time out.
+  // as the exit code instead of waiting for the test to time out. The payload
+  // runs in well under a second on a debug build; the exit itself is what an
+  // ASAN build makes slow.
   async function exitOutcome(proc: Bun.Subprocess<"ignore", "pipe", "pipe">) {
     const output = Promise.all([proc.stdout.text(), proc.stderr.text()]);
-    const exitCode = await Promise.race([proc.exited, delay(3000).then(() => "still running" as const)]);
+    const budget = isASAN || isDebug ? 15_000 : 3_000;
+    const exitCode = await Promise.race([proc.exited, delay(budget).then(() => "still running" as const)]);
     if (exitCode === "still running") proc.kill();
     const [stdout, stderr] = await output;
     return { stdout, stderr, exitCode };
@@ -1721,6 +1724,45 @@ describe("Valkey: Recovering After fail()", () => {
     }
   });
 
+  test("subscribe() on a fresh client with the offline queue off dials, rejects and registers no handler", async () => {
+    const fake = helloServer();
+    const port = await fake.listen();
+    const client = new RedisClient(`redis://127.0.0.1:${port}`, { enableOfflineQueue: false });
+    try {
+      const connected = Promise.withResolvers<void>();
+      client.onconnect = () => connected.resolve();
+      const delivered: string[] = [];
+      const firstDelivered = Promise.withResolvers<void>();
+      const listener = (message: string) => {
+        delivered.push(message);
+        firstDelivered.resolve();
+      };
+      // The rejection is the one get() gets on this client: the dial has been
+      // started, and the SUBSCRIBE cannot wait for it.
+      await expect(client.subscribe("ch", listener)).rejects.toMatchObject({
+        code: "ERR_REDIS_CONNECTION_CLOSED",
+        message: "Connection is closed and offline queue is disabled",
+      });
+      expect(() => client.unsubscribe("ch")).toThrow("can only be called while in subscriber mode");
+
+      // That dial completes on its own.
+      await connected.promise;
+      const connection = fake.sockets[0];
+      connection.on("data", chunk => {
+        if (chunk.toString("latin1").includes("SUBSCRIBE")) {
+          connection.write(push("subscribe", "ch", 1) + push("message", "ch", "m0"));
+        }
+      });
+      await client.subscribe("ch", listener);
+      await firstDelivered.promise;
+      await client.ping();
+      expect({ delivered, connections: fake.connections }).toEqual({ delivered: ["m0"], connections: 1 });
+    } finally {
+      client.close();
+      fake.server.close();
+    }
+  });
+
   test("the process exits after subscribe() is rejected by a failed client", async () => {
     const fake = helloServer();
     const port = await fake.listen();

From caf7776a44fe723ccc9baeb50d570fa9356bbd13 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Tue, 18 Aug 2026 20:53:15 +0000
Subject: [PATCH 9/9] valkey: check the channel argument's type before
 subscribe() dials

subscribe(123, fn) on a fresh client started the first dial and then threw.
Every other command checks its arguments before it reaches the dial. The
top level type check now runs next to the listener check, ahead of
ensure_dialing(). The walk over the argument is unchanged otherwise.
---
 src/runtime/valkey_jsc/js_valkey_functions.rs | 21 +++++++++++--------
 .../reliability/connection-failures.test.ts   | 20 ++++++++++++++++++
 2 files changed, 32 insertions(+), 9 deletions(-)

diff --git a/src/runtime/valkey_jsc/js_valkey_functions.rs b/src/runtime/valkey_jsc/js_valkey_functions.rs
index 2fcea8573e60..df3acf7e5580 100644
--- a/src/runtime/valkey_jsc/js_valkey_functions.rs
+++ b/src/runtime/valkey_jsc/js_valkey_functions.rs
@@ -1845,13 +1845,22 @@ impl JSValkeyClient {
         if !handler_callback.is_callable() {
             return Err(global.throw_invalid_argument_type("subscribe", "listener", "function"));
         }
+        if !channel_or_many.is_string() && !channel_or_many.is_array() {
+            return Err(global.throw_invalid_argument_type(
+                "subscribe",
+                "channel",
+                "string or array",
+            ));
+        }
 
         // The walk below stores each listener as it goes. A client that would
         // reject the SUBSCRIBE outright must not keep the listeners either: a
         // listener with no subscription behind it pins the event loop and the
         // client, and cannot be removed with unsubscribe(). The dial comes
-        // first, as in `send()`, so a fresh client with the offline queue off
-        // is rejected the way get() is: connecting, not never connected.
+        // after the argument checks, as for every other command, and before
+        // the state check, as in `send()`, so a fresh client with the offline
+        // queue off is rejected the way get() is: connecting, not never
+        // connected.
         this.ensure_dialing();
         if let Some(message) = this.client.get().send_rejection() {
             let error = valkey::ValkeyClient::send_rejection_error(global, message);
@@ -1886,7 +1895,7 @@ impl JSValkeyClient {
                 // handler.
                 this.upsert_receive_handler(global, channel_arg, handler_callback)?;
             }
-        } else if channel_or_many.is_string() {
+        } else {
             // It is a single string channel
             let Some(channel) = from_js(global, channel_or_many)? else {
                 return Err(global.throw_invalid_argument_type("subscribe", "channel", "string"));
@@ -1894,12 +1903,6 @@ impl JSValkeyClient {
             redis_channels.push(channel);
 
             this.upsert_receive_handler(global, channel_or_many, handler_callback)?;
-        } else {
-            return Err(global.throw_invalid_argument_type(
-                "subscribe",
-                "channel",
-                "string or array",
-            ));
         }
 
         let command = Command {
diff --git a/test/js/valkey/reliability/connection-failures.test.ts b/test/js/valkey/reliability/connection-failures.test.ts
index 181a481f913f..746a8a2b1781 100644
--- a/test/js/valkey/reliability/connection-failures.test.ts
+++ b/test/js/valkey/reliability/connection-failures.test.ts
@@ -1763,6 +1763,26 @@ describe("Valkey: Recovering After fail()", () => {
     }
   });
 
+  test("subscribe() with a channel of the wrong type throws before the fresh client dials", async () => {
+    const fake = helloServer();
+    const port = await fake.listen();
+    const client = new RedisClient(`redis://127.0.0.1:${port}`);
+    const probe = new RedisClient(`redis://127.0.0.1:${port}`);
+    try {
+      expect(() => client.subscribe(123 as never, () => {})).toThrow(
+        "Expected channel to be a string or array for 'subscribe'.",
+      );
+      // A dial made by the call above is ahead of the probe's in the stub's
+      // accept queue, so it has been counted by the time the probe is answered.
+      await probe.connect();
+      expect(fake.connections).toBe(1);
+    } finally {
+      client.close();
+      probe.close();
+      fake.server.close();
+    }
+  });
+
   test("the process exits after subscribe() is rejected by a failed client", async () => {
     const fake = helloServer();
     const port = await fake.listen();