fix(client): preserve Buffer replies in the client-side cache

packages/client/lib/client/cache.spec.ts

@@ -5,6 +5,7 @@ import { REDIS_FLUSH_MODES } from "../commands/FLUSHALL";
 import { once } from 'events';
 import RedisClient from "./index";
 import { BasicCommandParser } from "./parser";
+import { VerbatimString } from "../RESP/verbatim-string";
 import { ClientSideCacheCommandError, ClientSideCacheMarkError } from "../errors";
 
 describe("Client Side Cache", () => {
@@ -27,6 +28,115 @@ describe("Client Side Cache", () => {
     assert.ok(cache.get("k"), "a never-connected client must not flush the shared cache");
   });
 
+  describe("Reply cloning", () => {
+    function fakeParser(args: Array<string> = ["GET", "k"]) {
+      return {
+        redisArgs: args,
+        keys: [args[1]].filter(Boolean),
+        preserve: undefined
+      } as unknown as BasicCommandParser;
+    }
+
+    const shouldNotRun = async () => {
+      throw new Error("must be served from the cache");
+    };
+
+    it("preserves Buffer replies on miss and hit, with independent copies", async () => {
+      const cache = new BasicClientSideCache({ maxEntries: 10 });
+      const client = RedisClient.create({ RESP: 3, clientSideCache: cache });
+      const parser = fakeParser();
+      const transformReply = () => Buffer.from("hello");
+
+      const first = await cache.handleCache(client as never, parser, async () => "raw", transformReply as never);
+      assert.ok(Buffer.isBuffer(first), "the read that fills the cache should return a Buffer");
+      assert.strictEqual((first as Buffer).toString(), "hello");
+
+      const second = await cache.handleCache(client as never, parser, shouldNotRun, transformReply as never);
+      assert.ok(Buffer.isBuffer(second), "a cache hit should return a Buffer");
+      assert.strictEqual((second as Buffer).toString(), "hello");
+
+      // mutating a returned copy must not corrupt the cached entry
+      (second as Buffer)[0] = 0x21;
+      const third = await cache.handleCache(client as never, parser, shouldNotRun, transformReply as never);
+      assert.strictEqual((third as Buffer).toString(), "hello");
+    });
+
+    it("preserves Buffers nested in arrays, maps, sets and plain objects", async () => {
+      const cache = new BasicClientSideCache({ maxEntries: 10 });
+      const client = RedisClient.create({ RESP: 3, clientSideCache: cache });
+      const parser = fakeParser(["CMD"]);
+      const nested = new Map<string, unknown>([
+        ["list", [Buffer.from("a"), 1, [Buffer.from("b")]]],
+        ["set", new Set<unknown>([Buffer.from("c"), "d"])],
+        ["plain", { x: Buffer.from("e") }]
+      ]);
+      const transformReply = () => nested;
+
+      const first = await cache.handleCache(client as never, parser, async () => "raw", transformReply as never);
+      const second = await cache.handleCache(client as never, parser, shouldNotRun, transformReply as never);
+
+      for (const value of [first, second] as Array<Map<string, unknown>>) {
+        assert.ok(value instanceof Map);
+        const [a, one, [b]] = value.get("list") as [Buffer, number, [Buffer]];
+        assert.ok(Buffer.isBuffer(a) && a.toString() === "a");
+        assert.strictEqual(one, 1);
+        assert.ok(Buffer.isBuffer(b) && b.toString() === "b");
+        const set = value.get("set") as Set<unknown>;
+        assert.ok(set instanceof Set);
+        assert.ok([...set].some(v => Buffer.isBuffer(v) && (v as Buffer).toString() === "c"));
+        assert.ok(set.has("d"));
+        const plain = value.get("plain") as { x: Buffer };
+        assert.ok(Buffer.isBuffer(plain.x) && plain.x.toString() === "e");
+      }
+    });
+
+    it("clones VerbatimString replies without walking their read-only string properties", async () => {
+      const cache = new BasicClientSideCache({ maxEntries: 10 });
+      const client = RedisClient.create({ RESP: 3, clientSideCache: cache });
+      const parser = fakeParser(["CMD"]);
+      const reply = { verbatim: new VerbatimString("txt", "hello"), buf: Buffer.from("x") };
+      const transformReply = () => reply;
+
+      const first = await cache.handleCache(client as never, parser, async () => "raw", transformReply as never) as {
+        verbatim: unknown;
+        buf: Buffer;
+      };
+      // the clone keeps structuredClone's behavior for String objects (no throw, no format)
+      assert.ok(first.verbatim instanceof String);
+      assert.strictEqual(first.verbatim.toString(), "hello");
+      assert.ok(Buffer.isBuffer(first.buf) && first.buf.toString() === "x");
+
+      const second = await cache.handleCache(client as never, parser, shouldNotRun, transformReply as never) as {
+        verbatim: unknown;
+      };
+      assert.strictEqual(second.verbatim.toString(), "hello");
+    });
+
+    it("keeps structuredClone semantics for the rest of the reply", async () => {
+      const cache = new BasicClientSideCache({ maxEntries: 10 });
+      const client = RedisClient.create({ RESP: 3, clientSideCache: cache });
+      const parser = fakeParser(["CMD"]);
+      const shared = Buffer.from("shared");
+      const reply = { twice: [shared, shared], n: 42, s: "str", nil: null };
+      const transformReply = () => reply;
+
+      const out = await cache.handleCache(client as never, parser, async () => "raw", transformReply as never) as {
+        twice: Array<Buffer>;
+        n: number;
+        s: string;
+        nil: null;
+      };
+      assert.strictEqual(out.n, 42);
+      assert.strictEqual(out.s, "str");
+      assert.strictEqual(out.nil, null);
+      assert.ok(Buffer.isBuffer(out.twice[0]) && Buffer.isBuffer(out.twice[1]));
+      // shared references stay one and the same object, like structuredClone
+      assert.strictEqual(out.twice[0], out.twice[1]);
+      // but the returned copy is independent of the cached entry
+      assert.notStrictEqual(out.twice[0], shared);
+    });
+  });
+
   describe('Basic Cache', () => {
     const csc = new BasicClientSideCache({ maxEntries: 10 });
 

packages/client/lib/client/cache.ts

@@ -492,6 +492,94 @@ abstract class ClientSideCacheEntryBase implements ClientSideCacheEntry {
   }
 }
 
+
+/**
+ * Clones a cached reply, preserving the `Buffer` subclass.
+ *
+ * `structuredClone` clones a `Buffer` as a plain `Uint8Array`, so every read from the
+ * cache (and the read that fills it) lost the configured `Buffer` type mapping
+ * (`BLOB_STRING -> Buffer`, the documented setup for caching with binary replies).
+ * The clone and the original have identical structure, so walk them in tandem and
+ * replace the cloned values with `Buffer`s exactly where the original reply had them.
+ * The memo keeps shared references (and cycles) intact, like `structuredClone` does.
+ */
+function cloneReply<T>(value: T): T {
+  return reapplyReplyBuffers(structuredClone(value), value, new WeakMap()) as T;
+}
+
+function reapplyReplyBuffers(clone: unknown, original: unknown, seen: WeakMap<object, unknown>): unknown {
+  if (original === null || typeof original !== 'object') {
+    return clone;
+  }
+
+  const known = seen.get(original);
+  if (known !== undefined) {
+    return known;
+  }
+
+  if (original instanceof Buffer) {
+    const view = clone as Uint8Array;
+    const buffer = Buffer.from(view.buffer, view.byteOffset, view.length);
+    seen.set(original, buffer);
+    return buffer;
+  }
+
+  if (Array.isArray(original)) {
+    seen.set(original, clone);
+    const clonedArray = clone as Array<unknown>;
+    for (let i = 0; i < original.length; i++) {
+      clonedArray[i] = reapplyReplyBuffers(clonedArray[i], original[i], seen);
+    }
+    return clone;
+  }
+
+  if (original instanceof Map) {
+    const out = new Map<unknown, unknown>();
+    seen.set(original, out);
+    const clonedEntries = (clone as Map<unknown, unknown>).entries();
+    for (const [key, val] of original) {
+      const clonedEntry = clonedEntries.next().value as [unknown, unknown];
+      out.set(
+        reapplyReplyBuffers(clonedEntry[0], key, seen),
+        reapplyReplyBuffers(clonedEntry[1], val, seen)
+      );
+    }
+    return out;
+  }
+
+  if (original instanceof Set) {
+    const out = new Set<unknown>();
+    seen.set(original, out);
+    const clonedValues = (clone as Set<unknown>).values();
+    for (const item of original) {
+      out.add(reapplyReplyBuffers(clonedValues.next().value, item, seen));
+    }
+    return out;
+  }
+
+  if (!isPlainObject(original)) {
+    // `String` objects (`VerbatimString`), errors, dates and custom mapped classes keep
+    // their `structuredClone` result: their indexed/own properties are not writable
+    // (a `String` object throws on assignment) and cannot hold buffers in practice
+    return clone;
+  }
+
+  seen.set(original, clone);
+  for (const key of Object.keys(original)) {
+    (clone as Record<string, unknown>)[key] = reapplyReplyBuffers(
+      (clone as Record<string, unknown>)[key],
+      (original as Record<string, unknown>)[key],
+      seen
+    );
+  }
+  return clone;
+}
+
+function isPlainObject(value: object): boolean {
+  const prototype = Object.getPrototypeOf(value);
+  return prototype === Object.prototype || prototype === null;
+}
+
 class ClientSideCacheEntryValue extends ClientSideCacheEntryBase {
   readonly #value: unknown;
 
@@ -632,7 +720,7 @@ export class BasicClientSideCache extends ClientSideCacheProvider {
         this.#statsCounter.recordHits(1);
         publish(CHANNELS.CACHE_REQUEST, () => ({ result: 'hit', clientId: client._clientId }));
 
-        return structuredClone(cacheEntry.value);
+        return cloneReply(cacheEntry.value);
       } else if (cacheEntry instanceof ClientSideCacheEntryPromise) { // 2b2
         // This counts as a miss since the value hasn't been fully loaded yet.
         this.#statsCounter.recordMisses(1);
@@ -691,7 +779,7 @@ export class BasicClientSideCache extends ClientSideCacheProvider {
       //   cache entry for key got invalidated between execution and saving, so not saving
     }
 
-    return structuredClone(val);
+    return cloneReply(val);
   }
 
   override trackingOn() {