redis: route RESP3 replies on the frame an attribute decorates

src/runtime/valkey_jsc/valkey.rs

@@ -1051,6 +1051,10 @@ impl ValkeyClient {
 
     /// Handle Valkey protocol response
     fn handle_response(&mut self, value: &mut RESPValue) -> JsTerminated<()> {
+        // Everything below (HELLO, SELECT, push routing, promise pairing)
+        // dispatches on the variant, so strip any attribute decoration first.
+        value.unwrap_attributes();
+
         // Special handling for the initial HELLO response
         if !self.flags.is_authenticated {
             self.handle_hello_response(value)?;

src/valkey/valkey_protocol.rs

@@ -123,6 +123,20 @@ pub enum RESPValue {
 
 // `deinit` deleted — all payloads are Box/Vec; Drop is automatic.
 
+impl RESPValue {
+    /// Replace an attribute frame with the reply it decorates.
+    ///
+    /// RESP3 attributes (`|N`) are out-of-band metadata that may prefix any
+    /// reply, pushes included, so callers that dispatch on the variant must
+    /// strip the decoration first.
+    pub fn unwrap_attributes(&mut self) {
+        while let RESPValue::Attribute(attribute) = self {
+            let decorated = core::mem::replace(attribute.value.as_mut(), RESPValue::Null);
+            *self = decorated;
+        }
+    }
+}
+
 impl fmt::Display for RESPValue {
     fn fmt(&self, writer: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {

test/js/valkey/reliability/connection-failures.test.ts

@@ -1,7 +1,7 @@
 import { RedisClient } from "bun";
 import { describe, expect, mock, test } from "bun:test";
 import net from "net";
-import { DEFAULT_REDIS_OPTIONS, DEFAULT_REDIS_URL, delay, isEnabled } from "../test-utils";
+import { DEFAULT_REDIS_OPTIONS, DEFAULT_REDIS_URL, delay, isEnabled, readRespCommands } from "../test-utils";
 
 /**
  * Test suite for connection failures, reconnection, and error handling
@@ -338,49 +338,6 @@ describe.skipIf(!isEnabled)("Valkey: Connection Failures", () => {
 });
 
 describe("Valkey: Auto-Reconnect In-Flight Commands", () => {
-  function readCommands(state: { buffer: Buffer }): string[][] {
-    const commands: string[][] = [];
-    while (true) {
-      const text = state.buffer.toString("latin1");
-      if (text[0] !== "*") break;
-      const headerEnd = text.indexOf("\r\n");
-      if (headerEnd === -1) break;
-      const argCount = parseInt(text.slice(1, headerEnd), 10);
-      if (!Number.isInteger(argCount) || argCount < 0) break;
-      let pos = headerEnd + 2;
-      const args: string[] = [];
-      let complete = true;
-      for (let i = 0; i < argCount; i++) {
-        if (text[pos] !== "$") {
-          complete = false;
-          break;
-        }
-        const lenEnd = text.indexOf("\r\n", pos);
-        if (lenEnd === -1) {
-          complete = false;
-          break;
-        }
-        const len = parseInt(text.slice(pos + 1, lenEnd), 10);
-        if (!Number.isInteger(len) || len < 0) {
-          complete = false;
-          break;
-        }
-        const dataStart = lenEnd + 2;
-        const dataEnd = dataStart + len;
-        if (text.length < dataEnd + 2) {
-          complete = false;
-          break;
-        }
-        args.push(text.slice(dataStart, dataEnd));
-        pos = dataEnd + 2;
-      }
-      if (!complete) break;
-      commands.push(args);
-      state.buffer = state.buffer.subarray(pos);
-    }
-    return commands;
-  }
-
   test("rejects commands that were in flight when the connection dropped instead of pairing them with replies from the next connection", async () => {
     const sockets: net.Socket[] = [];
     let connections = 0;
@@ -393,7 +350,7 @@ describe("Valkey: Auto-Reconnect In-Flight Commands", () => {
       const state = { buffer: Buffer.alloc(0) };
       socket.on("data", chunk => {
         state.buffer = Buffer.concat([state.buffer, chunk]);
-        for (const args of readCommands(state)) {
+        for (const args of readRespCommands(state)) {
           const name = (args[0] ?? "").toUpperCase();
           if (name === "HELLO") {
             socket.write("+OK\r\n");

test/js/valkey/reliability/protocol-handling.test.ts

@@ -1,5 +1,7 @@
+import { RedisClient } from "bun";
 import { beforeEach, describe, expect, test } from "bun:test";
-import { ConnectionType, createClient, ctx, isEnabled, testKey } from "../test-utils";
+import net from "net";
+import { ConnectionType, createClient, ctx, isEnabled, readRespCommands, testKey } from "../test-utils";
 
 /**
  * Test suite for RESP protocol handling, focusing on edge cases
@@ -399,3 +401,158 @@ describe.skipIf(!isEnabled)("Valkey: Protocol Handling", () => {
     });
   });
 });
+
+/**
+ * A RESP3 attribute (`|N`) is out-of-band metadata that may prefix *any* reply
+ * or push, so the outer frame type of a decorated reply says nothing about how
+ * the client should route it. These tests script the wire directly because no
+ * real server emits attributes on demand.
+ */
+describe("Valkey: RESP3 attributes", () => {
+  const CRLF = "\r\n";
+  const bulk = (value: string) => `$${Buffer.byteLength(value)}${CRLF}${value}${CRLF}`;
+  const pushFrame = (...frames: string[]) => `>${frames.length}${CRLF}${frames.join("")}`;
+  const HELLO_REPLY = `%3${CRLF}${bulk("server")}${bulk("redis")}${bulk("proto")}:3${CRLF}${bulk("version")}${bulk("7.4.0")}`;
+  // One-entry attribute map, the shape Redis/Valkey use for key-popularity hints.
+  const ATTRIBUTE = `|1${CRLF}${bulk("key-popularity")}:42${CRLF}`;
+
+  /**
+   * Run `body` against a client wired to a scripted RESP3 server. `reply` gets
+   * the upper-cased command name and its arguments and returns the raw bytes to
+   * answer with; returning `undefined` falls back to the HELLO handshake reply
+   * for HELLO and `+OK` for everything else.
+   */
+  async function withScriptedServer<T>(
+    reply: (name: string, args: string[]) => string | undefined,
+    body: (client: RedisClient) => Promise<T>,
+  ): Promise<T> {
+    const sockets = new Set<net.Socket>();
+    const server = net.createServer(socket => {
+      sockets.add(socket);
+      socket.on("close", () => sockets.delete(socket));
+      socket.on("error", () => {});
+      const state = { buffer: Buffer.alloc(0) };
+      socket.on("data", chunk => {
+        state.buffer = Buffer.concat([state.buffer, chunk]);
+        for (const args of readRespCommands(state)) {
+          const name = (args[0] ?? "").toUpperCase();
+          socket.write(reply(name, args.slice(1)) ?? (name === "HELLO" ? HELLO_REPLY : `+OK${CRLF}`));
+        }
+      });
+    });
+    await new Promise<void>(resolve => server.listen(0, "127.0.0.1", resolve));
+    const { port } = server.address() as net.AddressInfo;
+    const client = new RedisClient(`redis://127.0.0.1:${port}`, { autoReconnect: false });
+    try {
+      return await body(client);
+    } finally {
+      client.close();
+      for (const socket of sockets) socket.destroy();
+      server.close();
+    }
+  }
+
+  test("an attribute-decorated out-of-band push is not consumed as a command reply", async () => {
+    const values = ["A", "B", "C"];
+    const results = await withScriptedServer(
+      name => {
+        if (name !== "GET") return undefined;
+        // The push arrives before the first GET's reply, decorated by an attribute.
+        const push = values.length === 3 ? ATTRIBUTE + pushFrame(bulk("message"), bulk("chan"), bulk("hi")) : "";
+        return push + bulk(values.shift()!);
+      },
+      async client => {
+        await client.connect();
+        return [await client.get("a"), await client.get("b"), await client.get("c")];
+      },
+    );
+
+    // Without unwrapping the attribute the push is handed to `get("a")` and every
+    // later reply on the connection is shifted by one, forever.
+    expect(results).toEqual(["A", "B", "C"]);
+  });
+
+  test("an attribute-decorated subscribe confirmation enters subscriber mode", async () => {
+    const received = Promise.withResolvers<[string, string]>();
+    const subscribeResult = await withScriptedServer(
+      (name, args) => {
+        if (name !== "SUBSCRIBE") return undefined;
+        return (
+          ATTRIBUTE +
+          pushFrame(bulk("subscribe"), bulk(args[0]), `:1${CRLF}`) +
+          pushFrame(bulk("message"), bulk(args[0]), bulk("hi"))
+        );
+      },
+      async client => {
+        await client.connect();
+        const count = await client.subscribe("chan", (message, channel) => {
+          received.resolve([message, channel]);
+        });
+        // Without unwrapping, `subscribe()` resolves with the raw push object and
+        // the client never enters subscriber mode, so the listener never fires.
+        expect(count).toBe(1);
+        expect(await received.promise).toEqual(["hi", "chan"]);
+        return count;
+      },
+    );
+
+    expect(subscribeResult).toBe(1);
+  });
+
+  test("an attribute-decorated HELLO reply completes the handshake", async () => {
+    const value = await withScriptedServer(
+      name => {
+        if (name === "HELLO") return ATTRIBUTE + HELLO_REPLY;
+        if (name === "GET") return bulk("value");
+        return undefined;
+      },
+      async client => {
+        await client.connect();
+        return client.get("key");
+      },
+    );
+
+    expect(value).toBe("value");
+  });
+
+  test("an attribute-decorated error reply rejects the command", async () => {
+    await withScriptedServer(
+      name => (name === "GET" ? `${ATTRIBUTE}-ERR decorated failure${CRLF}` : undefined),
+      async client => {
+        await client.connect();
+        // Without unwrapping, the error is *resolved* as an Error object.
+        await expect(client.get("key")).rejects.toThrow("ERR decorated failure");
+      },
+    );
+  });
+
+  test("an attribute-decorated integer reply still coerces to boolean for EXISTS", async () => {
+    const exists = await withScriptedServer(
+      name => (name === "EXISTS" ? `${ATTRIBUTE}:1${CRLF}` : undefined),
+      async client => {
+        await client.connect();
+        return client.exists("key");
+      },
+    );
+
+    expect(exists).toBe(true);
+  });
+
+  test("attributes are transparent metadata around the value they decorate", async () => {
+    const [value, array] = await withScriptedServer(
+      name => {
+        if (name === "GET") return ATTRIBUTE + bulk("value");
+        // Attributes can also decorate an element nested inside an aggregate.
+        if (name === "LRANGE") return `*2${CRLF}${ATTRIBUTE}${bulk("a")}${bulk("b")}`;
+        return undefined;
+      },
+      async client => {
+        await client.connect();
+        return [await client.get("key"), await client.send("LRANGE", ["list", "0", "-1"])];
+      },
+    );
+
+    expect(value).toBe("value");
+    expect(array).toEqual(["a", "b"]);
+  });
+});

test/js/valkey/test-utils.ts

@@ -498,6 +498,53 @@ if (!isEnabled) {
   console.warn("Redis is not enabled, skipping tests");
 }
 
+/**
+ * Pull every complete `*N` RESP command frame out of `state.buffer`, leaving any
+ * trailing partial frame behind. For mock servers that script the wire directly.
+ */
+export function readRespCommands(state: { buffer: Buffer }): string[][] {
+  const commands: string[][] = [];
+  for (;;) {
+    const text = state.buffer.toString("latin1");
+    if (text[0] !== "*") break;
+    const headerEnd = text.indexOf("\r\n");
+    if (headerEnd === -1) break;
+    const argCount = parseInt(text.slice(1, headerEnd), 10);
+    if (!Number.isInteger(argCount) || argCount < 0) break;
+    let pos = headerEnd + 2;
+    const args: string[] = [];
+    let complete = true;
+    for (let i = 0; i < argCount; i++) {
+      if (text[pos] !== "$") {
+        complete = false;
+        break;
+      }
+      const lenEnd = text.indexOf("\r\n", pos);
+      if (lenEnd === -1) {
+        complete = false;
+        break;
+      }
+      const len = parseInt(text.slice(pos + 1, lenEnd), 10);
+      if (!Number.isInteger(len) || len < 0) {
+        complete = false;
+        break;
+      }
+      const dataStart = lenEnd + 2;
+      const dataEnd = dataStart + len;
+      if (text.length < dataEnd + 2) {
+        complete = false;
+        break;
+      }
+      args.push(text.slice(dataStart, dataEnd));
+      pos = dataEnd + 2;
+    }
+    if (!complete) break;
+    commands.push(args);
+    state.buffer = state.buffer.subarray(pos);
+  }
+  return commands;
+}
+
 /**
  * Verify that a value is of a specific type
  */