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..8ffe8cf --- /dev/null +++ b/test/redis-adapters.integration.test.ts @@ -0,0 +1,203 @@ +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 REDIS_IMAGE = "redis:6.2-alpine"; + +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("DialCache Redis adapter conformance on Redis 6.2", () => { + 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(REDIS_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/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); + }); +});