From 9e60cdf232ff7f476cce5b44ebb132a84b6b2b64 Mon Sep 17 00:00:00 2001 From: Lev Neiman Date: Thu, 23 Jul 2026 19:17:09 -0700 Subject: [PATCH] feat: enforce per-use-case Redis read deadlines --- README.md | 68 ++- scripts/benchmark-request-local.mjs | 84 ++- scripts/test-package.mjs | 28 +- src/dialcache.ts | 166 +++--- src/errors.ts | 11 + src/index.ts | 2 + src/internal/cache-result.ts | 4 + src/internal/deadline.ts | 106 ++++ src/internal/redis-cache.ts | 46 +- src/node-redis.ts | 24 +- src/redis-client.ts | 22 +- src/valkey-glide.ts | 4 +- test/datadog.test.ts | 2 +- test/dialcache-coalescing.test.ts | 8 +- test/dialcache-config-ramp.test.ts | 24 +- test/dialcache-invalidation.test.ts | 30 +- test/dialcache-liveness.test.ts | 28 +- test/dialcache-logger.test.ts | 11 +- test/dialcache-metrics.test.ts | 18 +- .../dialcache-observability-internals.test.ts | 2 +- test/dialcache-redis-read-deadline.test.ts | 513 ++++++++++++++++++ test/dialcache-redis.test.ts | 41 +- test/node-redis.test.ts | 25 +- test/prometheus.test.ts | 4 +- test/redis-cluster.integration.test.ts | 6 +- test/redis-real.integration.test.ts | 8 +- test/valkey-glide.test.ts | 17 + 27 files changed, 1085 insertions(+), 217 deletions(-) create mode 100644 src/internal/deadline.ts create mode 100644 test/dialcache-redis-read-deadline.test.ts diff --git a/README.md b/README.md index 3edfbef..9112568 100644 --- a/README.md +++ b/README.md @@ -79,7 +79,8 @@ request-local cache -> process-local cache -> Redis cache -> fallback function - Process-local hits return immediately. - Process-local misses try Redis and populate the process-local cache on a Redis hit. - Redis misses call the fallback and attempt to populate Redis and, when active, the process-local cache. Tracked invalidation may suppress both publications. -- Redis cache read/write failures are logged, counted in metrics, and fail open; fallback results still return when fallback succeeds. `invalidateRemote` logs/counts Redis failures and rethrows them so callers do not assume invalidation succeeded. +- Redis read failures and core read timeouts are logged, counted in metrics, and fail open without attempting a Redis write after the fallback. An active process-local miss may retain the fallback value for an untracked key; tracked keys suppress that publication because the failed read did not establish invalidation-watermark safety. +- Other Redis cache failures also fail open; fallback results still return when fallback succeeds. `invalidateRemote` logs/counts Redis failures and rethrows them so callers do not assume invalidation succeeded. - Cache-key construction and config-provider failures also fail open and run the fallback uncached. - A missing effective process-local/Redis TTL disables that layer by policy; a configured TTL with no ramp defaults to 100%. Disabled layers record a disabled reason and fall through to the next layer/fallback. @@ -119,6 +120,7 @@ await dialcache.enable(async () => { | `defaultConfig` | no | `DialCacheKeyConfig` baseline policy that runtime config overlays field by field (see [Runtime config](#runtime-config-and-ramp-controls)). | | `serializer` | when the return type is not statically JSON-compatible | Per-function `Serializer` for Redis values (see [Serialization](#serialization)). | | `trackForInvalidation` | no (default `false`) | Opts this use case's Redis entries into watermark-based targeted invalidation. | +| `redisReadTimeoutMs` | no (inherits the Redis instance value) | Static per-use-case Redis read wait deadline in milliseconds, from 1 through 2,147,483,647 (see [Redis read deadlines](#redis-read-deadlines-and-async-liveness)). | | `fallbackTimeoutMs` | no (default `60_000`) | Fallback deadline in milliseconds, at most 2,147,483,647; `null` disables it (see [Fallback deadlines](#fallback-deadlines)). | `useCase` is validated at registration: a duplicate within one `DialCache` instance throws `UseCaseIsAlreadyRegisteredError`, and the internal name `watermark` throws `UseCaseNameIsReservedError`. @@ -147,7 +149,7 @@ await dialcache.enable(() => searchPosts("u1", 2, "active")); ```ts const dialcache = new DialCache({ namespace: "users-api", - redis: { client: redisClient }, + redis: { client: redisClient, readTimeoutMs: 200 }, }); ``` @@ -168,7 +170,7 @@ Instance-wide behavior is set through the `DialCache` constructor: | `DialCacheConfig` option | Default | Description | | --- | --- | --- | | `namespace` | `"urn"` | Logical cache namespace and first key component (see [Keys, ids, and extra dimensions](#keys-ids-and-extra-dimensions)). | -| `redis` | none | `{ client: DialCacheRedisClient }`; enables the Redis layer (see [Redis-backed TTL cache](#redis-backed-ttl-cache)). | +| `redis` | none | `{ client: DialCacheRedisClient, readTimeoutMs: number }`; enables Redis with a required application-selected read wait deadline (see [Redis-backed TTL cache](#redis-backed-ttl-cache)). | | `localMaxSize` | `10_000` | Global process-local entry cap; `0` disables process-local storage. Nonnegative safe integer. | | `cacheConfigProvider` | none | Resolves runtime config per enabled invocation as a sparse overlay on the function's `defaultConfig`; `null` applies no overrides. | | `rampSampler` | deterministic by key and layer | Percentage sampler for partial ramps; `randomRampSampler` is also exported. | @@ -297,7 +299,10 @@ await redisClient.connect(); const dialcache = new DialCache({ namespace: "users-api", - redis: { client: createNodeRedisDialCacheClient(redisClient) }, + redis: { + client: createNodeRedisDialCacheClient(redisClient), + readTimeoutMs: 200, + }, }); async function shutdown(): Promise { @@ -306,7 +311,7 @@ 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. +`redis.client` and `redis.readTimeoutMs` are required when Redis is configured. The timeout is an application-selected positive safe integer no greater than 2,147,483,647; DialCache has no universal duration or unbounded escape hatch. Create and connect the underlying semantic `DialCacheRedisClient` 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 and its module namespace to the GLIDE adapter: @@ -324,7 +329,7 @@ const glideClient = await valkeyGlide.GlideClient.createClient({ const redisClient = createValkeyGlideDialCacheClient(glideClient, valkeyGlide); const dialcache = new DialCache({ namespace: "users-api", - redis: { client: redisClient }, + redis: { client: redisClient, readTimeoutMs: 200 }, }); function shutdown(): void { @@ -339,21 +344,41 @@ Pass the same module namespace that created the client. DialCache uses its 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 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. A read that crossed the core wait deadline may still be active inside the client after the cached call settles, so also use the client's native budgets/telemetry to wait for or terminate that work before disposing adapter resources and closing the 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. +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 foreground and any late underlying 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, so a core-timed-out read may require waiting for GLIDE's native `requestTimeout` before retrying disposal. 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. Tracked reads are deliberately routed to primaries so a lagging replica cannot hide an invalidation watermark. -#### Async liveness contract +#### Migrating from v0.8 + +Redis-enabled construction now requires an explicit core read deadline: + +```ts +// Before +new DialCache({ redis: { client } }); + +// Now +new DialCache({ redis: { client, readTimeoutMs: 200 } }); +``` + +Choose this value from each application's latency budget. Add `redisReadTimeoutMs` to selected `cached()` definitions when one use case needs a tighter or looser bound. Existing custom clients with `read(request)` remain structurally compatible because the new `read(request, context?)` argument is optional, but they still need finite native resource budgets as described below. + +#### Redis read deadlines and async liveness + +DialCache starts one monotonic, referenced timer immediately before a shared leader invokes the semantic `DialCacheRedisClient.read()`. The boundary includes client queueing, reconnection, retry, response waiting, and adapter-level DialCache payload decoding. It excludes key/config/ramp work, local reads, user serializer loading, fallback execution, Redis writes, and invalidation. Same-key process and request-local followers allocate no additional read timer and inherit the leader's remaining budget. Local hits and disabled or ramped-out Redis layers allocate none. -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. +When the deadline expires, DialCache first aborts the optional `RedisReadContext.signal`, records one bounded `cache_read` failure, and starts the source fallback. Late read fulfillment or rejection is consumed and ignored: it cannot publish a cache value, suppress the fallback, create an unhandled rejection, or emit another DialCache log/metric. `RedisReadTimeoutError` is exported for typed logger inspection and exposes only `useCase` and `timeoutMs`, never the raw cache key. -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. +The read context is cooperative, and the core timer is a caller-visible wait/publication boundary rather than guaranteed cancellation. A queued or dispatched command may continue after DialCache starts the fallback. Keep finite client-native connection, retry, reconnect, offline-queue, dispatch, and response budgets so that underlying work eventually releases its resources. -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. +The bundled node-redis adapter passes the signal through per-command options, allowing node-redis to remove work that is still waiting to be sent where supported. In node-redis 4.7, `socket.connectTimeout`, `disableOfflineQueue`, and `commandsQueueMaxLength` bound connection or queue behavior but do not impose a strict response deadline on an already-dispatched command; aborting does not unsend it or prove the server stopped executing it. Do not put Redis writes or invalidations behind a bare `Promise.race`: rejecting the outer promise neither removes queued work nor proves that a dispatched mutation did not execute. -The same finite-settlement responsibility applies to async `cacheConfigProvider`, `rampSampler`, and custom `Serializer` methods. DialCache's [fallback deadline](#fallback-deadlines) below covers only the wrapped application function. Prefer resource-native budgets and cooperative cancellation for every injected async operation. Native budgets can bound client waiting and may prevent queued work; only a cooperating operation can guarantee that underlying work stops. +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. The current `invokeScript` API has no per-invocation signal, so the core may return first while GLIDE continues to its caller-owned request budget. That client timeout is still not server-side cancellation. + +After any Redis read exception or timeout, DialCache runs the fallback but skips the post-fallback Redis write. Untracked keys may populate an active process-local miss; tracked keys suppress process-local publication because the read did not establish watermark safety. A normal Redis miss and a serializer-load rejection remain refreshable misses and may write. A later invocation can start a fresh Redis read. + +`fallbackTimeoutMs` starts separately only when the source fallback begins. A call may therefore consume the Redis read budget and then the fallback budget; neither setting creates a whole-call or Redis-write deadline. Redis writes, invalidations, async `cacheConfigProvider`, `rampSampler`, and custom `Serializer` methods must still settle under application-owned budgets. Prefer resource-native limits and cooperative cancellation for every injected async operation. #### Serialization @@ -428,7 +453,10 @@ import { createNodeRedisDialCacheClient } from "dialcache/node-redis"; const dialcache = new DialCache({ namespace: "users-api", - redis: { client: createNodeRedisDialCacheClient(redisClient) }, + redis: { + client: createNodeRedisDialCacheClient(redisClient), + readTimeoutMs: 200, + }, }); // Chosen from this application's clock-skew bound and measured worst-case source/fallback timings. @@ -485,7 +513,7 @@ await dialcache.enable(async () => { }); ``` -With Redis configured, an instance-scoped leader that misses the process-local cache runs the Redis read and, on miss, the fallback/cache write; followers await that result. Process-local-only misses share the leader's fallback/cache write. This protects Redis and the source of truth from a thundering herd on hot keys. +With Redis configured, an instance-scoped leader that misses the process-local cache runs one bounded Redis read and, on a normal miss, the fallback/cache write; followers share its remaining read budget and await the same result. Process-local-only misses share the leader's fallback/cache write. This protects Redis and the source of truth from a thundering herd on hot keys. Coalescing only applies when at least one cache layer is active. Calls outside `enable()` are true pass-through. Calls where request-local, process-local, and Redis are all disabled are uncached and uncoalesced, but because they were initially enabled, the fallback deadline below still applies. @@ -521,7 +549,7 @@ try { } ``` -The timer starts only when the fallback begins. Same-key followers share the process or request-local leader's remaining budget and receive its `FallbackTimeoutError`; pass-through invocations where every layer is disabled have independent timers. Cache hits create no fallback timer. Calls that were initially outside an enabled context remain true pass-through and are not timed out, even when the wrapper configures `fallbackTimeoutMs`. +The timer starts only when the fallback begins, including after a Redis read deadline has already elapsed. Same-key followers share the process or request-local leader's remaining budget and receive its `FallbackTimeoutError`; pass-through invocations where every layer is disabled have independent timers. Cache hits create no fallback timer. Calls that were initially outside an enabled context remain true pass-through and are not timed out, even when the wrapper configures `fallbackTimeoutMs`. Deadline delivery requires the JavaScript event loop to make progress. It cannot preempt a synchronous fallback prefix or other event-loop blocking, so rejection can arrive later than the configured duration; when control returns, DialCache checks the monotonic deadline before accepting the result. The deadline timer remains referenced until the fallback settles or times out. Consequently, an abandoned enabled fallback can keep an otherwise idle short-lived process alive until that deadline; shutdown code should drain outstanding DialCache work rather than discarding its promises. @@ -541,7 +569,7 @@ state.process.activeFollowers; state.process.oldestLeaderAgeMs; // null when idle ``` -A leader is one exact cache key currently tracked by the instance-scoped coalescer. A follower is each later invocation that joined that pending leader; the initiating invocation is not counted as a follower. Followers remain counted until their leader settles because abandoning a JavaScript promise is not observable. Request-local flights are deliberately excluded because their lifecycle is bounded by the outer `enable()` scope. `oldestLeaderAgeMs` uses a monotonic clock and is computed when the snapshot is requested. +A leader is one exact cache key currently tracked by the instance-scoped coalescer. A follower is each later invocation that joined that pending leader; the initiating invocation is not counted as a follower. Followers remain counted until their leader settles because abandoning a JavaScript promise is not observable. Request-local flights are deliberately excluded because their lifecycle is bounded by the outer `enable()` scope. After a core Redis read deadline, the foreground flight clears when its fallback settles even if the underlying client operation continues; use client-native telemetry for that late work. `oldestLeaderAgeMs` uses a monotonic clock and is computed when the snapshot is requested. There is no library-wide flight cap or age-based replacement. A registry cap would bound only DialCache metadata while overflow or eviction could still create unbounded source work and unsafe duplicate publication. Finite operation deadlines provide eventual cleanup; application admission control and backpressure remain responsible for bounding simultaneous distinct-key work. Monitor leader count and oldest age to verify that those budgets hold in production. @@ -660,7 +688,7 @@ The `error` label reports where an operation failed rather than copying the thro | --- | --- | | `key_construction` | The cache-key selector or `DialCacheKey` construction failed | | `config_resolution` | Runtime or layer configuration, or ramp resolution, failed | -| `cache_read` | A local-cache or Redis read failed | +| `cache_read` | A local-cache or Redis read failed, including a core Redis read timeout | | `cache_write` | A local-cache or Redis write failed | | `serialization_load` | Deserializing a Redis payload failed | | `serialization_dump` | Serializing a value for Redis failed | @@ -668,7 +696,7 @@ The `error` label reports where an operation failed rather than copying the thro | `fallback` | The wrapped application function failed or exceeded its DialCache deadline | | `unknown` | Reserved for an otherwise unclassified future failure site | -These values are defined by the backend-neutral core and are identical for every metrics adapter. Raw thrown values, error names, messages, cache IDs, arguments, and Redis keys are never included in metric labels. Operational errors are still passed to the configured logger where the existing failure path logs them. `in_fallback` remains the explicit cache-plumbing-versus-application distinction. +These values are defined by the backend-neutral core and are identical for every metrics adapter. Raw thrown values, error names, messages, timeout values, cache IDs, arguments, and Redis keys are never included in metric labels. Operational errors are still passed to the configured logger where the existing failure path logs them. Redis read deadlines log a root-exported `RedisReadTimeoutError` with stable `useCase` and `timeoutMs` fields. `in_fallback` remains the explicit cache-plumbing-versus-application distinction. ### Custom adapters @@ -684,7 +712,7 @@ From a repository checkout, run the semantic microbenchmark after installing dep pnpm benchmark:request-local ``` -The command builds `dist` before reporting five scenarios: sequential request-local hits, sequential process-local hits, enabled bounded fallbacks, request-local coalescing fan-out, and process coalescing fan-out. The benchmark is a maintainer tool and is not included in the published package. It asserts fallback counts, coalescing state, and returned values but deliberately applies no timing threshold. Override its work sizes with `DIALCACHE_BENCH_ITERATIONS` and `DIALCACHE_BENCH_FANOUT`. +The command builds `dist` before reporting six scenarios: sequential request-local hits, sequential process-local hits, enabled bounded fallbacks, request-local coalescing fan-out, process coalescing fan-out, and Redis read-deadline coalescing fan-out. The Redis scenario asserts one semantic read, one timer allocation, and one timer clear for the shared leader regardless of follower count. The benchmark is a maintainer tool and is not included in the published package. It asserts semantic behavior but deliberately applies no timing threshold. Override its work sizes with `DIALCACHE_BENCH_ITERATIONS` and `DIALCACHE_BENCH_FANOUT`. ### Releasing diff --git a/scripts/benchmark-request-local.mjs b/scripts/benchmark-request-local.mjs index f89be83..d607dcd 100644 --- a/scripts/benchmark-request-local.mjs +++ b/scripts/benchmark-request-local.mjs @@ -13,15 +13,18 @@ const results = [ await benchmarkEnabledFallbacks(sequentialIterations), await benchmarkRequestLocalCoalescing(coalescingFanout), await benchmarkProcessCoalescing(coalescingFanout), + await benchmarkRedisReadDeadlineCoalescing(coalescingFanout), ]; console.table( - results.map(({ scenario, operations, elapsedMs, fallbackCalls }) => ({ + results.map(({ scenario, operations, elapsedMs, fallbackCalls, redisReadCalls = 0, deadlineTimers = 0 }) => ({ scenario, operations, "elapsed (ms)": elapsedMs.toFixed(2), "operations/sec": Math.round((operations / elapsedMs) * 1_000).toLocaleString("en-US"), "fallback calls": fallbackCalls, + "Redis reads": redisReadCalls, + "deadline timers": deadlineTimers, })), ); @@ -207,6 +210,85 @@ async function benchmarkProcessCoalescing(fanout) { return { scenario: "process coalescing", operations: fanout, elapsedMs, fallbackCalls }; } +async function benchmarkRedisReadDeadlineCoalescing(fanout) { + const gate = deferred(); + const started = deferred(); + let redisReadCalls = 0; + let deadlineTimers = 0; + let clearedDeadlineTimers = 0; + const originalSetTimeout = globalThis.setTimeout; + const originalClearTimeout = globalThis.clearTimeout; + const redisClient = { + async read() { + redisReadCalls += 1; + started.resolve(); + await gate.promise; + return JSON.stringify("shared"); + }, + async write() { + return true; + }, + async invalidate() {}, + }; + const dialcache = new DialCache({ + redis: { client: redisClient, readTimeoutMs: 60_000 }, + }); + let fallbackCalls = 0; + const getValue = dialcache.cached( + async (id) => { + fallbackCalls += 1; + return id; + }, + { + keyType: "benchmark_id", + useCase: "BenchmarkRedisReadDeadlineCoalescing", + cacheKey: (id) => id, + defaultConfig: new DialCacheKeyConfig({ + ttlSec: { [CacheLayer.REMOTE]: 60 }, + ramp: { [CacheLayer.REMOTE]: 100 }, + }), + }, + ); + + try { + globalThis.setTimeout = (...args) => { + deadlineTimers += 1; + return originalSetTimeout(...args); + }; + globalThis.clearTimeout = (timer) => { + clearedDeadlineTimers += 1; + originalClearTimeout(timer); + }; + + const start = performance.now(); + const valuesPromise = Promise.all( + Array.from({ length: fanout }, () => dialcache.enable(async () => await getValue("shared"))), + ); + await started.promise; + await nextTurn(); + gate.resolve(); + const values = await valuesPromise; + const elapsedMs = performance.now() - start; + + assert.deepEqual(new Set(values), new Set(["shared"])); + assert.equal(fallbackCalls, 0, "a Redis hit should not execute the fallback"); + assert.equal(redisReadCalls, 1, "Redis followers should share one semantic read"); + assert.equal(deadlineTimers, 1, "one Redis read leader should allocate one deadline timer"); + assert.equal(clearedDeadlineTimers, 1, "the Redis read deadline timer should be cleared exactly once"); + return { + scenario: "Redis read deadline coalescing", + operations: fanout, + elapsedMs, + fallbackCalls, + redisReadCalls, + deadlineTimers, + }; + } finally { + globalThis.setTimeout = originalSetTimeout; + globalThis.clearTimeout = originalClearTimeout; + } +} + function deferred() { let resolve; const promise = new Promise((resolvePromise) => { diff --git a/scripts/test-package.mjs b/scripts/test-package.mjs index fe56536..868a19c 100644 --- a/scripts/test-package.mjs +++ b/scripts/test-package.mjs @@ -17,6 +17,7 @@ const rootConsumer = `import { DialCacheRedisProtocolError, FallbackTimeoutError, JsonSerializer, + RedisReadTimeoutError, type CacheMetricLabels, type CacheConfigProvider, type CachedOptions, @@ -32,6 +33,7 @@ const rootConsumer = `import { type MetricErrorKind, type ProcessCoalescingState, type RedisConfig, + type RedisReadContext, type Serializer, } from "dialcache"; // @ts-expect-error The unused MissingKeyConfigError class was removed instead of deprecated. @@ -79,6 +81,7 @@ const missingObservationType: DatadogMetricsOptions = { client: dogStatsDClient const cache = new DialCache({ namespace: "consumer-cache", metrics }); const redisProtocolError = new DialCacheRedisProtocolError("Invalid DialCache Redis write reply"); const fallbackTimeoutError = new FallbackTimeoutError("Load", 1_000); +const redisReadTimeoutError = new RedisReadTimeoutError("Load", 100); const coalescingState: CoalescingState = cache.getCoalescingState(); const processCoalescingState: ProcessCoalescingState = coalescingState.process; const disabledOverlay: DialCacheKeyConfig = DialCacheKeyConfig.disabled(); @@ -87,6 +90,7 @@ const load = cache.cached(async (id: string) => id, { useCase: "Load", cacheKey: (id) => id, fallbackTimeoutMs: 1_000, + redisReadTimeoutMs: 100, defaultConfig: DialCacheKeyConfig.enabled(60), }); const loadWithoutFallbackDeadline = cache.cached(async (id: string) => id, { @@ -221,6 +225,7 @@ const metricErrorKinds: Readonly> = { const unboundedErrorKind: MetricErrorKind = "Tenant123Error"; const customRedisClient: DialCacheRedisClient = { + // The optional second read argument preserves one-argument custom clients. read: async () => Buffer.from([0, 255]), write: async ({ value }) => typeof value === "string" || Buffer.isBuffer(value), invalidate: async () => undefined, @@ -237,13 +242,20 @@ const configHasNoUrnPrefix: "urnPrefix" extends keyof DialCacheConfig ? false : const legacyNamespaceConfig: DialCacheConfig = { urnPrefix: "consumer-cache" }; const redisConfigHasNoKeyPrefix: "keyPrefix" extends keyof RedisConfig ? false : true = true; // @ts-expect-error keyPrefix was removed in favor of DialCacheConfig.namespace. -const legacyKeyPrefixConfig: RedisConfig = { client: customRedisClient, keyPrefix: "legacy:" }; +const legacyKeyPrefixConfig: RedisConfig = { client: customRedisClient, readTimeoutMs: 100, keyPrefix: "legacy:" }; const redisConfigRequiresClient: {} extends Pick ? false : true = true; +const redisConfigRequiresReadTimeout: {} extends Pick ? false : true = true; const redisConfigHasNoCreateClient: "createClient" extends keyof RedisConfig ? false : true = true; // @ts-expect-error Redis requires a caller-owned client. const missingRedisClientConfig: RedisConfig = {}; +// @ts-expect-error Redis requires an application-selected core read deadline. +const missingRedisReadTimeoutConfig: RedisConfig = { client: customRedisClient }; // @ts-expect-error createClient was removed; construct and pass RedisConfig.client instead. const legacyRedisFactoryConfig: RedisConfig = { createClient: () => customRedisClient }; +const redisReadContext: RedisReadContext = { + timeoutMs: 100, + signal: new AbortController().signal, +}; // @ts-expect-error RedisClientFactory was removed with RedisConfig.createClient. type LegacyRedisClientFactory = import("dialcache").RedisClientFactory; type DialCacheRoot = typeof import("dialcache"); @@ -270,6 +282,7 @@ void legacyKeyInit; void namespacedKey.namespace; void redisProtocolError.name; void fallbackTimeoutError.timeoutMs; +void redisReadTimeoutError.timeoutMs; void coalescingState.process; void processCoalescingState.activeLeaders; void requestLocalCoalescingScope; @@ -288,7 +301,7 @@ const globalSerializer: Serializer = { load: () => ({ source: "global" }), }; const cacheWithGlobalSerializer = new DialCache({ - redis: { client: customRedisClient, serializer: globalSerializer }, + redis: { client: customRedisClient, readTimeoutMs: 1_000, serializer: globalSerializer }, }); // @ts-expect-error A global serializer cannot establish per-function Date compatibility. cacheWithGlobalSerializer.cached(async (_id: string) => new Date(0), optionsFor("GlobalSerializerNeedsTypedOverride")); @@ -304,9 +317,12 @@ void legacyNamespaceConfig; void redisConfigHasNoKeyPrefix; void legacyKeyPrefixConfig; void redisConfigRequiresClient; +void redisConfigRequiresReadTimeout; void redisConfigHasNoCreateClient; void missingRedisClientConfig; +void missingRedisReadTimeoutConfig; void legacyRedisFactoryConfig; +void redisReadContext.signal; void rootHasNoPrometheusFactory; void rootHasNoDatadogFactory; void datadogMetrics; @@ -432,6 +448,10 @@ const fallbackTimeoutError = new root.FallbackTimeoutError("PackageRuntime", 100 if (!(fallbackTimeoutError instanceof root.DialCacheError) || fallbackTimeoutError.timeoutMs !== 1000) { throw new Error("The root ESM fallback-timeout error export is invalid"); } +const redisReadTimeoutError = new root.RedisReadTimeoutError("PackageRuntime", 100); +if (!(redisReadTimeoutError instanceof root.DialCacheError) || redisReadTimeoutError.timeoutMs !== 100) { + throw new Error("The root ESM Redis-read-timeout error export is invalid"); +} const coalescingState = new root.DialCache().getCoalescingState(); const idleCoalescingState = { process: { activeLeaders: 0, activeFollowers: 0, oldestLeaderAgeMs: null } }; if (JSON.stringify(coalescingState) !== JSON.stringify(idleCoalescingState)) { @@ -508,6 +528,10 @@ const fallbackTimeoutError = new root.FallbackTimeoutError("PackageRuntime", 100 if (!(fallbackTimeoutError instanceof root.DialCacheError) || fallbackTimeoutError.timeoutMs !== 1000) { throw new Error("The root CommonJS fallback-timeout error export is invalid"); } +const redisReadTimeoutError = new root.RedisReadTimeoutError("PackageRuntime", 100); +if (!(redisReadTimeoutError instanceof root.DialCacheError) || redisReadTimeoutError.timeoutMs !== 100) { + throw new Error("The root CommonJS Redis-read-timeout error export is invalid"); +} const coalescingState = new root.DialCache().getCoalescingState(); const idleCoalescingState = { process: { activeLeaders: 0, activeFollowers: 0, oldestLeaderAgeMs: null } }; if (JSON.stringify(coalescingState) !== JSON.stringify(idleCoalescingState)) { diff --git a/src/dialcache.ts b/src/dialcache.ts index 0f8b133..bc5cbf5 100644 --- a/src/dialcache.ts +++ b/src/dialcache.ts @@ -24,7 +24,8 @@ import { type MetricLayer, } from "./metrics.js"; import type { Serializer } from "./serializer.js"; -import type { CacheGetResult } from "./internal/cache-result.js"; +import type { CacheGetResult, RemoteCacheGetResult } from "./internal/cache-result.js"; +import { assertValidDeadlineMs, MAX_TIMER_DELAY_MS, withMonotonicDeadline } from "./internal/deadline.js"; import { LocalCache } from "./internal/local-cache.js"; import { RedisCache } from "./internal/redis-cache.js"; import { @@ -117,6 +118,12 @@ interface CachedOptionsBase { * published by DialCache, but does not cancel the underlying operation. */ readonly fallbackTimeoutMs?: number | null; + /** + * Static per-use-case override for the Redis read wait deadline, in + * milliseconds. Omission inherits the required Redis instance default. + * There is no unbounded escape hatch. + */ + readonly redisReadTimeoutMs?: number; /** * Select every input dimension that can affect the returned value. Concurrent * enabled calls with the same cache key may share one in-flight execution. @@ -162,7 +169,6 @@ interface ProcessFlight { const DEFAULT_LOCAL_MAX_SIZE = 10_000; const DEFAULT_FALLBACK_TIMEOUT_MS = 60_000; -const MAX_TIMER_DELAY_MS = 2_147_483_647; const defaultConfigProvider: CacheConfigProvider = () => null; const defaultLogger: Logger = console; @@ -249,6 +255,10 @@ export class DialCache { cached(fn: Fn, options: CachedOptions): CachedFn { const defaultConfig = snapshotDefaultConfig(options.defaultConfig); const fallbackTimeoutMs = resolveFallbackTimeoutMs(options.fallbackTimeoutMs); + const redisReadTimeoutMs = resolveRedisReadTimeoutMs( + options.redisReadTimeoutMs, + this.redisCache?.readTimeoutMs, + ); this.registerUseCase(options.useCase); const run = async (...args: Parameters): Promise> => { @@ -304,11 +314,23 @@ export class DialCache { if (keyConfig?.requestLocal === true) { const requestLocalCache = getOrCreateRequestLocalCache(this.context); if (requestLocalCache !== null) { - return await this.getThroughRequestLocal(requestLocalCache, key, keyConfig, fallback); + return await this.getThroughRequestLocal( + requestLocalCache, + key, + keyConfig, + fallback, + redisReadTimeoutMs, + ); } } - return await this.getThroughSharedLayers(key, keyConfig, fallback, CacheLayer.LOCAL); + return await this.getThroughSharedLayers( + key, + keyConfig, + fallback, + CacheLayer.LOCAL, + redisReadTimeoutMs, + ); }; return run; @@ -373,6 +395,7 @@ export class DialCache { key: DialCacheKey, keyConfig: DialCacheKeyConfig, fallback: () => Promise, + redisReadTimeoutMs: number | undefined, ): Promise { return await this.singleFlightRequestLocal(requestLocalCache.inFlight, key, async () => { const start = performance.now(); @@ -384,7 +407,13 @@ export class DialCache { } this.metrics?.miss(labelsFor(key, REQUEST_LOCAL_CACHE_LAYER)); - const value = await this.getThroughSharedLayers(key, keyConfig, fallback, REQUEST_LOCAL_CACHE_LAYER); + const value = await this.getThroughSharedLayers( + key, + keyConfig, + fallback, + REQUEST_LOCAL_CACHE_LAYER, + redisReadTimeoutMs, + ); requestLocalCache.set(key.urn, value); return value; }); @@ -395,11 +424,18 @@ export class DialCache { keyConfig: DialCacheKeyConfig | null, fallback: () => Promise, fallbackMetricLayer: MetricLayer, + redisReadTimeoutMs: number | undefined, ): Promise { const localLayer = await this.resolveLocalLayerConfig(key, keyConfig); if (localLayer.status === "enabled") { return await this.singleFlightProcess(key, async () => - await this.getThroughActiveLocal(key, keyConfig, localLayer.config, fallback), + await this.getThroughActiveLocal( + key, + keyConfig, + localLayer.config, + fallback, + redisReadTimeoutMs, + ), ); } @@ -415,7 +451,12 @@ export class DialCache { } return await this.singleFlightProcess(key, async () => { - const remote = await this.readRemoteWithResolvedConfig(redisCache, key, remoteLayer.config); + const remote = await this.readRemoteWithResolvedConfig( + redisCache, + key, + remoteLayer.config, + redisReadTimeoutMs, + ); return await this.finishRedisChain(redisCache, key, localLayer, remote, fallback, remoteLayer.config); }); } @@ -425,6 +466,7 @@ export class DialCache { keyConfig: DialCacheKeyConfig | null, localConfig: ResolvedLayerConfig, fallback: () => Promise, + redisReadTimeoutMs: number | undefined, ): Promise { const local = this.readLocalWithResolvedConfig(key, localConfig); if (local.status === "hit") { @@ -441,7 +483,12 @@ export class DialCache { return await this.finishRedisChain(redisCache, key, local, remoteLayer, fallback); } - const remote = await this.readRemoteWithResolvedConfig(redisCache, key, remoteLayer.config); + const remote = await this.readRemoteWithResolvedConfig( + redisCache, + key, + remoteLayer.config, + redisReadTimeoutMs, + ); return await this.finishRedisChain(redisCache, key, local, remote, fallback, remoteLayer.config); } @@ -457,7 +504,7 @@ export class DialCache { redisCache: RedisCache, key: DialCacheKey, local: CacheGetResult, - remote: CacheGetResult, + remote: RemoteCacheGetResult, fallback: () => Promise, resolvedRemoteConfig?: ResolvedLayerConfig, ): Promise { @@ -468,6 +515,14 @@ export class DialCache { return remote.value; } + if (remote.status === "error") { + const value = await this.callFallback(labelsFor(key, CacheLayer.REMOTE), fallback); + if (!key.trackForInvalidation && local.status === "miss") { + await this.putLocalFailOpen(key, value, local.config); + } + return value; + } + const remoteErrored = remote.status === "disabled" && remote.reason === "config_error"; const remoteWriteConfig = remote.status === "miss" ? remote.config : remoteErrored ? resolvedRemoteConfig : undefined; const fallbackLayer = remote.status === "miss" || remoteErrored ? CacheLayer.REMOTE : CacheLayer.LOCAL; @@ -547,10 +602,19 @@ export class DialCache { } } - private async readRemoteWithResolvedConfig(redisCache: RedisCache, key: DialCacheKey, layerConfig: ResolvedLayerConfig) { + private async readRemoteWithResolvedConfig( + redisCache: RedisCache, + key: DialCacheKey, + layerConfig: ResolvedLayerConfig, + redisReadTimeoutMs: number | undefined, + ): Promise> { const start = performance.now(); try { - const result = await redisCache.getWithResolvedConfig(key, layerConfig); + const result = await redisCache.getWithResolvedConfig( + key, + layerConfig, + redisReadTimeoutMs, + ); this.metrics?.request(labelsFor(key, CacheLayer.REMOTE)); this.metrics?.observeGet(labelsFor(key, CacheLayer.REMOTE), elapsedSeconds(start)); if (result.status === "miss") { @@ -559,11 +623,7 @@ export class DialCache { return result; } catch (error) { this.logger.warn("Error getting value from Redis cache", error); - return { - status: "disabled", - reason: "config_error", - ...(key.trackForInvalidation ? { skipCacheWrite: true } : {}), - } as const; + return { status: "error", operation: "read" }; } } @@ -783,6 +843,14 @@ function resolveFallbackTimeoutMs(value: number | null | undefined): number | nu return timeoutMs; } +function resolveRedisReadTimeoutMs(value: unknown, instanceDefault: number | undefined): number | undefined { + if (value === undefined) { + return instanceDefault; + } + assertValidDeadlineMs(value, "DialCache redisReadTimeoutMs"); + return value; +} + function withFallbackTimeout( fallback: () => Promise, useCase: string, @@ -791,68 +859,10 @@ function withFallbackTimeout( if (timeoutMs === null) { return fallback(); } - - const startedAtMs = performance.now(); - const operation = fallback(); - return new Promise((resolve, reject) => { - let settled = false; - let timer: NodeJS.Timeout | null = null; - const elapsedMs = (): number => Math.max(performance.now() - startedAtMs, 0); - const clearTimer = (): void => { - if (timer !== null) { - clearTimeout(timer); - timer = null; - } - }; - const rejectTimeout = (): void => { - if (settled) { - return; - } - settled = true; - clearTimer(); - reject(new FallbackTimeoutError(useCase, timeoutMs)); - }; - const onTimer = (): void => { - timer = null; - if (settled) { - return; - } - - const remainingMs = timeoutMs - elapsedMs(); - if (remainingMs > 0) { - timer = setTimeout(onTimer, Math.ceil(remainingMs)); - return; - } - rejectTimeout(); - }; - const settleBeforeDeadline = (settle: () => void): void => { - if (settled) { - return; - } - if (elapsedMs() >= timeoutMs) { - rejectTimeout(); - return; - } - settled = true; - clearTimer(); - settle(); - }; - - const remainingMs = timeoutMs - elapsedMs(); - if (remainingMs <= 0) { - rejectTimeout(); - } else { - timer = setTimeout(onTimer, Math.ceil(remainingMs)); - } - - void operation.then( - (value) => { - settleBeforeDeadline(() => resolve(value)); - }, - (error: unknown) => { - settleBeforeDeadline(() => reject(error)); - }, - ); + return withMonotonicDeadline({ + operation: fallback, + timeoutMs, + timeoutError: () => new FallbackTimeoutError(useCase, timeoutMs), }); } diff --git a/src/errors.ts b/src/errors.ts index 6153c2a..2cf896c 100644 --- a/src/errors.ts +++ b/src/errors.ts @@ -16,6 +16,17 @@ export class FallbackTimeoutError extends DialCacheError { } } +export class RedisReadTimeoutError extends DialCacheError { + readonly timeoutMs: number; + readonly useCase: string; + + constructor(useCase: string, timeoutMs: number) { + super(`DialCache Redis read for use case "${useCase}" timed out after ${timeoutMs} ms`); + this.useCase = useCase; + this.timeoutMs = timeoutMs; + } +} + export class UseCaseIsAlreadyRegisteredError extends DialCacheError { constructor(useCase: string) { super(`Use case already registered: ${useCase}`); diff --git a/src/index.ts b/src/index.ts index 38d8726..daa91cd 100644 --- a/src/index.ts +++ b/src/index.ts @@ -17,6 +17,7 @@ export type { export { DialCacheError, FallbackTimeoutError, + RedisReadTimeoutError, UseCaseIsAlreadyRegisteredError, UseCaseNameIsReservedError, } from "./errors.js"; @@ -41,6 +42,7 @@ export type { DialCacheRedisClient, RedisCachePayload, RedisInvalidationRequest, + RedisReadContext, RedisReadRequest, RedisWriteRequest, } from "./redis-client.js"; diff --git a/src/internal/cache-result.ts b/src/internal/cache-result.ts index 77d92fd..5771997 100644 --- a/src/internal/cache-result.ts +++ b/src/internal/cache-result.ts @@ -9,3 +9,7 @@ export type CacheGetResult = readonly reason: DisabledReason; readonly skipCacheWrite?: boolean; }; + +export type RemoteCacheGetResult = + | CacheGetResult + | { readonly status: "error"; readonly operation: "read" }; diff --git a/src/internal/deadline.ts b/src/internal/deadline.ts new file mode 100644 index 0000000..c87fbda --- /dev/null +++ b/src/internal/deadline.ts @@ -0,0 +1,106 @@ +import { performance } from "node:perf_hooks"; + +import type { Awaitable } from "../config.js"; + +export const MAX_TIMER_DELAY_MS = 2_147_483_647; + +interface MonotonicDeadlineOptions { + readonly operation: () => Awaitable; + readonly timeoutMs: number; + readonly timeoutError: () => Error; + readonly onTimeout?: () => void; +} + +/** + * Bound how long a caller waits for an operation without claiming ownership of + * the operation's underlying resources. Late fulfillment and rejection are + * consumed after the returned promise settles. + */ +export function withMonotonicDeadline({ + operation, + timeoutMs, + timeoutError, + onTimeout, +}: MonotonicDeadlineOptions): Promise { + const startedAtMs = performance.now(); + const pending = Promise.resolve(operation()); + + return new Promise((resolve, reject) => { + let settled = false; + let timer: NodeJS.Timeout | null = null; + const elapsedMs = (): number => Math.max(performance.now() - startedAtMs, 0); + const clearTimer = (): void => { + if (timer !== null) { + clearTimeout(timer); + timer = null; + } + }; + const rejectTimeout = (): void => { + if (settled) { + return; + } + settled = true; + clearTimer(); + try { + onTimeout?.(); + } catch { + // Cooperative cancellation must not replace the stable deadline error. + } + reject(timeoutError()); + }; + const onTimer = (): void => { + timer = null; + if (settled) { + return; + } + + const remainingMs = timeoutMs - elapsedMs(); + if (remainingMs > 0) { + timer = setTimeout(onTimer, Math.ceil(remainingMs)); + return; + } + rejectTimeout(); + }; + const settleBeforeDeadline = (settle: () => void): void => { + if (settled) { + return; + } + if (elapsedMs() >= timeoutMs) { + rejectTimeout(); + return; + } + settled = true; + clearTimer(); + settle(); + }; + + const remainingMs = timeoutMs - elapsedMs(); + if (remainingMs <= 0) { + rejectTimeout(); + } else { + // Keep the timer referenced so it can deliver the deadline as the only + // remaining active handle. + timer = setTimeout(onTimer, Math.ceil(remainingMs)); + } + + void pending.then( + (value) => { + settleBeforeDeadline(() => resolve(value)); + }, + (error: unknown) => { + settleBeforeDeadline(() => reject(error)); + }, + ); + }); +} + +export function assertValidDeadlineMs(value: unknown, name: string): asserts value is number { + if ( + typeof value !== "number" + || !Number.isSafeInteger(value) + || value <= 0 + || value > MAX_TIMER_DELAY_MS + ) { + throw new RangeError(`${name} must be a positive safe integer no greater than ${MAX_TIMER_DELAY_MS}`); + } +} diff --git a/src/internal/redis-cache.ts b/src/internal/redis-cache.ts index bb1bf72..f2982e2 100644 --- a/src/internal/redis-cache.ts +++ b/src/internal/redis-cache.ts @@ -1,21 +1,28 @@ import { performance } from "node:perf_hooks"; import { CacheLayer, DEFAULT_WATERMARK_TTL_SEC, type CacheConfigProvider, type CacheRampSampler, type DialCacheKeyConfig } from "../config.js"; +import { RedisReadTimeoutError } from "../errors.js"; import { invalidationPrefix, redisClusterHashTag, type DialCacheKey } from "../key.js"; import { labelsFor, type DialCacheMetricsAdapter, type MetricErrorKind } from "../metrics.js"; import type { DialCacheRedisClient, RedisCachePayload } from "../redis-client.js"; import { JsonSerializer, type Serializer } from "../serializer.js"; import type { CacheGetResult } from "./cache-result.js"; +import { assertValidDeadlineMs, withMonotonicDeadline } from "./deadline.js"; import { fetchKeyConfig, resolveLayerConfigResult, type ResolvedLayerConfig } from "./runtime-config.js"; export interface RedisConfig { /** * Caller-created, connected, and lifecycle-owned semantic Redis client. - * DialCache borrows it and never adds command deadlines or drains, disposes, - * or closes it. Every client operation must settle within a finite - * application-defined budget. + * DialCache borrows it and never drains, disposes, or closes it. The client + * still needs finite native budgets to bound work that outlives DialCache's + * read wait deadline. */ readonly client: DialCacheRedisClient; + /** + * Required application-selected default for how long DialCache waits for a + * semantic Redis read. Individual cached use cases may override it. + */ + readonly readTimeoutMs: number; readonly serializer?: Serializer; readonly watermarkTtlSec?: number; } @@ -37,6 +44,7 @@ export class RedisCache { private readonly watermarkTtlMs: number; private readonly client: DialCacheRedisClient; private readonly metrics: DialCacheMetricsAdapter | null; + readonly readTimeoutMs: number; constructor(options: RedisCacheOptions) { if (Object.hasOwn(options.redis, "keyPrefix")) { @@ -48,6 +56,14 @@ export class RedisCache { ); } + if (options.redis.client === undefined) { + throw new TypeError("Redis config requires client"); + } + if (options.redis.readTimeoutMs === undefined) { + throw new TypeError("Redis config requires readTimeoutMs"); + } + assertValidDeadlineMs(options.redis.readTimeoutMs, "Redis readTimeoutMs"); + this.configProvider = options.configProvider; this.rampSampler = options.rampSampler; this.defaultSerializer = options.redis.serializer ?? defaultSerializer; @@ -60,12 +76,8 @@ export class RedisCache { throw new RangeError("Redis watermarkTtlSec is too large"); } this.metrics = options.metrics; - - if (options.redis.client === undefined) { - throw new TypeError("Redis config requires client"); - } - this.client = options.redis.client; + this.readTimeoutMs = options.redis.readTimeoutMs; } async get(key: DialCacheKey): Promise { @@ -82,13 +94,27 @@ export class RedisCache { return await this.getWithResolvedConfig(key, layerConfig.config); } - async getWithResolvedConfig(key: DialCacheKey, layerConfig: ResolvedLayerConfig): Promise> { + async getWithResolvedConfig( + key: DialCacheKey, + layerConfig: ResolvedLayerConfig, + readTimeoutMs = this.readTimeoutMs, + ): Promise> { let payload: RedisCachePayload | null; try { const redisKey = this.redisKey(key); - payload = await this.client.read({ + const request = { valueKey: redisKey, ...(key.trackForInvalidation ? { watermarkKey: this.redisWatermarkKeyFromKey(key) } : {}), + }; + const controller = new AbortController(); + payload = await withMonotonicDeadline({ + operation: () => this.client.read(request, { + timeoutMs: readTimeoutMs, + signal: controller.signal, + }), + timeoutMs: readTimeoutMs, + timeoutError: () => new RedisReadTimeoutError(key.useCase, readTimeoutMs), + onTimeout: () => controller.abort(), }); } catch (error) { this.recordError(key, "cache_read"); diff --git a/src/node-redis.ts b/src/node-redis.ts index b84d95f..a20e633 100644 --- a/src/node-redis.ts +++ b/src/node-redis.ts @@ -14,7 +14,12 @@ import { } from "./internal/redis-script-reply.js"; import type { DialCacheRedisClient } from "./redis-client.js"; -type BufferReplyOptions = ReturnType>; +type BufferReplyOptions = ReturnType< + typeof commandOptions<{ + readonly returnBuffers: true; + readonly signal?: AbortSignal; + }> +>; // Redis bulk strings are binary data; decoding them as UTF-8 would corrupt arbitrary serializer output. const bufferReplyOptions: BufferReplyOptions = commandOptions({ returnBuffers: true }); const readReply = (reply: string | null): string | null => reply; @@ -151,17 +156,20 @@ interface NodeRedisScriptClient { /** * Create a resource-free semantic view over a caller-owned node-redis client. - * The adapter preserves the wrapped script commands' lifetimes; node-redis - * `connectTimeout` alone is not a command deadline. The caller remains - * responsible for finite command settlement, draining work, and closing the - * client. + * Read signals are passed to node-redis so queued commands can be removed when + * supported. Aborting after dispatch does not unsend a command or prove the + * server stopped executing it. The caller remains responsible for finite + * native command budgets, draining work, and closing the client. */ export function createNodeRedisDialCacheClient(client: NodeRedisScriptClient): DialCacheRedisClient { return { - async read({ valueKey, watermarkKey }) { + async read({ valueKey, watermarkKey }, context) { + const options: BufferReplyOptions = context === undefined + ? bufferReplyOptions + : commandOptions({ returnBuffers: true, signal: context.signal }); const raw = watermarkKey === undefined - ? await client.dialcacheRead(bufferReplyOptions, valueKey) - : await client.dialcacheReadTracked(bufferReplyOptions, valueKey, watermarkKey); + ? await client.dialcacheRead(options, valueKey) + : await client.dialcacheReadTracked(options, valueKey, watermarkKey); return raw === null ? null : decodeRedisPayload(raw); }, async write(request) { diff --git a/src/redis-client.ts b/src/redis-client.ts index aa565ed..ac81377 100644 --- a/src/redis-client.ts +++ b/src/redis-client.ts @@ -51,6 +51,15 @@ interface UntrackedRedisValueRequest extends RedisValueRequest { export type RedisReadRequest = TrackedRedisValueRequest | UntrackedRedisValueRequest; +/** + * Per-use-case read policy supplied by DialCache. Adapters may use the signal + * for cooperative cancellation, but the core deadline remains authoritative. + */ +export interface RedisReadContext { + readonly timeoutMs: number; + readonly signal: AbortSignal; +} + interface RedisWriteBase extends RedisValueRequest { readonly cacheTtlMs: number; readonly value: RedisCachePayload; @@ -76,15 +85,16 @@ export interface RedisInvalidationRequest { * Caller-owned semantic Redis boundary. DialCache borrows this client and does * not create, connect, drain, dispose, or close it. * - * Every method must settle within a finite application-defined deadline that - * includes connection, retry, offline-queue, dispatch, and response time. - * DialCache does not add Redis deadlines or server-side cancellation. A - * command that times out after dispatch may still have executed, so adapters - * must document their queue-removal and ambiguous-write semantics. + * Clients must use finite application-defined connection, retry, reconnect, + * offline-queue, dispatch, and response budgets to bound underlying resource + * lifetime. DialCache additionally bounds how long it waits for reads, but + * does not claim server-side cancellation. A command that times out after + * dispatch may still have executed, so adapters must document their + * queue-removal and ambiguous-write semantics. */ export interface DialCacheRedisClient { /** Atomically read and validate a value against its watermark when tracked. */ - read(request: RedisReadRequest): Awaitable; + read(request: RedisReadRequest, context?: RedisReadContext): Awaitable; /** Atomically write using server time. False means invalidation blocked the write. */ write(request: RedisWriteRequest): Awaitable; /** Advance the watermark monotonically after the source mutation commits. */ diff --git a/src/valkey-glide.ts b/src/valkey-glide.ts index 10c910f..3daf9f6 100644 --- a/src/valkey-glide.ts +++ b/src/valkey-glide.ts @@ -58,7 +58,9 @@ export interface ValkeyGlideDialCacheClient extends DialCacheRedisClient { * 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. + * client waiting but is not server-side command cancellation. GLIDE's current + * script API has no per-invocation signal, so DialCache's core read deadline + * may return before this adapter's invocation settles. */ export function createValkeyGlideDialCacheClient( client: ValkeyGlideScriptingClient, diff --git a/test/datadog.test.ts b/test/datadog.test.ts index 9a6368a..a4c76fa 100644 --- a/test/datadog.test.ts +++ b/test/datadog.test.ts @@ -321,7 +321,7 @@ describe("Datadog metrics adapter", () => { const dialcache = new DialCache({ namespace: "private-cache", metrics, - redis: { client: redis }, + redis: { client: redis, readTimeoutMs: 1_000 }, logger: { debug: vi.fn(), warn: vi.fn(), error: vi.fn() }, }); const load = dialcache.cached(async (id: string, filter: string) => ({ id, filter }), { diff --git a/test/dialcache-coalescing.test.ts b/test/dialcache-coalescing.test.ts index 0f4d90f..04d0278 100644 --- a/test/dialcache-coalescing.test.ts +++ b/test/dialcache-coalescing.test.ts @@ -259,7 +259,7 @@ describe("DialCache request coalescing", () => { const fallbackGate = deferred(); const redis = new FakeRedis(); redis.getGate = redisGate.promise; - const dialcache = new DialCache({ redis: { client: redis } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); let calls = 0; const getUser = dialcache.cached( async (id: string) => { @@ -299,7 +299,7 @@ describe("DialCache request coalescing", () => { it("coalesces concurrent Redis hits when the remote layer is active", async () => { const redisGate = deferred(); const redis = new FakeRedis(); - const dialcache = new DialCache({ redis: { client: redis } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); let calls = 0; const getUser = dialcache.cached( async (id: string) => { @@ -451,7 +451,7 @@ describe("DialCache request coalescing", () => { const gate = deferred(); const redis = new FakeRedis(); const { metrics, coalesced } = spyMetrics(); - const dialcache = new DialCache({ redis: { client: redis }, metrics }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 }, metrics }); let calls = 0; const getUser = dialcache.cached( async (id: string) => { @@ -490,7 +490,7 @@ describe("DialCache request coalescing", () => { const gate = deferred(); const redis = new FailingReadRedis(); const logger = { debug: vi.fn(), error: vi.fn(), warn: vi.fn() }; - const dialcache = new DialCache({ redis: { client: redis }, logger }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 }, logger }); let calls = 0; const getUser = dialcache.cached( async (id: string) => { diff --git a/test/dialcache-config-ramp.test.ts b/test/dialcache-config-ramp.test.ts index e24d992..5549f06 100644 --- a/test/dialcache-config-ramp.test.ts +++ b/test/dialcache-config-ramp.test.ts @@ -85,7 +85,7 @@ describe("DialCache runtime config and ramp controls", () => { ramp: { [CacheLayer.LOCAL]: 100, [CacheLayer.REMOTE]: 100 }, }); const cacheConfigProvider = vi.fn(async () => keyConfig); - const dialcache = new DialCache({ redis: { client: redis }, cacheConfigProvider }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 }, cacheConfigProvider }); const getUser = dialcache.cached(async (userId: string) => ({ userId }), { keyType: "user_id", useCase: "SingleRuntimeConfigSnapshot", @@ -108,7 +108,7 @@ describe("DialCache runtime config and ramp controls", () => { }); const cacheConfigProvider = vi.fn(async () => keyConfig); const dialcache = new DialCache({ - redis: { client: redis }, + redis: { client: redis, readTimeoutMs: 1_000 }, cacheConfigProvider, rampSampler: () => { throw new Error("ramp source unavailable"); @@ -180,7 +180,7 @@ describe("DialCache runtime config and ramp controls", () => { throw samplerError; }); const dialcache = new DialCache({ - redis: { client: redis }, + redis: { client: redis, readTimeoutMs: 1_000 }, rampSampler, logger, }); @@ -219,7 +219,7 @@ describe("DialCache runtime config and ramp controls", () => { throw samplerError; }); const dialcache = new DialCache({ - redis: { client: redis }, + redis: { client: redis, readTimeoutMs: 1_000 }, rampSampler, logger, }); @@ -295,7 +295,7 @@ describe("DialCache runtime config and ramp controls", () => { // while runtime config changes only the local ramp and remote TTL. const redis = new FakeRedis(); const dialcache = new DialCache({ - redis: { client: redis }, + redis: { client: redis, readTimeoutMs: 1_000 }, cacheConfigProvider: async () => new DialCacheKeyConfig({ ttlSec: { [CacheLayer.REMOTE]: 30 }, ramp: { [CacheLayer.LOCAL]: 0 }, @@ -490,7 +490,7 @@ describe("DialCache runtime config and ramp controls", () => { ])("uses %s to disable every inherited layer", async (_name, overlayFor) => { const redis = new FakeRedis(); const dialcache = new DialCache({ - redis: { client: redis }, + redis: { client: redis, readTimeoutMs: 1_000 }, cacheConfigProvider: async () => overlayFor(), }); let calls = 0; @@ -643,12 +643,12 @@ describe("DialCache runtime config and ramp controls", () => { const passingSampler = vi.fn(() => 49); const blockedSampler = vi.fn(() => 50); const passingCache = new DialCache({ - redis: { client: passingRedis }, + redis: { client: passingRedis, readTimeoutMs: 1_000 }, rampSampler: passingSampler, cacheConfigProvider: async () => configFor({ [CacheLayer.REMOTE]: 60 }, { [CacheLayer.REMOTE]: 50 }), }); const blockedCache = new DialCache({ - redis: { client: blockedRedis }, + redis: { client: blockedRedis, readTimeoutMs: 1_000 }, rampSampler: blockedSampler, cacheConfigProvider: async () => configFor({ [CacheLayer.REMOTE]: 60 }, { [CacheLayer.REMOTE]: 50 }), }); @@ -796,7 +796,7 @@ describe("DialCache runtime config and ramp controls", () => { const redis = new FakeRedis(); const rampSampler = vi.fn().mockReturnValueOnce(49).mockReturnValueOnce(50); const dialcache = new DialCache({ - redis: { client: redis }, + redis: { client: redis, readTimeoutMs: 1_000 }, rampSampler, cacheConfigProvider: async () => configFor({ [CacheLayer.REMOTE]: 60 }, { [CacheLayer.REMOTE]: 50 }), }); @@ -833,7 +833,7 @@ describe("DialCache runtime config and ramp controls", () => { // When each cached function is called twice. for (const ttl of badTtls) { const dialcache = new DialCache({ - redis: { client: redis }, + redis: { client: redis, readTimeoutMs: 1_000 }, cacheConfigProvider: async () => configFor( { [CacheLayer.LOCAL]: ttl, [CacheLayer.REMOTE]: ttl }, { [CacheLayer.LOCAL]: 100, [CacheLayer.REMOTE]: 100 }, @@ -862,7 +862,7 @@ describe("DialCache runtime config and ramp controls", () => { // Given runtime config only enables the remote layer. const redis = new FakeRedis(); const dialcache = new DialCache({ - redis: { client: redis }, + redis: { client: redis, readTimeoutMs: 1_000 }, cacheConfigProvider: async () => new DialCacheKeyConfig({ ttlSec: { [CacheLayer.REMOTE]: 60 } }), }); let calls = 0; @@ -888,7 +888,7 @@ describe("DialCache runtime config and ramp controls", () => { // Given Redis exists but runtime config only enables the local layer. const redis = new FakeRedis(); const dialcache = new DialCache({ - redis: { client: redis }, + redis: { client: redis, readTimeoutMs: 1_000 }, cacheConfigProvider: async () => configFor({ [CacheLayer.LOCAL]: 60 }, { [CacheLayer.LOCAL]: 100 }), }); let calls = 0; diff --git a/test/dialcache-invalidation.test.ts b/test/dialcache-invalidation.test.ts index 0b8040c..1f7a6af 100644 --- a/test/dialcache-invalidation.test.ts +++ b/test/dialcache-invalidation.test.ts @@ -79,7 +79,7 @@ describe("DialCache targeted invalidation watermarks", () => { it("invalidates all tracked use cases sharing a key type and id", async () => { const redis = new FakeRedis(); - const dialcache = new DialCache({ redis: { client: redis } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); let profileVersion = 1; let permissionsVersion = 1; const getProfile = dialcache.cached(async (userId: string) => ({ userId, profileVersion }), { @@ -117,7 +117,7 @@ describe("DialCache targeted invalidation watermarks", () => { it("does not write remote or local cache during a future invalidation window", async () => { const redis = new FakeRedis(); - const dialcache = new DialCache({ redis: { client: redis } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); let calls = 0; const getUser = dialcache.cached(async (userId: string) => ({ userId, calls: ++calls }), { keyType: "user_id", @@ -139,7 +139,7 @@ describe("DialCache targeted invalidation watermarks", () => { it("rejects a write when invalidation arrives during fallback", async () => { const redis = new FakeRedis(); - const dialcache = new DialCache({ redis: { client: redis } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); let calls = 0; const getUser = dialcache.cached( async (userId: string) => { @@ -185,7 +185,7 @@ describe("DialCache targeted invalidation watermarks", () => { return JSON.parse(payload) as { userId: string; calls: number }; }, }; - const dialcache = new DialCache({ redis: { client: redis } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); let calls = 0; const getUser = dialcache.cached(async (userId: string) => ({ userId, calls: ++calls }), { keyType: "user_id", @@ -211,7 +211,7 @@ describe("DialCache targeted invalidation watermarks", () => { it("blocks a same-millisecond write for a zero-length future buffer", async () => { const redis = new FakeRedis(); - const dialcache = new DialCache({ redis: { client: redis } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); let calls = 0; const getUser = dialcache.cached(async (userId: string) => ({ userId, calls: ++calls }), { keyType: "user_id", @@ -235,7 +235,7 @@ describe("DialCache targeted invalidation watermarks", () => { it("resumes tracked writes after the future buffer", async () => { const redis = new FakeRedis(); - const dialcache = new DialCache({ redis: { client: redis } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); let calls = 0; const getUser = dialcache.cached(async (userId: string) => ({ userId, calls: ++calls }), { keyType: "user_id", @@ -258,7 +258,7 @@ describe("DialCache targeted invalidation watermarks", () => { it("treats a tracked value with a missing watermark marker as a miss", async () => { const redis = new FakeRedis(); redis.setRaw(valueKey("MissingWatermark"), encodeFrame({ source: "stale" })); - const dialcache = new DialCache({ redis: { client: redis } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); let calls = 0; const getUser = dialcache.cached(async (userId: string) => ({ userId, source: `fallback-${++calls}` }), { keyType: "user_id", @@ -278,7 +278,7 @@ describe("DialCache targeted invalidation watermarks", () => { it("preserves the furthest watermark across repeated invalidations", async () => { const redis = new FakeRedis(); - const dialcache = new DialCache({ redis: { client: redis } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); await dialcache.invalidateRemote("user_id", "123", 5_000); const first = redis.readWatermarkValue(watermarkKey); @@ -290,7 +290,7 @@ describe("DialCache targeted invalidation watermarks", () => { it("extends watermark lifetime on writes but not reads", async () => { const redis = new FakeRedis(); - const dialcache = new DialCache({ redis: { client: redis, watermarkTtlSec: 60 } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000, watermarkTtlSec: 60 } }); const getUser = dialcache.cached(async (userId: string) => ({ userId }), { keyType: "user_id", useCase: "WatermarkLifetime", @@ -314,7 +314,7 @@ describe("DialCache targeted invalidation watermarks", () => { redis.failWatermarkGet = true; const logger = { debug: vi.fn(), warn: vi.fn(), error: vi.fn() }; const metrics = new RecordingMetrics(); - const dialcache = new DialCache({ redis: { client: redis }, logger, metrics }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 }, logger, metrics }); let calls = 0; const getUser = dialcache.cached(async (userId: string) => ({ userId, calls: ++calls }), { keyType: "user_id", @@ -348,7 +348,7 @@ describe("DialCache targeted invalidation watermarks", () => { redis.failSet = true; const logger = { debug: vi.fn(), warn: vi.fn(), error: vi.fn() }; const metrics = new RecordingMetrics(); - const dialcache = new DialCache({ redis: { client: redis }, logger, metrics }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 }, logger, metrics }); await expect(dialcache.invalidateRemote("user_id", "123")).rejects.toThrow("redis set failed"); @@ -372,7 +372,7 @@ describe("DialCache targeted invalidation watermarks", () => { it("rejects invalid future buffers before calling Redis", async () => { const redis = new FakeRedis(); - const dialcache = new DialCache({ redis: { client: redis } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); await expect(dialcache.invalidateRemote("user_id", "123", -1)).rejects.toThrow("futureBufferMs"); await expect(dialcache.invalidateRemote("user_id", "123", 1.5)).rejects.toThrow("futureBufferMs"); @@ -385,7 +385,7 @@ describe("DialCache targeted invalidation watermarks", () => { const redis = new FakeRedis(); const dialcache = new DialCache({ namespace: "urn:galileo:test", - redis: { client: redis }, + redis: { client: redis, readTimeoutMs: 1_000 }, }); const getUser = dialcache.cached(async (userId: string, locale: string) => ({ userId, locale }), { keyType: "User", @@ -410,7 +410,7 @@ describe("DialCache targeted invalidation watermarks", () => { it("documents that Redis invalidation does not evict local cache", async () => { const redis = new FakeRedis(); - const dialcache = new DialCache({ redis: { client: redis } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); let version = 1; const getUser = dialcache.cached(async (userId: string) => ({ userId, version }), { keyType: "user_id", @@ -431,7 +431,7 @@ describe("DialCache targeted invalidation watermarks", () => { it("keeps a memoized request-local value after remote invalidation and refreshes in the next request", async () => { const redis = new FakeRedis(); - const dialcache = new DialCache({ redis: { client: redis } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); let version = 1; const getUser = dialcache.cached(async (userId: string) => ({ userId, version }), { keyType: "user_id", diff --git a/test/dialcache-liveness.test.ts b/test/dialcache-liveness.test.ts index dcabf58..dd41f28 100644 --- a/test/dialcache-liveness.test.ts +++ b/test/dialcache-liveness.test.ts @@ -514,23 +514,22 @@ describe("DialCache fallback liveness", () => { expect(fallback).toHaveBeenCalledTimes(1); }); - it("does not apply the fallback deadline to a pending Redis read", async () => { - const readGate = deferred(); + it("bounds a pending Redis read before starting the separate fallback deadline", async () => { const readStarted = deferred(); const fallback = vi.fn(async () => "value"); const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); const redis: DialCacheRedisClient = { read: async () => { readStarted.resolve(); - return await readGate.promise; + return await new Promise(() => undefined); }, write: async () => true, invalidate: async () => undefined, }; - const dialcache = new DialCache({ redis: { client: redis } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 10 } }); const load = dialcache.cached(fallback, { keyType: "id", - useCase: "PendingRedisReadHasCallerOwnedDeadline", + useCase: "PendingRedisReadHasCoreDeadline", cacheKey: () => "123", fallbackTimeoutMs: 5, defaultConfig: remoteConfig, @@ -538,15 +537,17 @@ describe("DialCache fallback liveness", () => { const result = dialcache.enable(async () => await load()); await readStarted.promise; - await vi.advanceTimersByTimeAsync(100); + await vi.advanceTimersByTimeAsync(9); expect(fallback).not.toHaveBeenCalled(); - expect(setTimeoutSpy).not.toHaveBeenCalled(); + expect(setTimeoutSpy).toHaveBeenCalledTimes(1); expect(dialcache.getCoalescingState().process.activeLeaders).toBe(1); - readGate.resolve(null); + await vi.advanceTimersByTimeAsync(1); await expect(result).resolves.toBe("value"); expect(fallback).toHaveBeenCalledTimes(1); + expect(setTimeoutSpy).toHaveBeenCalledTimes(2); + expect(vi.getTimerCount()).toBe(0); expect(dialcache.getCoalescingState().process.activeLeaders).toBe(0); }); @@ -567,7 +568,7 @@ describe("DialCache fallback liveness", () => { write: async () => true, invalidate: async () => undefined, }; - const dialcache = new DialCache({ redis: { client: redis } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); const load = dialcache.cached(async () => await fallback(), { keyType: "id", useCase: "PendingSerializerLoadHasCallerOwnedDeadline", @@ -582,7 +583,8 @@ describe("DialCache fallback liveness", () => { await vi.advanceTimersByTimeAsync(100); expect(fallback).not.toHaveBeenCalled(); - expect(setTimeoutSpy).not.toHaveBeenCalled(); + expect(setTimeoutSpy).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); expect(dialcache.getCoalescingState().process.activeLeaders).toBe(1); loadGate.resolve("cached"); @@ -611,7 +613,7 @@ describe("DialCache fallback liveness", () => { }, invalidate: async () => undefined, }; - const dialcache = new DialCache({ redis: { client: redis } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); const load = dialcache.cached(async () => "value", { keyType: "id", useCase: "PendingPublicationHasCallerOwnedDeadline", @@ -632,7 +634,7 @@ describe("DialCache fallback liveness", () => { }, ); await dumpStarted.promise; - expect(setTimeoutSpy).toHaveBeenCalledTimes(1); + expect(setTimeoutSpy).toHaveBeenCalledTimes(2); expect(vi.getTimerCount()).toBe(0); await vi.advanceTimersByTimeAsync(100); @@ -728,7 +730,7 @@ describe("DialCache fallback liveness", () => { const firstGate = deferred<{ readonly id: string; readonly version: number }>(); const firstStarted = deferred(); const redis = new FakeRedis(); - const dialcache = new DialCache({ redis: { client: redis } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); let calls = 0; const getUser = dialcache.cached(async (id: string) => { calls += 1; diff --git a/test/dialcache-logger.test.ts b/test/dialcache-logger.test.ts index 4a5c27c..31e8cd8 100644 --- a/test/dialcache-logger.test.ts +++ b/test/dialcache-logger.test.ts @@ -108,12 +108,12 @@ describe("DialCache logger isolation", () => { expect(logger.warn).toHaveBeenCalledWith("Error putting value in local cache", expect.any(Error)); }); - it("preserves fallback behavior when Redis reads, writes, and logging fail", async () => { + it("preserves fallback behavior and skips Redis writes when reads and logging fail", async () => { const logger = throwingLogger(); const redis = new FakeRedis(); redis.failGet = true; redis.failSet = true; - const dialcache = new DialCache({ logger, redis: { client: redis } }); + const dialcache = new DialCache({ logger, redis: { client: redis, readTimeoutMs: 1_000 } }); const getUser = dialcache.cached(async () => ({ source: "fallback" }), { keyType: "user_id", useCase: "ThrowingLoggerRedisReadWrite", @@ -123,10 +123,9 @@ describe("DialCache logger isolation", () => { await expect(dialcache.enable(async () => await getUser())).resolves.toEqual({ source: "fallback" }); expect(redis.getCalls).toBe(1); - expect(redis.setCalls).toBe(1); - expect(logger.warn).toHaveBeenCalledTimes(2); + expect(redis.setCalls).toBe(0); + expect(logger.warn).toHaveBeenCalledOnce(); expect(logger.warn).toHaveBeenCalledWith("Error getting value from Redis cache", expect.any(Error)); - expect(logger.warn).toHaveBeenCalledWith("Error putting value in Redis cache", expect.any(Error)); }); it("preserves the original invalidation error when logging also fails", async () => { @@ -139,7 +138,7 @@ describe("DialCache logger isolation", () => { throw invalidationError; }), } satisfies DialCacheRedisClient; - const dialcache = new DialCache({ logger, redis: { client: redis } }); + const dialcache = new DialCache({ logger, redis: { client: redis, readTimeoutMs: 1_000 } }); await expect(dialcache.invalidateRemote("user_id", "123")).rejects.toBe(invalidationError); expect(logger.warn).toHaveBeenCalledOnce(); diff --git a/test/dialcache-metrics.test.ts b/test/dialcache-metrics.test.ts index 93d394a..f96c25f 100644 --- a/test/dialcache-metrics.test.ts +++ b/test/dialcache-metrics.test.ts @@ -89,7 +89,7 @@ describe("DialCache observability metrics", () => { const dialcache = new DialCache({ namespace: "users-cache", metrics, - redis: { client: redis }, + redis: { client: redis, readTimeoutMs: 1_000 }, logger: { debug: vi.fn(), warn: vi.fn(), error: vi.fn() }, }); const getUser = dialcache.cached(async (userId: string) => { @@ -359,14 +359,14 @@ describe("DialCache observability metrics", () => { write: vi.fn(async () => true), invalidate: vi.fn(async () => undefined), }; - const cacheFailure = new DialCache({ redis: { client: failingRedis }, metrics, logger }); + const cacheFailure = new DialCache({ redis: { client: failingRedis, readTimeoutMs: 1_000 }, metrics, logger }); const readThroughFailure = cacheFailure.cached(async (userId: string) => ({ userId }), { keyType: "user_id", useCase: "CacheErrorClassification", cacheKey: (userId) => userId, defaultConfig: remoteOnly(), }); - const fallbackCache = new DialCache({ redis: { client: new FakeRedis() }, metrics, logger }); + const fallbackCache = new DialCache({ redis: { client: new FakeRedis(), readTimeoutMs: 1_000 }, metrics, logger }); const fallbackFailure = fallbackCache.cached(async (userId: string) => { const fallbackError = new TypeError("database failed for tenant-456"); fallbackError.name = "Tenant456DatabaseError"; @@ -428,7 +428,7 @@ describe("DialCache observability metrics", () => { const failingWriteRedis = new FakeRedis(); failingWriteRedis.failSet = true; - const writeFailure = new DialCache({ redis: { client: failingWriteRedis }, metrics, logger }); + const writeFailure = new DialCache({ redis: { client: failingWriteRedis, readTimeoutMs: 1_000 }, metrics, logger }); const writeValue = writeFailure.cached(async (id: string) => id, { keyType: "user_id", useCase: "WriteErrorClassification", @@ -445,7 +445,7 @@ describe("DialCache observability metrics", () => { }, load: async (value) => value.toString(), }; - const dumpFailure = new DialCache({ redis: { client: new FakeRedis() }, metrics, logger }); + const dumpFailure = new DialCache({ redis: { client: new FakeRedis(), readTimeoutMs: 1_000 }, metrics, logger }); const dumpValue = dumpFailure.cached(async (id: string) => id, { keyType: "user_id", useCase: "SerializationDumpClassification", @@ -456,7 +456,7 @@ describe("DialCache observability metrics", () => { await dumpFailure.enable(async () => await dumpValue("123")); const loadRedis = new FakeRedis(); - const loadWriter = new DialCache({ redis: { client: loadRedis } }); + const loadWriter = new DialCache({ redis: { client: loadRedis, readTimeoutMs: 1_000 } }); const writeLoadFixture = loadWriter.cached(async (id: string) => id, { keyType: "user_id", useCase: "SerializationLoadClassification", @@ -472,7 +472,7 @@ describe("DialCache observability metrics", () => { throw loadError; }, }; - const loadFailure = new DialCache({ redis: { client: loadRedis }, metrics, logger }); + const loadFailure = new DialCache({ redis: { client: loadRedis, readTimeoutMs: 1_000 }, metrics, logger }); const loadValue = loadFailure.cached(async (id: string) => id, { keyType: "user_id", useCase: "SerializationLoadClassification", @@ -555,7 +555,7 @@ describe("DialCache observability metrics", () => { const remoteError = new Error("remote ramp failed for tenant-456"); remoteError.name = "Tenant456RemoteRampError"; const remoteFailure = new DialCache({ - redis: { client: new FakeRedis() }, + redis: { client: new FakeRedis(), readTimeoutMs: 1_000 }, metrics, logger, rampSampler: ({ layer }) => { @@ -694,7 +694,7 @@ describe("DialCache observability metrics", () => { cacheKey: (id: string) => id, defaultConfig: localOnly(), }); - const remoteFailure = new DialCache({ redis: { client: new FakeRedis() }, metrics }); + const remoteFailure = new DialCache({ redis: { client: new FakeRedis(), readTimeoutMs: 1_000 }, metrics }); const remoteFallback = remoteFailure.cached(async (_id: string) => { throw new Error("remote fallback failed"); }, { diff --git a/test/dialcache-observability-internals.test.ts b/test/dialcache-observability-internals.test.ts index 4beb703..1b96e91 100644 --- a/test/dialcache-observability-internals.test.ts +++ b/test/dialcache-observability-internals.test.ts @@ -72,7 +72,7 @@ describe("DialCache observability internal compatibility paths", () => { const redisCache = new RedisCache({ configProvider: async () => null, rampSampler: () => 0, - redis: { client: redis }, + redis: { client: redis, readTimeoutMs: 1_000 }, metrics: null, }); const enabledKey = key(); diff --git a/test/dialcache-redis-read-deadline.test.ts b/test/dialcache-redis-read-deadline.test.ts new file mode 100644 index 0000000..527e3b5 --- /dev/null +++ b/test/dialcache-redis-read-deadline.test.ts @@ -0,0 +1,513 @@ +import { performance } from "node:perf_hooks"; + +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { + CacheLayer, + DialCache, + DialCacheError, + DialCacheKeyConfig, + RedisReadTimeoutError, + type CachedOptions, + type DialCacheMetricsAdapter, + type DialCacheRedisClient, + type RedisCachePayload, + type RedisConfig, + type RedisReadContext, +} from "../src/index.js"; + +interface Deferred { + readonly promise: Promise; + resolve: (value: T) => void; + reject: (error: unknown) => void; +} + +function deferred(): Deferred { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +const remoteConfig = new DialCacheKeyConfig({ + ttlSec: { [CacheLayer.REMOTE]: 60 }, + ramp: { [CacheLayer.REMOTE]: 100 }, +}); + +const localAndRemoteConfig = new DialCacheKeyConfig({ + ttlSec: { [CacheLayer.LOCAL]: 60, [CacheLayer.REMOTE]: 60 }, + ramp: { [CacheLayer.LOCAL]: 100, [CacheLayer.REMOTE]: 100 }, +}); + +function metricsWithError(error: DialCacheMetricsAdapter["error"]): DialCacheMetricsAdapter { + return { + request: vi.fn(), + miss: vi.fn(), + disabled: vi.fn(), + error, + invalidation: vi.fn(), + observeGet: vi.fn(), + observeFallback: vi.fn(), + observeSerialization: vi.fn(), + observeSize: vi.fn(), + }; +} + +function redisClient(read: DialCacheRedisClient["read"]): { + readonly client: DialCacheRedisClient; + readonly read: ReturnType>; + readonly write: ReturnType>; +} { + const readMock = vi.fn(read); + const write = vi.fn(async () => true); + return { + client: { + read: readMock, + write, + invalidate: async () => undefined, + }, + read: readMock, + write, + }; +} + +function useFakeTimersWithMonotonicClock(): void { + vi.useFakeTimers(); + const clockOriginMs = Date.now(); + vi.spyOn(performance, "now").mockImplementation(() => Date.now() - clockOriginMs); +} + +describe("DialCache Redis read deadlines", () => { + beforeEach(() => { + useFakeTimersWithMonotonicClock(); + }); + + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it("resolves the instance default and static per-use-case override at registration", async () => { + const contexts: RedisReadContext[] = []; + const redis = redisClient(async (_request, context) => { + if (context === undefined) { + throw new Error("missing read context"); + } + contexts.push(context); + return null; + }); + const dialcache = new DialCache({ redis: { client: redis.client, readTimeoutMs: 100 } }); + const inherited = dialcache.cached(async () => "inherited", { + keyType: "id", + useCase: "InheritedRedisReadDeadline", + cacheKey: () => "1", + defaultConfig: remoteConfig, + }); + const overriddenOptions = { + keyType: "id", + useCase: "OverriddenRedisReadDeadline", + cacheKey: () => "2", + redisReadTimeoutMs: 25, + defaultConfig: remoteConfig, + }; + const overridden = dialcache.cached(async () => "overridden", overriddenOptions); + overriddenOptions.redisReadTimeoutMs = 50; + + await expect(dialcache.enable(async () => await inherited())).resolves.toBe("inherited"); + await expect(dialcache.enable(async () => await overridden())).resolves.toBe("overridden"); + + expect(contexts.map(({ timeoutMs }) => timeoutMs)).toEqual([100, 25]); + expect(contexts.every(({ signal }) => signal.aborted === false)).toBe(true); + expect(vi.getTimerCount()).toBe(0); + }); + + it("rejects missing and invalid instance defaults", () => { + const client = redisClient(async () => null).client; + expect( + () => new DialCache({ redis: { client } as RedisConfig }), + ).toThrow(new TypeError("Redis config requires readTimeoutMs")); + + const invalidValues: readonly unknown[] = [ + null, + 0, + -1, + 1.5, + Number.NaN, + Number.POSITIVE_INFINITY, + Number.MAX_SAFE_INTEGER + 1, + 2_147_483_648, + "100", + ]; + for (const readTimeoutMs of invalidValues) { + expect( + () => new DialCache({ + redis: { client, readTimeoutMs } as unknown as RedisConfig, + }), + ).toThrow( + new RangeError( + "Redis readTimeoutMs must be a positive safe integer no greater than 2147483647", + ), + ); + } + expect( + () => new DialCache({ redis: { client, readTimeoutMs: 2_147_483_647 } }), + ).not.toThrow(); + }); + + it("rejects invalid use-case overrides before reserving the use-case name", () => { + const client = redisClient(async () => null).client; + const invalidValues: readonly unknown[] = [ + null, + 0, + -1, + 1.5, + Number.NaN, + Number.POSITIVE_INFINITY, + Number.MAX_SAFE_INTEGER + 1, + 2_147_483_648, + "100", + ]; + + for (const [index, redisReadTimeoutMs] of invalidValues.entries()) { + const useCase = `InvalidRedisReadDeadline${index}`; + const options = { + keyType: "id", + useCase, + cacheKey: () => String(index), + redisReadTimeoutMs, + } as unknown as CachedOptions<() => Promise>; + const dialcache = new DialCache({ redis: { client, readTimeoutMs: 100 } }); + + expect(() => dialcache.cached(async () => "value", options)).toThrow( + new RangeError( + "DialCache redisReadTimeoutMs must be a positive safe integer no greater than 2147483647", + ), + ); + expect( + () => dialcache.cached(async () => "value", { ...options, redisReadTimeoutMs: 10 }), + ).not.toThrow(); + } + + const dialcache = new DialCache({ redis: { client, readTimeoutMs: 100 } }); + expect( + () => dialcache.cached(async () => "value", { + keyType: "id", + useCase: "MaximumRedisReadDeadline", + cacheKey: () => "max", + redisReadTimeoutMs: 2_147_483_647, + }), + ).not.toThrow(); + }); + + it("cleans up the read timer after hits and misses", async () => { + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const redis = redisClient( + vi.fn() + .mockResolvedValueOnce(JSON.stringify({ source: "redis" })) + .mockResolvedValueOnce(null), + ); + const dialcache = new DialCache({ redis: { client: redis.client, readTimeoutMs: 100 } }); + const hit = dialcache.cached(async () => ({ source: "fallback" }), { + keyType: "id", + useCase: "RedisReadDeadlineHit", + cacheKey: () => "1", + defaultConfig: remoteConfig, + }); + const miss = dialcache.cached(async () => ({ source: "fallback" }), { + keyType: "id", + useCase: "RedisReadDeadlineMiss", + cacheKey: () => "2", + defaultConfig: remoteConfig, + }); + + await expect(dialcache.enable(async () => await hit())).resolves.toEqual({ source: "redis" }); + expect(vi.getTimerCount()).toBe(0); + await expect(dialcache.enable(async () => await miss())).resolves.toEqual({ source: "fallback" }); + + expect(redis.read).toHaveBeenCalledTimes(2); + expect(redis.write).toHaveBeenCalledTimes(1); + expect(setTimeoutSpy).toHaveBeenCalledTimes(3); + expect(vi.getTimerCount()).toBe(0); + }); + + it("times out one shared leader, aborts cooperatively, and fails open once", async () => { + const readStarted = deferred(); + const contexts: RedisReadContext[] = []; + const redis = redisClient(async (_request, context) => { + if (context === undefined) { + throw new Error("missing read context"); + } + contexts.push(context); + readStarted.resolve(); + return await new Promise(() => undefined); + }); + const error = vi.fn(); + const logger = { debug: vi.fn(), warn: vi.fn(), error: vi.fn() }; + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const metrics = metricsWithError(error); + const dialcache = new DialCache({ + redis: { client: redis.client, readTimeoutMs: 10 }, + metrics, + logger, + }); + const fallback = vi.fn(async () => { + expect(contexts[0]?.signal.aborted).toBe(true); + return { source: "fallback" }; + }); + const load = dialcache.cached(fallback, { + keyType: "id", + useCase: "SharedRedisReadDeadline", + cacheKey: () => "123", + fallbackTimeoutMs: 100, + defaultConfig: remoteConfig, + }); + + const result = dialcache.enable(async () => + await Promise.all([load(), load(), load()]), + ); + await readStarted.promise; + + expect(redis.read).toHaveBeenCalledTimes(1); + expect(setTimeoutSpy).toHaveBeenCalledTimes(1); + expect((setTimeoutSpy.mock.results[0]?.value as NodeJS.Timeout).hasRef()).toBe(true); + expect(dialcache.getCoalescingState().process).toMatchObject({ + activeLeaders: 1, + activeFollowers: 2, + }); + + await vi.advanceTimersByTimeAsync(9); + expect(fallback).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + + await expect(result).resolves.toEqual([ + { source: "fallback" }, + { source: "fallback" }, + { source: "fallback" }, + ]); + expect(fallback).toHaveBeenCalledTimes(1); + expect(redis.write).not.toHaveBeenCalled(); + expect(logger.warn).toHaveBeenCalledTimes(1); + const timeout = logger.warn.mock.calls[0]?.[1]; + expect(timeout).toBeInstanceOf(RedisReadTimeoutError); + expect(timeout).toBeInstanceOf(DialCacheError); + expect(timeout).toMatchObject({ + useCase: "SharedRedisReadDeadline", + timeoutMs: 10, + }); + expect(String((timeout as Error).message)).not.toContain("123"); + expect(error).toHaveBeenCalledTimes(1); + expect(error).toHaveBeenCalledWith({ + cacheNamespace: "urn", + useCase: "SharedRedisReadDeadline", + keyType: "id", + layer: CacheLayer.REMOTE, + error: "cache_read", + inFallback: false, + }); + expect(metrics.disabled).not.toHaveBeenCalledWith( + expect.objectContaining({ reason: "config_error" }), + ); + expect(dialcache.getCoalescingState().process).toEqual({ + activeLeaders: 0, + activeFollowers: 0, + oldestLeaderAgeMs: null, + }); + expect(vi.getTimerCount()).toBe(0); + }); + + it("gives a late process follower only the leader's remaining read budget", async () => { + const readStarted = deferred(); + const redis = redisClient(async () => { + readStarted.resolve(); + return await new Promise(() => undefined); + }); + const dialcache = new DialCache({ redis: { client: redis.client, readTimeoutMs: 10 } }); + const fallback = vi.fn(async () => "fallback"); + const load = dialcache.cached(fallback, { + keyType: "id", + useCase: "LateRedisReadFollower", + cacheKey: () => "123", + defaultConfig: remoteConfig, + }); + + const first = dialcache.enable(async () => await load()); + await readStarted.promise; + await vi.advanceTimersByTimeAsync(7); + const second = dialcache.enable(async () => await load()); + await vi.advanceTimersByTimeAsync(0); + + expect(redis.read).toHaveBeenCalledTimes(1); + expect(dialcache.getCoalescingState().process.activeFollowers).toBe(1); + + await vi.advanceTimersByTimeAsync(3); + await expect(Promise.all([first, second])).resolves.toEqual(["fallback", "fallback"]); + expect(fallback).toHaveBeenCalledTimes(1); + }); + + it("shares one read deadline with request-local followers", async () => { + const readStarted = deferred(); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const redis = redisClient(async () => { + readStarted.resolve(); + return await new Promise(() => undefined); + }); + const dialcache = new DialCache({ redis: { client: redis.client, readTimeoutMs: 10 } }); + const fallback = vi.fn(async () => "fallback"); + const load = dialcache.cached(fallback, { + keyType: "id", + useCase: "RequestLocalRedisReadDeadline", + cacheKey: () => "123", + defaultConfig: new DialCacheKeyConfig({ + requestLocal: true, + ttlSec: { [CacheLayer.REMOTE]: 60 }, + ramp: { [CacheLayer.REMOTE]: 100 }, + }), + }); + + const result = dialcache.enable(async () => await Promise.all([load(), load(), load()])); + await readStarted.promise; + + expect(redis.read).toHaveBeenCalledTimes(1); + expect(setTimeoutSpy).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(10); + + await expect(result).resolves.toEqual(["fallback", "fallback", "fallback"]); + expect(fallback).toHaveBeenCalledTimes(1); + }); + + it.each(["fulfillment", "rejection"] as const)( + "consumes late read %s and lets a later invocation recover", + async (settlement) => { + const firstRead = deferred(); + let readCalls = 0; + const redis = redisClient(async () => { + readCalls += 1; + return readCalls === 1 + ? await firstRead.promise + : JSON.stringify({ source: "redis" }); + }); + const error = vi.fn(); + const logger = { debug: vi.fn(), warn: vi.fn(), error: vi.fn() }; + const dialcache = new DialCache({ + redis: { client: redis.client, readTimeoutMs: 10 }, + metrics: metricsWithError(error), + logger, + }); + const fallback = vi.fn(async () => ({ source: "fallback" })); + const load = dialcache.cached(fallback, { + keyType: "id", + useCase: `LateRedisRead${settlement}`, + cacheKey: () => "123", + defaultConfig: remoteConfig, + }); + + const first = dialcache.enable(async () => await load()); + await vi.advanceTimersByTimeAsync(10); + await expect(first).resolves.toEqual({ source: "fallback" }); + await expect(dialcache.enable(async () => await load())).resolves.toEqual({ source: "redis" }); + + if (settlement === "fulfillment") { + firstRead.resolve(JSON.stringify({ source: "late" })); + } else { + firstRead.reject(new Error("late Redis failure")); + } + await vi.advanceTimersByTimeAsync(0); + + expect(redis.read).toHaveBeenCalledTimes(2); + expect(redis.write).not.toHaveBeenCalled(); + expect(fallback).toHaveBeenCalledTimes(1); + expect(logger.warn).toHaveBeenCalledTimes(1); + expect(error).toHaveBeenCalledTimes(1); + }, + ); + + it.each([ + { failure: "exception", tracked: false, expectedReadCalls: 1, expectedFallbackCalls: 1 }, + { failure: "exception", tracked: true, expectedReadCalls: 2, expectedFallbackCalls: 2 }, + { failure: "timeout", tracked: false, expectedReadCalls: 1, expectedFallbackCalls: 1 }, + { failure: "timeout", tracked: true, expectedReadCalls: 2, expectedFallbackCalls: 2 }, + ])( + "skips Redis writes and applies safe local publication after a read $failure (tracked=$tracked)", + async ({ failure, tracked, expectedReadCalls, expectedFallbackCalls }) => { + const redis = redisClient( + failure === "exception" + ? async () => { + throw new Error("Redis unavailable"); + } + : async () => await new Promise(() => undefined), + ); + const dialcache = new DialCache({ + redis: { client: redis.client, readTimeoutMs: 10 }, + logger: { debug: vi.fn(), warn: vi.fn(), error: vi.fn() }, + }); + let fallbackCalls = 0; + const load = dialcache.cached(async () => ({ call: ++fallbackCalls }), { + keyType: "id", + useCase: tracked ? "TrackedReadFailure" : "UntrackedReadFailure", + cacheKey: () => "123", + trackForInvalidation: tracked, + defaultConfig: localAndRemoteConfig, + }); + + const firstResult = dialcache.enable(async () => await load()); + if (failure === "timeout") { + await vi.advanceTimersByTimeAsync(10); + } + const first = await firstResult; + const secondResult = dialcache.enable(async () => await load()); + if (failure === "timeout" && tracked) { + await vi.advanceTimersByTimeAsync(10); + } + const second = await secondResult; + + expect(first).toEqual({ call: 1 }); + expect(second).toEqual({ call: expectedFallbackCalls }); + expect(redis.read).toHaveBeenCalledTimes(expectedReadCalls); + expect(redis.write).not.toHaveBeenCalled(); + expect(fallbackCalls).toBe(expectedFallbackCalls); + }, + ); + + it("allocates no read timer for disabled calls, ramped-out Redis, or local hits", async () => { + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const redis = redisClient(async () => null); + const dialcache = new DialCache({ redis: { client: redis.client, readTimeoutMs: 100 } }); + const disabled = dialcache.cached(async () => "disabled", { + keyType: "id", + useCase: "DisabledRedisReadDeadline", + cacheKey: () => "disabled", + defaultConfig: remoteConfig, + }); + const rampedOut = dialcache.cached(async () => "ramped", { + keyType: "id", + useCase: "RampedOutRedisReadDeadline", + cacheKey: () => "ramped", + fallbackTimeoutMs: null, + defaultConfig: new DialCacheKeyConfig({ + ttlSec: { [CacheLayer.REMOTE]: 60 }, + ramp: { [CacheLayer.REMOTE]: 0 }, + }), + }); + const localHit = dialcache.cached(async () => "local", { + keyType: "id", + useCase: "LocalHitRedisReadDeadline", + cacheKey: () => "local", + defaultConfig: localAndRemoteConfig, + }); + + await expect(disabled()).resolves.toBe("disabled"); + await expect(dialcache.enable(async () => await rampedOut())).resolves.toBe("ramped"); + expect(redis.read).not.toHaveBeenCalled(); + expect(setTimeoutSpy).not.toHaveBeenCalled(); + + await expect(dialcache.enable(async () => await localHit())).resolves.toBe("local"); + expect(redis.read).toHaveBeenCalledTimes(1); + setTimeoutSpy.mockClear(); + await expect(dialcache.enable(async () => await localHit())).resolves.toBe("local"); + expect(redis.read).toHaveBeenCalledTimes(1); + expect(setTimeoutSpy).not.toHaveBeenCalled(); + }); +}); diff --git a/test/dialcache-redis.test.ts b/test/dialcache-redis.test.ts index 5ef0e74..aafeb2e 100644 --- a/test/dialcache-redis.test.ts +++ b/test/dialcache-redis.test.ts @@ -25,7 +25,7 @@ describe("DialCache Redis TTL layer", () => { it("reads local miss from Redis and populates the local layer", async () => { // Given one process has already written a value into the shared Redis cache. const redis = new FakeRedis(); - const writer = new DialCache({ redis: { client: redis } }); + const writer = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); let writerCalls = 0; const writeUser = writer.cached(async (userId: string) => ({ userId, calls: ++writerCalls }), { keyType: "user_id", @@ -35,7 +35,7 @@ describe("DialCache Redis TTL layer", () => { }); await writer.enable(async () => await writeUser("123")); - const reader = new DialCache({ redis: { client: redis } }); + const reader = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); let readerCalls = 0; const readUser = reader.cached(async (userId: string) => ({ userId, calls: ++readerCalls }), { keyType: "user_id", @@ -60,7 +60,7 @@ describe("DialCache Redis TTL layer", () => { it("does not populate local cache when the local layer was disabled for the read", async () => { // Given one process has already written a value into the shared Redis cache. const redis = new FakeRedis(); - const writer = new DialCache({ redis: { client: redis } }); + const writer = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); const writeUser = writer.cached(async (userId: string) => ({ userId, source: "redis" }), { keyType: "user_id", useCase: "RedisNoDisabledLocalPopulate", @@ -77,7 +77,7 @@ describe("DialCache Redis TTL layer", () => { }); const cacheConfigProvider = vi.fn(async () => (++providerCalls <= 2 ? remoteOnlyConfig : DialCacheKeyConfig.enabled(60))); const logger = { debug: vi.fn(), warn: vi.fn(), error: vi.fn() }; - const reader = new DialCache({ redis: { client: redis }, cacheConfigProvider, logger }); + const reader = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 }, cacheConfigProvider, logger }); let readerCalls = 0; const readUser = reader.cached(async (userId: string) => ({ userId, source: `fallback-${++readerCalls}` }), { keyType: "user_id", @@ -107,7 +107,7 @@ describe("DialCache Redis TTL layer", () => { }); const cacheConfigProvider = vi.fn(async () => remoteOnlyConfig); const logger = { debug: vi.fn(), warn: vi.fn(), error: vi.fn() }; - const dialcache = new DialCache({ redis: { client: redis }, cacheConfigProvider, logger }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 }, cacheConfigProvider, logger }); const getUser = dialcache.cached(async (userId: string) => ({ userId }), { keyType: "user_id", useCase: "RemoteFailureConfigSnapshot", @@ -118,7 +118,7 @@ describe("DialCache Redis TTL layer", () => { expect(cacheConfigProvider).toHaveBeenCalledTimes(1); expect(redis.getCalls).toBe(1); - expect(redis.setCalls).toBe(1); + expect(redis.setCalls).toBe(0); }); it("writes Redis misses with a timestamped binary frame", async () => { @@ -126,7 +126,7 @@ describe("DialCache Redis TTL layer", () => { vi.useFakeTimers(); vi.setSystemTime(new Date("2026-05-12T17:00:00.000Z")); const redis = new FakeRedis(); - const dialcache = new DialCache({ redis: { client: redis } }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); let calls = 0; const getUser = dialcache.cached(async (userId: string) => ({ userId, calls: ++calls }), { keyType: "user_id", @@ -170,13 +170,13 @@ describe("DialCache Redis TTL layer", () => { }, ); - it("fails open when Redis operations fail before fallback", async () => { + it("fails open without attempting a second Redis operation after a read failure", async () => { // Given Redis is unavailable and local caching is not configured for this key. const redis = new FakeRedis(); redis.failGet = true; redis.failSet = true; const logger = { debug: vi.fn(), warn: vi.fn(), error: vi.fn() }; - const dialcache = new DialCache({ redis: { client: redis }, logger }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 }, logger }); let calls = 0; const getUser = dialcache.cached(async (userId: string) => ({ userId, calls: ++calls }), { keyType: "user_id", @@ -196,7 +196,8 @@ describe("DialCache Redis TTL layer", () => { expect(first).toEqual({ userId: "123", calls: 1 }); expect(second).toEqual({ userId: "123", calls: 2 }); expect(logger.warn).toHaveBeenCalledWith("Error getting value from Redis cache", expect.any(Error)); - expect(logger.warn).toHaveBeenCalledWith("Error putting value in Redis cache", expect.any(Error)); + expect(redis.setCalls).toBe(0); + expect(logger.warn).not.toHaveBeenCalledWith("Error putting value in Redis cache", expect.any(Error)); }); it("round-trips Redis values through a custom serializer", async () => { @@ -209,7 +210,7 @@ describe("DialCache Redis TTL layer", () => { return { id: id ?? "", source: source ?? "" }; }), }; - const writer = new DialCache({ redis: { client: redis } }); + const writer = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); const readFromWriter = writer.cached(async (userId: string) => ({ id: userId, source: "fallback" }), { keyType: "user_id", useCase: "RedisCustomSerializer", @@ -219,7 +220,7 @@ describe("DialCache Redis TTL layer", () => { }); await writer.enable(async () => await readFromWriter("123")); - const reader = new DialCache({ redis: { client: redis } }); + const reader = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); let readerCalls = 0; const readFromRedis = reader.cached(async (userId: string) => ({ id: userId, source: `fallback-${++readerCalls}` }), { keyType: "user_id", @@ -248,7 +249,7 @@ describe("DialCache Redis TTL layer", () => { dump: vi.fn((value) => value.toISOString()), load: vi.fn((value) => new Date(Buffer.isBuffer(value) ? value.toString("utf8") : value)), }; - const writer = new DialCache({ redis: { client: redis } }); + const writer = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); const writeUpdatedAt = writer.cached(async () => new Date("2026-07-17T12:00:00.000Z"), { keyType: "user_id", useCase: "RedisDateSerializer", @@ -258,7 +259,7 @@ describe("DialCache Redis TTL layer", () => { }); await writer.enable(async () => await writeUpdatedAt()); - const reader = new DialCache({ redis: { client: redis } }); + const reader = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 } }); let readerCalls = 0; const readUpdatedAt = reader.cached( async () => { @@ -307,7 +308,7 @@ describe("DialCache Redis TTL layer", () => { }), load: vi.fn(async () => ({ userId: "never", calls: 0 })), }; - const dialcache = new DialCache({ redis: { client: redis }, logger }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 }, logger }); let calls = 0; const getUser = dialcache.cached(async (userId: string) => ({ userId, calls: ++calls }), { keyType: "user_id", @@ -346,7 +347,7 @@ describe("DialCache Redis TTL layer", () => { }), }; const logger = { debug: vi.fn(), warn: vi.fn(), error: vi.fn() }; - const dialcache = new DialCache({ redis: { client: redis }, logger }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 }, logger }); let calls = 0; const getUser = dialcache.cached(async (userId: string) => ({ userId, source: `fallback-${++calls}` }), { keyType: "user_id", @@ -383,7 +384,7 @@ describe("DialCache Redis TTL layer", () => { redis.setRaw(badKey, Buffer.from([2, 0, 0, 0, 0, 0, 0, 0, 0, 0])); redis.setRaw(nonFiniteKey, Buffer.from([1, 0])); const logger = { debug: vi.fn(), warn: vi.fn(), error: vi.fn() }; - const dialcache = new DialCache({ redis: { client: redis }, logger }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 }, logger }); let calls = 0; const getUser = dialcache.cached(async (userId: string) => ({ userId, calls: ++calls }), { keyType: "user_id", @@ -427,7 +428,7 @@ describe("DialCache Redis TTL layer", () => { observeSerialization: vi.fn(), observeSize: vi.fn(), }; - const dialcache = new DialCache({ redis: { client: redisClient }, logger, metrics }); + const dialcache = new DialCache({ redis: { client: redisClient, readTimeoutMs: 1_000 }, logger, metrics }); let calls = 0; const getUser = dialcache.cached(async (userId: string) => ({ userId, calls: ++calls }), { keyType: "user_id", @@ -476,7 +477,7 @@ describe("DialCache Redis TTL layer", () => { observeSerialization: vi.fn(), observeSize: vi.fn(), }; - const dialcache = new DialCache({ redis: { client: redis }, logger, metrics }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 }, logger, metrics }); let calls = 0; const getUser = dialcache.cached(async (userId: string) => ({ userId, calls: ++calls }), { keyType: "user_id", @@ -511,7 +512,7 @@ describe("DialCache Redis TTL layer", () => { // Given Redis is configured but the key has no remote TTL. const redis = new FakeRedis(); const logger = { debug: vi.fn(), warn: vi.fn(), error: vi.fn() }; - const dialcache = new DialCache({ redis: { client: redis }, logger }); + const dialcache = new DialCache({ redis: { client: redis, readTimeoutMs: 1_000 }, logger }); let calls = 0; const getUser = dialcache.cached(async (userId: string) => ({ userId, calls: ++calls }), { keyType: "user_id", diff --git a/test/node-redis.test.ts b/test/node-redis.test.ts index 8858a94..abb86fb 100644 --- a/test/node-redis.test.ts +++ b/test/node-redis.test.ts @@ -115,6 +115,29 @@ describe("node-redis adapter", () => { ).resolves.toBeUndefined(); }); + it("passes the cooperative read signal through node-redis command options", async () => { + const client = fakeClient(); + const adapter = createNodeRedisDialCacheClient(client as never); + const controller = new AbortController(); + const context = { timeoutMs: 25, signal: controller.signal } as const; + + await adapter.read({ valueKey: "plain:value" }, context); + await adapter.read( + { valueKey: "tracked:{id}:value", watermarkKey: "tracked:{id}:watermark" }, + context, + ); + + expect(client.dialcacheRead).toHaveBeenCalledWith( + expect.objectContaining({ returnBuffers: true, signal: controller.signal }), + "plain:value", + ); + expect(client.dialcacheReadTracked).toHaveBeenCalledWith( + expect.objectContaining({ returnBuffers: true, signal: controller.signal }), + "tracked:{id}:value", + "tracked:{id}:watermark", + ); + }); + it("rejects every out-of-domain reply returned by a node-redis client", async () => { const writeMessage = "Invalid DialCache Redis write reply; expected integer 0 or 1"; const invalidationMessage = "Invalid DialCache Redis invalidate reply; expected integer 1"; @@ -204,7 +227,7 @@ describe("node-redis adapter", () => { observeSerialization: vi.fn(), observeSize: vi.fn(), }; - const dialcache = new DialCache({ redis: { client: redisClient }, logger, metrics }); + const dialcache = new DialCache({ redis: { client: redisClient, readTimeoutMs: 1_000 }, logger, metrics }); const load = dialcache.cached(async (id: string) => ({ id }), { keyType: "user_id", useCase: "ProtocolFailure", diff --git a/test/prometheus.test.ts b/test/prometheus.test.ts index 4688301..29ac212 100644 --- a/test/prometheus.test.ts +++ b/test/prometheus.test.ts @@ -351,7 +351,7 @@ describe("Prometheus metrics adapter", () => { const registry = new Registry(); const redis = new FakeRedis(); const metrics = createPrometheusDialCacheMetrics({ registry, prefix: "test_" }); - const dialcache = new DialCache({ namespace: "metrics-cache", redis: { client: redis }, metrics }); + const dialcache = new DialCache({ namespace: "metrics-cache", redis: { client: redis, readTimeoutMs: 1_000 }, metrics }); let calls = 0; const getUser = dialcache.cached(async (userId: string) => ({ userId, calls: ++calls }), { keyType: "user_id", @@ -380,7 +380,7 @@ describe("Prometheus metrics adapter", () => { const registry = new Registry(); const redis = new FakeRedis(); const metrics = createPrometheusDialCacheMetrics({ registry, prefix: "test_" }); - const dialcache = new DialCache({ namespace: "metrics-cache", redis: { client: redis }, metrics }); + const dialcache = new DialCache({ namespace: "metrics-cache", redis: { client: redis, readTimeoutMs: 1_000 }, metrics }); const contextDisabled = dialcache.cached(async (userId: string) => userId, { keyType: "user_id", useCase: "PrometheusDisabledMetric", diff --git a/test/redis-cluster.integration.test.ts b/test/redis-cluster.integration.test.ts index 7cf395f..dda4fee 100644 --- a/test/redis-cluster.integration.test.ts +++ b/test/redis-cluster.integration.test.ts @@ -115,7 +115,7 @@ describe("DialCache Lua protocol on Redis Cluster", () => { const scriptClient: DialCacheRedisClient = createNodeRedisDialCacheClient(activeCluster); const dialcache = new DialCache({ namespace: "cluster-cache", - redis: { client: scriptClient }, + redis: { client: scriptClient, readTimeoutMs: 10_000 }, }); const ids = Array.from({ length: 30 }, (_, index) => `item-${index}`); let calls = 0; @@ -141,7 +141,7 @@ describe("DialCache Lua protocol on Redis Cluster", () => { ); const recoveryDialcache = new DialCache({ namespace: "cluster-cache", - redis: { client: scriptClient }, + redis: { client: scriptClient, readTimeoutMs: 10_000 }, }); const recoverValue = recoveryDialcache.cached(async (id: string) => ({ id, calls: ++calls }), { keyType: "item_id", @@ -168,7 +168,7 @@ describe("DialCache Lua protocol on Redis Cluster", () => { const scriptClient: DialCacheRedisClient = createNodeRedisDialCacheClient(cluster); const dialcache = new DialCache({ namespace: "cluster-cache", - redis: { client: scriptClient }, + redis: { client: scriptClient, readTimeoutMs: 10_000 }, }); let version = 1; const getUser = dialcache.cached(async (id: string) => ({ id, version }), { diff --git a/test/redis-real.integration.test.ts b/test/redis-real.integration.test.ts index 04725a7..d344e0b 100644 --- a/test/redis-real.integration.test.ts +++ b/test/redis-real.integration.test.ts @@ -194,7 +194,7 @@ describe.each(engines)("DialCache Lua protocol on $name", ({ image }) => { throw new Error("Redis test clients did not start"); } const scriptClient: DialCacheRedisClient = client.adapter; - const dialcache = new DialCache({ namespace: "real", redis: { client: scriptClient } }); + const dialcache = new DialCache({ namespace: "real", redis: { client: scriptClient, readTimeoutMs: 10_000 } }); let jsonCalls = 0; let binaryCalls = 0; const getJson = dialcache.cached(async (id: string) => ({ id, calls: ++jsonCalls }), { @@ -413,7 +413,7 @@ describe.each(engines)("DialCache Lua protocol on $name", ({ image }) => { throw new Error("Redis test clients did not start"); } const scriptClient: DialCacheRedisClient = client.adapter; - const dialcache = new DialCache({ namespace: "tracked", redis: { client: scriptClient } }); + const dialcache = new DialCache({ namespace: "tracked", redis: { client: scriptClient, readTimeoutMs: 10_000 } }); let version = 1; let calls = 0; const getUser = dialcache.cached(async (id: string) => ({ id, version, calls: ++calls }), { @@ -445,7 +445,7 @@ describe.each(engines)("DialCache Lua protocol on $name", ({ image }) => { } const scriptClient = client.adapter; const logger = { debug: () => undefined, warn: () => undefined, error: () => undefined }; - const dialcache = new DialCache({ namespace: "malformed", redis: { client: scriptClient }, logger }); + const dialcache = new DialCache({ namespace: "malformed", redis: { client: scriptClient, readTimeoutMs: 10_000 }, logger }); let calls = 0; const getUser = dialcache.cached(async (id: string) => ({ id, calls: ++calls }), { keyType: "user_id", @@ -507,7 +507,7 @@ describe.each(engines)("DialCache Lua protocol on $name", ({ image }) => { 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 }); + const dialcache = new DialCache({ namespace, redis: { client: redisClient, readTimeoutMs: 10_000 }, logger, metrics }); let calls = 0; const getUser = dialcache.cached(async (id: string) => ({ id, calls: ++calls }), { keyType: "user_id", diff --git a/test/valkey-glide.test.ts b/test/valkey-glide.test.ts index 9e9d323..1a7ab06 100644 --- a/test/valkey-glide.test.ts +++ b/test/valkey-glide.test.ts @@ -93,6 +93,23 @@ describe("Valkey GLIDE adapter", () => { expect(scriptInstances).toHaveLength(5); }); + it("preserves GLIDE invocation options when given a core read context", async () => { + const client = fakeClient(null); + const adapter = createValkeyGlideDialCacheClient(client, mockGlide); + const controller = new AbortController(); + + await adapter.read( + { valueKey: "plain:value" }, + { timeoutMs: 25, signal: controller.signal }, + ); + + expect(client.invokeScript).toHaveBeenCalledWith( + expect.any(MockScript), + { keys: ["plain:value"], args: [], decoder: decoderBytes }, + ); + adapter.dispose(); + }); + it("passes string and Buffer writes directly to GLIDE", async () => { const binary = Buffer.from([0, 0xff, 0x80]); const client = fakeClient(1, 0, 1);