redis: accept unsubscribe acks that arrive outside subscriber mode

src/runtime/valkey_jsc/js_valkey.rs

@@ -1317,7 +1317,10 @@ impl JSValkeyClient {
     }
 
     pub(crate) fn on_valkey_unsubscribe(&self) -> JsResult<()> {
-        debug_assert!(self.is_subscriber());
+        // `is_subscriber()` may already be false here: `unsubscribe()` drops the
+        // handlers when called, so with several UNSUBSCRIBEs in flight the first
+        // ack leaves subscriber mode and the rest are acked afterwards. The same
+        // goes for `punsubscribe()` / a raw UNSUBSCRIBE from a non-subscriber.
         debug_assert!(self.this_value.get().is_strong());
 
         self.client_mut().on_writable();

test/js/valkey/reliability/resp-nesting-depth.test.ts

@@ -357,4 +357,67 @@ describe("Valkey: RESP push frame routing", () => {
       server.close();
     }
   });
+
+  // The acks below reach the client while it is not (or no longer) in
+  // subscriber mode. The ack handler used to assert that it was, which
+  // aborted debug and ASAN builds (release builds compile the check out).
+  const bulk = (s: string) => `$${Buffer.byteLength(s)}\r\n${s}\r\n`;
+  const pushAck = (kind: string, subject: string | null, remaining: number) =>
+    Buffer.from(`>3\r\n${bulk(kind)}${subject === null ? "_\r\n" : bulk(subject)}:${remaining}\r\n`);
+  const OK = Buffer.from("+OK\r\n");
+
+  async function withMockClient<T>(payloads: Buffer[], body: (client: Bun.RedisClient) => Promise<T>): Promise<T> {
+    const { server, port } = await createMockRedisServer(payloads);
+    try {
+      const client = new Bun.RedisClient(`redis://127.0.0.1:${port}`, {
+        autoReconnect: false,
+        connectionTimeout: 2000,
+      });
+      try {
+        return await body(client);
+      } finally {
+        client.close();
+      }
+    } finally {
+      server.close();
+    }
+  }
+
+  test("acks for several in-flight UNSUBSCRIBEs all resolve after the first one leaves subscriber mode", async () => {
+    // unsubscribe() drops the channel's handlers when it is called, so by the
+    // time the first ack arrives the handler map is already empty and the
+    // client leaves subscriber mode; the second UNSUBSCRIBE is acked afterwards.
+    const payloads = [
+      pushAck("subscribe", "a", 1),
+      pushAck("subscribe", "b", 2),
+      pushAck("unsubscribe", "a", 1),
+      pushAck("unsubscribe", "b", 0),
+      OK,
+    ];
+    await withMockClient(payloads, async client => {
+      const noop = () => {};
+      await client.subscribe("a", noop);
+      await client.subscribe("b", noop);
+
+      expect(await Promise.all([client.unsubscribe("a"), client.unsubscribe("b")])).toEqual([undefined, undefined]);
+      // set() throws synchronously while in subscriber mode, so this also
+      // checks that the client is back in normal command mode.
+      expect(await client.set("key", "value")).toBe("OK");
+    });
+  });
+
+  test("punsubscribe() from a client that never subscribed resolves with the ack", async () => {
+    await withMockClient([pushAck("punsubscribe", "news.*", 0), OK], async client => {
+      expect(await client.punsubscribe("news.*")).toEqual({ type: "punsubscribe", data: ["news.*", 0] });
+      expect(await client.set("key", "value")).toBe("OK");
+    });
+  });
+
+  test("a raw UNSUBSCRIBE sent through send() from a client that never subscribed resolves", async () => {
+    // With nothing to unsubscribe from, the server acks with a null channel.
+    await withMockClient([pushAck("unsubscribe", null, 0), OK], async client => {
+      expect(await client.send("UNSUBSCRIBE", [])).toBeUndefined();
+      expect(await client.set("key", "value")).toBe("OK");
+    });
+  });
 });

test/js/valkey/valkey.test.ts

@@ -6635,6 +6635,35 @@ for (const connectionType of [ConnectionType.TLS, ConnectionType.TCP]) {
         expect(value).toBe("value");
       });
 
+      test("overlapping unsubscribe() calls all resolve and restore normal command mode", async () => {
+        const channel1 = testChannel();
+        const channel2 = testChannel();
+
+        const subscriber = await ctx.newSubscriberClient(connectionType);
+        await subscriber.subscribe(channel1, () => {});
+        await subscriber.subscribe(channel2, () => {});
+
+        // Both UNSUBSCRIBEs are issued before either ack arrives. The first ack
+        // already takes the client out of subscriber mode, and the second ack
+        // used to trip an assertion in debug builds instead of resolving.
+        expect(await Promise.all([subscriber.unsubscribe(channel1), subscriber.unsubscribe(channel2)])).toEqual([
+          undefined,
+          undefined,
+        ]);
+
+        expect(await ctx.redis.publish(channel1, testMessage())).toBe(0);
+        expect(await ctx.redis.publish(channel2, testMessage())).toBe(0);
+        expect(await subscriber.set(testKey(), testValue())).toBe("OK");
+      });
+
+      test("punsubscribe() from a client that is not in subscriber mode resolves", async () => {
+        const pattern = `${testChannel()}*`;
+        const client = await ctx.newSubscriberClient(connectionType);
+
+        expect(await client.punsubscribe(pattern)).toEqual({ type: "punsubscribe", data: [pattern, 0] });
+        expect(await client.set(testKey(), testValue())).toBe("OK");
+      });
+
       test("publishing without subscribers succeeds", async () => {
         const channel = "no-subscribers-channel";