From a547a483bcb86e67ac571753932b6b0de5a26c8c Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Mon, 6 Jul 2026 10:42:08 +0000
Subject: [PATCH 01/11] redis: re-issue SUBSCRIBE after an auto-reconnect

A reconnect opens a brand-new server-side connection, which carries none
of the previous connection's subscriptions. The client only sent HELLO,
so a subscriber reported connected === true with its listeners still
registered while no message ever arrived again.

Replay SUBSCRIBE for every channel in the subscription callback map as
soon as the HELLO handshake completes. The command is written directly to
the write buffer, ahead of whatever the offline queue still holds, and
carries no promise; resubscribe_pending counts off the confirmations it
produces so they do not consume a queued command's promise pair.
---
 docs/runtime/redis.mdx                        |   2 +
 src/jsc/JSMap.rs                              |   6 +
 src/jsc/bindings/bindings.cpp                 |  22 ++
 src/runtime/valkey_jsc/js_valkey.rs           |  57 ++++
 src/runtime/valkey_jsc/valkey.rs              |  14 +
 .../reliability/subscriber-reconnect.test.ts  | 257 ++++++++++++++++++
 6 files changed, 358 insertions(+)
 create mode 100644 test/js/valkey/reliability/subscriber-reconnect.test.ts

diff --git a/docs/runtime/redis.mdx b/docs/runtime/redis.mdx
index 5510173205b4..1c3d43f9b8cc 100644
--- a/docs/runtime/redis.mdx
+++ b/docs/runtime/redis.mdx
@@ -417,6 +417,8 @@ When a connection is lost, the client automatically attempts to reconnect with e
 4. Commands executed during disconnection are:
    - Queued if `enableOfflineQueue` is true (default)
    - Rejected immediately if `enableOfflineQueue` is false
+5. Channels registered with `.subscribe()` are re-subscribed as soon as the new
+   connection finishes its handshake, so listeners keep receiving messages
 
 ---
 
diff --git a/src/jsc/JSMap.rs b/src/jsc/JSMap.rs
index 522756df9f0b..52bc6a96d576 100644
--- a/src/jsc/JSMap.rs
+++ b/src/jsc/JSMap.rs
@@ -58,6 +58,12 @@ impl JSMap {
         crate::cpp::JSC__JSMap__size(self, global)
     }
 
+    /// Retrieve this JS Map object's keys as a JS `Array`, in insertion order.
+    #[track_caller]
+    pub fn keys(&mut self, global: &JSGlobalObject) -> JsResult<JSValue> {
+        crate::cpp::JSC__JSMap__keys(self, global)
+    }
+
     /// Attempt to convert a `JSValue` to a `*JSMap`.
     ///
     /// Returns `None` if the value is not a Map.
diff --git a/src/jsc/bindings/bindings.cpp b/src/jsc/bindings/bindings.cpp
index 7ea92a6e4e69..6d7a6d8b50f4 100644
--- a/src/jsc/bindings/bindings.cpp
+++ b/src/jsc/bindings/bindings.cpp
@@ -5943,6 +5943,28 @@ CPP_DECL [[ZIG_EXPORT(check_slow)]] uint32_t JSC__JSMap__size(JSC::JSMap* map, J
     return map->size();
 }
 
