Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion src/runtime/valkey_jsc/js_valkey.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
63 changes: 63 additions & 0 deletions test/js/valkey/reliability/resp-nesting-depth.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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");
});
});
});
29 changes: 29 additions & 0 deletions test/js/valkey/valkey.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";

Expand Down
Loading