diff --git a/docs/runtime/redis.mdx b/docs/runtime/redis.mdx index 628fb996ffd0..a88a1de42930 100644 --- a/docs/runtime/redis.mdx +++ b/docs/runtime/redis.mdx @@ -367,8 +367,10 @@ The following commands disable automatic pipelining: - `PIPELINE` - `SUBSCRIBE` - `PSUBSCRIBE` +- `SSUBSCRIBE` - `UNSUBSCRIBE` -- `UNPSUBSCRIBE` +- `PUNSUBSCRIBE` +- `SUNSUBSCRIBE` --- diff --git a/src/runtime/valkey_jsc/ValkeyCommand.rs b/src/runtime/valkey_jsc/ValkeyCommand.rs index abcc35864f62..3dcf94ae2b6b 100644 --- a/src/runtime/valkey_jsc/ValkeyCommand.rs +++ b/src/runtime/valkey_jsc/ValkeyCommand.rs @@ -150,7 +150,9 @@ impl Default for Meta { } bun_core::comptime_string_set! { - /// Commands that must not be auto-pipelined. + /// Commands that must not be auto-pipelined. Subscription commands are + /// not listed here: `Meta::check` excludes everything + /// `is_subscription_command` matches. static AUTO_PIPELINE_DISALLOWED_COMMANDS = { b"AUTH", b"EXEC", @@ -164,10 +166,6 @@ bun_core::comptime_string_set! { b"DISCARD", b"UNWATCH", b"PIPELINE", - b"SUBSCRIBE", - b"PSUBSCRIBE", - b"UNSUBSCRIBE", - b"UNPSUBSCRIBE", }; } @@ -181,6 +179,9 @@ impl Meta { if is_subscription_command(command.command) { new.insert(Meta::SUBSCRIPTION_REQUEST); } + if new.contains(Meta::SUBSCRIPTION_REQUEST) { + new.remove(Meta::SUPPORTS_AUTO_PIPELINING); + } new } } diff --git a/test/js/valkey/valkey-auto-pipelining.test.ts b/test/js/valkey/valkey-auto-pipelining.test.ts new file mode 100644 index 000000000000..04072d2b3504 --- /dev/null +++ b/test/js/valkey/valkey-auto-pipelining.test.ts @@ -0,0 +1,139 @@ +import { RedisClient, type TCPSocketListener } from "bun"; +import { describe, expect, test } from "bun:test"; + +const CRLF = "\r\n"; +const bulk = (s: string) => `$${Buffer.byteLength(s)}${CRLF}${s}${CRLF}`; +// Minimal RESP3 HELLO reply so the client enters the Connected state. +const HELLO = `%1${CRLF}${bulk("proto")}:3${CRLF}`; + +const SUBSCRIPTION_COMMANDS = [ + "SUBSCRIBE", + "PSUBSCRIBE", + "SSUBSCRIBE", + "UNSUBSCRIBE", + "PUNSUBSCRIBE", + "SUNSUBSCRIBE", +] as const; + +/** Split `buf` into the complete `*N\r\n($len\r\n...\r\n){N}` command frames it holds. */ +function parseFrames(buf: Buffer): { frames: string[][]; rest: Buffer } { + const frames: string[][] = []; + for (;;) { + if (buf.length === 0 || buf[0] !== 0x2a /* '*' */) break; + const headerEnd = buf.indexOf(CRLF); + if (headerEnd < 0) break; + const argc = parseInt(buf.subarray(1, headerEnd).toString("latin1"), 10); + let pos = headerEnd + 2; + const fields: string[] = []; + for (let i = 0; i < argc; i++) { + const lenEnd = buf.indexOf(CRLF, pos); + if (lenEnd < 0 || buf[pos] !== 0x24 /* '$' */) return { frames, rest: buf }; + const len = parseInt(buf.subarray(pos + 1, lenEnd).toString("latin1"), 10); + const next = lenEnd + 2 + len + 2; + if (next > buf.length) return { frames, rest: buf }; + fields.push(buf.subarray(lenEnd + 2, lenEnd + 2 + len).toString("latin1")); + pos = next; + } + frames.push(fields); + buf = buf.subarray(pos); + } + return { frames, rest: buf }; +} + +function reply([command, target = ""]: string[]): string { + const upper = command.toUpperCase(); + if (upper === "HELLO") return HELLO; + if (upper === "PING") return `+PONG${CRLF}`; + if ((SUBSCRIPTION_COMMANDS as readonly string[]).includes(upper)) { + // Servers confirm each (un)subscribe with a push frame named after the + // command: >3 . + const count = upper.includes("UNSUBSCRIBE") ? 0 : 1; + return `>3${CRLF}${bulk(command.toLowerCase())}${bulk(target)}:${count}${CRLF}`; + } + return `+OK${CRLF}`; +} + +/** + * Mock server that logs the order in which commands arrive (`> CMD`) and + * replies go out (`< CMD`). Every complete frame in a socket read is logged + * before any of them is answered, so two commands the client flushed as one + * batch show up as `> A`, `> B`, `< A`, `< B`, while a command the client held + * back until A's reply arrived shows up as `> A`, `< A`, `> B`, `< B`. + */ +function listen(log: string[]): TCPSocketListener { + return Bun.listen({ + hostname: "127.0.0.1", + port: 0, + socket: { + open(socket) { + socket.data = Buffer.alloc(0); + }, + error() {}, + close() {}, + data(socket, chunk) { + const { frames, rest } = parseFrames(Buffer.concat([socket.data, chunk])); + socket.data = rest; + for (const frame of frames) log.push(`> ${frame[0]}`); + for (const frame of frames) { + socket.write(reply(frame)); + log.push(`< ${frame[0]}`); + } + }, + }, + }); +} + +/** + * Connects a client, puts it in subscriber mode (the state an unsubscribe is + * normally issued from), then issues a pipelinable PING and `issue` in the + * same tick. Returns the server's log for those two commands and whatever + * `issue` resolved with. + */ +async function issueBehindPing(issue: (client: RedisClient) => Promise) { + const log: string[] = []; + const server = listen(log); + const client = new RedisClient(`redis://127.0.0.1:${server.port}`, { autoReconnect: false }); + client.onconnect = client.onclose = () => {}; + try { + await client.connect(); + expect(await client.psubscribe("setup.*")).toEqual({ type: "psubscribe", data: ["setup.*", 1] }); + expect(log).toEqual(["> HELLO", "< HELLO", "> PSUBSCRIBE", "< PSUBSCRIBE"]); + log.length = 0; + + const [pong, acked] = await Promise.all([client.send("PING", []), issue(client)]); + expect(pong).toBe("PONG"); + return { log, acked }; + } finally { + client.close(); + server.stop(true); + } +} + +describe.concurrent("RedisClient auto-pipelining", () => { + test("pipelinable commands issued in the same tick are flushed as one batch", async () => { + const { log, acked } = await issueBehindPing(client => client.send("PING", [])); + expect(log).toEqual(["> PING", "> PING", "< PING", "< PING"]); + expect(acked).toBe("PONG"); + }); + + test("punsubscribe() is held back until the commands ahead of it are answered", async () => { + const { log, acked } = await issueBehindPing(client => client.punsubscribe("news.*")); + expect(log).toEqual(["> PING", "< PING", "> PUNSUBSCRIBE", "< PUNSUBSCRIBE"]); + expect(acked).toEqual({ type: "punsubscribe", data: ["news.*", 0] }); + }); + + test("psubscribe() is held back until the commands ahead of it are answered", async () => { + const { log, acked } = await issueBehindPing(client => client.psubscribe("news.*")); + expect(log).toEqual(["> PING", "< PING", "> PSUBSCRIBE", "< PSUBSCRIBE"]); + expect(acked).toEqual({ type: "psubscribe", data: ["news.*", 1] }); + }); + + // send() forwards the command name as spelled by the caller, and the server + // accepts any casing, so the client has to recognize every spelling. + const rawSpellings = SUBSCRIPTION_COMMANDS.flatMap(command => [command, command.toLowerCase()]); + + test.each(rawSpellings)('send("%s") is held back until the commands ahead of it are answered', async command => { + const { log } = await issueBehindPing(client => client.send(command, ["target"])); + expect(log).toEqual(["> PING", "< PING", `> ${command}`, `< ${command}`]); + }); +});