diff --git a/README.md b/README.md index a2f35e3..4b12ff3 100644 --- a/README.md +++ b/README.md @@ -90,18 +90,18 @@ const dialcache = new DialCache({ }); ``` -The core Redis boundary is the client-agnostic `DialCacheRedisClient` interface. Other clients can implement that semantic interface without changing DialCache; distinct untracked/tracked read and write Lua sources, the invalidation source, and wire constants are available from `dialcache/redis-protocol`, so a Valkey GLIDE adapter can be added without depending on node-redis. Custom adapters can use the root-exported `DialCacheRedisPayloadError` and `DialCacheRedisPayloadEncodingError` classes to preserve the standard metrics labels. +The core Redis boundary is the client-agnostic `DialCacheRedisClient` interface. It exchanges serialized values as `string | Buffer` and does not expose node-redis commands or wire encodings. Other clients can implement that semantic interface without changing DialCache; distinct untracked/tracked read and write Lua sources, the invalidation source, and wire constants are available from `dialcache/redis-protocol`, so a Valkey GLIDE adapter can use its own script lifecycle, cluster routing, and byte decoder without depending on node-redis. Custom adapters can use the root-exported `DialCacheRedisPayloadError` and `DialCacheRedisPayloadEncodingError` classes to preserve the standard metrics labels. Redis values use a compact binary frame: ```text byte 1 format version bytes 2-9 Redis-created timestamp in milliseconds (uint64, big-endian) -byte 10 payload encoding (0 = UTF-8, 1 = base64) +byte 10 payload encoding (0 = UTF-8, 1 = raw binary) bytes 11... serialized payload ``` -Redis's Lua `struct` library packs and unpacks the timestamp. Redis TTL is authoritative, so expiry metadata is not duplicated in the frame. `payload` is produced by the cached function's serializer, or by `JsonSerializer` by default. Custom serializers can return either `string` or `Buffer`; Buffer payloads are base64 encoded and reconstructed before `serializer.load`. +Redis's Lua `struct` library packs and unpacks the timestamp. Redis TTL is authoritative, so expiry metadata is not duplicated in the frame. `payload` is produced by the cached function's serializer, or by `JsonSerializer` by default. Custom serializers can return either `string` or `Buffer`; strings are stored as UTF-8 and Buffers are stored byte-for-byte without base64 expansion. Adapters restore the same representation before calling `serializer.load`. ## Targeted invalidation and watermarks diff --git a/scripts/test-package.mjs b/scripts/test-package.mjs index f67225d..e7f9e6d 100644 --- a/scripts/test-package.mjs +++ b/scripts/test-package.mjs @@ -8,7 +8,7 @@ import { promisify } from "node:util"; const exec = promisify(execFile); const root = dirname(dirname(fileURLToPath(import.meta.url))); const workspace = await mkdtemp(join(tmpdir(), "dialcache-package-")); -const consumer = `import { DialCache, DialCacheKeyConfig } from "dialcache"; +const consumer = `import { DialCache, DialCacheKeyConfig, type DialCacheRedisClient } from "dialcache"; import { createNodeRedisDialCacheClient } from "dialcache/node-redis"; import { READ_CACHE_SCRIPT } from "dialcache/redis-protocol"; @@ -23,6 +23,14 @@ const load = cache.cached(async (id: string) => id, { void load; void createNodeRedisDialCacheClient; void READ_CACHE_SCRIPT; + +const customRedisClient: DialCacheRedisClient = { + read: async () => Buffer.from([0, 255]), + write: async ({ value }) => typeof value === "string" || Buffer.isBuffer(value), + invalidate: async () => undefined, + flushAll: async () => undefined, +}; +void customRedisClient; `; try { diff --git a/src/index.ts b/src/index.ts index e33787b..6785d4a 100644 --- a/src/index.ts +++ b/src/index.ts @@ -31,7 +31,6 @@ export type { RedisCachePayload, RedisClientFactory, RedisInvalidationRequest, - RedisPayloadEncoding, RedisReadRequest, RedisWriteRequest, } from "./redis-client.js"; diff --git a/src/internal/redis-cache.ts b/src/internal/redis-cache.ts index 45fe152..5396a34 100644 --- a/src/internal/redis-cache.ts +++ b/src/internal/redis-cache.ts @@ -4,7 +4,7 @@ import { CacheLayer, DEFAULT_WATERMARK_TTL_SEC, type CacheConfigProvider, type C import { invalidationPrefix, redisClusterHashTag, type DialCacheKey } from "../key.js"; import type { DialCacheMetricsAdapter } from "../metrics.js"; import { errorName, labelsFor } from "../metrics.js"; -import type { DialCacheRedisClient, RedisCachePayload, RedisClientFactory } from "../redis-client.js"; +import type { DialCacheRedisClient, RedisClientFactory } from "../redis-client.js"; import { JsonSerializer, type Serializer } from "../serializer.js"; import type { CacheGetResult } from "./cache-result.js"; import { fetchKeyConfig, resolveLayerConfigResult, type ResolvedLayerConfig } from "./runtime-config.js"; @@ -87,7 +87,7 @@ export class RedisCache { const start = performance.now(); try { - const value = (await this.serializerFor(key).load(decodePayload(payload))) as T; + const value = (await this.serializerFor(key).load(payload)) as T; return { status: "hit", value }; } catch (error) { this.recordMetric((metrics) => metrics.error({ ...labelsFor(key, CacheLayer.REMOTE), error: errorName(error), inFallback: false })); @@ -120,8 +120,7 @@ export class RedisCache { const request = { valueKey: this.redisKey(key), cacheTtlMs, - encoding: Buffer.isBuffer(serialized) ? "base64" : "utf8", - value: Buffer.isBuffer(serialized) ? serialized.toString("base64") : serialized, + value: serialized, } as const; return key.trackForInvalidation ? await client.write({ @@ -206,10 +205,6 @@ export class RedisCache { } } -function decodePayload(payload: RedisCachePayload): string | Buffer { - return payload.encoding === "base64" ? Buffer.from(payload.value, "base64") : payload.value; -} - function payloadSize(payload: string | Buffer): number { return Buffer.isBuffer(payload) ? payload.byteLength : Buffer.byteLength(payload); } diff --git a/src/internal/redis-scripts.ts b/src/internal/redis-scripts.ts index ca4219c..f0e1094 100644 --- a/src/internal/redis-scripts.ts +++ b/src/internal/redis-scripts.ts @@ -1,6 +1,6 @@ export const REDIS_FRAME_VERSION = 1; export const REDIS_ENCODING_UTF8 = 0; -export const REDIS_ENCODING_BASE64 = 1; +export const REDIS_ENCODING_BINARY = 1; const WATERMARK_TTL_MARGIN_MS = 60_000; @@ -39,7 +39,7 @@ local encoding = tonumber(ARGV[2]) if not cache_ttl_ms or cache_ttl_ms <= 0 then return redis.error_reply("ERR invalid DialCache TTL") end -if not encoding or (encoding ~= ${REDIS_ENCODING_UTF8} and encoding ~= ${REDIS_ENCODING_BASE64}) then +if not encoding or (encoding ~= ${REDIS_ENCODING_UTF8} and encoding ~= ${REDIS_ENCODING_BINARY}) then return redis.error_reply("ERR invalid DialCache payload encoding") end`; diff --git a/src/node-redis.ts b/src/node-redis.ts index f5de72d..8853821 100644 --- a/src/node-redis.ts +++ b/src/node-redis.ts @@ -1,10 +1,10 @@ -import { defineScript } from "redis"; +import { commandOptions, defineScript } from "redis"; import { INVALIDATE_CACHE_SCRIPT, READ_CACHE_SCRIPT, READ_TRACKED_CACHE_SCRIPT, - REDIS_ENCODING_BASE64, + REDIS_ENCODING_BINARY, REDIS_ENCODING_UTF8, WRITE_CACHE_SCRIPT, WRITE_TRACKED_CACHE_SCRIPT, @@ -12,8 +12,12 @@ import { import { DialCacheRedisPayloadEncodingError, DialCacheRedisPayloadError } from "./redis-client.js"; import type { DialCacheRedisClient, RedisCachePayload } from "./redis-client.js"; +type BufferReplyOptions = ReturnType>; +// Redis bulk strings are binary data; decoding them as UTF-8 would corrupt arbitrary serializer output. +const bufferReplyOptions: BufferReplyOptions = commandOptions({ returnBuffers: true }); const readReply = (reply: string | null): string | null => reply; const integerReply = (reply: number): number => reply; +type NodeRedisArgument = string | Buffer; interface NodeRedisScript, Reply> { readonly SCRIPT: string; @@ -21,7 +25,7 @@ interface NodeRedisScript, Reply> { readonly NUMBER_OF_KEYS: number; readonly FIRST_KEY_INDEX: number; readonly IS_READ_ONLY: boolean; - transformArguments(...args: Args): Array; + transformArguments(...args: Args): Array; transformReply(reply: Reply): Reply; } @@ -37,7 +41,7 @@ export type DialCacheNodeRedisScripts = { readonly dialcacheRead: NodeRedisScript<[valueKey: string], string | null>; readonly dialcacheReadTracked: NodeRedisScript<[valueKey: string, watermarkKey: string], string | null>; readonly dialcacheWrite: NodeRedisScript< - [valueKey: string, cacheTtlMs: number, encoding: number, payload: string], + [valueKey: string, cacheTtlMs: number, encoding: number, payload: string | Buffer], number >; readonly dialcacheWriteTracked: NodeRedisScript< @@ -46,7 +50,7 @@ export type DialCacheNodeRedisScripts = { watermarkKey: string, cacheTtlMs: number, encoding: number, - payload: string, + payload: string | Buffer, watermarkTtlFloorMs: number, ], number @@ -84,7 +88,12 @@ export const dialcacheRedisScripts: DialCacheNodeRedisScripts = { NUMBER_OF_KEYS: 1, FIRST_KEY_INDEX: 0, IS_READ_ONLY: false, - transformArguments(valueKey: string, cacheTtlMs: number, encoding: number, payload: string): Array { + transformArguments( + valueKey: string, + cacheTtlMs: number, + encoding: number, + payload: string | Buffer, + ): Array { return [valueKey, String(cacheTtlMs), String(encoding), payload]; }, transformReply: integerReply, @@ -99,9 +108,9 @@ export const dialcacheRedisScripts: DialCacheNodeRedisScripts = { watermarkKey: string, cacheTtlMs: number, encoding: number, - payload: string, + payload: string | Buffer, watermarkTtlFloorMs: number, - ): Array { + ): Array { return [valueKey, watermarkKey, String(cacheTtlMs), String(encoding), payload, String(watermarkTtlFloorMs)]; }, transformReply: integerReply, @@ -119,15 +128,19 @@ export const dialcacheRedisScripts: DialCacheNodeRedisScripts = { }; interface NodeRedisScriptClient { - dialcacheRead(valueKey: string): Promise; - dialcacheReadTracked(valueKey: string, watermarkKey: string): Promise; - dialcacheWrite(valueKey: string, cacheTtlMs: number, encoding: number, payload: string): Promise; + dialcacheRead(options: BufferReplyOptions, valueKey: string): Promise; + dialcacheReadTracked( + options: BufferReplyOptions, + valueKey: string, + watermarkKey: string, + ): Promise; + dialcacheWrite(valueKey: string, cacheTtlMs: number, encoding: number, payload: string | Buffer): Promise; dialcacheWriteTracked( valueKey: string, watermarkKey: string, cacheTtlMs: number, encoding: number, - payload: string, + payload: string | Buffer, watermarkTtlFloorMs: number, ): Promise; dialcacheInvalidate(watermarkKey: string, futureBufferMs: number, watermarkTtlFloorMs: number): Promise; @@ -143,13 +156,13 @@ export function createNodeRedisDialCacheClient(client: NodeRedisScriptClient): D return { async read({ valueKey, watermarkKey }) { const raw = watermarkKey === undefined - ? await client.dialcacheRead(valueKey) - : await client.dialcacheReadTracked(valueKey, watermarkKey); + ? await client.dialcacheRead(bufferReplyOptions, valueKey) + : await client.dialcacheReadTracked(bufferReplyOptions, valueKey, watermarkKey); return raw === null ? null : decodePayload(raw); }, async write(request) { - const { valueKey, watermarkKey, cacheTtlMs, encoding, value } = request; - const encodingByte = encoding === "base64" ? REDIS_ENCODING_BASE64 : REDIS_ENCODING_UTF8; + const { valueKey, watermarkKey, cacheTtlMs, value } = request; + const encodingByte = Buffer.isBuffer(value) ? REDIS_ENCODING_BINARY : REDIS_ENCODING_UTF8; const result = watermarkKey === undefined ? await client.dialcacheWrite(valueKey, cacheTtlMs, encodingByte, value) : await client.dialcacheWriteTracked( @@ -187,17 +200,18 @@ export function createNodeRedisDialCacheClient(client: NodeRedisScriptClient): D }; } -function decodePayload(raw: string): RedisCachePayload { +function decodePayload(raw: Buffer): RedisCachePayload { if (raw.length === 0) { throw new DialCacheRedisPayloadError("Invalid DialCache Redis payload"); } - const encoding = raw.charCodeAt(0); + const encoding = raw[0]; + const payload = raw.subarray(1); if (encoding === REDIS_ENCODING_UTF8) { - return { encoding: "utf8", value: raw.slice(1) }; + return payload.toString("utf8"); } - if (encoding === REDIS_ENCODING_BASE64) { - return { encoding: "base64", value: raw.slice(1) }; + if (encoding === REDIS_ENCODING_BINARY) { + return payload; } throw new DialCacheRedisPayloadEncodingError("Invalid DialCache Redis payload encoding"); } diff --git a/src/redis-client.ts b/src/redis-client.ts index 395d9d8..cd05543 100644 --- a/src/redis-client.ts +++ b/src/redis-client.ts @@ -14,12 +14,8 @@ export class DialCacheRedisPayloadEncodingError extends Error { } } -export type RedisPayloadEncoding = "utf8" | "base64"; - -export interface RedisCachePayload { - readonly encoding: RedisPayloadEncoding; - readonly value: string; -} +/** Serialized cache data, independent of any Redis client or wire framing. */ +export type RedisCachePayload = string | Buffer; interface RedisValueRequest { readonly valueKey: string; @@ -35,8 +31,9 @@ interface UntrackedRedisValueRequest extends RedisValueRequest { export type RedisReadRequest = TrackedRedisValueRequest | UntrackedRedisValueRequest; -interface RedisWriteBase extends RedisValueRequest, RedisCachePayload { +interface RedisWriteBase extends RedisValueRequest { readonly cacheTtlMs: number; + readonly value: RedisCachePayload; } interface TrackedRedisWriteRequest extends RedisWriteBase, TrackedRedisValueRequest { diff --git a/src/redis-protocol.ts b/src/redis-protocol.ts index 4742713..5a9c23b 100644 --- a/src/redis-protocol.ts +++ b/src/redis-protocol.ts @@ -2,7 +2,7 @@ export { INVALIDATE_CACHE_SCRIPT, READ_CACHE_SCRIPT, READ_TRACKED_CACHE_SCRIPT, - REDIS_ENCODING_BASE64, + REDIS_ENCODING_BINARY, REDIS_ENCODING_UTF8, REDIS_FRAME_VERSION, WRITE_CACHE_SCRIPT, diff --git a/test/dialcache-redis.test.ts b/test/dialcache-redis.test.ts index c9f4377..b408080 100644 --- a/test/dialcache-redis.test.ts +++ b/test/dialcache-redis.test.ts @@ -259,11 +259,26 @@ describe("DialCache Redis TTL layer", () => { expect(value).toEqual({ id: "123", source: "fallback" }); expect(readerCalls).toBe(0); expect(frame.encoding).toBe(1); - expect(Buffer.from(frame.payload, "base64").toString("utf8")).toBe("123|fallback"); + expect(frame.payload).toEqual(Buffer.from("123|fallback")); expect(serializer.dump).toHaveBeenCalledOnce(); expect(serializer.load).toHaveBeenCalledOnce(); }); + it("isolates binary read payloads from the stored Redis frame", async () => { + const redis = new FakeRedis(); + const valueKey = "binary-isolation:{item:123}:value"; + const payload = Buffer.from([0, 1, 2, 0xff]); + await redis.write({ valueKey, cacheTtlMs: 60_000, value: payload }); + + const firstRead = await redis.read({ valueKey }); + if (!Buffer.isBuffer(firstRead)) { + throw new Error("Expected a binary Redis payload"); + } + firstRead[0] = 0xff; + + expect(await redis.read({ valueKey })).toEqual(payload); + }); + it("fails open when Redis serializer dump fails", async () => { // Given Redis serialization fails after the fallback returns but local cache is still configured. const redis = new FakeRedis(); diff --git a/test/fake-redis.ts b/test/fake-redis.ts index 5bde3c0..5c7cb3a 100644 --- a/test/fake-redis.ts +++ b/test/fake-redis.ts @@ -44,7 +44,6 @@ export class FakeRedis implements DialCacheRedisClient { valueKey, watermarkKey, cacheTtlMs, - encoding, value, watermarkTtlFloorMs, }: RedisWriteRequest): Promise { @@ -55,14 +54,14 @@ export class FakeRedis implements DialCacheRedisClient { if (watermark >= Date.now()) { return false; } - this.storeFrame(valueKey, cacheTtlMs, encoding, value); + this.storeFrame(valueKey, cacheTtlMs, value); const currentTtlMs = this.remainingTtlMs(watermarkKey); const desiredTtlMs = Math.max(currentTtlMs, watermarkTtlFloorMs, cacheTtlMs + WATERMARK_TTL_MARGIN_MS); this.storeWatermark(watermarkKey, watermark, desiredTtlMs); return true; } - this.storeFrame(valueKey, cacheTtlMs, encoding, value); + this.storeFrame(valueKey, cacheTtlMs, value); return true; } @@ -155,22 +154,22 @@ export class FakeRedis implements DialCacheRedisClient { const encoding = raw[ENCODING_OFFSET]; if (encoding === 0) { - return { encoding: "utf8", value: raw.subarray(PAYLOAD_OFFSET).toString("utf8") }; + return raw.subarray(PAYLOAD_OFFSET).toString("utf8"); } if (encoding === 1) { - return { encoding: "base64", value: raw.subarray(PAYLOAD_OFFSET).toString("utf8") }; + return Buffer.from(raw.subarray(PAYLOAD_OFFSET)); } throw new DialCacheRedisPayloadEncodingError("Invalid DialCache Redis payload encoding"); } - private storeFrame(key: string, ttlMs: number, encoding: "utf8" | "base64", payload: string): void { + private storeFrame(key: string, ttlMs: number, payload: RedisCachePayload): void { const timestamp = Buffer.alloc(8); timestamp.writeBigUInt64BE(BigInt(Date.now())); this.values.set(key, { value: Buffer.concat([ Buffer.from([FRAME_VERSION]), timestamp, - Buffer.from([encoding === "base64" ? 1 : 0]), + Buffer.from([Buffer.isBuffer(payload) ? 1 : 0]), Buffer.from(payload), ]), expiresAtMs: Date.now() + ttlMs, @@ -221,18 +220,24 @@ export class FakeRedis implements DialCacheRedisClient { export function encodeFrame(value: unknown, createdAtMs = Date.now(), encoding = 0): Buffer { const timestamp = Buffer.alloc(8); timestamp.writeBigUInt64BE(BigInt(createdAtMs)); - const payload = typeof value === "string" ? value : JSON.stringify(value); - return Buffer.concat([Buffer.from([FRAME_VERSION]), timestamp, Buffer.from([encoding]), Buffer.from(payload)]); + const payload = Buffer.isBuffer(value) + ? value + : Buffer.from(typeof value === "string" ? value : JSON.stringify(value)); + return Buffer.concat([Buffer.from([FRAME_VERSION]), timestamp, Buffer.from([encoding]), payload]); } -export function decodeFrame(raw: Buffer): { readonly createdAtMs: number; readonly encoding: number; readonly payload: string } { +export function decodeFrame( + raw: Buffer, +): { readonly createdAtMs: number; readonly encoding: number; readonly payload: string | Buffer } { if (raw.length < PAYLOAD_OFFSET || raw[0] !== FRAME_VERSION) { throw new Error("Invalid DialCache frame"); } + const encoding = raw[ENCODING_OFFSET] ?? -1; + const payload = raw.subarray(PAYLOAD_OFFSET); return { createdAtMs: Number(readTimestamp(raw)), - encoding: raw[ENCODING_OFFSET] ?? -1, - payload: raw.subarray(PAYLOAD_OFFSET).toString("utf8"), + encoding, + payload: encoding === 0 ? payload.toString("utf8") : payload, }; } diff --git a/test/redis-cluster.integration.test.ts b/test/redis-cluster.integration.test.ts index d568cec..7a264e2 100644 --- a/test/redis-cluster.integration.test.ts +++ b/test/redis-cluster.integration.test.ts @@ -1,4 +1,4 @@ -import { createCluster, type RedisClusterOptions } from "redis"; +import { commandOptions, createCluster, type RedisClusterOptions } from "redis"; import { GenericContainer, Network, @@ -184,4 +184,35 @@ describe("DialCache Lua protocol on Redis Cluster", () => { expect(after).toEqual({ id: "123", version: 2 }); await expect(cluster.dialcacheReadTracked("{slot-a}:value", "{slot-b}:watermark")).rejects.toThrow(/CROSSSLOT/); }); + + it("round-trips binary payloads through cluster script routing", async () => { + if (cluster === undefined) { + throw new Error("Redis Cluster did not start"); + } + const scriptClient = createNodeRedisDialCacheClient(cluster); + const valueKey = "binary-cluster:{item:untracked}:value"; + const payload = Buffer.from(Array.from({ length: 256 }, (_, index) => index)); + + expect(await scriptClient.write({ valueKey, cacheTtlMs: 60_000, value: payload })).toBe(true); + expect(await scriptClient.read({ valueKey })).toEqual(payload); + + const stored = await cluster.get(commandOptions({ returnBuffers: true }), valueKey); + expect(stored?.length).toBe(10 + payload.length); + expect(stored?.[9]).toBe(1); + expect(stored?.subarray(10)).toEqual(payload); + + const trackedValueKey = "binary-cluster:{item:tracked}:value"; + const watermarkKey = "binary-cluster:{item:tracked}:watermark"; + const trackedPayload = Buffer.from([0, 0xff, 0xc3, 0x28, 0x80]); + expect( + await scriptClient.write({ + valueKey: trackedValueKey, + watermarkKey, + cacheTtlMs: 60_000, + value: trackedPayload, + watermarkTtlFloorMs: 60_000, + }), + ).toBe(true); + expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toEqual(trackedPayload); + }); }); diff --git a/test/redis-real.integration.test.ts b/test/redis-real.integration.test.ts index 4b6a087..4d26117 100644 --- a/test/redis-real.integration.test.ts +++ b/test/redis-real.integration.test.ts @@ -1,4 +1,4 @@ -import { createClient } from "redis"; +import { commandOptions, createClient } from "redis"; import { GenericContainer, type StartedTestContainer, Wait } from "testcontainers"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; @@ -24,7 +24,7 @@ const remoteOnly = new DialCacheKeyConfig({ const createTestClient = (url: string) => createClient({ url, scripts: dialcacheRedisScripts }); -function encodeFrame(payload: string, encoding: number, createdAtMs = Date.now(), version = 1): Buffer { +function encodeFrame(payload: string | Buffer, encoding: number, createdAtMs = Date.now(), version = 1): Buffer { const timestamp = Buffer.alloc(8); timestamp.writeBigUInt64BE(BigInt(createdAtMs)); return Buffer.concat([Buffer.from([version]), timestamp, Buffer.from([encoding]), Buffer.from(payload)]); @@ -88,6 +88,48 @@ describe.each(engines)("DialCache Lua protocol on $name", ({ image }) => { expect(binaryCalls).toBe(1); }); + it("stores arbitrary binary payloads without base64 expansion", async () => { + if (client === undefined) { + throw new Error("Redis client did not start"); + } + const scriptClient = createNodeRedisDialCacheClient(client); + const payloads = [ + Buffer.alloc(0), + Buffer.from(Array.from({ length: 256 }, (_, index) => index)), + Buffer.alloc(2 * 1024 * 1024, 0xa5), + ]; + + for (const [index, payload] of payloads.entries()) { + const valueKey = `binary-raw:{item:${index}}:value`; + expect(await scriptClient.write({ valueKey, cacheTtlMs: 60_000, value: payload })).toBe(true); + + const roundTrip = await scriptClient.read({ valueKey }); + const stored = await client.get(commandOptions({ returnBuffers: true }), valueKey); + + expect(Buffer.isBuffer(roundTrip)).toBe(true); + expect(roundTrip).toEqual(payload); + expect(stored).not.toBeNull(); + expect(stored?.length).toBe(10 + payload.length); + expect(stored?.[0]).toBe(1); + expect(stored?.[9]).toBe(1); + expect(stored?.subarray(10)).toEqual(payload); + } + + const trackedValueKey = "binary-raw:{item:tracked}:value"; + const watermarkKey = "binary-raw:{item:tracked}:watermark"; + const trackedPayload = Buffer.from([0, 0xff, 0xc3, 0x28, 0x80]); + expect( + await scriptClient.write({ + valueKey: trackedValueKey, + watermarkKey, + cacheTtlMs: 60_000, + value: trackedPayload, + watermarkTtlFloorMs: 60_000, + }), + ).toBe(true); + expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toEqual(trackedPayload); + }); + it("recovers every distinct read and write script after SCRIPT FLUSH", async () => { if (client === undefined) { throw new Error("Redis client did not start"); @@ -101,17 +143,14 @@ describe.each(engines)("DialCache Lua protocol on $name", ({ image }) => { await client.scriptFlush(); expect(await client.dialcacheWrite(valueKey, 60_000, 0, "untracked")).toBe(1); await client.scriptFlush(); - expect(await scriptClient.read({ valueKey })).toEqual({ encoding: "utf8", value: "untracked" }); + expect(await scriptClient.read({ valueKey })).toBe("untracked"); const trackedValueKey = "script-recovery:{item:tracked}:value"; const watermarkKey = "script-recovery:{item:tracked}:watermark"; await client.scriptFlush(); expect(await client.dialcacheWriteTracked(trackedValueKey, watermarkKey, 60_000, 0, "tracked", 60_000)).toBe(1); await client.scriptFlush(); - expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toEqual({ - encoding: "utf8", - value: "tracked", - }); + expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toBe("tracked"); }); it("treats every invalid read frame and watermark state as a miss", async () => { @@ -143,7 +182,7 @@ describe.each(engines)("DialCache Lua protocol on $name", ({ image }) => { expect(await scriptClient.read({ valueKey, watermarkKey })).toBeNull(); await client.set(watermarkKey, "999.5"); - expect(await scriptClient.read({ valueKey, watermarkKey })).toEqual({ encoding: "utf8", value: "tracked" }); + expect(await scriptClient.read({ valueKey, watermarkKey })).toBe("tracked"); }); it("rejects invalid raw script arguments before mutating Redis", async () => { @@ -285,7 +324,7 @@ describe.each(engines)("DialCache Lua protocol on $name", ({ image }) => { await expect(client.dialcacheWriteTracked(valueKey, watermarkKey, 60_000, 0, "replacement", 60_000)).rejects.toThrow( "invalid DialCache watermark", ); - expect(await scriptClient.read({ valueKey })).toEqual({ encoding: "utf8", value: "original" }); + expect(await scriptClient.read({ valueKey })).toBe("original"); } }); @@ -431,7 +470,6 @@ describe.each(engines)("DialCache Lua protocol on $name", ({ image }) => { valueKey, watermarkKey, cacheTtlMs: 2_000, - encoding: "utf8", value: "cached", watermarkTtlFloorMs: 1_000, }); @@ -439,7 +477,7 @@ describe.each(engines)("DialCache Lua protocol on $name", ({ image }) => { expect(wrote).toBe(true); expect(await client.get(watermarkKey)).toBe("1.75"); expect(await client.pTTL(watermarkKey)).toBeGreaterThanOrEqual(61_000); - expect(await scriptClient.read({ valueKey, watermarkKey })).toEqual({ encoding: "utf8", value: "cached" }); + expect(await scriptClient.read({ valueKey, watermarkKey })).toBe("cached"); }); it("does not rewrite sufficient or persistent watermarks on tracked writes", async () => { @@ -457,7 +495,6 @@ describe.each(engines)("DialCache Lua protocol on $name", ({ image }) => { valueKey: sufficientValueKey, watermarkKey: sufficientWatermarkKey, cacheTtlMs: 2_000, - encoding: "utf8", value: "cached", watermarkTtlFloorMs: 1_000, }), @@ -476,7 +513,6 @@ describe.each(engines)("DialCache Lua protocol on $name", ({ image }) => { valueKey: persistentValueKey, watermarkKey: persistentWatermarkKey, cacheTtlMs: 2_000, - encoding: "utf8", value: "cached", watermarkTtlFloorMs: 1_000, }), @@ -496,7 +532,6 @@ describe.each(engines)("DialCache Lua protocol on $name", ({ image }) => { valueKey, watermarkKey, cacheTtlMs: 2_000, - encoding: "utf8" as const, value: "cached", watermarkTtlFloorMs: 1_000, }; @@ -509,13 +544,13 @@ describe.each(engines)("DialCache Lua protocol on $name", ({ image }) => { await scriptClient.invalidate({ watermarkKey, futureBufferMs: 100, watermarkTtlFloorMs: 1_000 }); expect(await scriptClient.read({ valueKey, watermarkKey })).toBeNull(); expect(await scriptClient.write({ ...writeRequest, value: "blocked" })).toBe(false); - expect(await scriptClient.read({ valueKey })).toEqual({ encoding: "utf8", value: "cached" }); + expect(await scriptClient.read({ valueKey })).toBe("cached"); const ttlBeforeRead = await client.pTTL(watermarkKey); await scriptClient.read({ valueKey, watermarkKey }); expect(await client.pTTL(watermarkKey)).toBeLessThanOrEqual(ttlBeforeRead); await new Promise((resolve) => setTimeout(resolve, 110)); expect(await scriptClient.write({ ...writeRequest, value: "fresh" })).toBe(true); - expect(await scriptClient.read({ valueKey, watermarkKey })).toEqual({ encoding: "utf8", value: "fresh" }); + expect(await scriptClient.read({ valueKey, watermarkKey })).toBe("fresh"); }); });