From 5e0d83442487faa74b6686e6a3d23699229b9dc2 Mon Sep 17 00:00:00 2001 From: Lev Neiman Date: Sun, 12 Jul 2026 19:21:32 -0700 Subject: [PATCH 1/2] codex: add Valkey GLIDE adapter --- README.md | 33 ++- package.json | 17 +- pnpm-lock.yaml | 71 +++++ scripts/test-package.mjs | 52 +++- src/internal/redis-payload.ts | 26 ++ src/node-redis.ts | 26 +- src/valkey-glide.ts | 127 +++++++++ test/redis-adapters.integration.test.ts | 206 +++++++++++++++ ...redis-cluster-adapters.integration.test.ts | 250 ++++++++++++++++++ test/redis-cluster.integration.test.ts | 218 --------------- test/valkey-glide.test.ts | 183 +++++++++++++ 11 files changed, 963 insertions(+), 246 deletions(-) create mode 100644 src/internal/redis-payload.ts create mode 100644 src/valkey-glide.ts create mode 100644 test/redis-adapters.integration.test.ts create mode 100644 test/redis-cluster-adapters.integration.test.ts delete mode 100644 test/redis-cluster.integration.test.ts create mode 100644 test/valkey-glide.test.ts diff --git a/README.md b/README.md index 5b9a520..5e35aea 100644 --- a/README.md +++ b/README.md @@ -6,8 +6,10 @@ Fine-grained TypeScript caching with explicit enabled contexts, stable key const ```bash pnpm add dialcache -# For Redis-backed caching: +# Choose a Redis client when using the remote layer: pnpm add redis@~4.7.1 +# or +pnpm add @valkey/valkey-glide@^2.4.2 ``` DialCache requires Node.js 20 or Node.js 22 and newer. @@ -61,6 +63,30 @@ const dialcache = new DialCache({ The `redis.client` and `redis.createClient` options accept the semantic `DialCacheRedisClient` interface. Node-redis users should register the supplied scripts and wrap their client with `createNodeRedisDialCacheClient` as shown above. +Valkey GLIDE users pass an already-created standalone or cluster client to the GLIDE adapter: + +```ts +import { GlideClient } from "@valkey/valkey-glide"; +import { DialCache } from "dialcache"; +import { createValkeyGlideDialCacheClient } from "dialcache/valkey-glide"; + +const glideClient = await GlideClient.createClient({ + addresses: [{ host: "127.0.0.1", port: 6379 }], +}); +const redisClient = createValkeyGlideDialCacheClient(glideClient); +const dialcache = new DialCache({ + redis: { client: redisClient, keyPrefix: "dialcache:" }, +}); + +function shutdown(): void { + // Release adapter-owned scripts before closing GLIDE. + redisClient.dispose(); + glideClient.close(); +} +``` + +DialCache does not create, connect, or close the underlying Redis client. After outstanding cache operations finish, the GLIDE adapter's `dispose()` method releases its five native `Script` handles; it is idempotent and does not close the wrapped GLIDE connection. + When caching is enabled, reads flow through: ```text @@ -81,7 +107,7 @@ const dialcache = new DialCache({ localMaxSize: 25_000 }); The limit counts entries rather than estimating JavaScript object memory. Recently read entries stay resident ahead of less recently used entries when the limit is reached. -Node-redis computes each script's SHA, uses `EVALSHA`, and retries with `EVAL` after `NOSCRIPT`. Its cluster client routes scripts by their first key and performs that fallback on the selected shard. Tracked reads are deliberately routed to primaries so a lagging Redis replica cannot hide an invalidation watermark. The adapter also fans `flushAll()` across every cluster master instead of silently clearing one shard. +Node-redis computes each script's SHA, uses `EVALSHA`, and retries with `EVAL` after `NOSCRIPT`. Its cluster client routes scripts by their first key and performs that fallback on the selected shard. The GLIDE adapter uses GLIDE's native `Script` lifecycle and byte decoder; GLIDE routes scripts from their declared keys and the adapter broadcasts `flushAll()` to all cluster primaries. Tracked reads are deliberately routed to primaries so a lagging replica cannot hide an invalidation watermark. You can also provide a lazy factory that returns a script-enabled client: @@ -100,7 +126,7 @@ const dialcache = new DialCache({ }); ``` -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. +The core Redis boundary is the client-agnostic `DialCacheRedisClient` interface. It exchanges serialized values as `string | Buffer` and does not expose client commands or wire encodings. Distinct untracked/tracked read and write Lua sources, the invalidation source, and wire constants are available from `dialcache/redis-protocol`. Custom adapters can use the root-exported `DialCacheRedisPayloadError` and `DialCacheRedisPayloadEncodingError` classes to preserve the standard metrics labels. Redis values use a compact binary frame: @@ -294,6 +320,7 @@ Included: - Lua-backed Redis reads and writes with Redis-generated timestamps - Versioned binary Redis frames for UTF-8 and Buffer serializer output - Native node-redis script registration with automatic `NOSCRIPT` recovery +- Native Valkey GLIDE adapter with explicit script disposal and automatic script-cache recovery - Standalone Redis, Valkey, and Redis Cluster support - JSON and custom serializer support for Redis values - Duplicate and reserved use-case validation diff --git a/package.json b/package.json index 32bfec4..31f66c0 100644 --- a/package.json +++ b/package.json @@ -29,6 +29,16 @@ "default": "./dist/node-redis.cjs" } }, + "./valkey-glide": { + "import": { + "types": "./dist/valkey-glide.d.ts", + "default": "./dist/valkey-glide.js" + }, + "require": { + "types": "./dist/valkey-glide.d.cts", + "default": "./dist/valkey-glide.cjs" + } + }, "./redis-protocol": { "import": { "types": "./dist/redis-protocol.d.ts", @@ -46,7 +56,7 @@ "LICENSE" ], "scripts": { - "build": "tsup src/index.ts src/node-redis.ts src/redis-protocol.ts --format esm,cjs --dts --clean", + "build": "tsup src/index.ts src/node-redis.ts src/redis-protocol.ts src/valkey-glide.ts --format esm,cjs --dts --clean", "check": "pnpm typecheck && pnpm test && pnpm build && pnpm test:package", "typecheck": "tsc --noEmit", "test": "vitest run --coverage", @@ -64,6 +74,7 @@ "prom-client": "^15.1.3" }, "devDependencies": { + "@valkey/valkey-glide": "^2.4.2", "@vitest/coverage-v8": "^4.0.14", "redis": "~4.7.1", "testcontainers": "^12.0.4", @@ -72,9 +83,13 @@ "vitest": "^4.0.14" }, "peerDependencies": { + "@valkey/valkey-glide": "^2.4.2", "redis": "~4.7.1" }, "peerDependenciesMeta": { + "@valkey/valkey-glide": { + "optional": true + }, "redis": { "optional": true } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index f596511..b846019 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -18,6 +18,9 @@ importers: specifier: ^15.1.3 version: 15.1.3 devDependencies: + '@valkey/valkey-glide': + specifier: ^2.4.2 + version: 2.4.2 '@vitest/coverage-v8': specifier: ^4.0.14 version: 4.1.10(vitest@4.1.10) @@ -610,6 +613,44 @@ packages: '@types/ssh2@1.15.5': resolution: {integrity: sha512-N1ASjp/nXH3ovBHddRJpli4ozpk6UdDYIX4RJWFa9L1YKnzdhTlVmiGHm4DZnj/jLbqZpes4aeR30EFGQtvhQQ==} + '@valkey/valkey-glide-darwin-arm64@2.4.2': + resolution: {integrity: sha512-ILJt4/nGWvzWbrNTUFCvOmUnUoLhOHDKAdCQBTTZtt61JRv4z1ctS8kSkHXtdiwWYCPZIsqjjIsb43HdujkuKQ==} + cpu: [arm64] + os: [darwin] + + '@valkey/valkey-glide-darwin-x64@2.4.2': + resolution: {integrity: sha512-qd68iRRAdQ3maWJMgf44JE6v4dk9S5mCWXcRwBa6BzjqzLNoqiHa13lf+YOzaEj4EQEPKz6leBIfQEqIs1ZNMw==} + cpu: [x64] + os: [darwin] + + '@valkey/valkey-glide-linux-arm64-gnu@2.4.2': + resolution: {integrity: sha512-zzstOJvpXHHZSnwBdkk3J0fz8v3+HjPGa/4PktldBTR5VZ3ZfS5+HFuXdRR6/nNpww5fcTQ0KRYdAONUWDiWsw==} + cpu: [arm64] + os: [linux] + libc: [glibc] + + '@valkey/valkey-glide-linux-arm64-musl@2.4.2': + resolution: {integrity: sha512-Li7htG1mAikPRN2v9IggIuFED6zGjMRlcYS+42qbNeXRjyYG/V4M4Y0TjAhS2qte6jRqpgCbuyEQSIVt5JSWEQ==} + cpu: [arm64] + os: [linux] + libc: [musl] + + '@valkey/valkey-glide-linux-x64-gnu@2.4.2': + resolution: {integrity: sha512-3cJfKpJSzXgsVwhNdeQcHO73L5K42V1iZaLGpLsXPwtuEO8olSu/tslN24AYgjwTM3QghOQb0k66x6JonB+3rA==} + cpu: [x64] + os: [linux] + libc: [glibc] + + '@valkey/valkey-glide-linux-x64-musl@2.4.2': + resolution: {integrity: sha512-YBPVVW84VSOFfIME6jqE6DAQKiV688oWkhjNmakJ/xm4Pxae78QWNBI0s2Rbwpfy7PQ59gJ/ZfwPVbMHOXp/vw==} + cpu: [x64] + os: [linux] + libc: [musl] + + '@valkey/valkey-glide@2.4.2': + resolution: {integrity: sha512-ugE9hxB+0QoXhix2PgmtJZqWAEFqS8iJsWRqn8dgqdBrOwKUv9LidWuEzF/WBGyYZDj9Ji/g8CvaskA2fHZvTw==} + engines: {node: '>=16'} + '@vitest/coverage-v8@4.1.10': resolution: {integrity: sha512-IM49HmthevbgAO4anp1hwtoT9wYe59w0LR00gr+eagHE+ZJ5lK4sLPeO0ubgoJcwLk6dehU3R24N+FbEEKDc8g==} peerDependencies: @@ -2026,6 +2067,36 @@ snapshots: dependencies: '@types/node': 18.19.130 + '@valkey/valkey-glide-darwin-arm64@2.4.2': + optional: true + + '@valkey/valkey-glide-darwin-x64@2.4.2': + optional: true + + '@valkey/valkey-glide-linux-arm64-gnu@2.4.2': + optional: true + + '@valkey/valkey-glide-linux-arm64-musl@2.4.2': + optional: true + + '@valkey/valkey-glide-linux-x64-gnu@2.4.2': + optional: true + + '@valkey/valkey-glide-linux-x64-musl@2.4.2': + optional: true + + '@valkey/valkey-glide@2.4.2': + dependencies: + long: 5.3.2 + protobufjs: 7.6.5 + optionalDependencies: + '@valkey/valkey-glide-darwin-arm64': 2.4.2 + '@valkey/valkey-glide-darwin-x64': 2.4.2 + '@valkey/valkey-glide-linux-arm64-gnu': 2.4.2 + '@valkey/valkey-glide-linux-arm64-musl': 2.4.2 + '@valkey/valkey-glide-linux-x64-gnu': 2.4.2 + '@valkey/valkey-glide-linux-x64-musl': 2.4.2 + '@vitest/coverage-v8@4.1.10(vitest@4.1.10)': dependencies: '@bcoe/v8-coverage': 1.0.2 diff --git a/scripts/test-package.mjs b/scripts/test-package.mjs index e7f9e6d..f3a0ee6 100644 --- a/scripts/test-package.mjs +++ b/scripts/test-package.mjs @@ -11,6 +11,7 @@ const workspace = await mkdtemp(join(tmpdir(), "dialcache-package-")); const consumer = `import { DialCache, DialCacheKeyConfig, type DialCacheRedisClient } from "dialcache"; import { createNodeRedisDialCacheClient } from "dialcache/node-redis"; import { READ_CACHE_SCRIPT } from "dialcache/redis-protocol"; +import { createValkeyGlideDialCacheClient, type ValkeyGlideDialCacheClient } from "dialcache/valkey-glide"; const cache = new DialCache(); const load = cache.cached(async (id: string) => id, { @@ -22,6 +23,7 @@ const load = cache.cached(async (id: string) => id, { void load; void createNodeRedisDialCacheClient; +void createValkeyGlideDialCacheClient; void READ_CACHE_SCRIPT; const customRedisClient: DialCacheRedisClient = { @@ -31,6 +33,8 @@ const customRedisClient: DialCacheRedisClient = { flushAll: async () => undefined, }; void customRedisClient; +const glideRedisClient: ValkeyGlideDialCacheClient | undefined = undefined; +void glideRedisClient; `; try { @@ -54,6 +58,47 @@ try { { cwd: workspace }, ); + let glideWasInstalled = false; + try { + await exec(process.execPath, ["--eval", "require.resolve('@valkey/valkey-glide')"], { cwd: workspace }); + glideWasInstalled = true; + } catch { + // Optional peers remain absent until the consumer selects the corresponding adapter. + } + if (glideWasInstalled) { + throw new Error("The optional Valkey GLIDE peer was installed automatically"); + } + + await exec( + process.execPath, + [ + "--input-type=module", + "--eval", + "await import('dialcache'); await import('dialcache/redis-protocol'); await import('dialcache/node-redis')", + ], + { cwd: workspace }, + ); + await exec( + process.execPath, + ["--eval", "require('dialcache'); require('dialcache/redis-protocol'); require('dialcache/node-redis')"], + { cwd: workspace }, + ); + + await exec( + "npm", + [ + "install", + "--ignore-scripts", + "--no-package-lock", + "--no-save", + join(workspace, tarball), + "redis@~4.7.1", + "typescript@5.9.3", + "@valkey/valkey-glide@^2.4.2", + ], + { cwd: workspace }, + ); + await Promise.all([ writeFile(join(workspace, "consumer.mts"), consumer), writeFile(join(workspace, "consumer.cts"), consumer), @@ -80,13 +125,16 @@ try { [ "--input-type=module", "--eval", - "await import('dialcache'); await import('dialcache/redis-protocol'); await import('dialcache/node-redis')", + "await import('dialcache'); await import('dialcache/redis-protocol'); await import('dialcache/node-redis'); await import('dialcache/valkey-glide')", ], { cwd: workspace }, ); await exec( process.execPath, - ["--eval", "require('dialcache'); require('dialcache/redis-protocol'); require('dialcache/node-redis')"], + [ + "--eval", + "require('dialcache'); require('dialcache/redis-protocol'); require('dialcache/node-redis'); require('dialcache/valkey-glide')", + ], { cwd: workspace }, ); await exec( diff --git a/src/internal/redis-payload.ts b/src/internal/redis-payload.ts new file mode 100644 index 0000000..dccfdea --- /dev/null +++ b/src/internal/redis-payload.ts @@ -0,0 +1,26 @@ +import { + DialCacheRedisPayloadEncodingError, + DialCacheRedisPayloadError, + type RedisCachePayload, +} from "../redis-client.js"; +import { REDIS_ENCODING_BINARY, REDIS_ENCODING_UTF8 } from "./redis-scripts.js"; + +export function redisPayloadEncoding(value: RedisCachePayload): number { + return Buffer.isBuffer(value) ? REDIS_ENCODING_BINARY : REDIS_ENCODING_UTF8; +} + +export function decodeRedisPayload(raw: Buffer): RedisCachePayload { + if (raw.length === 0) { + throw new DialCacheRedisPayloadError("Invalid DialCache Redis payload"); + } + + const encoding = raw[0]; + const payload = raw.subarray(1); + if (encoding === REDIS_ENCODING_UTF8) { + return payload.toString("utf8"); + } + if (encoding === REDIS_ENCODING_BINARY) { + return payload; + } + throw new DialCacheRedisPayloadEncodingError("Invalid DialCache Redis payload encoding"); +} diff --git a/src/node-redis.ts b/src/node-redis.ts index 8853821..e38b415 100644 --- a/src/node-redis.ts +++ b/src/node-redis.ts @@ -4,13 +4,11 @@ import { INVALIDATE_CACHE_SCRIPT, READ_CACHE_SCRIPT, READ_TRACKED_CACHE_SCRIPT, - REDIS_ENCODING_BINARY, - REDIS_ENCODING_UTF8, WRITE_CACHE_SCRIPT, WRITE_TRACKED_CACHE_SCRIPT, } from "./internal/redis-scripts.js"; -import { DialCacheRedisPayloadEncodingError, DialCacheRedisPayloadError } from "./redis-client.js"; -import type { DialCacheRedisClient, RedisCachePayload } from "./redis-client.js"; +import { decodeRedisPayload, redisPayloadEncoding } from "./internal/redis-payload.js"; +import type { DialCacheRedisClient } from "./redis-client.js"; type BufferReplyOptions = ReturnType>; // Redis bulk strings are binary data; decoding them as UTF-8 would corrupt arbitrary serializer output. @@ -158,11 +156,11 @@ export function createNodeRedisDialCacheClient(client: NodeRedisScriptClient): D const raw = watermarkKey === undefined ? await client.dialcacheRead(bufferReplyOptions, valueKey) : await client.dialcacheReadTracked(bufferReplyOptions, valueKey, watermarkKey); - return raw === null ? null : decodePayload(raw); + return raw === null ? null : decodeRedisPayload(raw); }, async write(request) { const { valueKey, watermarkKey, cacheTtlMs, value } = request; - const encodingByte = Buffer.isBuffer(value) ? REDIS_ENCODING_BINARY : REDIS_ENCODING_UTF8; + const encodingByte = redisPayloadEncoding(value); const result = watermarkKey === undefined ? await client.dialcacheWrite(valueKey, cacheTtlMs, encodingByte, value) : await client.dialcacheWriteTracked( @@ -200,22 +198,6 @@ export function createNodeRedisDialCacheClient(client: NodeRedisScriptClient): D }; } -function decodePayload(raw: Buffer): RedisCachePayload { - if (raw.length === 0) { - throw new DialCacheRedisPayloadError("Invalid DialCache Redis payload"); - } - - const encoding = raw[0]; - const payload = raw.subarray(1); - if (encoding === REDIS_ENCODING_UTF8) { - return payload.toString("utf8"); - } - if (encoding === REDIS_ENCODING_BINARY) { - return payload; - } - throw new DialCacheRedisPayloadEncodingError("Invalid DialCache Redis payload encoding"); -} - function clusterCommands(client: NodeRedisScriptClient): NodeRedisClusterCommands | null { if (!("masters" in client) || !("nodeClient" in client) || !Array.isArray(client.masters)) { return null; diff --git a/src/valkey-glide.ts b/src/valkey-glide.ts new file mode 100644 index 0000000..59f6037 --- /dev/null +++ b/src/valkey-glide.ts @@ -0,0 +1,127 @@ +import { + Decoder, + GlideClusterClient, + Script, + type GlideClient, + type GlideReturnType, + type GlideString, +} from "@valkey/valkey-glide"; + +import { decodeRedisPayload, redisPayloadEncoding } from "./internal/redis-payload.js"; +import { + INVALIDATE_CACHE_SCRIPT, + READ_CACHE_SCRIPT, + READ_TRACKED_CACHE_SCRIPT, + WRITE_CACHE_SCRIPT, + WRITE_TRACKED_CACHE_SCRIPT, +} from "./internal/redis-scripts.js"; +import { DialCacheRedisPayloadError, type DialCacheRedisClient } from "./redis-client.js"; + +type SupportedValkeyGlideClient = GlideClient | GlideClusterClient; + +interface DialCacheGlideScripts { + readonly read: Script; + readonly readTracked: Script; + readonly write: Script; + readonly writeTracked: Script; + readonly invalidate: Script; +} + +export interface ValkeyGlideDialCacheClient extends DialCacheRedisClient { + /** Release the adapter-owned GLIDE Script handles. Does not close the wrapped GLIDE client. */ + dispose(): void; +} + +export function createValkeyGlideDialCacheClient( + client: SupportedValkeyGlideClient, +): ValkeyGlideDialCacheClient { + const scripts: DialCacheGlideScripts = { + read: new Script(READ_CACHE_SCRIPT), + readTracked: new Script(READ_TRACKED_CACHE_SCRIPT), + write: new Script(WRITE_CACHE_SCRIPT), + writeTracked: new Script(WRITE_TRACKED_CACHE_SCRIPT), + invalidate: new Script(INVALIDATE_CACHE_SCRIPT), + }; + let disposed = false; + let activeInvocations = 0; + + const invoke = async ( + script: Script, + keys: GlideString[], + args: GlideString[] = [], + ): Promise => { + if (disposed) { + throw new Error("Valkey GLIDE DialCache client is disposed"); + } + activeInvocations += 1; + try { + return await client.invokeScript(script, { keys, args, decoder: Decoder.Bytes }); + } finally { + activeInvocations -= 1; + } + }; + + return { + async read({ valueKey, watermarkKey }) { + const raw = watermarkKey === undefined + ? await invoke(scripts.read, [valueKey]) + : await invoke(scripts.readTracked, [valueKey, watermarkKey]); + if (raw === null) { + return null; + } + if (!Buffer.isBuffer(raw)) { + throw new DialCacheRedisPayloadError("Invalid DialCache Redis payload reply"); + } + return decodeRedisPayload(raw); + }, + async write(request) { + const { valueKey, watermarkKey, cacheTtlMs, value } = request; + const encoding = redisPayloadEncoding(value); + const raw = watermarkKey === undefined + ? await invoke(scripts.write, [valueKey], [String(cacheTtlMs), String(encoding), value]) + : await invoke( + scripts.writeTracked, + [valueKey, watermarkKey], + [String(cacheTtlMs), String(encoding), value, String(request.watermarkTtlFloorMs)], + ); + return integerReply(raw, "write") === 1; + }, + async invalidate({ watermarkKey, futureBufferMs, watermarkTtlFloorMs }) { + const raw = await invoke( + scripts.invalidate, + [watermarkKey], + [String(futureBufferMs), String(watermarkTtlFloorMs)], + ); + integerReply(raw, "invalidate"); + }, + async flushAll() { + if (disposed) { + throw new Error("Valkey GLIDE DialCache client is disposed"); + } + if (client instanceof GlideClusterClient) { + await client.flushall({ route: "allPrimaries" }); + } else { + await client.flushall(); + } + }, + dispose() { + if (disposed) { + return; + } + if (activeInvocations > 0) { + throw new Error("Cannot dispose Valkey GLIDE DialCache client while operations are in flight"); + } + disposed = true; + for (const script of Object.values(scripts)) { + script.release(); + } + }, + }; +} + +function integerReply(reply: GlideReturnType, operation: string): number { + if (typeof reply !== "number") { + throw new Error(`Invalid DialCache Redis ${operation} reply`); + } + return reply; +} diff --git a/test/redis-adapters.integration.test.ts b/test/redis-adapters.integration.test.ts new file mode 100644 index 0000000..2fd470f --- /dev/null +++ b/test/redis-adapters.integration.test.ts @@ -0,0 +1,206 @@ +import { GlideClient } from "@valkey/valkey-glide"; +import { createClient } from "redis"; +import { GenericContainer, type StartedTestContainer, Wait } from "testcontainers"; +import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; + +import { + CacheLayer, + DialCache, + DialCacheKeyConfig, + DialCacheRedisPayloadEncodingError, + type DialCacheRedisClient, + type Serializer, +} from "../src/index.js"; +import { createNodeRedisDialCacheClient, dialcacheRedisScripts } from "../src/node-redis.js"; +import { + createValkeyGlideDialCacheClient, + type ValkeyGlideDialCacheClient, +} from "../src/valkey-glide.js"; + +const engines = [ + { name: "Redis 6.2", image: "redis:6.2-alpine" }, + { name: "Valkey 8", image: "valkey/valkey:8-alpine" }, +] as const; + +const adapterKinds = [ + { kind: "nodeRedis", name: "node-redis" }, + { kind: "valkeyGlide", name: "Valkey GLIDE" }, +] as const; +type AdapterKind = (typeof adapterKinds)[number]["kind"]; + +const remoteOnly = new DialCacheKeyConfig({ + ttlSec: { [CacheLayer.REMOTE]: 60 }, + ramp: { [CacheLayer.REMOTE]: 100 }, +}); + +const createNodeRedisClient = (url: string) => createClient({ url, scripts: dialcacheRedisScripts }); + +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)]); +} + +describe.each(engines)("DialCache Redis adapter conformance on $name", ({ image }) => { + let container: StartedTestContainer | undefined; + let admin: ReturnType | undefined; + let glide: GlideClient | undefined; + let glideAdapter: ValkeyGlideDialCacheClient | undefined; + let adapters: Record | undefined; + + beforeAll(async () => { + container = await new GenericContainer(image) + .withExposedPorts(6379) + .withWaitStrategy(Wait.forListeningPorts()) + .start(); + const host = container.getHost(); + const port = container.getMappedPort(6379); + admin = createNodeRedisClient(`redis://${host}:${port}`); + admin.on("error", () => undefined); + await admin.connect(); + glide = await GlideClient.createClient({ addresses: [{ host, port }] }); + glideAdapter = createValkeyGlideDialCacheClient(glide); + adapters = { + nodeRedis: createNodeRedisDialCacheClient(admin), + valkeyGlide: glideAdapter, + }; + }); + + beforeEach(async () => { + await admin?.flushAll(); + }); + + afterAll(async () => { + glideAdapter?.dispose(); + glide?.close(); + await admin?.quit(); + await container?.stop(); + }); + + describe.each(adapterKinds)("$name adapter", ({ kind }) => { + function adapter(): DialCacheRedisClient { + const active = adapters?.[kind]; + if (active === undefined) { + throw new Error("Redis adapters did not start"); + } + return active; + } + + it("round-trips UTF-8 and arbitrary binary payloads", async () => { + const client = adapter(); + const binary = Buffer.from(Array.from({ length: 256 }, (_, index) => index)); + + await expect(client.write({ valueKey: `${kind}:utf8`, cacheTtlMs: 60_000, value: "hello" })).resolves.toBe(true); + await expect(client.read({ valueKey: `${kind}:utf8` })).resolves.toBe("hello"); + await expect(client.write({ valueKey: `${kind}:binary`, cacheTtlMs: 60_000, value: binary })).resolves.toBe(true); + await expect(client.read({ valueKey: `${kind}:binary` })).resolves.toEqual(binary); + await expect( + client.write({ valueKey: `${kind}:empty-binary`, cacheTtlMs: 60_000, value: Buffer.alloc(0) }), + ).resolves.toBe(true); + await expect(client.read({ valueKey: `${kind}:empty-binary` })).resolves.toEqual(Buffer.alloc(0)); + }); + + it("round-trips tracked values and honors invalidation watermarks", async () => { + const client = adapter(); + const valueKey = `${kind}:{tracked}:value`; + const watermarkKey = `${kind}:{tracked}:watermark`; + const request = { + valueKey, + watermarkKey, + cacheTtlMs: 60_000, + value: Buffer.from([0, 0xff, 0x80]), + watermarkTtlFloorMs: 60_000, + } as const; + + await expect(client.write(request)).resolves.toBe(true); + await expect(client.read({ valueKey, watermarkKey })).resolves.toEqual(request.value); + await client.invalidate({ watermarkKey, futureBufferMs: 1_000, watermarkTtlFloorMs: 60_000 }); + await expect(client.read({ valueKey, watermarkKey })).resolves.toBeNull(); + await expect(client.write(request)).resolves.toBe(false); + }); + + it("recovers its scripts after the server script cache is flushed", async () => { + const client = adapter(); + const valueKey = `${kind}:script-recovery`; + if (admin === undefined) { + throw new Error("Redis admin client did not start"); + } + + await admin.scriptFlush(); + await expect(client.write({ valueKey, cacheTtlMs: 60_000, value: "cached" })).resolves.toBe(true); + await admin.scriptFlush(); + await expect(client.read({ valueKey })).resolves.toBe("cached"); + }); + + it("treats invalid frames as misses and preserves encoding errors", async () => { + const client = adapter(); + if (admin === undefined) { + throw new Error("Redis admin client did not start"); + } + + await admin.set(`${kind}:empty-frame`, Buffer.alloc(9)); + await expect(client.read({ valueKey: `${kind}:empty-frame` })).resolves.toBeNull(); + + await admin.set(`${kind}:bad-encoding`, encodeFrame("bad", 2), { PX: 60_000 }); + await expect(client.read({ valueKey: `${kind}:bad-encoding` })).rejects.toBeInstanceOf( + DialCacheRedisPayloadEncodingError, + ); + }); + + it("backs a complete DialCache serializer round trip", async () => { + const dialcache = new DialCache({ redis: { client: adapter(), keyPrefix: `${kind}:cache:` } }); + let calls = 0; + const serializer: Serializer = { + dump: async (value) => Buffer.from(value, "utf8"), + load: async (value) => (Buffer.isBuffer(value) ? value.toString("utf8") : value), + }; + const load = dialcache.cached(async (id: string) => `${id}:${++calls}`, { + keyType: "item_id", + useCase: `Adapter${kind}`, + cacheKey: (id) => id, + defaultConfig: remoteOnly, + serializer, + }); + + const first = await dialcache.enable(async () => await load("one")); + const second = await dialcache.enable(async () => await load("one")); + expect(first).toBe("one:1"); + expect(second).toBe(first); + expect(calls).toBe(1); + }); + + it("flushes all cached values", async () => { + const client = adapter(); + if (admin === undefined) { + throw new Error("Redis admin client did not start"); + } + await client.write({ valueKey: `${kind}:flush`, cacheTtlMs: 60_000, value: "cached" }); + await client.flushAll(); + await expect(admin.exists(`${kind}:flush`)).resolves.toBe(0); + }); + }); + + it("uses one wire format across node-redis and Valkey GLIDE", async () => { + if (adapters === undefined) { + throw new Error("Redis adapters did not start"); + } + const binary = Buffer.from([0, 0xff, 0xc3, 0x28, 0x80]); + + await adapters.nodeRedis.write({ valueKey: "interop:node-to-glide", cacheTtlMs: 60_000, value: binary }); + await expect(adapters.valkeyGlide.read({ valueKey: "interop:node-to-glide" })).resolves.toEqual(binary); + + await adapters.valkeyGlide.write({ valueKey: "interop:glide-to-node", cacheTtlMs: 60_000, value: "hello" }); + await expect(adapters.nodeRedis.read({ valueKey: "interop:glide-to-node" })).resolves.toBe("hello"); + + const valueKey = "interop:{tracked}:value"; + const watermarkKey = "interop:{tracked}:watermark"; + await adapters.nodeRedis.write({ + valueKey, + watermarkKey, + cacheTtlMs: 60_000, + value: binary, + watermarkTtlFloorMs: 60_000, + }); + await expect(adapters.valkeyGlide.read({ valueKey, watermarkKey })).resolves.toEqual(binary); + }); +}); diff --git a/test/redis-cluster-adapters.integration.test.ts b/test/redis-cluster-adapters.integration.test.ts new file mode 100644 index 0000000..041f17c --- /dev/null +++ b/test/redis-cluster-adapters.integration.test.ts @@ -0,0 +1,250 @@ +import { createServer } from "node:net"; + +import { GlideClusterClient } from "@valkey/valkey-glide"; +import { commandOptions, createCluster } from "redis"; +import { GenericContainer, type StartedTestContainer, Wait } from "testcontainers"; +import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; + +import { CacheLayer, DialCache, DialCacheKeyConfig, type DialCacheRedisClient } from "../src/index.js"; +import { createNodeRedisDialCacheClient, dialcacheRedisScripts } from "../src/node-redis.js"; +import { + createValkeyGlideDialCacheClient, + type ValkeyGlideDialCacheClient, +} from "../src/valkey-glide.js"; + +const adapterKinds = [ + { kind: "nodeRedis", name: "node-redis" }, + { kind: "valkeyGlide", name: "Valkey GLIDE" }, +] as const; +type AdapterKind = (typeof adapterKinds)[number]["kind"]; + +const remoteOnly = new DialCacheKeyConfig({ + ttlSec: { [CacheLayer.REMOTE]: 60 }, + ramp: { [CacheLayer.REMOTE]: 100 }, +}); + +const createNodeRedisCluster = (port: number) => + createCluster({ + rootNodes: [{ url: `redis://127.0.0.1:${port}` }], + scripts: dialcacheRedisScripts, + }); + +async function waitForCluster(container: StartedTestContainer, port: number): Promise { + for (let attempt = 0; attempt < 50; attempt += 1) { + const result = await container.exec(["redis-cli", "-p", String(port), "cluster", "info"]); + if (result.output.includes("cluster_state:ok")) { + return; + } + await new Promise((resolve) => setTimeout(resolve, 100)); + } + throw new Error("Redis Cluster did not become ready"); +} + +async function chooseClusterPorts(): Promise { + for (let attempt = 0; attempt < 100; attempt += 1) { + const base = 20_000 + Math.floor(Math.random() * 25_000); + const ports = [base, base + 1, base + 2] as const; + if ((await Promise.all(ports.map(portIsAvailable))).every(Boolean)) { + return ports; + } + } + throw new Error("Could not reserve ports for Redis Cluster integration tests"); +} + +async function portIsAvailable(port: number): Promise { + return await new Promise((resolve) => { + const server = createServer(); + server.once("error", () => resolve(false)); + server.listen(port, "127.0.0.1", () => { + server.close(() => resolve(true)); + }); + }); +} + +function redisClusterCommand(ports: readonly number[]): string[] { + const commands = ports.map( + (port) => + `redis-server --port ${port} --cluster-enabled yes ` + + `--cluster-config-file /tmp/nodes-${port}.conf --cluster-node-timeout 5000 ` + + `--appendonly no --protected-mode no --bind 0.0.0.0 ` + + `--cluster-announce-ip 127.0.0.1 --cluster-announce-port ${port} ` + + `--cluster-announce-bus-port ${port + 10_000} --daemonize yes`, + ); + return ["sh", "-c", `${commands.join(" && ")} && tail -f /dev/null`]; +} + +describe("DialCache Redis adapter conformance on Redis Cluster", () => { + let container: StartedTestContainer | undefined; + let nodeCluster: ReturnType | undefined; + let glideCluster: GlideClusterClient | undefined; + let glideAdapter: ValkeyGlideDialCacheClient | undefined; + let adapters: Record | undefined; + + beforeAll(async () => { + const ports = await chooseClusterPorts(); + const exposedPorts = ports.map((port) => ({ + container: port, + host: port, + })); + container = await new GenericContainer("redis:7-alpine") + .withExposedPorts(...exposedPorts) + .withCommand(redisClusterCommand(ports)) + .withWaitStrategy(Wait.forListeningPorts()) + .start(); + + const createResult = await container.exec([ + "redis-cli", + "--cluster", + "create", + ...ports.map((port) => `127.0.0.1:${port}`), + "--cluster-replicas", + "0", + "--cluster-yes", + ]); + if (createResult.exitCode !== 0) { + throw new Error(`Could not create Redis Cluster: ${createResult.output}`); + } + await waitForCluster(container, ports[0]); + + nodeCluster = createNodeRedisCluster(ports[0]); + nodeCluster.on("error", () => undefined); + await nodeCluster.connect(); + glideCluster = await GlideClusterClient.createClient({ + addresses: ports.map((port) => ({ host: "127.0.0.1", port })), + }); + glideAdapter = createValkeyGlideDialCacheClient(glideCluster); + adapters = { + nodeRedis: createNodeRedisDialCacheClient(nodeCluster), + valkeyGlide: glideAdapter, + }; + }); + + beforeEach(async () => { + await adapters?.nodeRedis.flushAll(); + }); + + afterAll(async () => { + glideAdapter?.dispose(); + glideCluster?.close(); + await nodeCluster?.quit(); + await container?.stop(); + }); + + describe.each(adapterKinds)("$name adapter", ({ kind }) => { + function adapter(): DialCacheRedisClient { + const active = adapters?.[kind]; + if (active === undefined) { + throw new Error("Redis Cluster adapters did not start"); + } + return active; + } + + it("routes cached operations across cluster slots", async () => { + const dialcache = new DialCache({ redis: { client: adapter(), keyPrefix: `${kind}:cluster:` } }); + const ids = Array.from({ length: 30 }, (_, index) => `item-${index}`); + let calls = 0; + const load = dialcache.cached(async (id: string) => ({ id, calls: ++calls }), { + keyType: "item_id", + useCase: `Cluster${kind}`, + cacheKey: (id) => id, + defaultConfig: remoteOnly, + }); + + const first = await dialcache.enable(async () => await Promise.all(ids.map(load))); + const second = await dialcache.enable(async () => await Promise.all(ids.map(load))); + expect(second).toEqual(first); + expect(calls).toBe(ids.length); + }); + + it("keeps tracked keys colocated and rejects mismatched hash slots", async () => { + const client = adapter(); + const valueKey = `${kind}:{same}:value`; + const watermarkKey = `${kind}:{same}:watermark`; + await expect( + client.write({ + valueKey, + watermarkKey, + cacheTtlMs: 60_000, + value: "tracked", + watermarkTtlFloorMs: 60_000, + }), + ).resolves.toBe(true); + await expect(client.read({ valueKey, watermarkKey })).resolves.toBe("tracked"); + await expect( + client.read({ valueKey: `${kind}:{slot-a}:value`, watermarkKey: `${kind}:{slot-b}:watermark` }), + ).rejects.toThrow(/CROSSSLOT|same slot/i); + }); + + it("round-trips binary payloads through cluster script routing", async () => { + const client = adapter(); + if (nodeCluster === undefined) { + throw new Error("Redis Cluster did not start"); + } + const payload = Buffer.from([0, 0xff, 0xc3, 0x28, 0x80]); + const valueKey = `${kind}:{binary}:value`; + await expect(client.write({ valueKey, cacheTtlMs: 60_000, value: payload })).resolves.toBe(true); + await expect(client.read({ valueKey })).resolves.toEqual(payload); + const stored = await nodeCluster.get(commandOptions({ returnBuffers: true }), valueKey); + expect(stored?.[0]).toBe(1); + expect(stored?.[9]).toBe(1); + expect(stored?.subarray(10)).toEqual(payload); + }); + + it("reloads scripts on every shard after SCRIPT FLUSH", async () => { + const client = adapter(); + if (nodeCluster === undefined) { + throw new Error("Redis Cluster did not start"); + } + const activeCluster = nodeCluster; + await Promise.all( + activeCluster.masters.map(async (master) => { + const node = await activeCluster.nodeClient(master); + await node.scriptFlush(); + }), + ); + + const keys = Array.from({ length: 30 }, (_, index) => `${kind}:recovery:{item-${index}}:value`); + await Promise.all(keys.map(async (valueKey) => await client.write({ valueKey, cacheTtlMs: 60_000, value: valueKey }))); + await expect(Promise.all(keys.map(async (valueKey) => await client.read({ valueKey })))).resolves.toEqual(keys); + }); + + it("flushes every primary shard", async () => { + const client = adapter(); + if (nodeCluster === undefined) { + throw new Error("Redis Cluster did not start"); + } + const activeCluster = nodeCluster; + const keys = Array.from({ length: 30 }, (_, index) => `${kind}:flush:{item-${index}}:value`); + await Promise.all(keys.map(async (valueKey) => await client.write({ valueKey, cacheTtlMs: 60_000, value: valueKey }))); + const sizesBefore = await Promise.all( + activeCluster.masters.map(async (master) => { + const node = await activeCluster.nodeClient(master); + return await node.dbSize(); + }), + ); + expect(sizesBefore.every((size) => size > 0)).toBe(true); + await client.flushAll(); + const sizes = await Promise.all( + activeCluster.masters.map(async (master) => { + const node = await activeCluster.nodeClient(master); + return await node.dbSize(); + }), + ); + expect(sizes).toEqual([0, 0, 0]); + }); + }); + + it("uses one cluster wire format across node-redis and Valkey GLIDE", async () => { + if (adapters === undefined) { + throw new Error("Redis Cluster adapters did not start"); + } + const nodeValueKey = "interop:{node-to-glide}:value"; + const glideValueKey = "interop:{glide-to-node}:value"; + const binary = Buffer.from([0, 0xff, 0x80]); + + await adapters.nodeRedis.write({ valueKey: nodeValueKey, cacheTtlMs: 60_000, value: binary }); + await expect(adapters.valkeyGlide.read({ valueKey: nodeValueKey })).resolves.toEqual(binary); + await adapters.valkeyGlide.write({ valueKey: glideValueKey, cacheTtlMs: 60_000, value: "hello" }); + await expect(adapters.nodeRedis.read({ valueKey: glideValueKey })).resolves.toBe("hello"); + }); +}); diff --git a/test/redis-cluster.integration.test.ts b/test/redis-cluster.integration.test.ts deleted file mode 100644 index 7a264e2..0000000 --- a/test/redis-cluster.integration.test.ts +++ /dev/null @@ -1,218 +0,0 @@ -import { commandOptions, createCluster, type RedisClusterOptions } from "redis"; -import { - GenericContainer, - Network, - type StartedNetwork, - type StartedTestContainer, - Wait, -} from "testcontainers"; -import { afterAll, beforeAll, describe, expect, it } from "vitest"; - -import { CacheLayer, DialCache, DialCacheKeyConfig, type DialCacheRedisClient } from "../src/index.js"; -import { createNodeRedisDialCacheClient, dialcacheRedisScripts } from "../src/node-redis.js"; - -const remoteOnly = new DialCacheKeyConfig({ - ttlSec: { [CacheLayer.REMOTE]: 60 }, - ramp: { [CacheLayer.REMOTE]: 100 }, -}); - -const createTestCluster = (options: RedisClusterOptions) => - createCluster({ - ...options, - scripts: dialcacheRedisScripts, - }); - -async function waitForCluster(container: StartedTestContainer): Promise { - for (let attempt = 0; attempt < 50; attempt += 1) { - const result = await container.exec(["redis-cli", "cluster", "info"]); - if (result.output.includes("cluster_state:ok")) { - return; - } - await new Promise((resolve) => setTimeout(resolve, 100)); - } - throw new Error("Redis Cluster did not become ready"); -} - -describe("DialCache Lua protocol on Redis Cluster", () => { - let network: StartedNetwork | undefined; - let containers: Array = []; - let cluster: ReturnType | undefined; - - beforeAll(async () => { - const startedNetwork = await new Network().start(); - network = startedNetwork; - for (const alias of ["redis-1", "redis-2", "redis-3"]) { - containers.push( - await new GenericContainer("redis:7-alpine") - .withNetwork(startedNetwork) - .withNetworkAliases(alias) - .withExposedPorts(6379) - .withCommand([ - "redis-server", - "--port", - "6379", - "--cluster-enabled", - "yes", - "--cluster-config-file", - "/tmp/nodes.conf", - "--cluster-node-timeout", - "5000", - "--appendonly", - "no", - "--protected-mode", - "no", - ]) - .withWaitStrategy(Wait.forListeningPorts()) - .start(), - ); - } - - const networkName = network.getName(); - const internalAddresses = containers.map((container) => `${container.getIpAddress(networkName)}:6379`); - const firstContainer = containers[0]; - if (firstContainer === undefined) { - throw new Error("Redis Cluster containers did not start"); - } - const createResult = await firstContainer.exec([ - "redis-cli", - "--cluster", - "create", - ...internalAddresses, - "--cluster-replicas", - "0", - "--cluster-yes", - ]); - if (createResult.exitCode !== 0) { - throw new Error(`Could not create Redis Cluster: ${createResult.output}`); - } - await Promise.all(containers.map(waitForCluster)); - - const nodeAddressMap = Object.fromEntries( - containers.map((container) => [ - `${container.getIpAddress(networkName)}:6379`, - { host: container.getHost(), port: container.getMappedPort(6379) }, - ]), - ); - cluster = createTestCluster({ - rootNodes: [{ url: `redis://${firstContainer.getHost()}:${firstContainer.getMappedPort(6379)}` }], - nodeAddressMap, - }); - cluster.on("error", () => undefined); - await cluster.connect(); - }); - - afterAll(async () => { - await cluster?.quit(); - await Promise.all(containers.map(async (container) => await container.stop())); - await network?.stop(); - }); - - it("routes scripts across slots and reloads them per node", async () => { - if (cluster === undefined) { - throw new Error("Redis Cluster did not start"); - } - const activeCluster = cluster; - const scriptClient: DialCacheRedisClient = createNodeRedisDialCacheClient(activeCluster); - const dialcache = new DialCache({ redis: { client: scriptClient, keyPrefix: "cluster:" } }); - const ids = Array.from({ length: 30 }, (_, index) => `item-${index}`); - let calls = 0; - const getValue = dialcache.cached(async (id: string) => ({ id, calls: ++calls }), { - keyType: "item_id", - useCase: "ClusterSlots", - cacheKey: (id) => id, - defaultConfig: remoteOnly, - }); - - const first = await dialcache.enable(async () => await Promise.all(ids.map(getValue))); - const sizesBeforeFlush = await Promise.all( - activeCluster.masters.map(async (master) => { - const client = await activeCluster.nodeClient(master); - return await client.dbSize(); - }), - ); - await Promise.all( - activeCluster.masters.map(async (master) => { - const client = await activeCluster.nodeClient(master); - await client.scriptFlush(); - }), - ); - const second = await dialcache.enable(async () => await Promise.all(ids.map(getValue))); - await dialcache.flushAll(); - const sizesAfterFlush = await Promise.all( - activeCluster.masters.map(async (master) => { - const client = await activeCluster.nodeClient(master); - return await client.dbSize(); - }), - ); - const afterFlush = await dialcache.enable(async () => await Promise.all(ids.map(getValue))); - - expect(first.map(({ id }) => id)).toEqual(ids); - expect(calls).toBe(60); - expect(sizesBeforeFlush.every((size) => size > 0)).toBe(true); - expect(second).toEqual(first); - expect(sizesAfterFlush).toEqual([0, 0, 0]); - expect(afterFlush.map(({ id }) => id)).toEqual(ids); - expect(new Set(afterFlush.map(({ calls: call }) => call))).toEqual(new Set(ids.map((_, index) => index + 31))); - }); - - it("keeps tracked keys colocated and rejects mismatched hash tags", async () => { - if (cluster === undefined) { - throw new Error("Redis Cluster did not start"); - } - expect(dialcacheRedisScripts.dialcacheRead.IS_READ_ONLY).toBe(true); - expect(dialcacheRedisScripts.dialcacheReadTracked.IS_READ_ONLY).toBe(false); - expect(dialcacheRedisScripts.dialcacheRead.SHA1).not.toBe(dialcacheRedisScripts.dialcacheReadTracked.SHA1); - expect(dialcacheRedisScripts.dialcacheWrite.SHA1).not.toBe(dialcacheRedisScripts.dialcacheWriteTracked.SHA1); - const scriptClient: DialCacheRedisClient = createNodeRedisDialCacheClient(cluster); - const dialcache = new DialCache({ redis: { client: scriptClient, keyPrefix: "cluster:" } }); - let version = 1; - const getUser = dialcache.cached(async (id: string) => ({ id, version }), { - keyType: "user_id", - useCase: "ClusterTracked", - cacheKey: (id) => id, - trackForInvalidation: true, - defaultConfig: remoteOnly, - }); - - const before = await dialcache.enable(async () => await getUser("123")); - version = 2; - await dialcache.invalidateRemote("user_id", "123"); - await new Promise((resolve) => setTimeout(resolve, 2)); - const after = await dialcache.enable(async () => await getUser("123")); - - expect(before).toEqual({ id: "123", version: 1 }); - 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/valkey-glide.test.ts b/test/valkey-glide.test.ts new file mode 100644 index 0000000..dac1e33 --- /dev/null +++ b/test/valkey-glide.test.ts @@ -0,0 +1,183 @@ +import { GlideClusterClient, Script } from "@valkey/valkey-glide"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { + DialCacheRedisPayloadEncodingError, + DialCacheRedisPayloadError, +} from "../src/redis-client.js"; +import { createValkeyGlideDialCacheClient } from "../src/valkey-glide.js"; + +const scriptInstances = vi.hoisted(() => [] as Array<{ code: string; release: ReturnType }>); + +vi.mock("@valkey/valkey-glide", () => { + class MockScript { + readonly release = vi.fn(); + + constructor(readonly code: string) { + scriptInstances.push(this); + } + } + + class MockGlideClusterClient {} + + return { + Decoder: { Bytes: 0 }, + GlideClusterClient: MockGlideClusterClient, + Script: MockScript, + }; +}); + +interface FakeGlideClient { + invokeScript: ReturnType; + flushall: ReturnType; +} + +function fakeClient(...replies: unknown[]): FakeGlideClient { + return { + invokeScript: vi.fn(async () => replies.shift()), + flushall: vi.fn(async () => "OK"), + }; +} + +describe("Valkey GLIDE adapter", () => { + beforeEach(() => { + scriptInstances.length = 0; + }); + + it("invokes distinct read scripts with byte decoding", async () => { + const client = fakeClient(Buffer.from([0, ...Buffer.from("plain")]), Buffer.from([1, 0, 0xff]), null); + const adapter = createValkeyGlideDialCacheClient(client as never); + + await expect(adapter.read({ valueKey: "plain:value" })).resolves.toBe("plain"); + await expect( + adapter.read({ valueKey: "tracked:{id}:value", watermarkKey: "tracked:{id}:watermark" }), + ).resolves.toEqual(Buffer.from([0, 0xff])); + await expect(adapter.read({ valueKey: "missing:value" })).resolves.toBeNull(); + + expect(client.invokeScript).toHaveBeenNthCalledWith( + 1, + expect.any(Script), + { keys: ["plain:value"], args: [], decoder: 0 }, + ); + expect(client.invokeScript).toHaveBeenNthCalledWith( + 2, + expect.any(Script), + { + keys: ["tracked:{id}:value", "tracked:{id}:watermark"], + args: [], + decoder: 0, + }, + ); + expect(scriptInstances).toHaveLength(5); + }); + + it("passes string and Buffer writes directly to GLIDE", async () => { + const binary = Buffer.from([0, 0xff, 0x80]); + const client = fakeClient(1, 0, 1); + const adapter = createValkeyGlideDialCacheClient(client as never); + + await expect( + adapter.write({ valueKey: "plain:value", cacheTtlMs: 1_000, value: "hello" }), + ).resolves.toBe(true); + await expect( + adapter.write({ + valueKey: "tracked:{id}:value", + watermarkKey: "tracked:{id}:watermark", + cacheTtlMs: 2_000, + value: binary, + watermarkTtlFloorMs: 3_000, + }), + ).resolves.toBe(false); + await expect( + adapter.invalidate({ watermarkKey: "tracked:{id}:watermark", futureBufferMs: 100, watermarkTtlFloorMs: 3_000 }), + ).resolves.toBeUndefined(); + + expect(client.invokeScript).toHaveBeenNthCalledWith( + 1, + expect.any(Script), + { keys: ["plain:value"], args: ["1000", "0", "hello"], decoder: 0 }, + ); + expect(client.invokeScript).toHaveBeenNthCalledWith( + 2, + expect.any(Script), + { + keys: ["tracked:{id}:value", "tracked:{id}:watermark"], + args: ["2000", "1", binary, "3000"], + decoder: 0, + }, + ); + expect(client.invokeScript).toHaveBeenNthCalledWith( + 3, + expect.any(Script), + { keys: ["tracked:{id}:watermark"], args: ["100", "3000"], decoder: 0 }, + ); + }); + + it("rejects malformed script replies", async () => { + const client = fakeClient("not-bytes", Buffer.alloc(0), Buffer.from([2, 1]), "not-an-integer", null); + const adapter = createValkeyGlideDialCacheClient(client as never); + + await expect(adapter.read({ valueKey: "wrong-type" })).rejects.toBeInstanceOf(DialCacheRedisPayloadError); + await expect(adapter.read({ valueKey: "empty" })).rejects.toBeInstanceOf(DialCacheRedisPayloadError); + await expect(adapter.read({ valueKey: "wrong-encoding" })).rejects.toBeInstanceOf( + DialCacheRedisPayloadEncodingError, + ); + await expect(adapter.write({ valueKey: "bad-write", cacheTtlMs: 1_000, value: "value" })).rejects.toThrow( + "Invalid DialCache Redis write reply", + ); + await expect( + adapter.invalidate({ watermarkKey: "bad-watermark", futureBufferMs: 0, watermarkTtlFloorMs: 1_000 }), + ).rejects.toThrow("Invalid DialCache Redis invalidate reply"); + }); + + it("flushes standalone and cluster clients with the appropriate routing", async () => { + const standalone = fakeClient(); + const standaloneAdapter = createValkeyGlideDialCacheClient(standalone as never); + await standaloneAdapter.flushAll(); + expect(standalone.flushall).toHaveBeenCalledWith(); + + const cluster = Object.assign(Object.create(GlideClusterClient.prototype) as FakeGlideClient, fakeClient()); + const clusterAdapter = createValkeyGlideDialCacheClient(cluster as never); + await clusterAdapter.flushAll(); + expect(cluster.flushall).toHaveBeenCalledWith({ route: "allPrimaries" }); + }); + + it("releases every script exactly once and rejects later operations", async () => { + const client = fakeClient(); + const adapter = createValkeyGlideDialCacheClient(client as never); + + adapter.dispose(); + adapter.dispose(); + + expect(scriptInstances).toHaveLength(5); + for (const script of scriptInstances) { + expect(script.release).toHaveBeenCalledTimes(1); + } + await expect(adapter.read({ valueKey: "disposed" })).rejects.toThrow("Valkey GLIDE DialCache client is disposed"); + await expect(adapter.flushAll()).rejects.toThrow("Valkey GLIDE DialCache client is disposed"); + expect(client.invokeScript).not.toHaveBeenCalled(); + expect(client.flushall).not.toHaveBeenCalled(); + }); + + it("does not release scripts while an invocation is in flight", async () => { + let resolveRead: ((value: Buffer) => void) | undefined; + const client = fakeClient(); + client.invokeScript.mockImplementationOnce( + async () => await new Promise((resolve) => { + resolveRead = resolve; + }), + ); + const adapter = createValkeyGlideDialCacheClient(client as never); + + const read = adapter.read({ valueKey: "in-flight" }); + expect(() => adapter.dispose()).toThrow( + "Cannot dispose Valkey GLIDE DialCache client while operations are in flight", + ); + expect(scriptInstances.every((script) => script.release.mock.calls.length === 0)).toBe(true); + + resolveRead?.(Buffer.from([0, ...Buffer.from("done")])); + await expect(read).resolves.toBe("done"); + adapter.dispose(); + expect(scriptInstances.every((script) => script.release.mock.calls.length === 1)).toBe(true); + }); +}); From 9c6f4fd943a91bfc3ebbdfcf5016725b39f7811b Mon Sep 17 00:00:00 2001 From: Lev Neiman Date: Sun, 12 Jul 2026 20:09:54 -0700 Subject: [PATCH 2/2] codex: simplify GLIDE integration matrix --- test/redis-adapters.integration.test.ts | 9 +- ...redis-cluster-adapters.integration.test.ts | 250 ------------------ test/redis-cluster.integration.test.ts | 218 +++++++++++++++ 3 files changed, 221 insertions(+), 256 deletions(-) delete mode 100644 test/redis-cluster-adapters.integration.test.ts create mode 100644 test/redis-cluster.integration.test.ts diff --git a/test/redis-adapters.integration.test.ts b/test/redis-adapters.integration.test.ts index 2fd470f..8ffe8cf 100644 --- a/test/redis-adapters.integration.test.ts +++ b/test/redis-adapters.integration.test.ts @@ -17,10 +17,7 @@ import { type ValkeyGlideDialCacheClient, } from "../src/valkey-glide.js"; -const engines = [ - { name: "Redis 6.2", image: "redis:6.2-alpine" }, - { name: "Valkey 8", image: "valkey/valkey:8-alpine" }, -] as const; +const REDIS_IMAGE = "redis:6.2-alpine"; const adapterKinds = [ { kind: "nodeRedis", name: "node-redis" }, @@ -41,7 +38,7 @@ function encodeFrame(payload: string | Buffer, encoding: number, createdAtMs = D return Buffer.concat([Buffer.from([version]), timestamp, Buffer.from([encoding]), Buffer.from(payload)]); } -describe.each(engines)("DialCache Redis adapter conformance on $name", ({ image }) => { +describe("DialCache Redis adapter conformance on Redis 6.2", () => { let container: StartedTestContainer | undefined; let admin: ReturnType | undefined; let glide: GlideClient | undefined; @@ -49,7 +46,7 @@ describe.each(engines)("DialCache Redis adapter conformance on $name", ({ image let adapters: Record | undefined; beforeAll(async () => { - container = await new GenericContainer(image) + container = await new GenericContainer(REDIS_IMAGE) .withExposedPorts(6379) .withWaitStrategy(Wait.forListeningPorts()) .start(); diff --git a/test/redis-cluster-adapters.integration.test.ts b/test/redis-cluster-adapters.integration.test.ts deleted file mode 100644 index 041f17c..0000000 --- a/test/redis-cluster-adapters.integration.test.ts +++ /dev/null @@ -1,250 +0,0 @@ -import { createServer } from "node:net"; - -import { GlideClusterClient } from "@valkey/valkey-glide"; -import { commandOptions, createCluster } from "redis"; -import { GenericContainer, type StartedTestContainer, Wait } from "testcontainers"; -import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; - -import { CacheLayer, DialCache, DialCacheKeyConfig, type DialCacheRedisClient } from "../src/index.js"; -import { createNodeRedisDialCacheClient, dialcacheRedisScripts } from "../src/node-redis.js"; -import { - createValkeyGlideDialCacheClient, - type ValkeyGlideDialCacheClient, -} from "../src/valkey-glide.js"; - -const adapterKinds = [ - { kind: "nodeRedis", name: "node-redis" }, - { kind: "valkeyGlide", name: "Valkey GLIDE" }, -] as const; -type AdapterKind = (typeof adapterKinds)[number]["kind"]; - -const remoteOnly = new DialCacheKeyConfig({ - ttlSec: { [CacheLayer.REMOTE]: 60 }, - ramp: { [CacheLayer.REMOTE]: 100 }, -}); - -const createNodeRedisCluster = (port: number) => - createCluster({ - rootNodes: [{ url: `redis://127.0.0.1:${port}` }], - scripts: dialcacheRedisScripts, - }); - -async function waitForCluster(container: StartedTestContainer, port: number): Promise { - for (let attempt = 0; attempt < 50; attempt += 1) { - const result = await container.exec(["redis-cli", "-p", String(port), "cluster", "info"]); - if (result.output.includes("cluster_state:ok")) { - return; - } - await new Promise((resolve) => setTimeout(resolve, 100)); - } - throw new Error("Redis Cluster did not become ready"); -} - -async function chooseClusterPorts(): Promise { - for (let attempt = 0; attempt < 100; attempt += 1) { - const base = 20_000 + Math.floor(Math.random() * 25_000); - const ports = [base, base + 1, base + 2] as const; - if ((await Promise.all(ports.map(portIsAvailable))).every(Boolean)) { - return ports; - } - } - throw new Error("Could not reserve ports for Redis Cluster integration tests"); -} - -async function portIsAvailable(port: number): Promise { - return await new Promise((resolve) => { - const server = createServer(); - server.once("error", () => resolve(false)); - server.listen(port, "127.0.0.1", () => { - server.close(() => resolve(true)); - }); - }); -} - -function redisClusterCommand(ports: readonly number[]): string[] { - const commands = ports.map( - (port) => - `redis-server --port ${port} --cluster-enabled yes ` + - `--cluster-config-file /tmp/nodes-${port}.conf --cluster-node-timeout 5000 ` + - `--appendonly no --protected-mode no --bind 0.0.0.0 ` + - `--cluster-announce-ip 127.0.0.1 --cluster-announce-port ${port} ` + - `--cluster-announce-bus-port ${port + 10_000} --daemonize yes`, - ); - return ["sh", "-c", `${commands.join(" && ")} && tail -f /dev/null`]; -} - -describe("DialCache Redis adapter conformance on Redis Cluster", () => { - let container: StartedTestContainer | undefined; - let nodeCluster: ReturnType | undefined; - let glideCluster: GlideClusterClient | undefined; - let glideAdapter: ValkeyGlideDialCacheClient | undefined; - let adapters: Record | undefined; - - beforeAll(async () => { - const ports = await chooseClusterPorts(); - const exposedPorts = ports.map((port) => ({ - container: port, - host: port, - })); - container = await new GenericContainer("redis:7-alpine") - .withExposedPorts(...exposedPorts) - .withCommand(redisClusterCommand(ports)) - .withWaitStrategy(Wait.forListeningPorts()) - .start(); - - const createResult = await container.exec([ - "redis-cli", - "--cluster", - "create", - ...ports.map((port) => `127.0.0.1:${port}`), - "--cluster-replicas", - "0", - "--cluster-yes", - ]); - if (createResult.exitCode !== 0) { - throw new Error(`Could not create Redis Cluster: ${createResult.output}`); - } - await waitForCluster(container, ports[0]); - - nodeCluster = createNodeRedisCluster(ports[0]); - nodeCluster.on("error", () => undefined); - await nodeCluster.connect(); - glideCluster = await GlideClusterClient.createClient({ - addresses: ports.map((port) => ({ host: "127.0.0.1", port })), - }); - glideAdapter = createValkeyGlideDialCacheClient(glideCluster); - adapters = { - nodeRedis: createNodeRedisDialCacheClient(nodeCluster), - valkeyGlide: glideAdapter, - }; - }); - - beforeEach(async () => { - await adapters?.nodeRedis.flushAll(); - }); - - afterAll(async () => { - glideAdapter?.dispose(); - glideCluster?.close(); - await nodeCluster?.quit(); - await container?.stop(); - }); - - describe.each(adapterKinds)("$name adapter", ({ kind }) => { - function adapter(): DialCacheRedisClient { - const active = adapters?.[kind]; - if (active === undefined) { - throw new Error("Redis Cluster adapters did not start"); - } - return active; - } - - it("routes cached operations across cluster slots", async () => { - const dialcache = new DialCache({ redis: { client: adapter(), keyPrefix: `${kind}:cluster:` } }); - const ids = Array.from({ length: 30 }, (_, index) => `item-${index}`); - let calls = 0; - const load = dialcache.cached(async (id: string) => ({ id, calls: ++calls }), { - keyType: "item_id", - useCase: `Cluster${kind}`, - cacheKey: (id) => id, - defaultConfig: remoteOnly, - }); - - const first = await dialcache.enable(async () => await Promise.all(ids.map(load))); - const second = await dialcache.enable(async () => await Promise.all(ids.map(load))); - expect(second).toEqual(first); - expect(calls).toBe(ids.length); - }); - - it("keeps tracked keys colocated and rejects mismatched hash slots", async () => { - const client = adapter(); - const valueKey = `${kind}:{same}:value`; - const watermarkKey = `${kind}:{same}:watermark`; - await expect( - client.write({ - valueKey, - watermarkKey, - cacheTtlMs: 60_000, - value: "tracked", - watermarkTtlFloorMs: 60_000, - }), - ).resolves.toBe(true); - await expect(client.read({ valueKey, watermarkKey })).resolves.toBe("tracked"); - await expect( - client.read({ valueKey: `${kind}:{slot-a}:value`, watermarkKey: `${kind}:{slot-b}:watermark` }), - ).rejects.toThrow(/CROSSSLOT|same slot/i); - }); - - it("round-trips binary payloads through cluster script routing", async () => { - const client = adapter(); - if (nodeCluster === undefined) { - throw new Error("Redis Cluster did not start"); - } - const payload = Buffer.from([0, 0xff, 0xc3, 0x28, 0x80]); - const valueKey = `${kind}:{binary}:value`; - await expect(client.write({ valueKey, cacheTtlMs: 60_000, value: payload })).resolves.toBe(true); - await expect(client.read({ valueKey })).resolves.toEqual(payload); - const stored = await nodeCluster.get(commandOptions({ returnBuffers: true }), valueKey); - expect(stored?.[0]).toBe(1); - expect(stored?.[9]).toBe(1); - expect(stored?.subarray(10)).toEqual(payload); - }); - - it("reloads scripts on every shard after SCRIPT FLUSH", async () => { - const client = adapter(); - if (nodeCluster === undefined) { - throw new Error("Redis Cluster did not start"); - } - const activeCluster = nodeCluster; - await Promise.all( - activeCluster.masters.map(async (master) => { - const node = await activeCluster.nodeClient(master); - await node.scriptFlush(); - }), - ); - - const keys = Array.from({ length: 30 }, (_, index) => `${kind}:recovery:{item-${index}}:value`); - await Promise.all(keys.map(async (valueKey) => await client.write({ valueKey, cacheTtlMs: 60_000, value: valueKey }))); - await expect(Promise.all(keys.map(async (valueKey) => await client.read({ valueKey })))).resolves.toEqual(keys); - }); - - it("flushes every primary shard", async () => { - const client = adapter(); - if (nodeCluster === undefined) { - throw new Error("Redis Cluster did not start"); - } - const activeCluster = nodeCluster; - const keys = Array.from({ length: 30 }, (_, index) => `${kind}:flush:{item-${index}}:value`); - await Promise.all(keys.map(async (valueKey) => await client.write({ valueKey, cacheTtlMs: 60_000, value: valueKey }))); - const sizesBefore = await Promise.all( - activeCluster.masters.map(async (master) => { - const node = await activeCluster.nodeClient(master); - return await node.dbSize(); - }), - ); - expect(sizesBefore.every((size) => size > 0)).toBe(true); - await client.flushAll(); - const sizes = await Promise.all( - activeCluster.masters.map(async (master) => { - const node = await activeCluster.nodeClient(master); - return await node.dbSize(); - }), - ); - expect(sizes).toEqual([0, 0, 0]); - }); - }); - - it("uses one cluster wire format across node-redis and Valkey GLIDE", async () => { - if (adapters === undefined) { - throw new Error("Redis Cluster adapters did not start"); - } - const nodeValueKey = "interop:{node-to-glide}:value"; - const glideValueKey = "interop:{glide-to-node}:value"; - const binary = Buffer.from([0, 0xff, 0x80]); - - await adapters.nodeRedis.write({ valueKey: nodeValueKey, cacheTtlMs: 60_000, value: binary }); - await expect(adapters.valkeyGlide.read({ valueKey: nodeValueKey })).resolves.toEqual(binary); - await adapters.valkeyGlide.write({ valueKey: glideValueKey, cacheTtlMs: 60_000, value: "hello" }); - await expect(adapters.nodeRedis.read({ valueKey: glideValueKey })).resolves.toBe("hello"); - }); -}); diff --git a/test/redis-cluster.integration.test.ts b/test/redis-cluster.integration.test.ts new file mode 100644 index 0000000..7a264e2 --- /dev/null +++ b/test/redis-cluster.integration.test.ts @@ -0,0 +1,218 @@ +import { commandOptions, createCluster, type RedisClusterOptions } from "redis"; +import { + GenericContainer, + Network, + type StartedNetwork, + type StartedTestContainer, + Wait, +} from "testcontainers"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; + +import { CacheLayer, DialCache, DialCacheKeyConfig, type DialCacheRedisClient } from "../src/index.js"; +import { createNodeRedisDialCacheClient, dialcacheRedisScripts } from "../src/node-redis.js"; + +const remoteOnly = new DialCacheKeyConfig({ + ttlSec: { [CacheLayer.REMOTE]: 60 }, + ramp: { [CacheLayer.REMOTE]: 100 }, +}); + +const createTestCluster = (options: RedisClusterOptions) => + createCluster({ + ...options, + scripts: dialcacheRedisScripts, + }); + +async function waitForCluster(container: StartedTestContainer): Promise { + for (let attempt = 0; attempt < 50; attempt += 1) { + const result = await container.exec(["redis-cli", "cluster", "info"]); + if (result.output.includes("cluster_state:ok")) { + return; + } + await new Promise((resolve) => setTimeout(resolve, 100)); + } + throw new Error("Redis Cluster did not become ready"); +} + +describe("DialCache Lua protocol on Redis Cluster", () => { + let network: StartedNetwork | undefined; + let containers: Array = []; + let cluster: ReturnType | undefined; + + beforeAll(async () => { + const startedNetwork = await new Network().start(); + network = startedNetwork; + for (const alias of ["redis-1", "redis-2", "redis-3"]) { + containers.push( + await new GenericContainer("redis:7-alpine") + .withNetwork(startedNetwork) + .withNetworkAliases(alias) + .withExposedPorts(6379) + .withCommand([ + "redis-server", + "--port", + "6379", + "--cluster-enabled", + "yes", + "--cluster-config-file", + "/tmp/nodes.conf", + "--cluster-node-timeout", + "5000", + "--appendonly", + "no", + "--protected-mode", + "no", + ]) + .withWaitStrategy(Wait.forListeningPorts()) + .start(), + ); + } + + const networkName = network.getName(); + const internalAddresses = containers.map((container) => `${container.getIpAddress(networkName)}:6379`); + const firstContainer = containers[0]; + if (firstContainer === undefined) { + throw new Error("Redis Cluster containers did not start"); + } + const createResult = await firstContainer.exec([ + "redis-cli", + "--cluster", + "create", + ...internalAddresses, + "--cluster-replicas", + "0", + "--cluster-yes", + ]); + if (createResult.exitCode !== 0) { + throw new Error(`Could not create Redis Cluster: ${createResult.output}`); + } + await Promise.all(containers.map(waitForCluster)); + + const nodeAddressMap = Object.fromEntries( + containers.map((container) => [ + `${container.getIpAddress(networkName)}:6379`, + { host: container.getHost(), port: container.getMappedPort(6379) }, + ]), + ); + cluster = createTestCluster({ + rootNodes: [{ url: `redis://${firstContainer.getHost()}:${firstContainer.getMappedPort(6379)}` }], + nodeAddressMap, + }); + cluster.on("error", () => undefined); + await cluster.connect(); + }); + + afterAll(async () => { + await cluster?.quit(); + await Promise.all(containers.map(async (container) => await container.stop())); + await network?.stop(); + }); + + it("routes scripts across slots and reloads them per node", async () => { + if (cluster === undefined) { + throw new Error("Redis Cluster did not start"); + } + const activeCluster = cluster; + const scriptClient: DialCacheRedisClient = createNodeRedisDialCacheClient(activeCluster); + const dialcache = new DialCache({ redis: { client: scriptClient, keyPrefix: "cluster:" } }); + const ids = Array.from({ length: 30 }, (_, index) => `item-${index}`); + let calls = 0; + const getValue = dialcache.cached(async (id: string) => ({ id, calls: ++calls }), { + keyType: "item_id", + useCase: "ClusterSlots", + cacheKey: (id) => id, + defaultConfig: remoteOnly, + }); + + const first = await dialcache.enable(async () => await Promise.all(ids.map(getValue))); + const sizesBeforeFlush = await Promise.all( + activeCluster.masters.map(async (master) => { + const client = await activeCluster.nodeClient(master); + return await client.dbSize(); + }), + ); + await Promise.all( + activeCluster.masters.map(async (master) => { + const client = await activeCluster.nodeClient(master); + await client.scriptFlush(); + }), + ); + const second = await dialcache.enable(async () => await Promise.all(ids.map(getValue))); + await dialcache.flushAll(); + const sizesAfterFlush = await Promise.all( + activeCluster.masters.map(async (master) => { + const client = await activeCluster.nodeClient(master); + return await client.dbSize(); + }), + ); + const afterFlush = await dialcache.enable(async () => await Promise.all(ids.map(getValue))); + + expect(first.map(({ id }) => id)).toEqual(ids); + expect(calls).toBe(60); + expect(sizesBeforeFlush.every((size) => size > 0)).toBe(true); + expect(second).toEqual(first); + expect(sizesAfterFlush).toEqual([0, 0, 0]); + expect(afterFlush.map(({ id }) => id)).toEqual(ids); + expect(new Set(afterFlush.map(({ calls: call }) => call))).toEqual(new Set(ids.map((_, index) => index + 31))); + }); + + it("keeps tracked keys colocated and rejects mismatched hash tags", async () => { + if (cluster === undefined) { + throw new Error("Redis Cluster did not start"); + } + expect(dialcacheRedisScripts.dialcacheRead.IS_READ_ONLY).toBe(true); + expect(dialcacheRedisScripts.dialcacheReadTracked.IS_READ_ONLY).toBe(false); + expect(dialcacheRedisScripts.dialcacheRead.SHA1).not.toBe(dialcacheRedisScripts.dialcacheReadTracked.SHA1); + expect(dialcacheRedisScripts.dialcacheWrite.SHA1).not.toBe(dialcacheRedisScripts.dialcacheWriteTracked.SHA1); + const scriptClient: DialCacheRedisClient = createNodeRedisDialCacheClient(cluster); + const dialcache = new DialCache({ redis: { client: scriptClient, keyPrefix: "cluster:" } }); + let version = 1; + const getUser = dialcache.cached(async (id: string) => ({ id, version }), { + keyType: "user_id", + useCase: "ClusterTracked", + cacheKey: (id) => id, + trackForInvalidation: true, + defaultConfig: remoteOnly, + }); + + const before = await dialcache.enable(async () => await getUser("123")); + version = 2; + await dialcache.invalidateRemote("user_id", "123"); + await new Promise((resolve) => setTimeout(resolve, 2)); + const after = await dialcache.enable(async () => await getUser("123")); + + expect(before).toEqual({ id: "123", version: 1 }); + 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); + }); +});