+// Returns the map's keys as a JSArray, in insertion order.
+CPP_DECL [[ZIG_EXPORT(zero_is_throw)]] JSC::EncodedJSValue JSC__JSMap__keys(JSC::JSMap* map, JSC::JSGlobalObject* globalObject)
+{
+    auto& vm = JSC::getVM(globalObject);
+    auto scope = DECLARE_THROW_SCOPE(vm);
+
+    JSC::JSArray* keys = JSC::constructEmptyArray(globalObject, nullptr, map->size());
+    RETURN_IF_EXCEPTION(scope, {});
+
+    auto* iterator = JSC::JSMapIterator::create(vm, globalObject->mapIteratorStructure(), map, JSC::IterationKind::Keys);
+    RETURN_IF_EXCEPTION(scope, {});
+
+    unsigned index = 0;
+    JSC::JSValue key;
+    while (iterator->next(globalObject, key)) {
+        keys->putDirectIndex(globalObject, index++, key);
+        RETURN_IF_EXCEPTION(scope, {});
+    }
+
+    return JSC::JSValue::encode(keys);
+}
+
 CPP_DECL void JSC__VM__setControlFlowProfiler(JSC::VM* vm, bool isEnabled)
 {
     if (isEnabled) {
diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs
index 2a5b66ae35e3..ebf15547dfa6 100644
--- a/src/runtime/valkey_jsc/js_valkey.rs
+++ b/src/runtime/valkey_jsc/js_valkey.rs
@@ -142,6 +142,11 @@ impl SubscriptionCtx {
         Ok(count)
     }
 
+    /// The channels this context has listeners for, as a JS array of strings.
+    pub fn subscribed_channels(&self, global_object: &JSGlobalObject) -> JsResult<JSValue> {
+        self.subscription_callback_map().keys(global_object)
+    }
+
     /// Test whether this context has any subscriptions. It is mandatory to
     /// guard deinit with this function.
     pub fn has_subscriptions(&self, global_object: &JSGlobalObject) -> JsResult<bool> {
@@ -722,6 +727,7 @@ impl JSValkeyClient {
                 read_buffer: Default::default(),
                 reply_scanner: Default::default(),
                 retry_attempts: 0,
+                resubscribe_pending: 0,
                 auto_flusher: Default::default(),
             }),
             global_object,
@@ -846,6 +852,7 @@ impl JSValkeyClient {
                 read_buffer: Default::default(),
                 reply_scanner: Default::default(),
                 retry_attempts: 0,
+                resubscribe_pending: 0,
                 auto_flusher: Default::default(),
             }),
             global_object,
@@ -1266,6 +1273,8 @@ impl JSValkeyClient {
         let global_object = self.global_object;
         let _exit = self.vm().enter_event_loop_scope();
 
+        self.resubscribe()?;
+
         if let Some(this_value) = self.this_value.get().try_get() {
             let hello_value: JSValue = 'js_hello: {
                 match protocol_jsc::resp_value_to_js(value, &global_object) {
@@ -1304,6 +1313,54 @@ impl JSValkeyClient {
         Ok(())
     }
 
+    /// Replay the subscription set onto a freshly (re)connected socket.
+    ///
+    /// The server forgets every subscription when the connection drops, but the
+    /// JS-side callback map survives it, so without this a reconnected
+    /// subscriber reports `connected` while silently receiving nothing.
+    ///
+    /// The command goes straight into the write buffer (ahead of anything the
+    /// offline queue still holds) and carries no promise, so the confirmations
+    /// it produces are counted off by `resubscribe_pending` rather than
+    /// consuming another command's promise.
+    fn resubscribe(&self) -> JsTerminatedResult<()> {
+        if !self.is_subscriber() || self.this_value.get().try_get().is_none() {
+            return Ok(());
+        }
+
+        let global_object = self.global_object;
+        let channels = self
+            ._subscription_ctx
+            .get()
+            .subscribed_channels(&global_object)?;
+        let mut iter = channels.array_iterator(&global_object)?;
+        let mut args: Vec<bun_core::ZigStringSlice> = Vec::with_capacity(iter.len as usize);
+        while let Some(channel) = iter.next()? {
+            let channel_str = channel.to_bun_string(&global_object)?;
+            args.push(channel_str.to_utf8());
+            channel_str.deref();
+        }
+        if args.is_empty() {
+            return Ok(());
+        }
+
+        debug!("Replaying SUBSCRIBE for {} channel(s)", args.len());
+        let subscribe_cmd = Command {
+            command: b"SUBSCRIBE",
+            args: command::Args::Slices(&args),
+            meta: command::Meta::default(),
+        };
+        let client = self.client_mut();
+        if subscribe_cmd.write(client.writer()).is_err() {
+            return narrow_terminated(client.fail(
+                b"Failed to write SUBSCRIBE command",
+                protocol::RedisError::OutOfMemory,
+            ));
+        }
+        client.resubscribe_pending = u32::try_from(args.len()).expect("int cast");
+        Ok(())
+    }
+
     /// Invoked when the Valkey client receives a new listener.
     ///
     /// `SubscriptionCtx` will invoke this to communicate that it has added a new listener.
diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs
index b909bff16982..ab84f2d558e0 100644
--- a/src/runtime/valkey_jsc/valkey.rs
+++ b/src/runtime/valkey_jsc/valkey.rs
@@ -264,6 +264,10 @@ pub struct ValkeyClient {
     /// Commands that are waiting to be sent to the server. When pipelining is implemented, this usually will be empty.
     pub queue: command::entry::Queue,
 
+    /// Subscribe confirmations still owed by the promise-less SUBSCRIBE that
+    /// `JSValkeyClient::resubscribe` replays after a reconnect. One per channel.
+    pub resubscribe_pending: u32,
+
     // Connection parameters
     // `connection_strings` is retained because `js_valkey.rs` still slices it
     // when constructing/duplicating clients.
@@ -1106,6 +1110,15 @@ impl ValkeyClient {
                     // Message pushes never need promise pairs
                     should_consume_promise_pair = false;
                 }
+                Some(protocol::SubscriptionPushMessage::Subscribe)
+                    if self.resubscribe_pending > 0 =>
+                {
+                    // Confirmation of the SUBSCRIBE replayed on reconnect. It was
+                    // written without a promise, and its confirmations are the first
+                    // replies on the new connection, so no pair belongs to it.
+                    self.resubscribe_pending -= 1;
+                    should_consume_promise_pair = false;
+                }
                 Some(
                     protocol::SubscriptionPushMessage::Subscribe
                     | protocol::SubscriptionPushMessage::Unsubscribe,
@@ -1282,6 +1295,7 @@ impl ValkeyClient {
         self.flags.failed = false;
         self.flags.is_authenticated = false;
         self.flags.is_selecting_db_internal = false;
+        self.resubscribe_pending = 0;
         if matches!(self.socket, AnySocket::SocketTcp(_)) {
             // if is tcp, we need to start the connection process
             // if is tls, we need to wait for the handshake to complete
diff --git a/test/js/valkey/reliability/subscriber-reconnect.test.ts b/test/js/valkey/reliability/subscriber-reconnect.test.ts
new file mode 100644
index 000000000000..1f0983b6c033
--- /dev/null
+++ b/test/js/valkey/reliability/subscriber-reconnect.test.ts
@@ -0,0 +1,257 @@
+import { RedisClient } from "bun";
+import { describe, expect, test } from "bun:test";
+
+/**
+ * A reconnect hands the client a brand-new server-side connection, which
+ * carries none of the previous connection's subscriptions. These tests pin the
+ * replay of the subscription set against a scripted RESP3 server so they run
+ * without a real Valkey/Redis instance.
+ */
+
+const CRLF = "\r\n";
+
+function bulk(value: string): string {
+  return `$${Buffer.byteLength(value)}${CRLF}${value}${CRLF}`;
+}
+
+const HELLO_REPLY =
+  `%3${CRLF}` + bulk("server") + bulk("redis") + bulk("proto") + `:3${CRLF}` + bulk("version") + bulk("7.4.0");
+
+type Connection = {
+  socket: { write(data: string): number; end(): void } | null;
+  commands: string[][];
+  channels: Set<string>;
+};
+
+/** Pull every complete RESP array-of-bulk-strings command out of `buffer`. */
+function parseCommands(buffer: Buffer): { commands: string[][]; rest: Buffer } {
+  const commands: string[][] = [];
+  let offset = 0;
+
+  while (offset < buffer.length && buffer[offset] === 0x2a /* '*' */) {
+    const headerEnd = buffer.indexOf(CRLF, offset);
+    if (headerEnd === -1) break;
+
+    const fieldCount = Number(buffer.subarray(offset + 1, headerEnd));
+    const fields: string[] = [];
+    let cursor = headerEnd + 2;
+    let complete = true;
+
+    for (let i = 0; i < fieldCount; i++) {
+      if (cursor >= buffer.length || buffer[cursor] !== 0x24 /* '$' */) {
+        complete = false;
+        break;
+      }
+      const lengthEnd = buffer.indexOf(CRLF, cursor);
+      if (lengthEnd === -1) {
+        complete = false;
+        break;
+      }
+      const byteLength = Number(buffer.subarray(cursor + 1, lengthEnd));
+      if (buffer.length < lengthEnd + 2 + byteLength + 2) {
+        complete = false;
+        break;
+      }
+      fields.push(buffer.subarray(lengthEnd + 2, lengthEnd + 2 + byteLength).toString());
+      cursor = lengthEnd + 2 + byteLength + 2;
+    }
+
+    if (!complete) break;
+    commands.push(fields);
+    offset = cursor;
+  }
+
+  return { commands, rest: buffer.subarray(offset) };
+}
+
+function startRespServer() {
+  const connections: Connection[] = [];
+  const waiters: { count: number; resolve: () => void }[] = [];
+
+  const server = Bun.listen<{ buffer: Buffer; connection: Connection }>({
+    hostname: "127.0.0.1",
+    port: 0,
+    socket: {
+      open(socket) {
+        const connection: Connection = { socket, commands: [], channels: new Set() };
+        connections.push(connection);
+        socket.data = { buffer: Buffer.alloc(0), connection };
+        for (const waiter of waiters.splice(0)) {
+          if (connections.length >= waiter.count) waiter.resolve();
+          else waiters.push(waiter);
+        }
+      },
+      close(socket) {
+        if (socket.data) socket.data.connection.socket = null;
+      },
+      error() {},
+      data(socket, chunk) {
+        const state = socket.data;
+        state.buffer = state.buffer.length ? Buffer.concat([state.buffer, chunk]) : chunk;
+        const { commands, rest } = parseCommands(state.buffer);
+        state.buffer = rest;
+
+        for (const command of commands) {
+          state.connection.commands.push(command);
+
+          switch (command[0].toUpperCase()) {
+            case "HELLO":
+              socket.write(HELLO_REPLY);
+              break;
+
+            case "SUBSCRIBE":
+              for (const channel of command.slice(1)) {
+                state.connection.channels.add(channel);
+                socket.write(
+                  `>3${CRLF}` + bulk("subscribe") + bulk(channel) + `:${state.connection.channels.size}${CRLF}`,
+                );
+              }
+              break;
+
+            case "UNSUBSCRIBE": {
+              const channels = command.length > 1 ? command.slice(1) : [...state.connection.channels];
+              for (const channel of channels) {
+                state.connection.channels.delete(channel);
+                socket.write(
+                  `>3${CRLF}` + bulk("unsubscribe") + bulk(channel) + `:${state.connection.channels.size}${CRLF}`,
+                );
+              }
+              break;
+            }
+
+            case "PING":
+              socket.write(`+PONG${CRLF}`);
+              break;
+
+            case "PUBLISH": {
+              let receivers = 0;
+              for (const target of connections) {
+                if (target.socket && target.channels.has(command[1])) {
+                  receivers++;
+                  target.socket.write(`>3${CRLF}` + bulk("message") + bulk(command[1]) + bulk(command[2]));
+                }
+              }
+              socket.write(`:${receivers}${CRLF}`);
+              break;
+            }
+
+            default:
+              socket.write(`+OK${CRLF}`);
+          }
+        }
+      },
+    },
+  });
+
+  return {
+    connections,
+    url: `redis://127.0.0.1:${server.port}`,
+    /** Resolves once the server has accepted at least `count` connections. */
+    waitForConnections(count: number): Promise<void> {
+      if (connections.length >= count) return Promise.resolve();
+      return new Promise<void>(resolve => waiters.push({ count, resolve }));
+    },
+    [Symbol.dispose]() {
+      server.stop(true);
+    },
+  };
+}
+
+/** An async queue so tests await the next delivered message instead of a timer. */
+function messageQueue() {
+  const buffered: string[] = [];
+  const waiters: ((message: string) => void)[] = [];
+  return {
+    push(message: string) {
+      const waiter = waiters.shift();
+      if (waiter) waiter(message);
+      else buffered.push(message);
+    },
+    next(): Promise<string> {
+      const message = buffered.shift();
+      if (message !== undefined) return Promise.resolve(message);
+      return new Promise<string>(resolve => waiters.push(resolve));
+    },
+  };
+}
+
+const commandLines = (connection: Connection) => connection.commands.map(command => command.join(" "));
+
+describe("Valkey: subscriber reconnect", () => {
+  test("replays SUBSCRIBE for every channel with a listener", async () => {
+    using server = startRespServer();
+    const messages = messageQueue();
+
+    const subscriber = new RedisClient(server.url, { autoReconnect: true, maxRetries: 10 });
+    const publisher = new RedisClient(server.url, { autoReconnect: false });
+
+    try {
+      await subscriber.connect();
+      await subscriber.subscribe(["news", "sports"], message => messages.push(message));
+      await publisher.connect();
+
+      expect(await publisher.publish("news", "before")).toBe(1);
+      expect(await messages.next()).toBe("before");
+
+      // Drop the subscriber's connection from the server side.
+      server.connections[0].socket!.end();
+      await server.waitForConnections(3);
+
+      // PING is written after the reconnect handshake, so its reply proves the
+      // server has already seen everything else the client sent on the new
+      // connection. Without a replay, SUBSCRIBE is simply missing below.
+      expect(await subscriber.ping()).toBe("PONG");
+
+      const reconnected = server.connections[2];
+      expect(commandLines(reconnected)).toEqual(["HELLO 3", "SUBSCRIBE news sports", "PING"]);
+      expect(subscriber.connected).toBe(true);
+
+      // Both channels are live again, and messages reach the original listener.
+      expect(await publisher.publish("sports", "after")).toBe(1);
+      expect(await messages.next()).toBe("after");
+    } finally {
+      subscriber.close();
+      publisher.close();
+    }
+  });
+
+  test("does not replay channels the client unsubscribed from", async () => {
+    using server = startRespServer();
+    const messages = messageQueue();
+
+    const subscriber = new RedisClient(server.url, { autoReconnect: true, maxRetries: 10 });
+
+    try {
+      await subscriber.connect();
+      await subscriber.subscribe(["news", "sports"], message => messages.push(message));
+      await subscriber.unsubscribe("sports");
+
+      server.connections[0].socket!.end();
+      await server.waitForConnections(2);
+      expect(await subscriber.ping()).toBe("PONG");
+
+      expect(commandLines(server.connections[1])).toEqual(["HELLO 3", "SUBSCRIBE news", "PING"]);
+    } finally {
+      subscriber.close();
+    }
+  });
+
+  test("a client that never subscribed sends no SUBSCRIBE on reconnect", async () => {
+    using server = startRespServer();
+
+    const client = new RedisClient(server.url, { autoReconnect: true, maxRetries: 10 });
+
+    try {
+      await client.connect();
+      expect(await client.ping()).toBe("PONG");
+
+      server.connections[0].socket!.end();
+      await server.waitForConnections(2);
+      expect(await client.ping()).toBe("PONG");
+
+      expect(commandLines(server.connections[1])).toEqual(["HELLO 3", "PING"]);
+    } finally {
+      client.close();
+    }
+  });
+});

From 953d661e815a3e4d62d8f71908e75fc2119e47ad Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Mon, 6 Jul 2026 11:44:05 +0000
Subject: [PATCH 02/11] test: pin the replay ordering against SELECT

With a database in the URL, the replayed SUBSCRIBE is written while SELECT's
reply is still outstanding. Assert it lands behind SELECT on the wire, or it
would subscribe on database 0.
---
 .../reliability/subscriber-reconnect.test.ts  | 29 +++++++++++++++++++
 1 file changed, 29 insertions(+)

diff --git a/test/js/valkey/reliability/subscriber-reconnect.test.ts b/test/js/valkey/reliability/subscriber-reconnect.test.ts
index 1f0983b6c033..2c39a1053046 100644
--- a/test/js/valkey/reliability/subscriber-reconnect.test.ts
+++ b/test/js/valkey/reliability/subscriber-reconnect.test.ts
@@ -254,4 +254,33 @@ describe("Valkey: subscriber reconnect", () => {
       client.close();
     }
   });
+
+  test("replays SUBSCRIBE after SELECT when the URL names a database", async () => {
+    using server = startRespServer();
+    const messages = messageQueue();
+
+    // The replay is written while SELECT's reply is still outstanding, so it has
+    // to land behind SELECT on the wire or it would subscribe on database 0.
+    const subscriber = new RedisClient(`${server.url}/3`, { autoReconnect: true, maxRetries: 10 });
+    const publisher = new RedisClient(`${server.url}/3`, { autoReconnect: false });
+
+    try {
+      await subscriber.connect();
+      await subscriber.subscribe("news", message => messages.push(message));
+      expect(commandLines(server.connections[0])).toEqual(["HELLO 3", "SELECT 3", "SUBSCRIBE news"]);
+
+      await publisher.connect();
+      server.connections[0].socket!.end();
+      await server.waitForConnections(3);
+      expect(await subscriber.ping()).toBe("PONG");
+
+      expect(commandLines(server.connections[2])).toEqual(["HELLO 3", "SELECT 3", "SUBSCRIBE news", "PING"]);
+
+      expect(await publisher.publish("news", "after")).toBe(1);
+      expect(await messages.next()).toBe("after");
+    } finally {
+      subscriber.close();
+      publisher.close();
+    }
+  });
 });

From 4961856a267e94f536616f489665641132077a60 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Mon, 6 Jul 2026 12:25:28 +0000
Subject: [PATCH 03/11] redis: don't let an error reply to the replayed
 SUBSCRIBE eat a reply slot

The replayed SUBSCRIBE carries no promise pair. If the server refuses it
(an ACL change during the outage, say) the reply is an Error rather than a
confirmation, so the push-kind accounting never ran and the error consumed
the pair of whatever user command was parked behind it in the in-flight
queue. The subscriber error path then dropped that pair without settling
it, hanging the caller's await.

Skip pair consumption for an error reply while confirmations are still
owed, and stop expecting them: the server rejects SUBSCRIBE as a whole.
---
 src/runtime/valkey_jsc/valkey.rs              |  8 +++++
 .../reliability/subscriber-reconnect.test.ts  | 34 ++++++++++++++++++-
 2 files changed, 41 insertions(+), 1 deletion(-)

diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs
index ab84f2d558e0..2103bfcc0b1f 100644
--- a/src/runtime/valkey_jsc/valkey.rs
+++ b/src/runtime/valkey_jsc/valkey.rs
@@ -1103,6 +1103,14 @@ impl ValkeyClient {
         let mut should_consume_promise_pair = true;
         let mut pair_maybe: Option<command::PromisePair> = None;
 
+        if self.resubscribe_pending > 0 && matches!(value, RESPValue::Error(_)) {
+            // The replayed SUBSCRIBE was refused (an ACL change during the outage,
+            // say). It owns no promise pair, so this error must not eat one, and no
+            // confirmation is coming for any of its channels.
+            self.resubscribe_pending = 0;
+            should_consume_promise_pair = false;
+        }
+
         // For subscription clients, check if this is a push message that doesn't need a promise pair
         if let RESPValue::Push(push) = value {
             match protocol::SubscriptionPushMessage::from_bytes(&push.kind) {
diff --git a/test/js/valkey/reliability/subscriber-reconnect.test.ts b/test/js/valkey/reliability/subscriber-reconnect.test.ts
index 2c39a1053046..0d26198447d1 100644
--- a/test/js/valkey/reliability/subscriber-reconnect.test.ts
+++ b/test/js/valkey/reliability/subscriber-reconnect.test.ts
@@ -64,7 +64,12 @@ function parseCommands(buffer: Buffer): { commands: string[][]; rest: Buffer } {
   return { commands, rest: buffer.subarray(offset) };
 }
 
-function startRespServer() {
+type ServerOptions = {
+  /** Refuse SUBSCRIBE with -NOPERM on every connection from this index onwards. */
+  refuseSubscribeFromConnection?: number;
+};
+
+function startRespServer({ refuseSubscribeFromConnection = Infinity }: ServerOptions = {}) {
   const connections: Connection[] = [];
   const waiters: { count: number; resolve: () => void }[] = [];
 
@@ -100,6 +105,11 @@ function startRespServer() {
               break;
 
             case "SUBSCRIBE":
+              // Redis rejects the whole command once, with no per-channel confirmations.
+              if (connections.indexOf(state.connection) >= refuseSubscribeFromConnection) {
+                socket.write(`-NOPERM this user has no permissions to access one of the channels${CRLF}`);
+                break;
+              }
               for (const channel of command.slice(1)) {
                 state.connection.channels.add(channel);
                 socket.write(
@@ -255,6 +265,28 @@ describe("Valkey: subscriber reconnect", () => {
     }
   });
 
+  test("an error reply to the replay settles the in-flight command instead of eating its reply slot", async () => {
+    using server = startRespServer({ refuseSubscribeFromConnection: 1 });
+
+    const subscriber = new RedisClient(server.url, { autoReconnect: true, maxRetries: 10 });
+
+    try {
+      await subscriber.connect();
+      await subscriber.subscribe("news", () => {});
+
+      server.connections[0].socket!.end();
+      await server.waitForConnections(2);
+
+      // The replayed SUBSCRIBE carries no promise, so its -NOPERM must not consume
+      // the pair PING parked in the in-flight queue behind it. It used to, and the
+      // pair was then dropped unsettled, hanging this await forever.
+      await expect(subscriber.ping()).rejects.toThrow(/NOPERM/);
+      expect(commandLines(server.connections[1])).toEqual(["HELLO 3", "SUBSCRIBE news", "PING"]);
+    } finally {
+      subscriber.close();
+    }
+  });
+
   test("replays SUBSCRIBE after SELECT when the URL names a database", async () => {
     using server = startRespServer();
     const messages = messageQueue();

From d044c4e90732e99e76291ae28baeb9b9f4d922d1 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Mon, 6 Jul 2026 12:59:49 +0000
Subject: [PATCH 04/11] redis: replay subscriptions after the connect promise
 settles

Running resubscribe() ahead of the promise meant an exception from it
(building the channel list allocates) returned early and left .connect()
pending forever. It still lands ahead of the offline queue, which only
drains once the microtask scope and on_writable run.

Also note in the docs that psubscribe patterns are not replayed, run the
reconnect tests concurrently, and unsubscribe before close() so teardown
does not depend on GC to release the event loop.
---
 docs/runtime/redis.mdx                        |  3 ++-
 src/runtime/valkey_jsc/js_valkey.rs           |  8 ++++---
 .../reliability/subscriber-reconnect.test.ts  | 23 +++++++++++++++----
 3 files changed, 25 insertions(+), 9 deletions(-)

diff --git a/docs/runtime/redis.mdx b/docs/runtime/redis.mdx
index 1c3d43f9b8cc..d5f92c5c4bcc 100644
--- a/docs/runtime/redis.mdx
+++ b/docs/runtime/redis.mdx
@@ -418,7 +418,8 @@ When a connection is lost, the client automatically attempts to reconnect with e
    - Queued if `enableOfflineQueue` is true (default)
    - Rejected immediately if `enableOfflineQueue` is false
 5. Channels registered with `.subscribe()` are re-subscribed as soon as the new
-   connection finishes its handshake, so listeners keep receiving messages
+   connection finishes its handshake, so listeners keep receiving messages.
+   Patterns sent with `psubscribe` are not replayed
 
 ---
 
diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs
index ebf15547dfa6..d28713b45f84 100644
--- a/src/runtime/valkey_jsc/js_valkey.rs
+++ b/src/runtime/valkey_jsc/js_valkey.rs
@@ -1273,8 +1273,6 @@ impl JSValkeyClient {
         let global_object = self.global_object;
         let _exit = self.vm().enter_event_loop_scope();
 
-        self.resubscribe()?;
-
         if let Some(this_value) = self.this_value.get().try_get() {
             let hello_value: JSValue = 'js_hello: {
                 match protocol_jsc::resp_value_to_js(value, &global_object) {
@@ -1310,7 +1308,11 @@ impl JSValkeyClient {
                 self.client_mut().flags.connection_promise_returns_client = false;
             }
         }
-        Ok(())
+
+        // Last, so a failure here cannot strand `.connect()`. Still ahead of the
+        // offline queue: nothing drains it until this scope's microtask drain and
+        // `on_writable` run, both of which happen after this returns.
+        self.resubscribe()
     }
 
     /// Replay the subscription set onto a freshly (re)connected socket.
diff --git a/test/js/valkey/reliability/subscriber-reconnect.test.ts b/test/js/valkey/reliability/subscriber-reconnect.test.ts
index 0d26198447d1..d55b772ff07a 100644
--- a/test/js/valkey/reliability/subscriber-reconnect.test.ts
+++ b/test/js/valkey/reliability/subscriber-reconnect.test.ts
@@ -187,7 +187,20 @@ function messageQueue() {
 
 const commandLines = (connection: Connection) => connection.commands.map(command => command.join(" "));
 
-describe("Valkey: subscriber reconnect", () => {
+/**
+ * Drop the subscriptions before closing. `close()` on a still-subscribed client
+ * leaves the event loop pinned by the handler map until GC (#33103). A failed
+ * client rejects the UNSUBSCRIBE it sends, which is fine: `unsubscribe()` clears
+ * the handlers before it reaches the socket.
+ */
+async function closeSubscriber(client: RedisClient) {
+  try {
+    await client.unsubscribe();
+  } catch {}
+  client.close();
+}
+
+describe.concurrent("Valkey: subscriber reconnect", () => {
   test("replays SUBSCRIBE for every channel with a listener", async () => {
     using server = startRespServer();
     const messages = messageQueue();
@@ -220,7 +233,7 @@ describe("Valkey: subscriber reconnect", () => {
       expect(await publisher.publish("sports", "after")).toBe(1);
       expect(await messages.next()).toBe("after");
     } finally {
-      subscriber.close();
+      await closeSubscriber(subscriber);
       publisher.close();
     }
   });
@@ -242,7 +255,7 @@ describe("Valkey: subscriber reconnect", () => {
 
       expect(commandLines(server.connections[1])).toEqual(["HELLO 3", "SUBSCRIBE news", "PING"]);
     } finally {
-      subscriber.close();
+      await closeSubscriber(subscriber);
     }
   });
 
@@ -283,7 +296,7 @@ describe("Valkey: subscriber reconnect", () => {
       await expect(subscriber.ping()).rejects.toThrow(/NOPERM/);
       expect(commandLines(server.connections[1])).toEqual(["HELLO 3", "SUBSCRIBE news", "PING"]);
     } finally {
-      subscriber.close();
+      await closeSubscriber(subscriber);
     }
   });
 
@@ -311,7 +324,7 @@ describe("Valkey: subscriber reconnect", () => {
       expect(await publisher.publish("news", "after")).toBe(1);
       expect(await messages.next()).toBe("after");
     } finally {
-      subscriber.close();
+      await closeSubscriber(subscriber);
       publisher.close();
     }
   });

From 45ff3b0e24a86c32b9a903ec7027947b45e7bc08 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Mon, 6 Jul 2026 13:14:55 +0000
Subject: [PATCH 05/11] test: pin the is_subscriber guard on the replay

subscribe() without an explicit connect() populates the handler map and
queues the SUBSCRIBE, so the map is non-empty before the handshake finishes.
Replaying it there would put SUBSCRIBE on the wire twice. Nothing caught
that, so assert the lazy-connect path sends it exactly once.
---
 .../reliability/subscriber-reconnect.test.ts  | 19 +++++++++++++++++++
 1 file changed, 19 insertions(+)

diff --git a/test/js/valkey/reliability/subscriber-reconnect.test.ts b/test/js/valkey/reliability/subscriber-reconnect.test.ts
index d55b772ff07a..e26eaf333502 100644
--- a/test/js/valkey/reliability/subscriber-reconnect.test.ts
+++ b/test/js/valkey/reliability/subscriber-reconnect.test.ts
@@ -278,6 +278,25 @@ describe.concurrent("Valkey: subscriber reconnect", () => {
     }
   });
 
+  test("does not replay a SUBSCRIBE the offline queue is already holding", async () => {
+    using server = startRespServer();
+
+    // No explicit connect(): subscribe() opens the socket and queues the command,
+    // so the handler map is populated before the handshake finishes. Replaying it
+    // here would put SUBSCRIBE on the wire twice, once from the replay and once
+    // from the queue.
+    const subscriber = new RedisClient(server.url, { autoReconnect: true, maxRetries: 10 });
+
+    try {
+      await subscriber.subscribe("news", () => {});
+      expect(await subscriber.ping()).toBe("PONG");
+
+      expect(commandLines(server.connections[0])).toEqual(["HELLO 3", "SUBSCRIBE news", "PING"]);
+    } finally {
+      await closeSubscriber(subscriber);
+    }
+  });
+
   test("an error reply to the replay settles the in-flight command instead of eating its reply slot", async () => {
     using server = startRespServer({ refuseSubscribeFromConnection: 1 });
 

From a3a57b48a8ca94e2694968e9fba53ffaade4f760 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Mon, 6 Jul 2026 16:15:46 +0000
Subject: [PATCH 06/11] test: run the subscriber reconnect cases serially
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Six of them at once drove six reconnects through six clients, and the
Windows agents blew the client's 10s connection timeout on three of them:

  ✗ an error reply to the replay ... [10081ms]
    Received message: "Connection timeout reached after 10000ms"
  ✗ replays SUBSCRIBE after SELECT ... [10080ms]
    RedisError: Connection has failed

The same file passed on windows x64, x64-baseline and win11-aarch64 before
it was made concurrent. Serial, it runs in 3s.
---
 test/js/valkey/reliability/subscriber-reconnect.test.ts | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/test/js/valkey/reliability/subscriber-reconnect.test.ts b/test/js/valkey/reliability/subscriber-reconnect.test.ts
index e26eaf333502..5becebd2eb0e 100644
--- a/test/js/valkey/reliability/subscriber-reconnect.test.ts
+++ b/test/js/valkey/reliability/subscriber-reconnect.test.ts
@@ -200,7 +200,10 @@ async function closeSubscriber(client: RedisClient) {
   client.close();
 }
 
-describe.concurrent("Valkey: subscriber reconnect", () => {
+// Deliberately not `describe.concurrent`. Each case drives a client through a
+// real reconnect, and six of those at once starved the 10s connection timeout on
+// the Windows CI agents. Serial, the whole file is under 4s.
+describe("Valkey: subscriber reconnect", () => {
   test("replays SUBSCRIBE for every channel with a listener", async () => {
     using server = startRespServer();
     const messages = messageQueue();

From 4e5b59d6c572840cd023bb345a875f54aa5d5a16 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Mon, 6 Jul 2026 16:48:41 +0000
Subject: [PATCH 07/11] test: answer UNSUBSCRIBE even when the connection holds
 no channels

Redis always confirms UNSUBSCRIBE, sending a single nil-channel reply when
the connection is subscribed to nothing. The scripted server stayed silent
in that case, so closeSubscriber()'s `await client.unsubscribe()` hung in
the teardown of every test whose reconnect carried no subscriptions.

That only bit on a build without the fix, where the replay never happens:
the four failing tests timed out after 5s apiece instead of reporting the
assertion that actually failed. Now they fail in 0.5s with the real diff.
---
 test/js/valkey/reliability/subscriber-reconnect.test.ts | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/test/js/valkey/reliability/subscriber-reconnect.test.ts b/test/js/valkey/reliability/subscriber-reconnect.test.ts
index 5becebd2eb0e..e0c195b30b9a 100644
--- a/test/js/valkey/reliability/subscriber-reconnect.test.ts
+++ b/test/js/valkey/reliability/subscriber-reconnect.test.ts
@@ -120,6 +120,13 @@ function startRespServer({ refuseSubscribeFromConnection = Infinity }: ServerOpt
 
             case "UNSUBSCRIBE": {
               const channels = command.length > 1 ? command.slice(1) : [...state.connection.channels];
+              // Redis always answers, even when the connection holds no subscriptions:
+              // in that case with a single nil-channel confirmation. Staying silent here
+              // hangs the caller's `await unsubscribe()`.
+              if (channels.length === 0) {
+                socket.write(`>3${CRLF}` + bulk("unsubscribe") + `_${CRLF}` + `:0${CRLF}`);
+                break;
+              }
               for (const channel of channels) {
                 state.connection.channels.delete(channel);
                 socket.write(

From f10533e5fac9e9483d81a8211368d72bb5701a4b Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Mon, 6 Jul 2026 17:42:12 +0000
Subject: [PATCH 08/11] test: respect socket backpressure in the scripted RESP3
 server
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

socket.write is unbuffered: it reports how many bytes it took, 0 when the
send buffer is full and -1 once the socket is closing. The server ignored
that and dropped the remainder, so a reply could vanish.

A freshly accepted socket is not always writable yet on Windows, which lost
the HELLO reply outright. The client then sat in Connecting until its 10s
timeout, and the NOPERM case failed on windows 11 aarch64 every attempt:

  ✗ an error reply to the replay ... [10073.15ms]
    Expected pattern: /NOPERM/
    Received message: "Connection timeout reached after 10000ms"

Queue what the socket will not take and flush it again on drain.
---
 .../reliability/subscriber-reconnect.test.ts  | 48 ++++++++++++++-----
 1 file changed, 37 insertions(+), 11 deletions(-)

diff --git a/test/js/valkey/reliability/subscriber-reconnect.test.ts b/test/js/valkey/reliability/subscriber-reconnect.test.ts
index e0c195b30b9a..e5c481e3674f 100644
--- a/test/js/valkey/reliability/subscriber-reconnect.test.ts
+++ b/test/js/valkey/reliability/subscriber-reconnect.test.ts
@@ -18,11 +18,32 @@ const HELLO_REPLY =
   `%3${CRLF}` + bulk("server") + bulk("redis") + bulk("proto") + `:3${CRLF}` + bulk("version") + bulk("7.4.0");
 
 type Connection = {
-  socket: { write(data: string): number; end(): void } | null;
+  socket: { write(data: Buffer): number; end(): void } | null;
   commands: string[][];
   channels: Set<string>;
+  /** Bytes `socket.write` would not take yet; flushed again on `drain`. */
+  pending: Buffer;
 };
 
+/**
+ * `socket.write` is unbuffered: it returns how many bytes it took, 0 under
+ * backpressure and -1 once the socket is closing. A freshly accepted socket is
+ * not always writable yet, so dropping the remainder silently loses replies.
+ */
+function send(connection: Connection, data: string): void {
+  const chunk = Buffer.from(data, "utf8");
+  connection.pending = connection.pending.length ? Buffer.concat([connection.pending, chunk]) : chunk;
+  flush(connection);
+}
+
+function flush(connection: Connection): void {
+  while (connection.socket && connection.pending.length > 0) {
+    const wrote = connection.socket.write(connection.pending);
+    if (wrote <= 0) return; // backpressure or closing; `drain` retries.
+    connection.pending = connection.pending.subarray(wrote);
+  }
+}
+
 /** Pull every complete RESP array-of-bulk-strings command out of `buffer`. */
 function parseCommands(buffer: Buffer): { commands: string[][]; rest: Buffer } {
   const commands: string[][] = [];
@@ -78,7 +99,7 @@ function startRespServer({ refuseSubscribeFromConnection = Infinity }: ServerOpt
     port: 0,
     socket: {
       open(socket) {
-        const connection: Connection = { socket, commands: [], channels: new Set() };
+        const connection: Connection = { socket, commands: [], channels: new Set(), pending: Buffer.alloc(0) };
         connections.push(connection);
         socket.data = { buffer: Buffer.alloc(0), connection };
         for (const waiter of waiters.splice(0)) {
@@ -89,6 +110,9 @@ function startRespServer({ refuseSubscribeFromConnection = Infinity }: ServerOpt
       close(socket) {
         if (socket.data) socket.data.connection.socket = null;
       },
+      drain(socket) {
+        if (socket.data) flush(socket.data.connection);
+      },
       error() {},
       data(socket, chunk) {
         const state = socket.data;
@@ -101,18 +125,19 @@ function startRespServer({ refuseSubscribeFromConnection = Infinity }: ServerOpt
 
           switch (command[0].toUpperCase()) {
             case "HELLO":
-              socket.write(HELLO_REPLY);
+              send(state.connection, HELLO_REPLY);
               break;
 
             case "SUBSCRIBE":
               // Redis rejects the whole command once, with no per-channel confirmations.
               if (connections.indexOf(state.connection) >= refuseSubscribeFromConnection) {
-                socket.write(`-NOPERM this user has no permissions to access one of the channels${CRLF}`);
+                send(state.connection, `-NOPERM this user has no permissions to access one of the channels${CRLF}`);
                 break;
               }
               for (const channel of command.slice(1)) {
                 state.connection.channels.add(channel);
-                socket.write(
+                send(
+                  state.connection,
                   `>3${CRLF}` + bulk("subscribe") + bulk(channel) + `:${state.connection.channels.size}${CRLF}`,
                 );
               }
@@ -124,12 +149,13 @@ function startRespServer({ refuseSubscribeFromConnection = Infinity }: ServerOpt
               // in that case with a single nil-channel confirmation. Staying silent here
               // hangs the caller's `await unsubscribe()`.
               if (channels.length === 0) {
-                socket.write(`>3${CRLF}` + bulk("unsubscribe") + `_${CRLF}` + `:0${CRLF}`);
+                send(state.connection, `>3${CRLF}` + bulk("unsubscribe") + `_${CRLF}` + `:0${CRLF}`);
                 break;
               }
               for (const channel of channels) {
                 state.connection.channels.delete(channel);
-                socket.write(
+                send(
+                  state.connection,
                   `>3${CRLF}` + bulk("unsubscribe") + bulk(channel) + `:${state.connection.channels.size}${CRLF}`,
                 );
               }
@@ -137,7 +163,7 @@ function startRespServer({ refuseSubscribeFromConnection = Infinity }: ServerOpt
             }
 
             case "PING":
-              socket.write(`+PONG${CRLF}`);
+              send(state.connection, `+PONG${CRLF}`);
               break;
 
             case "PUBLISH": {
@@ -145,15 +171,15 @@ function startRespServer({ refuseSubscribeFromConnection = Infinity }: ServerOpt
               for (const target of connections) {
                 if (target.socket && target.channels.has(command[1])) {
                   receivers++;
-                  target.socket.write(`>3${CRLF}` + bulk("message") + bulk(command[1]) + bulk(command[2]));
+                  send(target, `>3${CRLF}` + bulk("message") + bulk(command[1]) + bulk(command[2]));
                 }
               }
-              socket.write(`:${receivers}${CRLF}`);
+              send(state.connection, `:${receivers}${CRLF}`);
               break;
             }
 
             default:
-              socket.write(`+OK${CRLF}`);
+              send(state.connection, `+OK${CRLF}`);
           }
         }
       },

From cd6c95389253918143ef17ad18cfcaaba61ce777 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Mon, 6 Jul 2026 19:03:41 +0000
Subject: [PATCH 09/11] test: assert the wire alongside the error reply outcome

windows 11 aarch64 fails this case on every attempt with

  Expected pattern: /NOPERM/
  Received message: "Connection timeout reached after 10000ms"

which can only happen if the reconnect never finished its handshake, since
the -NOPERM path sets flags.failed and that suppresses the timeout. The
other five reconnect cases pass on the same agent, so the error text alone
does not say which half broke.

Assert the ping outcome, the connected flag and the server's per-connection
command log together, so a failure shows whether HELLO, SUBSCRIBE and PING
reached the server at all. Also catches the no-replay case immediately
instead of waiting for PING to hang.
---
 .../reliability/subscriber-reconnect.test.ts   | 18 ++++++++++++++++--
 1 file changed, 16 insertions(+), 2 deletions(-)

diff --git a/test/js/valkey/reliability/subscriber-reconnect.test.ts b/test/js/valkey/reliability/subscriber-reconnect.test.ts
index e5c481e3674f..210734ca0cfa 100644
--- a/test/js/valkey/reliability/subscriber-reconnect.test.ts
+++ b/test/js/valkey/reliability/subscriber-reconnect.test.ts
@@ -348,8 +348,22 @@ describe("Valkey: subscriber reconnect", () => {
       // The replayed SUBSCRIBE carries no promise, so its -NOPERM must not consume
       // the pair PING parked in the in-flight queue behind it. It used to, and the
       // pair was then dropped unsettled, hanging this await forever.
-      await expect(subscriber.ping()).rejects.toThrow(/NOPERM/);
-      expect(commandLines(server.connections[1])).toEqual(["HELLO 3", "SUBSCRIBE news", "PING"]);
+      //
+      // Asserted together with the wire so a failure says which half broke: the
+      // server's view of the reconnect, or where PING's reply went.
+      const pinged = await subscriber.ping().then(
+        value => ({ settled: "resolved", value }),
+        error => ({ settled: "rejected", message: String(error?.message ?? error) }),
+      );
+
+      expect({ pinged, connected: subscriber.connected, wire: server.connections.map(commandLines) }).toEqual({
+        pinged: { settled: "rejected", message: expect.stringContaining("NOPERM") },
+        connected: true,
+        wire: [
+          ["HELLO 3", "SUBSCRIBE news"],
+          ["HELLO 3", "SUBSCRIBE news", "PING"],
+        ],
+      });
     } finally {
       await closeSubscriber(subscriber);
     }

From 861f5554c5c4e493efab9ca4d70f525163b4f152 Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Mon, 6 Jul 2026 20:49:01 +0000
Subject: [PATCH 10/11] test: trigger the reconnect with close()/connect()
 instead of a socket drop

The socket-drop tests relied on the auto-reconnect backoff timer firing and
the handshake finishing inside the client's 10s connection timeout. On the
loaded Windows CI agents that raced, and the subscriber tests flaked (win11
aarch64 most often).

close() + await connect() exercises the identical replay path
(on_valkey_connect -> resubscribe) and the connect() promise resolves only
once HELLO completes, so there is no timer to race. The per-connection wire
assertions are unchanged, 4 of 6 still fail on a build without the fix, and
the no-longer-needed waitForConnections barrier is removed.
---
 .../reliability/subscriber-reconnect.test.ts  | 45 +++++++++----------
 1 file changed, 21 insertions(+), 24 deletions(-)

diff --git a/test/js/valkey/reliability/subscriber-reconnect.test.ts b/test/js/valkey/reliability/subscriber-reconnect.test.ts
index 210734ca0cfa..f4bca06da520 100644
--- a/test/js/valkey/reliability/subscriber-reconnect.test.ts
+++ b/test/js/valkey/reliability/subscriber-reconnect.test.ts
@@ -92,7 +92,6 @@ type ServerOptions = {
 
 function startRespServer({ refuseSubscribeFromConnection = Infinity }: ServerOptions = {}) {
   const connections: Connection[] = [];
-  const waiters: { count: number; resolve: () => void }[] = [];
 
   const server = Bun.listen<{ buffer: Buffer; connection: Connection }>({
     hostname: "127.0.0.1",
@@ -102,10 +101,6 @@ function startRespServer({ refuseSubscribeFromConnection = Infinity }: ServerOpt
         const connection: Connection = { socket, commands: [], channels: new Set(), pending: Buffer.alloc(0) };
         connections.push(connection);
         socket.data = { buffer: Buffer.alloc(0), connection };
-        for (const waiter of waiters.splice(0)) {
-          if (connections.length >= waiter.count) waiter.resolve();
-          else waiters.push(waiter);
-        }
       },
       close(socket) {
         if (socket.data) socket.data.connection.socket = null;
@@ -189,11 +184,6 @@ function startRespServer({ refuseSubscribeFromConnection = Infinity }: ServerOpt
   return {
     connections,
     url: `redis://127.0.0.1:${server.port}`,
-    /** Resolves once the server has accepted at least `count` connections. */
-    waitForConnections(count: number): Promise<void> {
-      if (connections.length >= count) return Promise.resolve();
-      return new Promise<void>(resolve => waiters.push({ count, resolve }));
-    },
     [Symbol.dispose]() {
       server.stop(true);
     },
@@ -233,9 +223,22 @@ async function closeSubscriber(client: RedisClient) {
   client.close();
 }
 
-// Deliberately not `describe.concurrent`. Each case drives a client through a
-// real reconnect, and six of those at once starved the 10s connection timeout on
-// the Windows CI agents. Serial, the whole file is under 4s.
+/**
+ * Force a reconnect deterministically. An explicit close()/connect() drives the
+ * same path as an auto-reconnect (on_valkey_connect -> resubscribe), but the
+ * connect() promise resolves only once the new HELLO handshake completes, so
+ * there is no race against the client's internal connection timeout. Dropping
+ * the socket server-side and waiting on the backoff-timer reconnect instead
+ * flaked on the loaded Windows CI agents, where that timer plus the handshake
+ * could exceed the 10s timeout.
+ */
+async function reconnect(client: RedisClient) {
+  client.close();
+  await client.connect();
+}
+
+// Serial on purpose: six clients reconnecting at once flaked on the Windows CI
+// agents. The whole file runs in a few seconds.
 describe("Valkey: subscriber reconnect", () => {
   test("replays SUBSCRIBE for every channel with a listener", async () => {
     using server = startRespServer();
@@ -252,9 +255,7 @@ describe("Valkey: subscriber reconnect", () => {
       expect(await publisher.publish("news", "before")).toBe(1);
       expect(await messages.next()).toBe("before");
 
-      // Drop the subscriber's connection from the server side.
-      server.connections[0].socket!.end();
-      await server.waitForConnections(3);
+      await reconnect(subscriber);
 
       // PING is written after the reconnect handshake, so its reply proves the
       // server has already seen everything else the client sent on the new
@@ -285,8 +286,7 @@ describe("Valkey: subscriber reconnect", () => {
       await subscriber.subscribe(["news", "sports"], message => messages.push(message));
       await subscriber.unsubscribe("sports");
 
-      server.connections[0].socket!.end();
-      await server.waitForConnections(2);
+      await reconnect(subscriber);
       expect(await subscriber.ping()).toBe("PONG");
 
       expect(commandLines(server.connections[1])).toEqual(["HELLO 3", "SUBSCRIBE news", "PING"]);
@@ -304,8 +304,7 @@ describe("Valkey: subscriber reconnect", () => {
       await client.connect();
       expect(await client.ping()).toBe("PONG");
 
-      server.connections[0].socket!.end();
-      await server.waitForConnections(2);
+      await reconnect(client);
       expect(await client.ping()).toBe("PONG");
 
       expect(commandLines(server.connections[1])).toEqual(["HELLO 3", "PING"]);
@@ -342,8 +341,7 @@ describe("Valkey: subscriber reconnect", () => {
       await subscriber.connect();
       await subscriber.subscribe("news", () => {});
 
-      server.connections[0].socket!.end();
-      await server.waitForConnections(2);
+      await reconnect(subscriber);
 
       // The replayed SUBSCRIBE carries no promise, so its -NOPERM must not consume
       // the pair PING parked in the in-flight queue behind it. It used to, and the
@@ -384,8 +382,7 @@ describe("Valkey: subscriber reconnect", () => {
       expect(commandLines(server.connections[0])).toEqual(["HELLO 3", "SELECT 3", "SUBSCRIBE news"]);
 
       await publisher.connect();
-      server.connections[0].socket!.end();
-      await server.waitForConnections(3);
+      await reconnect(subscriber);
       expect(await subscriber.ping()).toBe("PONG");
 
       expect(commandLines(server.connections[2])).toEqual(["HELLO 3", "SELECT 3", "SUBSCRIBE news", "PING"]);

From 11855d4a5e121a4d725fa1fdc71bec1c62ecd67e Mon Sep 17 00:00:00 2001
From: robobun <117481402+robobun@users.noreply.github.com>
Date: Tue, 7 Jul 2026 03:40:52 +0000
Subject: [PATCH 11/11] ci: retrigger