diff --git a/docs/runtime/redis.mdx b/docs/runtime/redis.mdx index 5510173205b4..d5f92c5c4bcc 100644 --- a/docs/runtime/redis.mdx +++ b/docs/runtime/redis.mdx @@ -417,6 +417,9 @@ 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. + Patterns sent with `psubscribe` are not replayed --- 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 { + 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..d28713b45f84 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 { + 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 { @@ -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, @@ -1301,6 +1308,58 @@ impl JSValkeyClient { self.client_mut().flags.connection_promise_returns_client = false; } } + + // 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. + /// + /// 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 = 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(()) } diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs index b909bff16982..2103bfcc0b1f 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. @@ -1099,6 +1103,14 @@ impl ValkeyClient { let mut should_consume_promise_pair = true; let mut pair_maybe: Option = 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) { @@ -1106,6 +1118,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 +1303,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..f4bca06da520 --- /dev/null +++ b/test/js/valkey/reliability/subscriber-reconnect.test.ts @@ -0,0 +1,397 @@ +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: Buffer): number; end(): void } | null; + commands: string[][]; + channels: Set; + /** 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[][] = []; + 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) }; +} + +type ServerOptions = { + /** Refuse SUBSCRIBE with -NOPERM on every connection from this index onwards. */ + refuseSubscribeFromConnection?: number; +}; + +function startRespServer({ refuseSubscribeFromConnection = Infinity }: ServerOptions = {}) { + const connections: Connection[] = []; + + 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(), pending: Buffer.alloc(0) }; + connections.push(connection); + socket.data = { buffer: Buffer.alloc(0), connection }; + }, + 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; + 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": + 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) { + 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); + send( + state.connection, + `>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]; + // 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) { + send(state.connection, `>3${CRLF}` + bulk("unsubscribe") + `_${CRLF}` + `:0${CRLF}`); + break; + } + for (const channel of channels) { + state.connection.channels.delete(channel); + send( + state.connection, + `>3${CRLF}` + bulk("unsubscribe") + bulk(channel) + `:${state.connection.channels.size}${CRLF}`, + ); + } + break; + } + + case "PING": + send(state.connection, `+PONG${CRLF}`); + break; + + case "PUBLISH": { + let receivers = 0; + for (const target of connections) { + if (target.socket && target.channels.has(command[1])) { + receivers++; + send(target, `>3${CRLF}` + bulk("message") + bulk(command[1]) + bulk(command[2])); + } + } + send(state.connection, `:${receivers}${CRLF}`); + break; + } + + default: + send(state.connection, `+OK${CRLF}`); + } + } + }, + }, + }); + + return { + connections, + url: `redis://127.0.0.1:${server.port}`, + [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 { + const message = buffered.shift(); + if (message !== undefined) return Promise.resolve(message); + return new Promise(resolve => waiters.push(resolve)); + }, + }; +} + +const commandLines = (connection: Connection) => connection.commands.map(command => command.join(" ")); + +/** + * 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(); +} + +/** + * 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(); + 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"); + + 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 + // 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 { + await closeSubscriber(subscriber); + 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"); + + await reconnect(subscriber); + expect(await subscriber.ping()).toBe("PONG"); + + expect(commandLines(server.connections[1])).toEqual(["HELLO 3", "SUBSCRIBE news", "PING"]); + } finally { + await closeSubscriber(subscriber); + } + }); + + 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"); + + await reconnect(client); + expect(await client.ping()).toBe("PONG"); + + expect(commandLines(server.connections[1])).toEqual(["HELLO 3", "PING"]); + } finally { + client.close(); + } + }); + + 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 }); + + const subscriber = new RedisClient(server.url, { autoReconnect: true, maxRetries: 10 }); + + try { + await subscriber.connect(); + await subscriber.subscribe("news", () => {}); + + 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 + // pair was then dropped unsettled, hanging this await forever. + // + // 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); + } + }); + + 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(); + await reconnect(subscriber); + 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 { + await closeSubscriber(subscriber); + publisher.close(); + } + }); +});