From 256eda81c3abfebade1be5cb721ce8fbe2aa9e36 Mon Sep 17 00:00:00 2001 From: Lev Neiman Date: Wed, 22 Jul 2026 21:56:12 -0700 Subject: [PATCH 1/2] feat: decouple GLIDE adapter runtime Require callers to pass their GLIDE module namespace so Script handles and Decoder.Bytes come from the same runtime as the client. The required second argument intentionally advances the pre-1.0 API through the repository's minor-release path. --- README.md | 18 +++-- package.json | 4 -- scripts/test-package.mjs | 53 ++++++++++++-- src/valkey-glide.ts | 83 +++++++++++++--------- test/redis-adapters.integration.test.ts | 8 +-- test/valkey-glide.test.ts | 91 +++++++++++++++---------- 6 files changed, 171 insertions(+), 86 deletions(-) diff --git a/README.md b/README.md index e39a8cb..3edfbef 100644 --- a/README.md +++ b/README.md @@ -31,7 +31,7 @@ pnpm add dialcache # Choose a Redis client when using the remote layer: pnpm add redis@~4.7.1 # or -pnpm add @valkey/valkey-glide@^2.4.2 +pnpm add @valkey/valkey-glide # Add a metrics client only when using its adapter: pnpm add prom-client@^15.1.3 # or @@ -308,19 +308,20 @@ async function shutdown(): Promise { `redis.client` is required when Redis is configured and accepts the semantic `DialCacheRedisClient` interface. Create and connect the underlying client before constructing `DialCache`. 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: +Valkey GLIDE users pass an already-created standalone or cluster client and its +module namespace to the GLIDE adapter: ```ts -import { GlideClient } from "@valkey/valkey-glide"; +import * as valkeyGlide from "@valkey/valkey-glide"; import { DialCache } from "dialcache"; import { createValkeyGlideDialCacheClient } from "dialcache/valkey-glide"; -const glideClient = await GlideClient.createClient({ +const glideClient = await valkeyGlide.GlideClient.createClient({ addresses: [{ host: "127.0.0.1", port: 6379 }], requestTimeout: 2_000, advancedConfiguration: { connectionTimeout: 2_000 }, }); -const redisClient = createValkeyGlideDialCacheClient(glideClient); +const redisClient = createValkeyGlideDialCacheClient(glideClient, valkeyGlide); const dialcache = new DialCache({ namespace: "users-api", redis: { client: redisClient }, @@ -333,6 +334,11 @@ function shutdown(): void { } ``` +Pass the same module namespace that created the client. DialCache uses its +`Script` constructor and `Decoder.Bytes` value without importing a GLIDE runtime +itself, so linked workspaces and applications with another installed GLIDE +version cannot accidentally mix native script handles. + The application owns the complete Redis lifecycle. It creates and connects the underlying client and passes the semantic adapter to DialCache. During shutdown, stop starting DialCache-backed work and await every outstanding cached-function and `invalidateRemote()` promise, including calls still running fallbacks that may later write Redis. Then dispose adapter-owned resources and close the underlying connection. DialCache only borrows `redis.client`; it has no close or drain method and never disposes or closes caller resources. The node-redis adapter owns no additional resources, so the application closes the underlying node-redis client after draining work. The GLIDE adapter owns five native `Script` handles but not the wrapped connection. After outstanding operations finish, call its idempotent `dispose()` before closing GLIDE as shown above; disposal while an adapter operation is in flight throws rather than releasing a live script. @@ -343,7 +349,7 @@ Node-redis computes each script's SHA, uses `EVALSHA`, and retries with `EVAL` a DialCache fail-open behavior is rejection-driven: a Redis promise that never settles cannot fall through. Every `DialCacheRedisClient.read`, `write`, and `invalidate` promise must therefore resolve or reject within a finite application-defined budget covering connection establishment, reconnection, retries, offline queueing, dispatch, and response time. A connection timeout alone does not satisfy this contract. A pending read prevents fallback from starting, while a pending write withholds an already-computed fallback result and retains its process flight. -Valkey GLIDE users should configure [`requestTimeout`](https://github.com/valkey-io/valkey-glide/blob/v2.4.2/node/src/BaseClient.ts#L770-L776) and [`advancedConfiguration.connectionTimeout`](https://github.com/valkey-io/valkey-glide/blob/v2.4.2/node/src/BaseClient.ts#L957-L964), as shown above. GLIDE applies the request timeout while sending, waiting for a response, reconnecting, and retrying. This bounds client waiting; it is not server-side cancellation, so a write dispatched before timeout may still execute. +Valkey GLIDE users should configure [`requestTimeout`](https://glide.valkey.io/languages/nodejs/api/interfaces/BaseClient.BaseClientConfiguration.html) and [`advancedConfiguration.connectionTimeout`](https://glide.valkey.io/languages/nodejs/api/interfaces/BaseClient.AdvancedBaseClientConfiguration.html), as shown above. GLIDE applies the request timeout while sending, waiting for a response, reconnecting, and retrying. This bounds client waiting; it is not server-side cancellation, so a write dispatched before timeout may still execute. In node-redis 4.7, `socket.connectTimeout`, `disableOfflineQueue`, and `commandsQueueMaxLength` bound connection or queue behavior but do not impose a response deadline on a dispatched command. A [per-command `AbortSignal`](https://redis.io/docs/latest/develop/clients/nodejs/produsage/) can remove work that is still waiting to be sent, but once dispatched it no longer controls the pending reply. Node-redis 4.7 has no built-in strict dispatched-response deadline, and the bundled adapter does not inject queue cancellation. Applications requiring finite node-redis settlement must supply and document a custom `DialCacheRedisClient` policy for queue removal, hung connections, and ambiguous writes. Do not put Redis writes or invalidations behind a bare `Promise.race`: rejecting the outer promise neither removes queued work nor proves that an already-dispatched command did not execute. diff --git a/package.json b/package.json index c361abd..625e1b1 100644 --- a/package.json +++ b/package.json @@ -110,14 +110,10 @@ "vitest": "^4.0.14" }, "peerDependencies": { - "@valkey/valkey-glide": "^2.4.2", "prom-client": "^15.1.3", "redis": "~4.7.1" }, "peerDependenciesMeta": { - "@valkey/valkey-glide": { - "optional": true - }, "prom-client": { "optional": true }, diff --git a/scripts/test-package.mjs b/scripts/test-package.mjs index fd2f86a..fe56536 100644 --- a/scripts/test-package.mjs +++ b/scripts/test-package.mjs @@ -313,7 +313,8 @@ void datadogMetrics; void datadogClassAdapter; void missingObservationType; `; -const integrationConsumer = `import { DialCache } from "dialcache"; +const integrationConsumer = `import * as valkeyGlide from "@valkey/valkey-glide"; +import { DialCache } from "dialcache"; import StatsD from "hot-shots"; import { DatadogDialCacheMetrics, @@ -330,6 +331,7 @@ import { import { createValkeyGlideDialCacheClient, type ValkeyGlideDialCacheClient, + type ValkeyGlideRuntime, } from "dialcache/valkey-glide"; import { Registry, type OpenMetricsContentType } from "prom-client"; @@ -343,6 +345,13 @@ openMetricsRegistry.setContentType(Registry.OPENMETRICS_CONTENT_TYPE); const openMetricsAdapter = new PrometheusDialCacheMetrics({ registry: openMetricsRegistry, prefix: "open_" }); const registryIsRequired: {} extends Pick ? false : true = true; const glideRedisClient: ValkeyGlideDialCacheClient | undefined = undefined; +const glideRuntime: ValkeyGlideRuntime = valkeyGlide; +declare const standaloneGlideClient: valkeyGlide.GlideClient; +declare const clusterGlideClient: valkeyGlide.GlideClusterClient; +const standaloneGlideAdapter = createValkeyGlideDialCacheClient(standaloneGlideClient, glideRuntime); +const clusterGlideAdapter = createValkeyGlideDialCacheClient(clusterGlideClient, glideRuntime); +// @ts-expect-error The caller's GLIDE runtime is required for native Script ownership. +createValkeyGlideDialCacheClient(standaloneGlideClient); const dogStatsD = new StatsD({ mock: true }); const compatibleDogStatsD: DatadogDogStatsDClient = dogStatsD; const observationMetricType: DatadogObservationMetricType = "distribution"; @@ -365,7 +374,8 @@ void classAdapter; void openMetricsAdapter; void registryIsRequired; void glideRedisClient; -void createValkeyGlideDialCacheClient; +void standaloneGlideAdapter; +void clusterGlideAdapter; void datadogClassAdapter; void datadogCache; void observationMetricTypeIsRequired; @@ -415,6 +425,7 @@ try { "--eval", `const root = await import("dialcache"); const nodeRedis = await import("dialcache/node-redis"); +await import("dialcache/valkey-glide"); await import("dialcache/datadog"); await import("dialcache/redis-protocol"); const fallbackTimeoutError = new root.FallbackTimeoutError("PackageRuntime", 1000); @@ -490,6 +501,7 @@ if (calls !== 1 || second !== first) { "--eval", `const root = require("dialcache"); const nodeRedis = require("dialcache/node-redis"); +require("dialcache/valkey-glide"); require("dialcache/datadog"); require("dialcache/redis-protocol"); const fallbackTimeoutError = new root.FallbackTimeoutError("PackageRuntime", 1000); @@ -582,7 +594,8 @@ void (async () => { "redis@~4.7.1", "typescript@5.9.3", "prom-client@^15.1.3", - "@valkey/valkey-glide@^2.4.2", + "@valkey/valkey-glide@2.2.10", + "dialcache-test-glide@npm:@valkey/valkey-glide@2.4.2", "hot-shots@^17.0.0", ], { cwd: workspace }, @@ -601,11 +614,26 @@ void (async () => { "--eval", `const root = await import("dialcache"); const glide = await import("dialcache/valkey-glide"); +const appGlide = await import("@valkey/valkey-glide"); +const otherGlide = await import("dialcache-test-glide"); await import("dialcache/datadog"); await import("dialcache/prometheus"); await import("dialcache/redis-protocol"); await import("dialcache/node-redis"); -const adapter = glide.createValkeyGlideDialCacheClient({ invokeScript: async () => 2 }); +if (appGlide.Script === otherGlide.Script) { + throw new Error("The package test requires two distinct GLIDE module instances"); +} +const adapter = glide.createValkeyGlideDialCacheClient({ + invokeScript: async (script, options) => { + if (!(script instanceof appGlide.Script) || script instanceof otherGlide.Script) { + throw new Error("The ESM adapter did not use the caller-supplied GLIDE Script constructor"); + } + if (options.decoder !== appGlide.Decoder.Bytes) { + throw new Error("The ESM adapter did not use the caller-supplied GLIDE byte decoder"); + } + return 2; + }, +}, appGlide); try { await adapter.write({ valueKey: "value", cacheTtlMs: 1_000, value: "payload" }); throw new Error("Expected an invalid GLIDE script reply to fail"); @@ -625,12 +653,27 @@ try { "--eval", `const root = require("dialcache"); const glide = require("dialcache/valkey-glide"); +const appGlide = require("@valkey/valkey-glide"); +const otherGlide = require("dialcache-test-glide"); require("dialcache/datadog"); require("dialcache/prometheus"); require("dialcache/redis-protocol"); require("dialcache/node-redis"); void (async () => { - const adapter = glide.createValkeyGlideDialCacheClient({ invokeScript: async () => 2 }); + if (appGlide.Script === otherGlide.Script) { + throw new Error("The package test requires two distinct GLIDE module instances"); + } + const adapter = glide.createValkeyGlideDialCacheClient({ + invokeScript: async (script, options) => { + if (!(script instanceof appGlide.Script) || script instanceof otherGlide.Script) { + throw new Error("The CommonJS adapter did not use the caller-supplied GLIDE Script constructor"); + } + if (options.decoder !== appGlide.Decoder.Bytes) { + throw new Error("The CommonJS adapter did not use the caller-supplied GLIDE byte decoder"); + } + return 2; + }, + }, appGlide); try { await adapter.write({ valueKey: "value", cacheTtlMs: 1_000, value: "payload" }); throw new Error("Expected an invalid GLIDE script reply to fail"); diff --git a/src/valkey-glide.ts b/src/valkey-glide.ts index 97fc6c4..10c910f 100644 --- a/src/valkey-glide.ts +++ b/src/valkey-glide.ts @@ -1,12 +1,3 @@ -import { - Decoder, - Script, - type GlideClient, - type GlideClusterClient, - type GlideReturnType, - type GlideString, -} from "@valkey/valkey-glide"; - import { decodeRedisPayload, redisPayloadEncoding } from "./internal/redis-payload.js"; import { INVALIDATE_CACHE_SCRIPT, @@ -21,14 +12,39 @@ import { } from "./internal/redis-script-reply.js"; import { DialCacheRedisPayloadError, type DialCacheRedisClient } from "./redis-client.js"; -type SupportedValkeyGlideClient = GlideClient | GlideClusterClient; +type ValkeyGlideString = string | Buffer; + +export interface ValkeyGlideScriptHandle { + /** Release the native GLIDE script registration. */ + release(): void; +} + +export interface ValkeyGlideScriptingClient { + invokeScript( + script: TScript, + options: { + keys: ValkeyGlideString[]; + args: ValkeyGlideString[]; + decoder: TDecoder; + }, + ): Promise; +} + +export interface ValkeyGlideRuntime { + /** The Script constructor exported by the same GLIDE module instance as the client. */ + readonly Script: new (source: string) => TScript; + /** The Decoder enum exported by the same GLIDE module instance as the client. */ + readonly Decoder: { + readonly Bytes: TDecoder; + }; +} -interface DialCacheGlideScripts { - readonly read: Script; - readonly readTracked: Script; - readonly write: Script; - readonly writeTracked: Script; - readonly invalidate: Script; +interface DialCacheGlideScripts { + readonly read: TScript; + readonly readTracked: TScript; + readonly write: TScript; + readonly writeTracked: TScript; + readonly invalidate: TScript; } export interface ValkeyGlideDialCacheClient extends DialCacheRedisClient { @@ -38,34 +54,37 @@ export interface ValkeyGlideDialCacheClient extends DialCacheRedisClient { /** * Wrap a caller-owned GLIDE connection. The returned adapter owns only its - * Script handles and preserves the connection's `requestTimeout`; callers - * dispose the handles after draining work, then close GLIDE. A request timeout - * bounds client waiting but is not server-side command cancellation. + * Script handles and preserves the connection's `requestTimeout`. Pass the + * same GLIDE module namespace used to create the client so native Script + * handles are registered with that client's runtime. Callers dispose the + * handles after draining work, then close GLIDE. A request timeout bounds + * client waiting but is not server-side command cancellation. */ -export function createValkeyGlideDialCacheClient( - client: SupportedValkeyGlideClient, +export function createValkeyGlideDialCacheClient( + client: ValkeyGlideScriptingClient, + glide: ValkeyGlideRuntime, ): 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), + const scripts: DialCacheGlideScripts = { + read: new glide.Script(READ_CACHE_SCRIPT), + readTracked: new glide.Script(READ_TRACKED_CACHE_SCRIPT), + write: new glide.Script(WRITE_CACHE_SCRIPT), + writeTracked: new glide.Script(WRITE_TRACKED_CACHE_SCRIPT), + invalidate: new glide.Script(INVALIDATE_CACHE_SCRIPT), }; let disposed = false; let activeInvocations = 0; const invoke = async ( - script: Script, - keys: GlideString[], - args: GlideString[] = [], - ): Promise => { + script: TScript, + keys: ValkeyGlideString[], + args: ValkeyGlideString[] = [], + ): 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 }); + return await client.invokeScript(script, { keys, args, decoder: glide.Decoder.Bytes }); } finally { activeInvocations -= 1; } diff --git a/test/redis-adapters.integration.test.ts b/test/redis-adapters.integration.test.ts index 806b21c..87ee227 100644 --- a/test/redis-adapters.integration.test.ts +++ b/test/redis-adapters.integration.test.ts @@ -1,4 +1,4 @@ -import { GlideClient } from "@valkey/valkey-glide"; +import * as valkeyGlide from "@valkey/valkey-glide"; import { createClient } from "redis"; import { GenericContainer, type StartedTestContainer, Wait } from "testcontainers"; import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; @@ -41,7 +41,7 @@ function encodeFrame(payload: string | Buffer, encoding: number, createdAtMs = D describe("DialCache Redis adapter conformance on Redis 6.2", () => { let container: StartedTestContainer | undefined; let admin: ReturnType | undefined; - let glide: GlideClient | undefined; + let glide: valkeyGlide.GlideClient | undefined; let glideAdapter: ValkeyGlideDialCacheClient | undefined; let adapters: Record | undefined; @@ -55,8 +55,8 @@ describe("DialCache Redis adapter conformance on Redis 6.2", () => { admin = createNodeRedisClient(`redis://${host}:${port}`); admin.on("error", () => undefined); await admin.connect(); - glide = await GlideClient.createClient({ addresses: [{ host, port }] }); - glideAdapter = createValkeyGlideDialCacheClient(glide); + glide = await valkeyGlide.GlideClient.createClient({ addresses: [{ host, port }] }); + glideAdapter = createValkeyGlideDialCacheClient(glide, valkeyGlide); adapters = { nodeRedis: createNodeRedisDialCacheClient(admin), valkeyGlide: glideAdapter, diff --git a/test/valkey-glide.test.ts b/test/valkey-glide.test.ts index e765b71..9e9d323 100644 --- a/test/valkey-glide.test.ts +++ b/test/valkey-glide.test.ts @@ -1,4 +1,3 @@ -import { Script } from "@valkey/valkey-glide"; import { beforeEach, describe, expect, it, vi } from "vitest"; import { @@ -23,30 +22,31 @@ const INVALID_WRITE_REPLIES: readonly unknown[] = [ ]; const INVALID_INVALIDATION_REPLIES: readonly unknown[] = [0, ...INVALID_WRITE_REPLIES]; -const scriptInstances = vi.hoisted(() => [] as Array<{ code: string; release: ReturnType }>); +const decoderBytes = Symbol("bytes"); +const scriptInstances: MockScript[] = []; -vi.mock("@valkey/valkey-glide", () => { - class MockScript { - readonly release = vi.fn(); +class MockScript { + readonly release = vi.fn(); - constructor(readonly code: string) { - scriptInstances.push(this); - } + constructor(readonly code: string) { + scriptInstances.push(this); } +} - return { - Decoder: { Bytes: 0 }, - Script: MockScript, - }; -}); +const mockGlide = { + Decoder: { Bytes: decoderBytes }, + Script: MockScript, +}; -interface FakeGlideClient { - invokeScript: ReturnType; +interface InvokeScriptOptions { + keys: Array; + args: Array; + decoder: typeof decoderBytes; } -function fakeClient(...replies: unknown[]): FakeGlideClient { +function fakeClient(...replies: unknown[]) { return { - invokeScript: vi.fn(async () => replies.shift()), + invokeScript: vi.fn(async (_script: MockScript, _options: InvokeScriptOptions) => replies.shift()), }; } @@ -68,7 +68,7 @@ describe("Valkey GLIDE adapter", () => { 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); + const adapter = createValkeyGlideDialCacheClient(client, mockGlide); await expect(adapter.read({ valueKey: "plain:value" })).resolves.toBe("plain"); await expect( @@ -78,16 +78,16 @@ describe("Valkey GLIDE adapter", () => { expect(client.invokeScript).toHaveBeenNthCalledWith( 1, - expect.any(Script), - { keys: ["plain:value"], args: [], decoder: 0 }, + expect.any(MockScript), + { keys: ["plain:value"], args: [], decoder: decoderBytes }, ); expect(client.invokeScript).toHaveBeenNthCalledWith( 2, - expect.any(Script), + expect.any(MockScript), { keys: ["tracked:{id}:value", "tracked:{id}:watermark"], args: [], - decoder: 0, + decoder: decoderBytes, }, ); expect(scriptInstances).toHaveLength(5); @@ -96,7 +96,7 @@ describe("Valkey GLIDE adapter", () => { 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); + const adapter = createValkeyGlideDialCacheClient(client, mockGlide); await expect( adapter.write({ valueKey: "plain:value", cacheTtlMs: 1_000, value: "hello" }), @@ -116,28 +116,28 @@ describe("Valkey GLIDE adapter", () => { expect(client.invokeScript).toHaveBeenNthCalledWith( 1, - expect.any(Script), - { keys: ["plain:value"], args: ["1000", "0", "hello"], decoder: 0 }, + expect.any(MockScript), + { keys: ["plain:value"], args: ["1000", "0", "hello"], decoder: decoderBytes }, ); expect(client.invokeScript).toHaveBeenNthCalledWith( 2, - expect.any(Script), + expect.any(MockScript), { keys: ["tracked:{id}:value", "tracked:{id}:watermark"], args: ["2000", "1", binary, "3000"], - decoder: 0, + decoder: decoderBytes, }, ); expect(client.invokeScript).toHaveBeenNthCalledWith( 3, - expect.any(Script), - { keys: ["tracked:{id}:watermark"], args: ["100", "3000"], decoder: 0 }, + expect.any(MockScript), + { keys: ["tracked:{id}:watermark"], args: ["100", "3000"], decoder: decoderBytes }, ); }); 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); + const adapter = createValkeyGlideDialCacheClient(client, mockGlide); await expect(adapter.read({ valueKey: "wrong-type" })).rejects.toBeInstanceOf(DialCacheRedisPayloadError); await expect(adapter.read({ valueKey: "empty" })).rejects.toBeInstanceOf(DialCacheRedisPayloadError); @@ -161,14 +161,14 @@ describe("Valkey GLIDE adapter", () => { const invalidationMessage = "Invalid DialCache Redis invalidate reply; expected integer 1"; for (const reply of INVALID_WRITE_REPLIES) { - const untracked = createValkeyGlideDialCacheClient(fakeClient(reply) as never); + const untracked = createValkeyGlideDialCacheClient(fakeClient(reply), mockGlide); await expectProtocolError( Promise.resolve(untracked.write({ valueKey: "plain:value", cacheTtlMs: 1_000, value: "plain" })), writeMessage, ); untracked.dispose(); - const tracked = createValkeyGlideDialCacheClient(fakeClient(reply) as never); + const tracked = createValkeyGlideDialCacheClient(fakeClient(reply), mockGlide); await expectProtocolError( Promise.resolve(tracked.write({ valueKey: "tracked:{id}:value", @@ -183,7 +183,7 @@ describe("Valkey GLIDE adapter", () => { } for (const reply of INVALID_INVALIDATION_REPLIES) { - const adapter = createValkeyGlideDialCacheClient(fakeClient(reply) as never); + const adapter = createValkeyGlideDialCacheClient(fakeClient(reply), mockGlide); await expectProtocolError( Promise.resolve(adapter.invalidate({ watermarkKey: "tracked:{id}:watermark", @@ -198,7 +198,7 @@ describe("Valkey GLIDE adapter", () => { it("releases every script exactly once and rejects later operations", async () => { const client = fakeClient(); - const adapter = createValkeyGlideDialCacheClient(client as never); + const adapter = createValkeyGlideDialCacheClient(client, mockGlide); adapter.dispose(); adapter.dispose(); @@ -219,7 +219,7 @@ describe("Valkey GLIDE adapter", () => { resolveRead = resolve; }), ); - const adapter = createValkeyGlideDialCacheClient(client as never); + const adapter = createValkeyGlideDialCacheClient(client, mockGlide); const read = adapter.read({ valueKey: "in-flight" }); expect(() => adapter.dispose()).toThrow( @@ -232,4 +232,25 @@ describe("Valkey GLIDE adapter", () => { adapter.dispose(); expect(scriptInstances.every((script) => script.release.mock.calls.length === 1)).toBe(true); }); + + it("uses Script and Decoder from the supplied GLIDE module instance", async () => { + class OtherScript { + readonly release = vi.fn(); + } + const otherGlide = { + Decoder: { Bytes: Symbol("other-bytes") }, + Script: OtherScript, + }; + const client = fakeClient(null); + const adapter = createValkeyGlideDialCacheClient(client, mockGlide); + + await adapter.read({ valueKey: "module-instance" }); + + const [script, options] = client.invokeScript.mock.calls[0] ?? []; + expect(script).toBeInstanceOf(MockScript); + expect(script).not.toBeInstanceOf(otherGlide.Script); + expect(options?.decoder).toBe(mockGlide.Decoder.Bytes); + expect(options?.decoder).not.toBe(otherGlide.Decoder.Bytes); + adapter.dispose(); + }); }); From 7058d6adb0fe6727d33496a9a50cf103418b4d15 Mon Sep 17 00:00:00 2001 From: Lev Neiman Date: Wed, 22 Jul 2026 22:29:03 -0700 Subject: [PATCH 2/2] test: run Lua protocol matrix through both clients --- test/redis-adapters.integration.test.ts | 194 ---- test/redis-real.integration.test.ts | 1126 +++++++++++++---------- 2 files changed, 664 insertions(+), 656 deletions(-) delete mode 100644 test/redis-adapters.integration.test.ts diff --git a/test/redis-adapters.integration.test.ts b/test/redis-adapters.integration.test.ts deleted file mode 100644 index 87ee227..0000000 --- a/test/redis-adapters.integration.test.ts +++ /dev/null @@ -1,194 +0,0 @@ -import * as valkeyGlide 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: valkeyGlide.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 valkeyGlide.GlideClient.createClient({ addresses: [{ host, port }] }); - glideAdapter = createValkeyGlideDialCacheClient(glide, valkeyGlide); - 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({ namespace: `${kind}-cache`, redis: { client: adapter() } }); - 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("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-real.integration.test.ts b/test/redis-real.integration.test.ts index 814a13d..04725a7 100644 --- a/test/redis-real.integration.test.ts +++ b/test/redis-real.integration.test.ts @@ -1,6 +1,7 @@ +import * as valkeyGlide from "@valkey/valkey-glide"; import { commandOptions, createClient } from "redis"; import { GenericContainer, type StartedTestContainer, Wait } from "testcontainers"; -import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { CacheLayer, @@ -10,19 +11,133 @@ import { type DialCacheRedisClient, type Serializer, } from "../src/index.js"; +import { + INVALIDATE_CACHE_SCRIPT, + WRITE_CACHE_SCRIPT, + WRITE_TRACKED_CACHE_SCRIPT, +} from "../src/internal/redis-scripts.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 createTestClient = (url: string) => createClient({ url, scripts: dialcacheRedisScripts }); +type NodeRedisTestClient = ReturnType; + +interface RawRedisScriptClient { + write( + valueKey: string, + cacheTtlMs: number, + encoding: number, + payload: string | Buffer, + ): Promise; + writeTracked( + valueKey: string, + watermarkKey: string, + cacheTtlMs: number, + encoding: number, + payload: string | Buffer, + watermarkTtlFloorMs: number, + ): Promise; + invalidate( + watermarkKey: string, + futureBufferMs: number, + watermarkTtlFloorMs: number, + ): Promise; +} + +interface RedisAdapterHarness { + readonly adapter: DialCacheRedisClient; + /** Exercise Lua argument validation that the semantic adapter cannot represent. */ + readonly raw: RawRedisScriptClient; + dispose(): void; +} + +function createNodeRedisHarness(client: NodeRedisTestClient): RedisAdapterHarness { + return { + adapter: createNodeRedisDialCacheClient(client), + raw: { + write: async (...args) => await client.dialcacheWrite(...args), + writeTracked: async (...args) => await client.dialcacheWriteTracked(...args), + invalidate: async (...args) => await client.dialcacheInvalidate(...args), + }, + dispose: () => undefined, + }; +} + +function createValkeyGlideHarness(client: valkeyGlide.GlideClient): RedisAdapterHarness { + const adapter: ValkeyGlideDialCacheClient = createValkeyGlideDialCacheClient(client, valkeyGlide); + const rawScripts = { + write: new valkeyGlide.Script(WRITE_CACHE_SCRIPT), + writeTracked: new valkeyGlide.Script(WRITE_TRACKED_CACHE_SCRIPT), + invalidate: new valkeyGlide.Script(INVALIDATE_CACHE_SCRIPT), + }; + const invoke = async ( + script: valkeyGlide.Script, + keys: Array, + args: Array, + ): Promise => { + const reply = await client.invokeScript(script, { + keys, + args, + decoder: valkeyGlide.Decoder.Bytes, + }); + if (typeof reply !== "number") { + throw new Error("Unexpected non-integer reply from DialCache test script"); + } + return reply; + }; + + return { + adapter, + raw: { + write: async (valueKey, cacheTtlMs, encoding, payload) => + await invoke(rawScripts.write, [valueKey], [String(cacheTtlMs), String(encoding), payload]), + writeTracked: async ( + valueKey, + watermarkKey, + cacheTtlMs, + encoding, + payload, + watermarkTtlFloorMs, + ) => + await invoke( + rawScripts.writeTracked, + [valueKey, watermarkKey], + [String(cacheTtlMs), String(encoding), payload, String(watermarkTtlFloorMs)], + ), + invalidate: async (watermarkKey, futureBufferMs, watermarkTtlFloorMs) => + await invoke( + rawScripts.invalidate, + [watermarkKey], + [String(futureBufferMs), String(watermarkTtlFloorMs)], + ), + }, + dispose() { + adapter.dispose(); + for (const script of Object.values(rawScripts)) { + script.release(); + } + }, + }; +} function encodeFrame(payload: string | Buffer, encoding: number, createdAtMs = Date.now(), version = 1): Buffer { const timestamp = Buffer.alloc(8); @@ -32,526 +147,613 @@ function encodeFrame(payload: string | Buffer, encoding: number, createdAtMs = D describe.each(engines)("DialCache Lua protocol on $name", ({ image }) => { let container: StartedTestContainer | undefined; - let client: ReturnType | undefined; + // This connection controls and inspects server state; cache operations use the selected adapter harness. + let admin: NodeRedisTestClient | undefined; + let glide: valkeyGlide.GlideClient | undefined; + let harnesses: Record | undefined; beforeAll(async () => { container = await new GenericContainer(image) .withExposedPorts(6379) .withWaitStrategy(Wait.forListeningPorts()) .start(); - client = createTestClient(`redis://${container.getHost()}:${container.getMappedPort(6379)}`); - client.on("error", () => undefined); - await client.connect(); + const host = container.getHost(); + const port = container.getMappedPort(6379); + admin = createTestClient(`redis://${host}:${port}`); + admin.on("error", () => undefined); + await admin.connect(); + glide = await valkeyGlide.GlideClient.createClient({ addresses: [{ host, port }] }); + harnesses = { + nodeRedis: createNodeRedisHarness(admin), + valkeyGlide: createValkeyGlideHarness(glide), + }; }); afterAll(async () => { - await client?.quit(); - await container?.stop(); - }); - - it("round-trips untracked UTF-8 and binary values", async () => { - if (client === undefined) { - throw new Error("Redis client did not start"); + for (const harness of Object.values(harnesses ?? {})) { + harness.dispose(); } - const scriptClient: DialCacheRedisClient = createNodeRedisDialCacheClient(client); - const dialcache = new DialCache({ namespace: "real", redis: { client: scriptClient } }); - let jsonCalls = 0; - let binaryCalls = 0; - const getJson = dialcache.cached(async (id: string) => ({ id, calls: ++jsonCalls }), { - keyType: "item_id", - useCase: "RealJson", - cacheKey: (id) => id, - defaultConfig: remoteOnly, - }); - const binarySerializer: Serializer = { - dump: async (value) => Buffer.from(value, "utf8"), - load: async (value) => (Buffer.isBuffer(value) ? value.toString("utf8") : value), - }; - const getBinary = dialcache.cached(async (id: string) => `binary:${id}:${++binaryCalls}`, { - keyType: "item_id", - useCase: "RealBinary", - cacheKey: (id) => id, - defaultConfig: remoteOnly, - serializer: binarySerializer, - }); - - const firstJson = await dialcache.enable(async () => await getJson("json")); - const secondJson = await dialcache.enable(async () => await getJson("json")); - const firstBinary = await dialcache.enable(async () => await getBinary("buffer")); - const secondBinary = await dialcache.enable(async () => await getBinary("buffer")); - - expect(firstJson).toEqual({ id: "json", calls: 1 }); - expect(secondJson).toEqual(firstJson); - expect(firstBinary).toBe("binary:buffer:1"); - expect(secondBinary).toBe(firstBinary); - expect(jsonCalls).toBe(1); - expect(binaryCalls).toBe(1); + glide?.close(); + await admin?.quit(); + await container?.stop(); }); - it("stores arbitrary binary payloads without base64 expansion", async () => { - if (client === undefined) { - throw new Error("Redis client did not start"); - } - const scriptClient = createNodeRedisDialCacheClient(client); - const payloads = [ - Buffer.alloc(0), - Buffer.from(Array.from({ length: 256 }, (_, index) => index)), - Buffer.alloc(2 * 1024 * 1024, 0xa5), - ]; - - for (const [index, payload] of payloads.entries()) { - const valueKey = `binary-raw:{item:${index}}:value`; - expect(await scriptClient.write({ valueKey, cacheTtlMs: 60_000, value: payload })).toBe(true); - - const roundTrip = await scriptClient.read({ valueKey }); - const stored = await client.get(commandOptions({ returnBuffers: true }), valueKey); - - expect(Buffer.isBuffer(roundTrip)).toBe(true); - expect(roundTrip).toEqual(payload); - expect(stored).not.toBeNull(); - expect(stored?.length).toBe(10 + payload.length); - expect(stored?.[0]).toBe(1); - expect(stored?.[9]).toBe(1); - expect(stored?.subarray(10)).toEqual(payload); - } - - const trackedValueKey = "binary-raw:{item:tracked}:value"; - const watermarkKey = "binary-raw:{item:tracked}:watermark"; - const trackedPayload = Buffer.from([0, 0xff, 0xc3, 0x28, 0x80]); - expect( - await scriptClient.write({ - valueKey: trackedValueKey, - watermarkKey, - cacheTtlMs: 60_000, - value: trackedPayload, - watermarkTtlFloorMs: 60_000, - }), - ).toBe(true); - expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toEqual(trackedPayload); - }); + describe.each(adapterKinds)("with $name", ({ kind }) => { + let client: RedisAdapterHarness | undefined; - it("recovers every distinct read and write script after SCRIPT FLUSH", async () => { - if (client === undefined) { - throw new Error("Redis client did not start"); - } - const scriptClient = createNodeRedisDialCacheClient(client); - const valueKey = "script-recovery:{item:untracked}:value"; - - expect(dialcacheRedisScripts.dialcacheRead.SHA1).not.toBe(dialcacheRedisScripts.dialcacheReadTracked.SHA1); - expect(dialcacheRedisScripts.dialcacheWrite.SHA1).not.toBe(dialcacheRedisScripts.dialcacheWriteTracked.SHA1); - - await client.scriptFlush(); - expect(await client.dialcacheWrite(valueKey, 60_000, 0, "untracked")).toBe(1); - await client.scriptFlush(); - expect(await scriptClient.read({ valueKey })).toBe("untracked"); - - const trackedValueKey = "script-recovery:{item:tracked}:value"; - const watermarkKey = "script-recovery:{item:tracked}:watermark"; - await client.scriptFlush(); - expect(await client.dialcacheWriteTracked(trackedValueKey, watermarkKey, 60_000, 0, "tracked", 60_000)).toBe(1); - await client.scriptFlush(); - expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toBe("tracked"); - }); + beforeEach(async () => { + if (admin === undefined || harnesses === undefined) { + throw new Error("Redis test clients did not start"); + } + await admin.flushAll(); + client = harnesses[kind]; + }); - it("treats every invalid read frame and watermark state as a miss", async () => { - if (client === undefined) { - throw new Error("Redis client did not start"); - } - const scriptClient = createNodeRedisDialCacheClient(client); - const valueKey = "read-paths:{item:read}:value"; - const watermarkKey = "read-paths:{item:read}:watermark"; + it("round-trips untracked UTF-8 and binary values", async () => { + if (client === undefined || admin === undefined) { + throw new Error("Redis test clients did not start"); + } + const scriptClient: DialCacheRedisClient = client.adapter; + const dialcache = new DialCache({ namespace: "real", redis: { client: scriptClient } }); + let jsonCalls = 0; + let binaryCalls = 0; + const getJson = dialcache.cached(async (id: string) => ({ id, calls: ++jsonCalls }), { + keyType: "item_id", + useCase: "RealJson", + cacheKey: (id) => id, + defaultConfig: remoteOnly, + }); + const binarySerializer: Serializer = { + dump: async (value) => Buffer.from(value, "utf8"), + load: async (value) => (Buffer.isBuffer(value) ? value.toString("utf8") : value), + }; + const getBinary = dialcache.cached(async (id: string) => `binary:${id}:${++binaryCalls}`, { + keyType: "item_id", + useCase: "RealBinary", + cacheKey: (id) => id, + defaultConfig: remoteOnly, + serializer: binarySerializer, + }); + + const firstJson = await dialcache.enable(async () => await getJson("json")); + const secondJson = await dialcache.enable(async () => await getJson("json")); + const firstBinary = await dialcache.enable(async () => await getBinary("buffer")); + const secondBinary = await dialcache.enable(async () => await getBinary("buffer")); + + expect(firstJson).toEqual({ id: "json", calls: 1 }); + expect(secondJson).toEqual(firstJson); + expect(firstBinary).toBe("binary:buffer:1"); + expect(secondBinary).toBe(firstBinary); + expect(jsonCalls).toBe(1); + expect(binaryCalls).toBe(1); + }); - expect(await scriptClient.read({ valueKey })).toBeNull(); + it("stores arbitrary binary payloads without base64 expansion", async () => { + if (client === undefined || admin === undefined) { + throw new Error("Redis test clients did not start"); + } + const scriptClient = client.adapter; + const payloads = [ + Buffer.alloc(0), + Buffer.from(Array.from({ length: 256 }, (_, index) => index)), + Buffer.alloc(2 * 1024 * 1024, 0xa5), + ]; + + for (const [index, payload] of payloads.entries()) { + const valueKey = `binary-raw:{item:${index}}:value`; + expect(await scriptClient.write({ valueKey, cacheTtlMs: 60_000, value: payload })).toBe(true); + + const roundTrip = await scriptClient.read({ valueKey }); + const stored = await admin.get(commandOptions({ returnBuffers: true }), valueKey); + + expect(Buffer.isBuffer(roundTrip)).toBe(true); + expect(roundTrip).toEqual(payload); + expect(stored).not.toBeNull(); + expect(stored?.length).toBe(10 + payload.length); + expect(stored?.[0]).toBe(1); + expect(stored?.[9]).toBe(1); + expect(stored?.subarray(10)).toEqual(payload); + } + + const trackedValueKey = "binary-raw:{item:tracked}:value"; + const watermarkKey = "binary-raw:{item:tracked}:watermark"; + const trackedPayload = Buffer.from([0, 0xff, 0xc3, 0x28, 0x80]); + expect( + await scriptClient.write({ + valueKey: trackedValueKey, + watermarkKey, + cacheTtlMs: 60_000, + value: trackedPayload, + watermarkTtlFloorMs: 60_000, + }), + ).toBe(true); + expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toEqual(trackedPayload); + }); - await client.set(valueKey, Buffer.alloc(9)); - expect(await scriptClient.read({ valueKey })).toBeNull(); + it("recovers every Lua script after SCRIPT FLUSH", async () => { + if (client === undefined || admin === undefined) { + throw new Error("Redis test clients did not start"); + } + const scriptClient = client.adapter; + const valueKey = "script-recovery:{item:untracked}:value"; + + await admin.scriptFlush(); + expect(await scriptClient.write({ valueKey, cacheTtlMs: 60_000, value: "untracked" })).toBe(true); + await admin.scriptFlush(); + expect(await scriptClient.read({ valueKey })).toBe("untracked"); + + const trackedValueKey = "script-recovery:{item:tracked}:value"; + const watermarkKey = "script-recovery:{item:tracked}:watermark"; + await admin.scriptFlush(); + expect( + await scriptClient.write({ + valueKey: trackedValueKey, + watermarkKey, + cacheTtlMs: 60_000, + value: "tracked", + watermarkTtlFloorMs: 60_000, + }), + ).toBe(true); + await admin.scriptFlush(); + expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toBe("tracked"); + await admin.scriptFlush(); + await expect( + scriptClient.invalidate({ + watermarkKey, + futureBufferMs: 0, + watermarkTtlFloorMs: 60_000, + }), + ).resolves.toBeUndefined(); + }); - await client.set(valueKey, encodeFrame("wrong-version", 0, 1_000, 2)); - expect(await scriptClient.read({ valueKey })).toBeNull(); + it("treats every invalid read frame and watermark state as a miss", async () => { + if (client === undefined || admin === undefined) { + throw new Error("Redis test clients did not start"); + } + const scriptClient = client.adapter; + const valueKey = "read-paths:{item:read}:value"; + const watermarkKey = "read-paths:{item:read}:watermark"; - await client.set(valueKey, encodeFrame("tracked", 0, 1_000)); - expect(await scriptClient.read({ valueKey, watermarkKey })).toBeNull(); + expect(await scriptClient.read({ valueKey })).toBeNull(); - await client.set(watermarkKey, "not-a-watermark"); - expect(await scriptClient.read({ valueKey, watermarkKey })).toBeNull(); + await admin.set(valueKey, Buffer.alloc(9)); + expect(await scriptClient.read({ valueKey })).toBeNull(); - await client.set(watermarkKey, "9".repeat(400)); - expect(await scriptClient.read({ valueKey, watermarkKey })).toBeNull(); + await admin.set(valueKey, encodeFrame("wrong-version", 0, 1_000, 2)); + expect(await scriptClient.read({ valueKey })).toBeNull(); - await client.set(watermarkKey, "1000"); - expect(await scriptClient.read({ valueKey, watermarkKey })).toBeNull(); + await admin.set(valueKey, encodeFrame("tracked", 0, 1_000)); + expect(await scriptClient.read({ valueKey, watermarkKey })).toBeNull(); - await client.set(watermarkKey, "999.5"); - expect(await scriptClient.read({ valueKey, watermarkKey })).toBe("tracked"); - }); + await admin.set(watermarkKey, "not-a-watermark"); + expect(await scriptClient.read({ valueKey, watermarkKey })).toBeNull(); - it("rejects invalid raw script arguments before mutating Redis", async () => { - if (client === undefined) { - throw new Error("Redis client did not start"); - } - const valueKey = "invalid-args:{item:invalid}:value"; - const watermarkKey = "invalid-args:{item:invalid}:watermark"; - const notANumber = "not-a-number" as unknown as number; - - await expect(client.dialcacheWrite(valueKey, 0, 0, "value")).rejects.toThrow("invalid DialCache TTL"); - await expect(client.dialcacheWrite(valueKey, notANumber, 0, "value")).rejects.toThrow("invalid DialCache TTL"); - await expect(client.dialcacheWrite(valueKey, Number.NaN, 0, "value")).rejects.toThrow("invalid DialCache TTL"); - await expect(client.dialcacheWrite(valueKey, Number.POSITIVE_INFINITY, 0, "value")).rejects.toThrow("invalid DialCache TTL"); - await expect(client.dialcacheWrite(valueKey, Number.NEGATIVE_INFINITY, 0, "value")).rejects.toThrow("invalid DialCache TTL"); - await expect(client.dialcacheWrite(valueKey, 1_000, notANumber, "value")).rejects.toThrow("invalid DialCache payload encoding"); - await expect(client.dialcacheWrite(valueKey, 1_000, Number.NaN, "value")).rejects.toThrow("invalid DialCache payload encoding"); - await expect(client.dialcacheWrite(valueKey, 1_000, 2, "value")).rejects.toThrow("invalid DialCache payload encoding"); - await expect(client.dialcacheWriteTracked(valueKey, watermarkKey, 1_000, 0, "value", 0)).rejects.toThrow( - "invalid DialCache watermark TTL", - ); - await expect(client.dialcacheWriteTracked(valueKey, watermarkKey, 1_000, 0, "value", Number.NaN)).rejects.toThrow( - "invalid DialCache watermark TTL", - ); - await expect(client.dialcacheInvalidate(watermarkKey, -1, 1_000)).rejects.toThrow("invalid DialCache future buffer"); - await expect(client.dialcacheInvalidate(watermarkKey, notANumber, 1_000)).rejects.toThrow("invalid DialCache future buffer"); - await expect(client.dialcacheInvalidate(watermarkKey, Number.NaN, 1_000)).rejects.toThrow("invalid DialCache future buffer"); - await expect(client.dialcacheInvalidate(watermarkKey, Number.POSITIVE_INFINITY, 1_000)).rejects.toThrow( - "invalid DialCache future buffer", - ); - await expect(client.dialcacheInvalidate(watermarkKey, Number.NEGATIVE_INFINITY, 1_000)).rejects.toThrow( - "invalid DialCache future buffer", - ); - await expect(client.dialcacheInvalidate(watermarkKey, 0, 0)).rejects.toThrow("invalid DialCache watermark TTL"); - await expect(client.dialcacheInvalidate(watermarkKey, 0, notANumber)).rejects.toThrow("invalid DialCache watermark TTL"); - await expect(client.dialcacheInvalidate(watermarkKey, 0, Number.NaN)).rejects.toThrow("invalid DialCache watermark TTL"); - await expect(client.dialcacheInvalidate(watermarkKey, 0, Number.POSITIVE_INFINITY)).rejects.toThrow( - "invalid DialCache watermark TTL", - ); - - expect(await client.exists([valueKey, watermarkKey])).toBe(0); - }); + await admin.set(watermarkKey, "9".repeat(400)); + expect(await scriptClient.read({ valueKey, watermarkKey })).toBeNull(); - it("rounds fractional raw protocol durations upward", async () => { - if (client === undefined) { - throw new Error("Redis client did not start"); - } - const valueKey = "fractional-args:{item:fractional}:value"; - const watermarkKey = "fractional-args:{item:fractional}:watermark"; - - expect(await client.dialcacheWrite(valueKey, 1_000.1, 0, "value")).toBe(1); - expect(await client.pTTL(valueKey)).toBeGreaterThan(900); - expect(await client.pTTL(valueKey)).toBeLessThanOrEqual(1_001); - - const trackedValueKey = "fractional-args:{item:tracked-floor}:value"; - const trackedWatermarkKey = "fractional-args:{item:tracked-floor}:watermark"; - expect( - await client.dialcacheWriteTracked(trackedValueKey, trackedWatermarkKey, 1_000, 0, "value", 70_000.1), - ).toBe(1); - expect(await client.get(trackedWatermarkKey)).toBe("0"); - expect(await client.pTTL(trackedWatermarkKey)).toBeGreaterThan(69_000); - expect(await client.pTTL(trackedWatermarkKey)).toBeLessThanOrEqual(70_001); - - const beforeMs = (await client.time()).getTime(); - expect(await client.dialcacheInvalidate(watermarkKey, 100.1, 70_000.1)).toBe(1); - const watermark = Number(await client.get(watermarkKey)); - expect(Number.isSafeInteger(watermark)).toBe(true); - expect(watermark).toBeGreaterThanOrEqual(beforeMs + 101); - expect(await client.pTTL(watermarkKey)).toBeGreaterThan(69_000); - expect(await client.pTTL(watermarkKey)).toBeLessThanOrEqual(70_001); - }); + await admin.set(watermarkKey, "1000"); + expect(await scriptClient.read({ valueKey, watermarkKey })).toBeNull(); - it("invalidates tracked entries and recovers after SCRIPT FLUSH", async () => { - if (client === undefined) { - throw new Error("Redis client did not start"); - } - const scriptClient: DialCacheRedisClient = createNodeRedisDialCacheClient(client); - const dialcache = new DialCache({ namespace: "tracked", redis: { client: scriptClient } }); - let version = 1; - let calls = 0; - const getUser = dialcache.cached(async (id: string) => ({ id, version, calls: ++calls }), { - keyType: "user_id", - useCase: "RealTracked", - cacheKey: (id) => id, - trackForInvalidation: true, - defaultConfig: remoteOnly, + await admin.set(watermarkKey, "999.5"); + expect(await scriptClient.read({ valueKey, watermarkKey })).toBe("tracked"); }); - const first = await dialcache.enable(async () => await getUser("123")); - version = 2; - const cached = await dialcache.enable(async () => await getUser("123")); - await dialcache.invalidateRemote("user_id", "123"); - await new Promise((resolve) => setTimeout(resolve, 2)); - const refreshed = await dialcache.enable(async () => await getUser("123")); - await client.scriptFlush(); - const afterScriptFlush = await dialcache.enable(async () => await getUser("123")); - - expect(first).toEqual({ id: "123", version: 1, calls: 1 }); - expect(cached).toEqual(first); - expect(refreshed).toEqual({ id: "123", version: 2, calls: 2 }); - expect(afterScriptFlush).toEqual(refreshed); - }); + it("rejects invalid raw script arguments before mutating Redis", async () => { + if (client === undefined || admin === undefined) { + throw new Error("Redis test clients did not start"); + } + const valueKey = "invalid-args:{item:invalid}:value"; + const watermarkKey = "invalid-args:{item:invalid}:watermark"; + const notANumber = "not-a-number" as unknown as number; + + await expect(client.raw.write(valueKey, 0, 0, "value")).rejects.toThrow("invalid DialCache TTL"); + await expect(client.raw.write(valueKey, notANumber, 0, "value")).rejects.toThrow("invalid DialCache TTL"); + await expect(client.raw.write(valueKey, Number.NaN, 0, "value")).rejects.toThrow("invalid DialCache TTL"); + await expect(client.raw.write(valueKey, Number.POSITIVE_INFINITY, 0, "value")).rejects.toThrow("invalid DialCache TTL"); + await expect(client.raw.write(valueKey, Number.NEGATIVE_INFINITY, 0, "value")).rejects.toThrow("invalid DialCache TTL"); + await expect(client.raw.write(valueKey, 1_000, notANumber, "value")).rejects.toThrow("invalid DialCache payload encoding"); + await expect(client.raw.write(valueKey, 1_000, Number.NaN, "value")).rejects.toThrow("invalid DialCache payload encoding"); + await expect(client.raw.write(valueKey, 1_000, 2, "value")).rejects.toThrow("invalid DialCache payload encoding"); + await expect(client.raw.writeTracked(valueKey, watermarkKey, 1_000, 0, "value", 0)).rejects.toThrow( + "invalid DialCache watermark TTL", + ); + await expect(client.raw.writeTracked(valueKey, watermarkKey, 1_000, 0, "value", Number.NaN)).rejects.toThrow( + "invalid DialCache watermark TTL", + ); + await expect(client.raw.invalidate(watermarkKey, -1, 1_000)).rejects.toThrow("invalid DialCache future buffer"); + await expect(client.raw.invalidate(watermarkKey, notANumber, 1_000)).rejects.toThrow("invalid DialCache future buffer"); + await expect(client.raw.invalidate(watermarkKey, Number.NaN, 1_000)).rejects.toThrow("invalid DialCache future buffer"); + await expect(client.raw.invalidate(watermarkKey, Number.POSITIVE_INFINITY, 1_000)).rejects.toThrow( + "invalid DialCache future buffer", + ); + await expect(client.raw.invalidate(watermarkKey, Number.NEGATIVE_INFINITY, 1_000)).rejects.toThrow( + "invalid DialCache future buffer", + ); + await expect(client.raw.invalidate(watermarkKey, 0, 0)).rejects.toThrow("invalid DialCache watermark TTL"); + await expect(client.raw.invalidate(watermarkKey, 0, notANumber)).rejects.toThrow("invalid DialCache watermark TTL"); + await expect(client.raw.invalidate(watermarkKey, 0, Number.NaN)).rejects.toThrow("invalid DialCache watermark TTL"); + await expect(client.raw.invalidate(watermarkKey, 0, Number.POSITIVE_INFINITY)).rejects.toThrow( + "invalid DialCache watermark TTL", + ); - it("fails open without caching malformed watermark state", async () => { - if (client === undefined) { - throw new Error("Redis client did not start"); - } - const scriptClient = createNodeRedisDialCacheClient(client); - const logger = { debug: () => undefined, warn: () => undefined, error: () => undefined }; - const dialcache = new DialCache({ namespace: "malformed", redis: { client: scriptClient }, logger }); - let calls = 0; - const getUser = dialcache.cached(async (id: string) => ({ id, calls: ++calls }), { - keyType: "user_id", - useCase: "MalformedWatermark", - cacheKey: (id) => id, - trackForInvalidation: true, - defaultConfig: remoteOnly, + expect(await admin.exists([valueKey, watermarkKey])).toBe(0); }); - await client.set("{malformed:user_id:bad}#watermark", "0x10"); - - const first = await dialcache.enable(async () => await getUser("bad")); - const second = await dialcache.enable(async () => await getUser("bad")); - expect(first).toEqual({ id: "bad", calls: 1 }); - expect(second).toEqual({ id: "bad", calls: 2 }); - }); + it("rounds fractional raw protocol durations upward", async () => { + if (client === undefined || admin === undefined) { + throw new Error("Redis test clients did not start"); + } + const valueKey = "fractional-args:{item:fractional}:value"; + const watermarkKey = "fractional-args:{item:fractional}:watermark"; + + expect(await client.raw.write(valueKey, 1_000.1, 0, "value")).toBe(1); + expect(await admin.pTTL(valueKey)).toBeGreaterThan(900); + expect(await admin.pTTL(valueKey)).toBeLessThanOrEqual(1_001); + + const trackedValueKey = "fractional-args:{item:tracked-floor}:value"; + const trackedWatermarkKey = "fractional-args:{item:tracked-floor}:watermark"; + expect( + await client.raw.writeTracked(trackedValueKey, trackedWatermarkKey, 1_000, 0, "value", 70_000.1), + ).toBe(1); + expect(await admin.get(trackedWatermarkKey)).toBe("0"); + expect(await admin.pTTL(trackedWatermarkKey)).toBeGreaterThan(69_000); + expect(await admin.pTTL(trackedWatermarkKey)).toBeLessThanOrEqual(70_001); + + const beforeMs = (await admin.time()).getTime(); + expect(await client.raw.invalidate(watermarkKey, 100.1, 70_000.1)).toBe(1); + const watermark = Number(await admin.get(watermarkKey)); + expect(Number.isSafeInteger(watermark)).toBe(true); + expect(watermark).toBeGreaterThanOrEqual(beforeMs + 101); + expect(await admin.pTTL(watermarkKey)).toBeGreaterThan(69_000); + expect(await admin.pTTL(watermarkKey)).toBeLessThanOrEqual(70_001); + }); - it("rejects malformed tracked watermark writes without overwriting the cached value", async () => { - if (client === undefined) { - throw new Error("Redis client did not start"); - } - const scriptClient = createNodeRedisDialCacheClient(client); - const valueKey = "malformed-write:{item:malformed}:value"; - const watermarkKey = "malformed-write:{item:malformed}:watermark"; - expect(await client.dialcacheWrite(valueKey, 60_000, 0, "original")).toBe(1); - - for (const malformed of ["not-a-watermark", "9".repeat(400)]) { - await client.set(watermarkKey, malformed, { PX: 60_000 }); - await expect(client.dialcacheWriteTracked(valueKey, watermarkKey, 60_000, 0, "replacement", 60_000)).rejects.toThrow( - "invalid DialCache watermark", - ); - expect(await scriptClient.read({ valueKey })).toBe("original"); - } - }); + it("invalidates tracked entries and recovers after SCRIPT FLUSH", async () => { + if (client === undefined || admin === undefined) { + throw new Error("Redis test clients did not start"); + } + const scriptClient: DialCacheRedisClient = client.adapter; + const dialcache = new DialCache({ namespace: "tracked", redis: { client: scriptClient } }); + let version = 1; + let calls = 0; + const getUser = dialcache.cached(async (id: string) => ({ id, version, calls: ++calls }), { + keyType: "user_id", + useCase: "RealTracked", + cacheKey: (id) => id, + trackForInvalidation: true, + defaultConfig: remoteOnly, + }); + + const first = await dialcache.enable(async () => await getUser("123")); + version = 2; + const cached = await dialcache.enable(async () => await getUser("123")); + await dialcache.invalidateRemote("user_id", "123"); + await new Promise((resolve) => setTimeout(resolve, 2)); + const refreshed = await dialcache.enable(async () => await getUser("123")); + await admin.scriptFlush(); + const afterScriptFlush = await dialcache.enable(async () => await getUser("123")); + + expect(first).toEqual({ id: "123", version: 1, calls: 1 }); + expect(cached).toEqual(first); + expect(refreshed).toEqual({ id: "123", version: 2, calls: 2 }); + expect(afterScriptFlush).toEqual(refreshed); + }); - it("labels malformed payload encoding through the production adapter", async () => { - if (client === undefined) { - throw new Error("Redis client did not start"); - } - const adapter = createNodeRedisDialCacheClient(client); - const write = vi.fn(adapter.write); - const redisClient: DialCacheRedisClient = { ...adapter, write }; - const metrics: DialCacheMetricsAdapter = { - request: vi.fn(), - miss: vi.fn(), - disabled: vi.fn(), - error: vi.fn(), - invalidation: vi.fn(), - coalesced: vi.fn(), - observeGet: vi.fn(), - observeFallback: vi.fn(), - observeSerialization: vi.fn(), - observeSize: vi.fn(), - }; - const logger = { debug: vi.fn(), warn: vi.fn(), error: vi.fn() }; - const namespace = "bad-encoding"; - const valueKey = `{${namespace}:user_id:bad}#RealMalformedEncoding:dialcache-frame-v1`; - const watermarkKey = `{${namespace}:user_id:bad}#watermark`; - await client.set(valueKey, encodeFrame("malformed", 2), { PX: 60_000 }); - await client.set(watermarkKey, "0", { PX: 60_000 }); - - const dialcache = new DialCache({ namespace, redis: { client: redisClient }, logger, metrics }); - let calls = 0; - const getUser = dialcache.cached(async (id: string) => ({ id, calls: ++calls }), { - keyType: "user_id", - useCase: "RealMalformedEncoding", - cacheKey: (id) => id, - trackForInvalidation: true, - defaultConfig: remoteOnly, + it("fails open without caching malformed watermark state", async () => { + if (client === undefined || admin === undefined) { + throw new Error("Redis test clients did not start"); + } + const scriptClient = client.adapter; + const logger = { debug: () => undefined, warn: () => undefined, error: () => undefined }; + const dialcache = new DialCache({ namespace: "malformed", redis: { client: scriptClient }, logger }); + let calls = 0; + const getUser = dialcache.cached(async (id: string) => ({ id, calls: ++calls }), { + keyType: "user_id", + useCase: "MalformedWatermark", + cacheKey: (id) => id, + trackForInvalidation: true, + defaultConfig: remoteOnly, + }); + await admin.set("{malformed:user_id:bad}#watermark", "0x10"); + + const first = await dialcache.enable(async () => await getUser("bad")); + const second = await dialcache.enable(async () => await getUser("bad")); + + expect(first).toEqual({ id: "bad", calls: 1 }); + expect(second).toEqual({ id: "bad", calls: 2 }); }); - const value = await dialcache.enable(async () => await getUser("bad")); - - expect(value).toEqual({ id: "bad", calls: 1 }); - expect(write).not.toHaveBeenCalled(); - expect(metrics.error).toHaveBeenCalledWith({ - cacheNamespace: namespace, - useCase: "RealMalformedEncoding", - keyType: "user_id", - layer: CacheLayer.REMOTE, - error: "cache_read", - inFallback: false, + it("rejects malformed tracked watermark writes without overwriting the cached value", async () => { + if (client === undefined || admin === undefined) { + throw new Error("Redis test clients did not start"); + } + const scriptClient = client.adapter; + const valueKey = "malformed-write:{item:malformed}:value"; + const watermarkKey = "malformed-write:{item:malformed}:watermark"; + expect(await client.raw.write(valueKey, 60_000, 0, "original")).toBe(1); + + for (const malformed of ["not-a-watermark", "9".repeat(400)]) { + await admin.set(watermarkKey, malformed, { PX: 60_000 }); + await expect(client.raw.writeTracked(valueKey, watermarkKey, 60_000, 0, "replacement", 60_000)).rejects.toThrow( + "invalid DialCache watermark", + ); + expect(await scriptClient.read({ valueKey })).toBe("original"); + } }); - expect(logger.warn).toHaveBeenCalledWith( - "Error getting value from Redis cache", - expect.objectContaining({ name: "DialCacheRedisPayloadEncodingError" }), - ); - }); - it("keeps future fractional watermarks alive without shortening longer TTLs", async () => { - if (client === undefined) { - throw new Error("Redis client did not start"); - } - const scriptClient = createNodeRedisDialCacheClient(client); - const redisNowMs = (await client.time()).getTime(); - const legacyWatermark = redisNowMs + 30_000.5; - const shortTtlKey = "legacy:{urn:user_id:short}#watermark"; - await client.set(shortTtlKey, String(legacyWatermark), { PX: 1_000 }); - - await scriptClient.invalidate({ - watermarkKey: shortTtlKey, - futureBufferMs: 1_000, - watermarkTtlFloorMs: 60_000, + it("labels malformed payload encoding through the production adapter", async () => { + if (client === undefined || admin === undefined) { + throw new Error("Redis test clients did not start"); + } + const adapter = client.adapter; + const write = vi.fn(adapter.write); + const redisClient: DialCacheRedisClient = { ...adapter, write }; + const metrics: DialCacheMetricsAdapter = { + request: vi.fn(), + miss: vi.fn(), + disabled: vi.fn(), + error: vi.fn(), + invalidation: vi.fn(), + coalesced: vi.fn(), + observeGet: vi.fn(), + observeFallback: vi.fn(), + observeSerialization: vi.fn(), + observeSize: vi.fn(), + }; + const logger = { debug: vi.fn(), warn: vi.fn(), error: vi.fn() }; + const namespace = "bad-encoding"; + const valueKey = `{${namespace}:user_id:bad}#RealMalformedEncoding:dialcache-frame-v1`; + const watermarkKey = `{${namespace}:user_id:bad}#watermark`; + await admin.set(valueKey, encodeFrame("malformed", 2), { PX: 60_000 }); + await admin.set(watermarkKey, "0", { PX: 60_000 }); + + const dialcache = new DialCache({ namespace, redis: { client: redisClient }, logger, metrics }); + let calls = 0; + const getUser = dialcache.cached(async (id: string) => ({ id, calls: ++calls }), { + keyType: "user_id", + useCase: "RealMalformedEncoding", + cacheKey: (id) => id, + trackForInvalidation: true, + defaultConfig: remoteOnly, + }); + + const value = await dialcache.enable(async () => await getUser("bad")); + + expect(value).toEqual({ id: "bad", calls: 1 }); + expect(write).not.toHaveBeenCalled(); + expect(metrics.error).toHaveBeenCalledWith({ + cacheNamespace: namespace, + useCase: "RealMalformedEncoding", + keyType: "user_id", + layer: CacheLayer.REMOTE, + error: "cache_read", + inFallback: false, + }); + expect(logger.warn).toHaveBeenCalledWith( + "Error getting value from Redis cache", + expect.objectContaining({ name: "DialCacheRedisPayloadEncodingError" }), + ); }); - expect(Number(await client.get(shortTtlKey))).toBeGreaterThanOrEqual(Math.ceil(legacyWatermark)); - expect(await client.pTTL(shortTtlKey)).toBeGreaterThan(89_000); + it("keeps future fractional watermarks alive without shortening longer TTLs", async () => { + if (client === undefined || admin === undefined) { + throw new Error("Redis test clients did not start"); + } + const scriptClient = client.adapter; + const redisNowMs = (await admin.time()).getTime(); + const legacyWatermark = redisNowMs + 30_000.5; + const shortTtlKey = "legacy:{urn:user_id:short}#watermark"; + await admin.set(shortTtlKey, String(legacyWatermark), { PX: 1_000 }); + + await scriptClient.invalidate({ + watermarkKey: shortTtlKey, + futureBufferMs: 1_000, + watermarkTtlFloorMs: 60_000, + }); - const longTtlKey = "legacy:{urn:user_id:long}#watermark"; - await client.set(longTtlKey, String(legacyWatermark), { PX: 120_000 }); - const ttlBefore = await client.pTTL(longTtlKey); + expect(Number(await admin.get(shortTtlKey))).toBeGreaterThanOrEqual(Math.ceil(legacyWatermark)); + expect(await admin.pTTL(shortTtlKey)).toBeGreaterThan(89_000); - await scriptClient.invalidate({ - watermarkKey: longTtlKey, - futureBufferMs: 1_000, - watermarkTtlFloorMs: 60_000, - }); + const longTtlKey = "legacy:{urn:user_id:long}#watermark"; + await admin.set(longTtlKey, String(legacyWatermark), { PX: 120_000 }); + const ttlBefore = await admin.pTTL(longTtlKey); - expect(Number(await client.get(longTtlKey))).toBeGreaterThanOrEqual(Math.ceil(legacyWatermark)); - const ttlAfter = await client.pTTL(longTtlKey); - expect(ttlAfter).toBeGreaterThan(ttlBefore - 1_000); - expect(ttlAfter).toBeLessThanOrEqual(ttlBefore); - }); + await scriptClient.invalidate({ + watermarkKey: longTtlKey, + futureBufferMs: 1_000, + watermarkTtlFloorMs: 60_000, + }); - it("creates missing and repairs malformed invalidation watermarks", async () => { - if (client === undefined) { - throw new Error("Redis client did not start"); - } - const scriptClient = createNodeRedisDialCacheClient(client); - const missingKey = "invalidate-paths:{item:missing}:watermark"; - const beforeMs = (await client.time()).getTime(); - - await scriptClient.invalidate({ watermarkKey: missingKey, futureBufferMs: 100, watermarkTtlFloorMs: 1_000 }); - - const created = Number(await client.get(missingKey)); - expect(Number.isSafeInteger(created)).toBe(true); - expect(created).toBeGreaterThanOrEqual(beforeMs + 100); - expect(await client.pTTL(missingKey)).toBeGreaterThan(60_000); - - for (const [suffix, malformed] of [ - ["syntax", "not-a-watermark"], - ["overflow", "9".repeat(400)], - ] as const) { - const watermarkKey = `invalidate-paths:{item:${suffix}}:watermark`; - await client.set(watermarkKey, malformed, { PX: 1_000 }); - await scriptClient.invalidate({ watermarkKey, futureBufferMs: 0, watermarkTtlFloorMs: 2_000 }); - expect(Number.isSafeInteger(Number(await client.get(watermarkKey)))).toBe(true); - expect(await client.pTTL(watermarkKey)).toBeGreaterThan(59_000); - } - }); + expect(Number(await admin.get(longTtlKey))).toBeGreaterThanOrEqual(Math.ceil(legacyWatermark)); + const ttlAfter = await admin.pTTL(longTtlKey); + expect(ttlAfter).toBeGreaterThan(ttlBefore - 1_000); + expect(ttlAfter).toBeLessThanOrEqual(ttlBefore); + }); - it("keeps persistent invalidation watermarks persistent", async () => { - if (client === undefined) { - throw new Error("Redis client did not start"); - } - const scriptClient = createNodeRedisDialCacheClient(client); - const watermarkKey = "invalidate-persistent:{item:persistent}:watermark"; - await client.set(watermarkKey, "1"); + it("creates missing and repairs malformed invalidation watermarks", async () => { + if (client === undefined || admin === undefined) { + throw new Error("Redis test clients did not start"); + } + const scriptClient = client.adapter; + const missingKey = "invalidate-paths:{item:missing}:watermark"; + const beforeMs = (await admin.time()).getTime(); + + await scriptClient.invalidate({ watermarkKey: missingKey, futureBufferMs: 100, watermarkTtlFloorMs: 1_000 }); + + const created = Number(await admin.get(missingKey)); + expect(Number.isSafeInteger(created)).toBe(true); + expect(created).toBeGreaterThanOrEqual(beforeMs + 100); + expect(await admin.pTTL(missingKey)).toBeGreaterThan(60_000); + + for (const [suffix, malformed] of [ + ["syntax", "not-a-watermark"], + ["overflow", "9".repeat(400)], + ] as const) { + const watermarkKey = `invalidate-paths:{item:${suffix}}:watermark`; + await admin.set(watermarkKey, malformed, { PX: 1_000 }); + await scriptClient.invalidate({ watermarkKey, futureBufferMs: 0, watermarkTtlFloorMs: 2_000 }); + expect(Number.isSafeInteger(Number(await admin.get(watermarkKey)))).toBe(true); + expect(await admin.pTTL(watermarkKey)).toBeGreaterThan(59_000); + } + }); - await scriptClient.invalidate({ watermarkKey, futureBufferMs: 0, watermarkTtlFloorMs: 60_000 }); + it("keeps persistent invalidation watermarks persistent", async () => { + if (client === undefined || admin === undefined) { + throw new Error("Redis test clients did not start"); + } + const scriptClient = client.adapter; + const watermarkKey = "invalidate-persistent:{item:persistent}:watermark"; + await admin.set(watermarkKey, "1"); - expect(Number(await client.get(watermarkKey))).toBeGreaterThan(1); - expect(await client.pTTL(watermarkKey)).toBe(-1); - }); + await scriptClient.invalidate({ watermarkKey, futureBufferMs: 0, watermarkTtlFloorMs: 60_000 }); - it("preserves a fractional legacy watermark while extending its TTL", async () => { - if (client === undefined) { - throw new Error("Redis client did not start"); - } - const scriptClient = createNodeRedisDialCacheClient(client); - const valueKey = "legacy-write:{urn:user_id:123}:value"; - const watermarkKey = "legacy-write:{urn:user_id:123}:watermark"; - await client.set(watermarkKey, "1.75", { PX: 1_000 }); - - const wrote = await scriptClient.write({ - valueKey, - watermarkKey, - cacheTtlMs: 2_000, - value: "cached", - watermarkTtlFloorMs: 1_000, + expect(Number(await admin.get(watermarkKey))).toBeGreaterThan(1); + expect(await admin.pTTL(watermarkKey)).toBe(-1); }); - expect(wrote).toBe(true); - expect(await client.get(watermarkKey)).toBe("1.75"); - expect(await client.pTTL(watermarkKey)).toBeGreaterThanOrEqual(61_000); - expect(await scriptClient.read({ valueKey, watermarkKey })).toBe("cached"); - }); - - it("does not rewrite sufficient or persistent watermarks on tracked writes", async () => { - if (client === undefined) { - throw new Error("Redis client did not start"); - } - const scriptClient = createNodeRedisDialCacheClient(client); - const sufficientValueKey = "write-sufficient:{item:sufficient}:value"; - const sufficientWatermarkKey = "write-sufficient:{item:sufficient}:watermark"; - await client.set(sufficientWatermarkKey, "1.75", { PX: 120_000 }); - const sufficientTtlBefore = await client.pTTL(sufficientWatermarkKey); - - expect( - await scriptClient.write({ - valueKey: sufficientValueKey, - watermarkKey: sufficientWatermarkKey, + it("preserves a fractional legacy watermark while extending its TTL", async () => { + if (client === undefined || admin === undefined) { + throw new Error("Redis test clients did not start"); + } + const scriptClient = client.adapter; + const valueKey = "legacy-write:{urn:user_id:123}:value"; + const watermarkKey = "legacy-write:{urn:user_id:123}:watermark"; + await admin.set(watermarkKey, "1.75", { PX: 1_000 }); + + const wrote = await scriptClient.write({ + valueKey, + watermarkKey, cacheTtlMs: 2_000, value: "cached", watermarkTtlFloorMs: 1_000, - }), - ).toBe(true); + }); - expect(await client.get(sufficientWatermarkKey)).toBe("1.75"); - expect(await client.pTTL(sufficientWatermarkKey)).toBeGreaterThan(sufficientTtlBefore - 1_000); - expect(await client.pTTL(sufficientWatermarkKey)).toBeLessThanOrEqual(sufficientTtlBefore); + expect(wrote).toBe(true); + expect(await admin.get(watermarkKey)).toBe("1.75"); + expect(await admin.pTTL(watermarkKey)).toBeGreaterThanOrEqual(61_000); + expect(await scriptClient.read({ valueKey, watermarkKey })).toBe("cached"); + }); - const persistentValueKey = "write-persistent:{item:persistent}:value"; - const persistentWatermarkKey = "write-persistent:{item:persistent}:watermark"; - await client.set(persistentWatermarkKey, "2.25"); + it("does not rewrite sufficient or persistent watermarks on tracked writes", async () => { + if (client === undefined || admin === undefined) { + throw new Error("Redis test clients did not start"); + } + const scriptClient = client.adapter; + const sufficientValueKey = "write-sufficient:{item:sufficient}:value"; + const sufficientWatermarkKey = "write-sufficient:{item:sufficient}:watermark"; + await admin.set(sufficientWatermarkKey, "1.75", { PX: 120_000 }); + const sufficientTtlBefore = await admin.pTTL(sufficientWatermarkKey); + + expect( + await scriptClient.write({ + valueKey: sufficientValueKey, + watermarkKey: sufficientWatermarkKey, + cacheTtlMs: 2_000, + value: "cached", + watermarkTtlFloorMs: 1_000, + }), + ).toBe(true); + + expect(await admin.get(sufficientWatermarkKey)).toBe("1.75"); + expect(await admin.pTTL(sufficientWatermarkKey)).toBeGreaterThan(sufficientTtlBefore - 1_000); + expect(await admin.pTTL(sufficientWatermarkKey)).toBeLessThanOrEqual(sufficientTtlBefore); + + const persistentValueKey = "write-persistent:{item:persistent}:value"; + const persistentWatermarkKey = "write-persistent:{item:persistent}:watermark"; + await admin.set(persistentWatermarkKey, "2.25"); + + expect( + await scriptClient.write({ + valueKey: persistentValueKey, + watermarkKey: persistentWatermarkKey, + cacheTtlMs: 2_000, + value: "cached", + watermarkTtlFloorMs: 1_000, + }), + ).toBe(true); + expect(await admin.get(persistentWatermarkKey)).toBe("2.25"); + expect(await admin.pTTL(persistentWatermarkKey)).toBe(-1); + }); - expect( - await scriptClient.write({ - valueKey: persistentValueKey, - watermarkKey: persistentWatermarkKey, + it("atomically blocks writes during the buffer and extends watermark TTL", async () => { + if (client === undefined || admin === undefined) { + throw new Error("Redis test clients did not start"); + } + const scriptClient = client.adapter; + const valueKey = "protocol:{item:ttl}:value"; + const watermarkKey = "protocol:{item:ttl}:watermark"; + const writeRequest = { + valueKey, + watermarkKey, cacheTtlMs: 2_000, value: "cached", watermarkTtlFloorMs: 1_000, - }), - ).toBe(true); - expect(await client.get(persistentWatermarkKey)).toBe("2.25"); - expect(await client.pTTL(persistentWatermarkKey)).toBe(-1); + }; + + expect(await scriptClient.write(writeRequest)).toBe(true); + expect(await admin.get(watermarkKey)).toBe("0"); + const ttlAfterWrite = await admin.pTTL(watermarkKey); + expect(ttlAfterWrite).toBeGreaterThanOrEqual(61_000); + + await scriptClient.invalidate({ watermarkKey, futureBufferMs: 100, watermarkTtlFloorMs: 1_000 }); + expect(await scriptClient.read({ valueKey, watermarkKey })).toBeNull(); + expect(await scriptClient.write({ ...writeRequest, value: "blocked" })).toBe(false); + expect(await scriptClient.read({ valueKey })).toBe("cached"); + const ttlBeforeRead = await admin.pTTL(watermarkKey); + await scriptClient.read({ valueKey, watermarkKey }); + expect(await admin.pTTL(watermarkKey)).toBeLessThanOrEqual(ttlBeforeRead); + + await new Promise((resolve) => setTimeout(resolve, 110)); + expect(await scriptClient.write({ ...writeRequest, value: "fresh" })).toBe(true); + expect(await scriptClient.read({ valueKey, watermarkKey })).toBe("fresh"); + }); }); - it("atomically blocks writes during the buffer and extends watermark TTL", async () => { - if (client === undefined) { - throw new Error("Redis client did not start"); + it("uses one wire format across node-redis and Valkey GLIDE", async () => { + if (admin === undefined || harnesses === undefined) { + throw new Error("Redis test clients did not start"); } - const scriptClient = createNodeRedisDialCacheClient(client); - const valueKey = "protocol:{item:ttl}:value"; - const watermarkKey = "protocol:{item:ttl}:watermark"; - const writeRequest = { - valueKey, - watermarkKey, - cacheTtlMs: 2_000, - value: "cached", - watermarkTtlFloorMs: 1_000, - }; - - expect(await scriptClient.write(writeRequest)).toBe(true); - expect(await client.get(watermarkKey)).toBe("0"); - const ttlAfterWrite = await client.pTTL(watermarkKey); - expect(ttlAfterWrite).toBeGreaterThanOrEqual(61_000); - - await scriptClient.invalidate({ watermarkKey, futureBufferMs: 100, watermarkTtlFloorMs: 1_000 }); - expect(await scriptClient.read({ valueKey, watermarkKey })).toBeNull(); - expect(await scriptClient.write({ ...writeRequest, value: "blocked" })).toBe(false); - expect(await scriptClient.read({ valueKey })).toBe("cached"); - const ttlBeforeRead = await client.pTTL(watermarkKey); - await scriptClient.read({ valueKey, watermarkKey }); - expect(await client.pTTL(watermarkKey)).toBeLessThanOrEqual(ttlBeforeRead); - - await new Promise((resolve) => setTimeout(resolve, 110)); - expect(await scriptClient.write({ ...writeRequest, value: "fresh" })).toBe(true); - expect(await scriptClient.read({ valueKey, watermarkKey })).toBe("fresh"); + await admin.flushAll(); + const nodeRedis = harnesses.nodeRedis.adapter; + const valkeyGlide = harnesses.valkeyGlide.adapter; + const binary = Buffer.from([0, 0xff, 0xc3, 0x28, 0x80]); + + await nodeRedis.write({ valueKey: "interop:node-to-glide", cacheTtlMs: 60_000, value: binary }); + await expect(valkeyGlide.read({ valueKey: "interop:node-to-glide" })).resolves.toEqual(binary); + + await valkeyGlide.write({ valueKey: "interop:glide-to-node", cacheTtlMs: 60_000, value: "hello" }); + await expect(nodeRedis.read({ valueKey: "interop:glide-to-node" })).resolves.toBe("hello"); + + const nodeTrackedValueKey = "interop:{node-tracked}:value"; + const nodeTrackedWatermarkKey = "interop:{node-tracked}:watermark"; + await nodeRedis.write({ + valueKey: nodeTrackedValueKey, + watermarkKey: nodeTrackedWatermarkKey, + cacheTtlMs: 60_000, + value: binary, + watermarkTtlFloorMs: 60_000, + }); + await expect( + valkeyGlide.read({ + valueKey: nodeTrackedValueKey, + watermarkKey: nodeTrackedWatermarkKey, + }), + ).resolves.toEqual(binary); + + const glideTrackedValueKey = "interop:{glide-tracked}:value"; + const glideTrackedWatermarkKey = "interop:{glide-tracked}:watermark"; + await valkeyGlide.write({ + valueKey: glideTrackedValueKey, + watermarkKey: glideTrackedWatermarkKey, + cacheTtlMs: 60_000, + value: "tracked", + watermarkTtlFloorMs: 60_000, + }); + await expect( + nodeRedis.read({ + valueKey: glideTrackedValueKey, + watermarkKey: glideTrackedWatermarkKey, + }), + ).resolves.toBe("tracked"); }); });