From db3de576dc6c57cedf450219be6ac43fc4f3ffa5 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 5 Jul 2026 09:58:18 +0000 Subject: [PATCH] redis: fix reply-desynchronization bugs that delivered the wrong value to the wrong command A command's promise must only ever be fulfilled with that command's own reply. Several independent paths in the Valkey client broke this. - SUBSCRIBE with N > 1 channels produces N confirmation pushes but only occupies one in-flight slot; confirmations 2..N consumed the slots of whatever unrelated commands followed (ping() and get() both resolved with the subscription count). - In subscriber mode a '-ERR' reply to any command (e.g. -LOADING to a PING) was routed through fail(), which rejected every other pending command while leaving the erroring command's own promise unsettled. - The pattern and sharded push kinds (psubscribe, ssubscribe, pmessage, smessage, punsubscribe, sunsubscribe) were not recognized as subscription confirmations, so a PSUBSCRIBE/SSUBSCRIBE resolved with the raw RESP push frame instead of the subscribed-channel count. - The raw client.send("SUBSCRIBE", [...]) escape hatch built its meta via Meta::check, which never derived the subscription flags, so its confirmation could not be paired against its own command. - An argless UNSUBSCRIBE emits one confirmation per subscribed channel; with a single undirected SUBSCRIPTION_REQUEST flag those extras could be charged to a following SUBSCRIBE drained into the in-flight queue. Push frames are now routed out-of-band unconditionally: the '>' type byte alone makes them unsolicited. SUBSCRIPTION_REQUEST is split into SUBSCRIBE_REQUEST and UNSUBSCRIBE_REQUEST so a confirmation only counts against an in-flight head of the matching direction, and each such request tracks how many confirmations it is owed. Meta::check derives those flags from the command name (case-insensitively, covering the plain, pattern and sharded families) so every entry point pairs the same way, and clears auto-pipelining for any casing that matches. An error reply in subscriber mode falls through to the same per-command rejection path as non-subscriber mode. on_close also drops the read buffer and reply scanner, which belong to the dead socket. Out-of-band push routing and in-flight rejection on reconnect were fixed independently in #33072; this keeps their behavior and replaces the `is_reply_kind` prefix heuristic with explicit push-kind map entries, which also covers the message kinds it did not. `psubscribe()` now resolves with the subscribed-channel count like `subscribe()` does, rather than the raw push frame; the assertion in resp-nesting-depth.test.ts is updated accordingly. Tested with a byte-scripted in-process RESP3 server so each case is deterministic and needs no real Redis. --- src/runtime/valkey_jsc/ValkeyCommand.rs | 59 ++- src/runtime/valkey_jsc/js_valkey_functions.rs | 25 +- src/runtime/valkey_jsc/valkey.rs | 227 ++++----- src/valkey/valkey_protocol.rs | 22 +- .../reliability/resp-nesting-depth.test.ts | 5 +- test/js/valkey/valkey-reply-desync.test.ts | 450 ++++++++++++++++++ 6 files changed, 622 insertions(+), 166 deletions(-) create mode 100644 test/js/valkey/valkey-reply-desync.test.ts diff --git a/src/runtime/valkey_jsc/ValkeyCommand.rs b/src/runtime/valkey_jsc/ValkeyCommand.rs index 9ac7dc39b674..891033cb3e87 100644 --- a/src/runtime/valkey_jsc/ValkeyCommand.rs +++ b/src/runtime/valkey_jsc/ValkeyCommand.rs @@ -102,12 +102,27 @@ impl<'a> Command<'a> { self.write(&mut buf)?; Ok(buf.into_boxed_slice()) } + + /// Number of top-level replies this command produces. + /// + /// Almost every Redis command produces exactly one reply. The exception + /// is `(P)SUBSCRIBE` / `(P)UNSUBSCRIBE`, which emit one confirmation push + /// per channel argument (RESP spec and `redis/src/pubsub.c`). + pub fn expected_reply_count(&self) -> u32 { + if self.meta.intersects(Meta::SUBSCRIPTION_REQUEST) { + u32::try_from(self.args.len()).unwrap_or(u32::MAX).max(1) + } else { + 1 + } + } } /// Command stored in offline queue when disconnected pub struct Entry { pub serialized_data: Box<[u8]>, // Pre-serialized RESP protocol bytes pub meta: Meta, + /// See [`PromisePair::remaining_replies`]. + pub remaining_replies: u32, pub promise: Promise, } @@ -125,6 +140,7 @@ impl Entry { // We should be calling .check against command here but due // to a hack introduced to let SUBSCRIBE work, we are not doing that for now. meta: command.meta, + remaining_replies: command.expected_reply_count(), promise, }) } @@ -134,11 +150,17 @@ bitflags::bitflags! { #[repr(transparent)] #[derive(Clone, Copy, PartialEq, Eq)] pub struct Meta: u8 { - const RETURN_AS_BOOL = 1 << 0; + const RETURN_AS_BOOL = 1 << 0; const SUPPORTS_AUTO_PIPELINING = 1 << 1; - const RETURN_AS_BUFFER = 1 << 2; - const SUBSCRIPTION_REQUEST = 1 << 3; - // bits 4..8 are padding + const RETURN_AS_BUFFER = 1 << 2; + /// Set on `(P)SUBSCRIBE`: reply is one `(p)subscribe` push per arg. + const SUBSCRIBE_REQUEST = 1 << 3; + /// Set on `(P)UNSUBSCRIBE`: reply is one `(p)unsubscribe` push per arg. + const UNSUBSCRIBE_REQUEST = 1 << 4; + /// Either direction of a subscription-state command. + const SUBSCRIPTION_REQUEST = Self::SUBSCRIBE_REQUEST.bits() + | Self::UNSUBSCRIBE_REQUEST.bits(); + // bits 5..8 are padding } } @@ -166,8 +188,10 @@ bun_core::comptime_string_set! { b"PIPELINE", b"SUBSCRIBE", b"PSUBSCRIBE", + b"SSUBSCRIBE", b"UNSUBSCRIBE", - b"UNPSUBSCRIBE", + b"PUNSUBSCRIBE", + b"SUNSUBSCRIBE", }; } @@ -178,6 +202,27 @@ impl Meta { Meta::SUPPORTS_AUTO_PIPELINING, !AUTO_PIPELINE_DISALLOWED_COMMANDS.contains(command.command), ); + // Derive subscription flags from the command name so the raw + // `client.send("SUBSCRIBE", [...])` escape hatch still pairs its + // confirmation pushes correctly. + let name = command.command; + if name.eq_ignore_ascii_case(b"SUBSCRIBE") + || name.eq_ignore_ascii_case(b"PSUBSCRIBE") + || name.eq_ignore_ascii_case(b"SSUBSCRIBE") + { + new |= Meta::SUBSCRIBE_REQUEST; + } else if name.eq_ignore_ascii_case(b"UNSUBSCRIBE") + || name.eq_ignore_ascii_case(b"PUNSUBSCRIBE") + || name.eq_ignore_ascii_case(b"SUNSUBSCRIBE") + { + new |= Meta::UNSUBSCRIBE_REQUEST; + } + // Subscription commands are in `AUTO_PIPELINE_DISALLOWED_COMMANDS`, + // but that lookup is case-sensitive; enforce the same for any + // casing that matched above. + if new.intersects(Meta::SUBSCRIPTION_REQUEST) { + new.remove(Meta::SUPPORTS_AUTO_PIPELINING); + } new } } @@ -227,6 +272,10 @@ impl Promise { // Command+Promise pair for tracking which command corresponds to which promise pub struct PromisePair { pub meta: Meta, + /// Number of further top-level replies the server will produce for + /// this command before its promise may be settled. Always 1 except for + /// `(P)SUBSCRIBE` / `(P)UNSUBSCRIBE`, which emit one push per channel. + pub remaining_replies: u32, pub promise: Promise, } diff --git a/src/runtime/valkey_jsc/js_valkey_functions.rs b/src/runtime/valkey_jsc/js_valkey_functions.rs index 5789403e5d6f..2a522f3cc3b1 100644 --- a/src/runtime/valkey_jsc/js_valkey_functions.rs +++ b/src/runtime/valkey_jsc/js_valkey_functions.rs @@ -358,6 +358,9 @@ macro_rules! cmd_key_value_value2 { macro_rules! cmd_strings_varargs { ($fn_name:ident, $name:literal, $command:literal, $state:ident) => { + cmd_strings_varargs!($fn_name, $name, $command, $state, CommandMeta::default()); + }; + ($fn_name:ident, $name:literal, $command:literal, $state:ident, $meta:expr) => { #[bun_jsc::host_fn(method)] pub fn $fn_name( this: &Self, @@ -386,7 +389,7 @@ macro_rules! cmd_strings_varargs { frame.this(), $command.as_bytes(), CommandArgs::Args(&args), - CommandMeta::default(), + $meta, concat!("Failed to send ", $command), ) } @@ -1595,8 +1598,20 @@ impl JSValkeyClient { NotSubscriber ); cmd_key_varargs!(zrevrank, b"zrevrank", "ZREVRANK", "key", NotSubscriber); - cmd_strings_varargs!(psubscribe, b"psubscribe", "PSUBSCRIBE", DontCare); - cmd_strings_varargs!(punsubscribe, b"punsubscribe", "PUNSUBSCRIBE", DontCare); + cmd_strings_varargs!( + psubscribe, + b"psubscribe", + "PSUBSCRIBE", + DontCare, + CommandMeta::default() | CommandMeta::SUBSCRIBE_REQUEST + ); + cmd_strings_varargs!( + punsubscribe, + b"punsubscribe", + "PUNSUBSCRIBE", + DontCare, + CommandMeta::default() | CommandMeta::UNSUBSCRIBE_REQUEST + ); cmd_strings_varargs!(pubsub, b"pubsub", "PUBSUB", DontCare); cmd_strings_varargs!(copy, b"copy", "COPY", NotSubscriber); cmd_key_varargs!(unlink, b"unlink", "UNLINK", "key", NotSubscriber); @@ -1707,7 +1722,7 @@ impl JSValkeyClient { let command = Command { command: b"SUBSCRIBE", args: CommandArgs::Args(&redis_channels), - meta: CommandMeta::default() | CommandMeta::SUBSCRIPTION_REQUEST, + meta: CommandMeta::default() | CommandMeta::SUBSCRIBE_REQUEST, }; let promise = match this.send(global, frame.this(), &command) { Ok(p) => p, @@ -1738,7 +1753,7 @@ impl JSValkeyClient { this_js, b"UNSUBSCRIBE", CommandArgs::Args(redis_channels), - CommandMeta::default(), + CommandMeta::default() | CommandMeta::UNSUBSCRIBE_REQUEST, "Failed to send UNSUBSCRIBE command", ) } diff --git a/src/runtime/valkey_jsc/valkey.rs b/src/runtime/valkey_jsc/valkey.rs index b909bff16982..9b63bdf9c731 100644 --- a/src/runtime/valkey_jsc/valkey.rs +++ b/src/runtime/valkey_jsc/valkey.rs @@ -292,12 +292,6 @@ pub struct ValkeyClient { pub vm: &'static VirtualMachine, } -#[derive(Copy, Clone, Eq, PartialEq)] -enum SubscribeHandled { - Handled, - Fallthrough, -} - pub(crate) struct DeferredFailure { message: Box<[u8]>, err: RedisError, @@ -447,6 +441,7 @@ impl ValkeyClient { self.in_flight .write_item(command::PromisePair { meta: cmd.meta, + remaining_replies: cmd.remaining_replies, promise: cmd.promise, }) .unwrap_or_oom(); @@ -665,6 +660,9 @@ impl ValkeyClient { pub fn on_close(&mut self) -> JsTerminated<()> { self.unregister_auto_flusher(); self.write_buffer.clear_and_free(); + // Any partially buffered reply belongs to the dead socket. + self.read_buffer.clear_and_free(); + self.reply_scanner.reset(); // If manually closing, don't attempt to reconnect if self.flags.is_manually_closed { @@ -898,14 +896,13 @@ impl ValkeyClient { Ok(()) } - /// Try handling this response as a subscriber-state response. - /// Returns `handled` if we handled it, `fallthrough` if we did not. + /// Dispatch a subscription push (message/subscribe/unsubscribe) and + /// settle the paired promise if the caller consumed one. fn handle_subscribe_response( &mut self, value: &mut RESPValue, pair: Option<&mut command::PromisePair>, - ) -> JsResult { - // Resolve the promise with the potentially transformed value + ) -> JsResult<()> { let global_this = self.global_object(); debug!("Handling a subscribe response: {}", value); @@ -913,73 +910,54 @@ impl ValkeyClient { // raw pointer (no long-lived `&mut`) and calls `exit()` on drop. let _exit = self.vm.enter_event_loop_scope(); - match value { - RESPValue::Error(_) => { - if let Some(p) = pair { - p.promise - .reject(&global_this, resp_value_to_js(value, &global_this))?; - } - Ok(SubscribeHandled::Handled) - } - RESPValue::Push(push) => { - let p = self.parent(); - let sub_count = p - ._subscription_ctx - .get() - .channels_subscribed_to_count(&global_this)?; - - if let Some(msg_type) = protocol::SubscriptionPushMessage::from_bytes(&push.kind) { - match msg_type { - protocol::SubscriptionPushMessage::Message => { - self.on_valkey_message(&mut push.data); - Ok(SubscribeHandled::Handled) - } - protocol::SubscriptionPushMessage::Subscribe => { - p.add_subscription(); - self.on_valkey_subscribe(value); - - // For SUBSCRIBE responses, only resolve the promise for the first channel confirmation - // Additional channel confirmations from multi-channel SUBSCRIBE commands don't need promise pairs - if let Some(req_pair) = pair { - req_pair.promise.promise.resolve( - &global_this, - JSValue::js_number(f64::from(sub_count)), - )?; - } - Ok(SubscribeHandled::Handled) - } - protocol::SubscriptionPushMessage::Unsubscribe => { - self.on_valkey_unsubscribe()?; - self.parent().remove_subscription(); - - // For UNSUBSCRIBE responses, only resolve the promise if we have one - // Additional channel confirmations from multi-channel UNSUBSCRIBE commands don't need promise pairs - if let Some(req_pair) = pair { - req_pair - .promise - .promise - .resolve(&global_this, JSValue::UNDEFINED)?; - } - Ok(SubscribeHandled::Handled) - } - } + let RESPValue::Push(push) = value else { + return Ok(()); + }; + let Some(msg_type) = protocol::SubscriptionPushMessage::from_bytes(&push.kind) else { + return Ok(()); + }; + + let p = self.parent(); + let sub_count = p + ._subscription_ctx + .get() + .channels_subscribed_to_count(&global_this)?; + + match msg_type { + protocol::SubscriptionPushMessage::Message => { + // `pmessage` payload is [pattern, channel, message]; skip the + // pattern so `on_valkey_message` sees [channel, message]. + let data = if push.kind.as_ref() == b"pmessage" && !push.data.is_empty() { + &mut push.data[1..] } else { - // We should rarely reach this point. If we're guaranteed to be handling a subscribe/unsubscribe, - // then this is an unexpected path. - bun_core::hint::cold(); - self.fail( - b"Push message is not a subscription message.", - RedisError::InvalidResponseType, - )?; - Ok(SubscribeHandled::Handled) + &mut push.data[..] + }; + self.on_valkey_message(data); + } + protocol::SubscriptionPushMessage::Subscribe => { + p.add_subscription(); + self.on_valkey_subscribe(value); + if let Some(req_pair) = pair { + req_pair + .promise + .promise + .resolve(&global_this, JSValue::js_number(f64::from(sub_count)))?; } } - _ => { - // This may be a regular command response. Let's pass it down - // to the next handler. - Ok(SubscribeHandled::Fallthrough) + protocol::SubscriptionPushMessage::Unsubscribe => { + if self.parent().is_subscriber() { + self.on_valkey_unsubscribe()?; + self.parent().remove_subscription(); + } + if let Some(req_pair) = pair { + req_pair + .promise + .promise + .resolve(&global_this, JSValue::UNDEFINED)?; + } } } + Ok(()) } fn handle_hello_response(&mut self, value: &mut RESPValue) -> JsTerminated<()> { @@ -1095,84 +1073,49 @@ impl ValkeyClient { } }; } - // Check if this is a subscription push message that might not need a promise pair - let mut should_consume_promise_pair = true; - let mut pair_maybe: Option = None; - - // For subscription clients, check if this is a push message that doesn't need a promise pair + // RESP3 push frames (`>`) are out-of-band: the type byte alone + // marks them unsolicited, so they never occupy a command's reply + // slot regardless of subscriber mode. if let RESPValue::Push(push) = value { - match protocol::SubscriptionPushMessage::from_bytes(&push.kind) { - Some(protocol::SubscriptionPushMessage::Message) => { - // Message pushes never need promise pairs - should_consume_promise_pair = false; + let msg_type = protocol::SubscriptionPushMessage::from_bytes(&push.kind); + // A `(p|s)subscribe`/`(p|s)unsubscribe` confirmation only counts + // against the in-flight head when that head is the matching + // direction; other push kinds consume no reply slot. + let mut pair_maybe: Option = None; + let head_flag = match msg_type { + Some(protocol::SubscriptionPushMessage::Subscribe) => { + Some(command::Meta::SUBSCRIBE_REQUEST) } - Some( - protocol::SubscriptionPushMessage::Subscribe - | protocol::SubscriptionPushMessage::Unsubscribe, - ) => { - // Subscribe/unsubscribe pushes only need promise pairs if we have pending commands - if self.in_flight.readable_length() == 0 { - should_consume_promise_pair = false; - } + Some(protocol::SubscriptionPushMessage::Unsubscribe) => { + Some(command::Meta::UNSUBSCRIBE_REQUEST) } - None => { - if !protocol::SubscriptionPushMessage::is_reply_kind(&push.kind) { - should_consume_promise_pair = false; + _ => None, + }; + if let Some(flag) = head_flag + && self.in_flight.readable_length() > 0 + { + let head = self.in_flight.peek_item_mut(0); + if head.meta.contains(flag) { + head.remaining_replies = head.remaining_replies.saturating_sub(1); + if head.remaining_replies == 0 { + pair_maybe = self.in_flight.read_item(); } } } - } - - // Only consume promise pair if we determined we need one - // The reaosn we consume pairs is that a SUBSCRIBE message may actually be followed by a number of SUBSCRIBE - // responses which indicate all the channels we have connected to. As a stop-gap, we currently ignore the - // actual of content of the SUBSCRIBE responses and just resolve the first one with the count of channels. - if should_consume_promise_pair { - pair_maybe = self.in_flight.read_item(); - } - - // We handle subscriptions specially because they are not regular commands and their failure will potentially - // cause the client to drop out of subscriber mode. - let request_is_subscribe = pair_maybe - .as_ref() - .map(|p| p.meta.contains(command::Meta::SUBSCRIPTION_REQUEST)) - .unwrap_or(false); - if self.parent().is_subscriber() || request_is_subscribe { - debug!("This client is a subscriber. Handling as subscriber..."); - - match value { - RESPValue::Error(err) => { - self.fail(err, RedisError::InvalidResponse)?; - return Ok(()); - } - RESPValue::Push(push) => { - if protocol::SubscriptionPushMessage::from_bytes(&push.kind).is_some() { - if self.handle_subscribe_response(value, pair_maybe.as_mut())? - == SubscribeHandled::Handled - { - return Ok(()); - } - } else { - bun_core::hint::cold(); - self.fail( - b"Unexpected push message kind without promise", - RedisError::InvalidResponseType, - )?; - return Ok(()); - } - } - _ => { - // In the else case, we fall through to the regular - // handler. Subscribers can send .Push commands which have - // the same semantics as regular commands. - } + if msg_type.is_some() { + self.handle_subscribe_response(value, pair_maybe.as_mut())?; + } else { + // Non-subscription push (e.g. `invalidate`): no command is + // waiting for it; drop so the next real reply pairs correctly. + debug!("Dropping out-of-band push: {}", bstr::BStr::new(&push.kind)); } - - debug!("Treating subscriber response as a regular command..."); + return Ok(()); } - // For regular commands, get the next command+promise pair from the queue - let Some(mut pair) = pair_maybe else { + // For regular (non-push) replies, the next in-flight command owns + // this reply. + let Some(mut pair) = self.in_flight.read_item() else { + debug!("Reply with no in-flight command: {}", value); return Ok(()); }; @@ -1326,6 +1269,7 @@ impl ValkeyClient { self.in_flight .write_item(command::PromisePair { meta: offline_cmd.meta, + remaining_replies: offline_cmd.remaining_replies, promise: offline_cmd.promise, }) .unwrap_or_oom(); @@ -1413,6 +1357,7 @@ impl ValkeyClient { let cmd_pair = command::PromisePair { meta: command.meta, + remaining_replies: command.expected_reply_count(), promise, }; diff --git a/src/valkey/valkey_protocol.rs b/src/valkey/valkey_protocol.rs index b87e084878bc..86d2780c9d62 100644 --- a/src/valkey/valkey_protocol.rs +++ b/src/valkey/valkey_protocol.rs @@ -802,10 +802,18 @@ pub enum SubscriptionPushMessage { } bun_core::comptime_string_map! { + /// Every push kind a subscription command can produce: the plain, pattern + /// (`p`-prefixed) and sharded (`s`-prefixed) variants. static SUBSCRIPTION_PUSH_MESSAGES: SubscriptionPushMessage = { b"message" => SubscriptionPushMessage::Message, + b"pmessage" => SubscriptionPushMessage::Message, + b"smessage" => SubscriptionPushMessage::Message, b"subscribe" => SubscriptionPushMessage::Subscribe, + b"psubscribe" => SubscriptionPushMessage::Subscribe, + b"ssubscribe" => SubscriptionPushMessage::Subscribe, b"unsubscribe" => SubscriptionPushMessage::Unsubscribe, + b"punsubscribe" => SubscriptionPushMessage::Unsubscribe, + b"sunsubscribe" => SubscriptionPushMessage::Unsubscribe, }; } @@ -814,18 +822,4 @@ impl SubscriptionPushMessage { pub fn from_bytes(bytes: &[u8]) -> Option { SUBSCRIPTION_PUSH_MESSAGES.get(bytes).copied() } - - /// Pattern (`p`-prefixed) and sharded (`s`-prefixed) variants of the - /// `Subscribe`/`Unsubscribe` push kinds; the unprefixed kinds are matched by - /// `from_bytes` before this is consulted. - #[inline] - pub fn is_reply_kind(kind: &[u8]) -> bool { - match kind.split_first() { - Some((b'p' | b's', base)) => matches!( - Self::from_bytes(base), - Some(Self::Subscribe | Self::Unsubscribe) - ), - _ => false, - } - } } diff --git a/test/js/valkey/reliability/resp-nesting-depth.test.ts b/test/js/valkey/reliability/resp-nesting-depth.test.ts index 4c757f755240..85aea3860abb 100644 --- a/test/js/valkey/reliability/resp-nesting-depth.test.ts +++ b/test/js/valkey/reliability/resp-nesting-depth.test.ts @@ -278,7 +278,10 @@ describe("Valkey: RESP push frame routing", () => { const psubscribed = client.psubscribe("news.*"); const pinged = client.send("PING", []); - expect(await psubscribed).toEqual({ type: "psubscribe", data: ["news.*", 1] }); + // Like `subscribe()`, a subscription command resolves with the + // subscribed-channel count. `psubscribe` registers no channel + // handler, so that count is 0. + expect(await psubscribed).toBe(0); expect(await pinged).toBe("PONG"); } finally { client.close(); diff --git a/test/js/valkey/valkey-reply-desync.test.ts b/test/js/valkey/valkey-reply-desync.test.ts new file mode 100644 index 000000000000..52d12accc919 --- /dev/null +++ b/test/js/valkey/valkey-reply-desync.test.ts @@ -0,0 +1,450 @@ +import { RedisClient } from "bun"; +import { describe, expect, test } from "bun:test"; +import net from "node:net"; + +// Byte-scripted RESP3 server for reply-integrity tests: a command's +// promise must never resolve with another command's data. + +const HELLO_MAP = "%1\r\n+proto\r\n:3\r\n"; + +type CommandFrame = { name: string; args: string[] }; + +function parseFrames(buffer: string, offset: number): { frames: CommandFrame[]; offset: number } { + const frames: CommandFrame[] = []; + while (offset < buffer.length) { + if (buffer[offset] !== "*") break; + const headerEnd = buffer.indexOf("\r\n", offset); + if (headerEnd === -1) break; + const argc = parseInt(buffer.slice(offset + 1, headerEnd), 10); + if (!Number.isInteger(argc) || argc < 0) break; + let pos = headerEnd + 2; + const args: string[] = []; + let complete = true; + for (let i = 0; i < argc; i++) { + if (buffer[pos] !== "$") { + complete = false; + break; + } + const lenEnd = buffer.indexOf("\r\n", pos); + if (lenEnd === -1) { + complete = false; + break; + } + const len = parseInt(buffer.slice(pos + 1, lenEnd), 10); + if (!Number.isInteger(len) || len < 0) { + complete = false; + break; + } + const next = lenEnd + 2 + len + 2; + if (next > buffer.length) { + complete = false; + break; + } + args.push(buffer.slice(lenEnd + 2, lenEnd + 2 + len)); + pos = next; + } + if (!complete) break; + frames.push({ name: args[0].toUpperCase(), args: args.slice(1) }); + offset = pos; + } + return { frames, offset }; +} + +type Mock = { + server: net.Server; + port: number; + sockets: net.Socket[]; + close: () => void; +}; + +function createMockServer( + onCommand: (frame: CommandFrame, socket: net.Socket, connectionIndex: number) => void, +): Promise { + return new Promise((resolve, reject) => { + const sockets: net.Socket[] = []; + let connectionIndex = 0; + const server = net.createServer(socket => { + const idx = connectionIndex++; + sockets.push(socket); + socket.setNoDelay(true); + socket.on("error", () => {}); + let received = ""; + let parsedOffset = 0; + socket.on("data", data => { + received += data.toString("latin1"); + const parsed = parseFrames(received, parsedOffset); + parsedOffset = parsed.offset; + for (const frame of parsed.frames) { + onCommand(frame, socket, idx); + } + }); + }); + server.listen(0, "127.0.0.1", () => { + const port = (server.address() as net.AddressInfo).port; + resolve({ + server, + port, + sockets, + close: () => { + for (const s of sockets) s.destroy(); + server.close(); + }, + }); + }); + server.on("error", reject); + }); +} + +function bulk(s: string): string { + return `$${s.length}\r\n${s}\r\n`; +} + +function settled(p: Promise): Promise<{ status: "fulfilled"; value: T } | { status: "rejected"; reason: any }> { + return p.then( + value => ({ status: "fulfilled" as const, value }), + reason => ({ status: "rejected" as const, reason }), + ); +} + +describe("Valkey reply/command pairing", () => { + // RESP3 push frames (`>`) are out-of-band: an `invalidate` push on a + // non-subscriber connection must not be delivered as the reply to the + // oldest in-flight command. + test("RESP3 push frame on a non-subscriber connection does not consume a command's reply slot", async () => { + const invalidatePush = ">2\r\n" + bulk("invalidate") + "*1\r\n" + bulk("some-key"); + const mock = await createMockServer((frame, socket) => { + if (frame.name === "HELLO") { + socket.write(HELLO_MAP); + return; + } + if (frame.name === "GET") { + // Interleave an unsolicited push before every GET reply. + socket.write(invalidatePush + bulk("value-of-" + frame.args[0])); + } + }); + const client = new RedisClient(`redis://127.0.0.1:${mock.port}`, { + autoReconnect: false, + connectionTimeout: 5000, + }); + try { + const [k1, k2, k3] = await Promise.all([client.get("k1"), client.get("k2"), client.get("k3")]); + // Each GET must resolve with its own value. When push frames are + // mis-routed as command replies, k1 resolves with the push object + // and k2/k3 shift one slot and receive the previous key's value. + expect({ k1, k2, k3 }).toEqual({ + k1: "value-of-k1", + k2: "value-of-k2", + k3: "value-of-k3", + }); + } finally { + client.close(); + mock.close(); + } + }); + + // Commands sent on a dropped socket must reject; if their slots survive + // the reconnect they pair with the new connection's replies. + // https://github.com/oven-sh/bun/issues/27861 + test("in-flight commands are rejected on disconnect instead of re-pairing with the next connection's replies", async () => { + const seen: Record = { 0: [], 1: [] }; + const conn1Ready = Promise.withResolvers(); + + const mock = await createMockServer((frame, socket, idx) => { + if (frame.name === "HELLO") { + socket.write(HELLO_MAP); + if (idx === 1) conn1Ready.resolve(); + return; + } + seen[idx].push(frame); + if (idx === 0) { + // First connection: accept commands but never reply, then drop. + if (seen[0].length === 2) socket.destroy(); + return; + } + // Second connection: reply to every GET with its own key's value. + if (frame.name === "GET") { + socket.write(bulk("value-of-" + frame.args[0])); + } + }); + + const client = new RedisClient(`redis://127.0.0.1:${mock.port}`, { + autoReconnect: true, + maxRetries: 5, + connectionTimeout: 5000, + }); + try { + await client.connect(); + // Pipeline two GETs on connection 0; the server will drop the socket + // after receiving them without replying. + const staleA = settled(client.get("a")); + const staleB = settled(client.get("b")); + + // After reconnect, issue two fresh commands. + await conn1Ready.promise; + const freshD = settled(client.get("d")); + const freshE = settled(client.get("e")); + + // The stale commands settle in both cases: rejected when the dead + // socket's in-flight queue is cleared (correct), or fulfilled with + // d/e's replies when it isn't (the bug). + const [resA, resB] = await Promise.all([staleA, staleB]); + expect({ a: resA.status, b: resB.status }).toEqual({ a: "rejected", b: "rejected" }); + expect((resA as any).value).toBeUndefined(); + expect((resB as any).value).toBeUndefined(); + + // Fresh commands must receive their own replies. This point is only + // reached once the stale commands have been proven rejected, so the + // fresh replies cannot have been consumed elsewhere. + const [resD, resE] = await Promise.all([freshD, freshE]); + expect({ d: resD, e: resE }).toEqual({ + d: { status: "fulfilled", value: "value-of-d" }, + e: { status: "fulfilled", value: "value-of-e" }, + }); + } finally { + client.close(); + mock.close(); + } + }); + + // SUBSCRIBE with N channels produces N per-channel `subscribe` push + // confirmations. Confirmations 2..N belong to the same SUBSCRIBE command + // and must not consume the reply slots of unrelated in-flight commands. + test("multi-channel SUBSCRIBE confirmations do not steal other in-flight commands' reply slots", async () => { + const afterSubscribe = Promise.withResolvers(); + const mock = await createMockServer((frame, socket) => { + if (frame.name === "HELLO") { + socket.write(HELLO_MAP); + return; + } + if (frame.name === "SUBSCRIBE") { + // One push per channel in the SUBSCRIBE. + let reply = ""; + for (let i = 0; i < frame.args.length; i++) { + reply += ">3\r\n" + bulk("subscribe") + bulk(frame.args[i]) + `:${i + 1}\r\n`; + } + socket.write(reply); + afterSubscribe.resolve(); + return; + } + if (frame.name === "PING") { + socket.write("+PONG\r\n"); + return; + } + if (frame.name === "GET") { + socket.write(bulk("value-of-" + frame.args[0])); + return; + } + }); + const client = new RedisClient(`redis://127.0.0.1:${mock.port}`, { + autoReconnect: false, + connectionTimeout: 5000, + }); + try { + const subP = client.subscribe(["ch-a", "ch-b", "ch-c"], () => {}); + await afterSubscribe.promise; + // These are sent after SUBSCRIBE. Their replies must not be displaced + // by the 2nd/3rd subscribe confirmations. + const pingP = client.ping(); + const getP = client.get("k1"); + + const [sub, ping, got] = await Promise.all([subP, pingP, getP]); + expect({ ping, got }).toEqual({ ping: "PONG", got: "value-of-k1" }); + expect(typeof sub).toBe("number"); + } finally { + client.close(); + mock.close(); + } + }); + + // A `-ERR ...` reply on a subscriber connection is still an ordinary + // per-command error. It rejects exactly the command it answers and leaves + // the rest of the in-flight queue and the connection intact. + test("an error reply in subscriber mode rejects only its own command", async () => { + let pingCount = 0; + const mock = await createMockServer((frame, socket) => { + if (frame.name === "HELLO") { + socket.write(HELLO_MAP); + return; + } + if (frame.name === "SUBSCRIBE") { + socket.write(">3\r\n" + bulk("subscribe") + bulk(frame.args[0]) + ":1\r\n"); + return; + } + if (frame.name === "PING") { + pingCount++; + if (pingCount === 1) { + socket.write("-LOADING Redis is loading the dataset in memory\r\n"); + } else { + socket.write("+PONG\r\n"); + } + return; + } + }); + const client = new RedisClient(`redis://127.0.0.1:${mock.port}`, { + autoReconnect: false, + connectionTimeout: 5000, + }); + try { + await client.subscribe("ch-a", () => {}); + // In subscriber mode only a restricted command set is allowed; PING + // is one of them. Issue two: the server answers the first with a + // transient -LOADING error and the second with +PONG. + const ping1 = settled(client.ping()); + const ping2 = settled(client.ping()); + + // ping2 settles on both paths: fulfilled "PONG" when the error is + // routed to its own command, or rejected when the error is treated as + // a connection failure that sweeps the whole queue. + const ping2Res = await ping2; + expect(ping2Res).toEqual({ status: "fulfilled", value: "PONG" }); + + // Once ping2 is known correct, ping1 has already been handled (its + // reply arrived first) and must have rejected with the server error. + const ping1Res = await ping1; + expect(ping1Res.status).toBe("rejected"); + if (ping1Res.status === "rejected") { + expect(String(ping1Res.reason?.message ?? ping1Res.reason)).toContain("LOADING"); + } + } finally { + client.close(); + mock.close(); + } + }); + + // PSUBSCRIBE/PUNSUBSCRIBE reply with `psubscribe`/`punsubscribe` pushes; + // each must pair with its own slot so the next PING reply isn't stolen. + test("PSUBSCRIBE and PUNSUBSCRIBE confirmations pair with their own reply slots", async () => { + const mock = await createMockServer((frame, socket) => { + if (frame.name === "HELLO") { + socket.write(HELLO_MAP); + return; + } + if (frame.name === "PSUBSCRIBE") { + let reply = ""; + for (let i = 0; i < frame.args.length; i++) { + reply += ">3\r\n" + bulk("psubscribe") + bulk(frame.args[i]) + `:${i + 1}\r\n`; + } + // Follow with an unsolicited pmessage to verify it doesn't + // consume a reply slot either. + reply += ">4\r\n" + bulk("pmessage") + bulk(frame.args[0]) + bulk("news.1") + bulk("hello"); + socket.write(reply); + return; + } + if (frame.name === "PUNSUBSCRIBE") { + let reply = ""; + for (let i = 0; i < frame.args.length; i++) { + reply += ">3\r\n" + bulk("punsubscribe") + bulk(frame.args[i]) + ":0\r\n"; + } + socket.write(reply); + return; + } + if (frame.name === "PING") { + socket.write("+PONG\r\n"); + return; + } + }); + const client = new RedisClient(`redis://127.0.0.1:${mock.port}`, { + autoReconnect: false, + connectionTimeout: 5000, + }); + try { + const sub = await client.psubscribe("news.*"); + expect(typeof sub).toBe("number"); + expect(await client.ping()).toBe("PONG"); + const unsub = await client.punsubscribe("news.*"); + expect(unsub).toBeUndefined(); + expect(await client.ping()).toBe("PONG"); + } finally { + client.close(); + mock.close(); + } + }); + + // The raw `send()` escape hatch must derive subscription flags from the + // command name so its confirmation push pairs with its own reply slot. + test.each([ + ["SUBSCRIBE", "subscribe"], + ["subscribe", "subscribe"], + ["SSUBSCRIBE", "ssubscribe"], + ])("raw send(%p, [...]) pairs its confirmation push with its own reply slot", async (commandName, pushKind) => { + const mock = await createMockServer((frame, socket) => { + if (frame.name === "HELLO") { + socket.write(HELLO_MAP); + return; + } + if (frame.name === commandName.toUpperCase()) { + socket.write(">3\r\n" + bulk(pushKind) + bulk(frame.args[0]) + ":1\r\n"); + return; + } + if (frame.name === "PING") { + socket.write("+PONG\r\n"); + return; + } + }); + const client = new RedisClient(`redis://127.0.0.1:${mock.port}`, { + autoReconnect: false, + connectionTimeout: 5000, + }); + try { + const sub = await client.send(commandName, ["ch"]); + expect(typeof sub).toBe("number"); + expect(await client.ping()).toBe("PONG"); + } finally { + client.close(); + mock.close(); + } + }); + + // An `unsubscribe` confirmation must only count against an UNSUBSCRIBE + // head. Argless UNSUBSCRIBE emits one push per channel; the extras must + // not be charged to a following SUBSCRIBE drained into in_flight. + test("extra unsubscribe confirmations do not consume a queued SUBSCRIBE's reply slot", async () => { + let subscribed = new Set(); + const mock = await createMockServer((frame, socket) => { + if (frame.name === "HELLO") { + socket.write(HELLO_MAP); + return; + } + if (frame.name === "SUBSCRIBE") { + let reply = ""; + for (const ch of frame.args) { + subscribed.add(ch); + reply += ">3\r\n" + bulk("subscribe") + bulk(ch) + `:${subscribed.size}\r\n`; + } + socket.write(reply); + return; + } + if (frame.name === "UNSUBSCRIBE") { + const channels = frame.args.length > 0 ? frame.args : [...subscribed]; + let reply = ""; + for (const ch of channels) { + subscribed.delete(ch); + reply += ">3\r\n" + bulk("unsubscribe") + bulk(ch) + `:${subscribed.size}\r\n`; + } + socket.write(reply); + return; + } + }); + const client = new RedisClient(`redis://127.0.0.1:${mock.port}`, { + autoReconnect: false, + connectionTimeout: 5000, + }); + try { + await client.subscribe(["a", "b", "c"], () => {}); + // Issue UNSUBSCRIBE-all (3 confirmation pushes) and a new SUBSCRIBE + // back to back so the SUBSCRIBE is drained into in_flight while + // unsubscribe confirmations are still arriving. + const unsubP = client.unsubscribe(); + const subP = client.subscribe(["d"], () => {}); + const [unsub, sub] = await Promise.all([unsubP, subP]); + expect(unsub).toBeUndefined(); + // The new SUBSCRIBE must resolve with its own channel count. If an + // unsubscribe push stole its slot it would resolve with `undefined` + // via the unsubscribe dispatch arm. + expect(typeof sub).toBe("number"); + } finally { + client.close(); + mock.close(); + } + }); +});