diff --git a/packages/client/lib/client/cache.spec.ts b/packages/client/lib/client/cache.spec.ts index 143c70f2267..17e2572407d 100644 --- a/packages/client/lib/client/cache.spec.ts +++ b/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 = ["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([ + ["list", [Buffer.from("a"), 1, [Buffer.from("b")]]], + ["set", new Set([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>) { + 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; + 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; + 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 }); diff --git a/packages/client/lib/client/cache.ts b/packages/client/lib/client/cache.ts index 5519edfba0b..efb80ebafac 100644 --- a/packages/client/lib/client/cache.ts +++ b/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(value: T): T { + return reapplyReplyBuffers(structuredClone(value), value, new WeakMap()) as T; +} + +function reapplyReplyBuffers(clone: unknown, original: unknown, seen: WeakMap): 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; + 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(); + seen.set(original, out); + const clonedEntries = (clone as Map).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(); + seen.set(original, out); + const clonedValues = (clone as Set).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)[key] = reapplyReplyBuffers( + (clone as Record)[key], + (original as Record)[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() {