diff --git a/src/runtime/valkey_jsc/js_valkey.rs b/src/runtime/valkey_jsc/js_valkey.rs index b2a479cb66e2..3dab5b6441a0 100644 --- a/src/runtime/valkey_jsc/js_valkey.rs +++ b/src/runtime/valkey_jsc/js_valkey.rs @@ -1317,7 +1317,6 @@ impl JSValkeyClient { } pub(crate) fn on_valkey_unsubscribe(&self) -> JsResult<()> { - debug_assert!(self.is_subscriber()); debug_assert!(self.this_value.get().is_strong()); self.client_mut().on_writable(); diff --git a/test/js/valkey/reliability/resp-nesting-depth.test.ts b/test/js/valkey/reliability/resp-nesting-depth.test.ts index 600f5005ffae..de8e44e32fd9 100644 --- a/test/js/valkey/reliability/resp-nesting-depth.test.ts +++ b/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(payloads: Buffer[], body: (client: Bun.RedisClient) => Promise): Promise { + 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"); + }); + }); }); diff --git a/test/js/valkey/valkey.test.ts b/test/js/valkey/valkey.test.ts index 22c1a30bbd32..2cab073eb244 100644 --- a/test/js/valkey/valkey.test.ts +++ b/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";