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 a3755ebbe731..df3acf7e5580 100644
--- a/src/runtime/valkey_jsc/js_valkey_functions.rs
+++ b/src/runtime/valkey_jsc/js_valkey_functions.rs
@@ -1845,6 +1845,27 @@ 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
+        // 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);
+            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() {
@@ -1874,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"));
@@ -1882,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/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..746a8a2b1781 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";
@@ -502,6 +502,26 @@ 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` : `$${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 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 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 };
+  }
+
   // 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 +937,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}`);
@@ -1659,6 +1675,172 @@ describe("Valkey: Recovering After fail()", () => {
       }
     },
   );
+
+  test("subscribe() on a failed client rejects and registers no handler", async () => {
+    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 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",
+      });
+      // 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);
+      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 {
+      client.close();
+      fake.server.close();
+    }
+  });
+
+  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("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();
+    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();
+          await closed.promise;
+          console.log("onclose");
+          await client.subscribe("ch", () => {}).catch(err => console.log("subscribe rejected", err.code));
+          `,
+        ],
+        env: bunEnv,
+        stdout: "pipe",
+        stderr: "pipe",
+      });
+      expect(await exitOutcome(proc)).toEqual({
+        stdout: "onclose\nsubscribe rejected ERR_REDIS_CONNECTION_CLOSED\n",
+        stderr: "",
+        exitCode: 0,
+      });
+    } finally {
+      fake.server.close();
+    }
+  });
+
+  // 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",
+        `
+        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));
+        `,
+      ],
+      env: bunEnv,
+      stdout: "pipe",
+      stderr: "pipe",
+    });
+    expect(await exitOutcome(proc)).toEqual({
+      stdout: "subscribe rejected ERR_REDIS_CONNECTION_CLOSED\n",
+      stderr: "",
+      exitCode: 0,
+    });
+  });
 });
 
 describe("Valkey: Offline Queue", () => {