From 9b1337020c1c563ad6c0c6d0e493aea2e96cdd04 Mon Sep 17 00:00:00 2001 From: Lev Neiman Date: Sun, 19 Jul 2026 14:58:54 -0700 Subject: [PATCH 1/4] feat: add fallback deadlines and coalescing state --- README.md | 75 +++- scripts/benchmark-request-local.mjs | 9 + scripts/test-package.mjs | 35 ++ src/config.ts | 8 + src/dialcache.ts | 174 ++++++++- src/errors.ts | 11 + src/index.ts | 10 +- src/internal/redis-cache.ts | 4 +- src/node-redis.ts | 5 +- src/prometheus.ts | 2 +- src/redis-client.ts | 6 + src/serializer.ts | 2 + src/valkey-glide.ts | 4 +- test/dialcache-liveness.test.ts | 566 ++++++++++++++++++++++++++++ 14 files changed, 892 insertions(+), 19 deletions(-) create mode 100644 test/dialcache-liveness.test.ts diff --git a/README.md b/README.md index 9ae3ef6..7953420 100644 --- a/README.md +++ b/README.md @@ -110,6 +110,9 @@ import { createNodeRedisDialCacheClient, dialcacheRedisScripts } from "dialcache const redisClient = createClient({ url: process.env.REDIS_URL, scripts: dialcacheRedisScripts, + disableOfflineQueue: true, + commandsQueueMaxLength: 1_000, + socket: { connectTimeout: 2_000 }, }); await redisClient.connect(); @@ -135,6 +138,8 @@ import { createValkeyGlideDialCacheClient } from "dialcache/valkey-glide"; const glideClient = await GlideClient.createClient({ addresses: [{ host: "127.0.0.1", port: 6379 }], + requestTimeout: 2_000, + advancedConfiguration: { connectionTimeout: 2_000 }, }); const redisClient = createValkeyGlideDialCacheClient(glideClient); const dialcache = new DialCache({ @@ -153,6 +158,16 @@ The application owns the complete Redis lifecycle. It creates and connects the u 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. +### Async liveness contract + +DialCache fail-open behavior is rejection-driven: a Redis promise that never settles cannot fall through. Every `DialCacheRedisClient.read`, `write`, and `invalidate` promise must therefore resolve or reject within a finite application-defined budget covering connection establishment, reconnection, retries, offline queueing, dispatch, and response time. A connection timeout alone does not satisfy this contract. A pending read prevents fallback from starting, while a pending write withholds an already-computed fallback result and retains its process flight. + +Valkey GLIDE users should configure [`requestTimeout`](https://github.com/valkey-io/valkey-glide/blob/v2.4.2/node/src/BaseClient.ts#L770-L776) and [`advancedConfiguration.connectionTimeout`](https://github.com/valkey-io/valkey-glide/blob/v2.4.2/node/src/BaseClient.ts#L957-L964), as shown above. GLIDE applies the request timeout while sending, waiting for a response, reconnecting, and retrying. This bounds client waiting; it is not server-side cancellation, so a write dispatched before timeout may still execute. + +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 same finite-settlement responsibility applies to async `cacheConfigProvider`, `rampSampler`, and custom `Serializer` methods. DialCache's fallback deadline 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. + `redis.createClient` and the `RedisClientFactory` type were removed. Move lazy factory work to application startup, retain the underlying client and adapter handles, and pass the resulting adapter as `redis.client`. Because `redis.client` was already supported and this change does not alter Redis keys or the wire protocol, migrated application construction works with both older and newer DialCache versions during rollout or rollback. When caching is enabled, reads flow through: @@ -321,6 +336,58 @@ const getUser = dialcache.cached( ); ``` +### Fallback deadlines + +Once an initially enabled invocation starts its fallback, DialCache waits at most 60 seconds by default. Set `fallbackTimeoutMs` once on a cached wrapper to choose a positive integer deadline in milliseconds, up to 2,147,483,647, or set it to `null` to preserve an intentionally unbounded fallback: + +```ts +import { FallbackTimeoutError } from "dialcache"; + +const getUser = dialcache.cached( + (userId: string) => db.fetchUser(userId), + { + keyType: "user_id", + useCase: "GetUserWithDeadline", + cacheKey: (userId) => userId, + defaultConfig: DialCacheKeyConfig.enabled(60), + fallbackTimeoutMs: 2_000, + }, +); + +try { + await dialcache.enable(() => getUser("123")); +} catch (error) { + if (error instanceof FallbackTimeoutError) { + logger.warn("source lookup exceeded its DialCache budget", { + useCase: error.useCase, + timeoutMs: error.timeoutMs, + }); + } +} +``` + +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`. + +Timing out rejects the DialCache chain and clears its flight normally. A later fallback resolution is ignored, so that timed-out invocation cannot proceed to serializer, Redis, or local-cache publication. The underlying function is not canceled and may continue its own I/O or side effects; give the source operation its own native timeout or `AbortSignal` whenever possible. `fallbackTimeoutMs: null` disables this guard and makes finite fallback settlement entirely application-owned. + +This default is a behavioral compatibility change for enabled calls whose fallback previously remained pending beyond 60 seconds. Use the `null` escape hatch only after intentionally accepting that liveness risk. Timeout failures retain the bounded metrics classification `error="fallback"` with `in_fallback="true"`; the typed error provides the timeout details without adding high-cardinality labels. + +### Coalescing state + +`getCoalescingState()` returns a detached, point-in-time snapshot of process-scoped flights owned by that `DialCache` instance: + +```ts +const state = dialcache.getCoalescingState(); + +state.process.activeLeaders; +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. + +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. + ## Enabled context Caching is **off by default** and only active inside a `dialcache.enable(...)` scope. This is deliberate: it lets you turn caching **off in write paths** so a stale read can't be cached around a write. DialCache uses Node `AsyncLocalStorage` to keep enabled state scoped to the current asynchronous call chain. @@ -457,7 +524,7 @@ The Prometheus adapter emits: | `dialcache_invalidation_counter` | Counter | `cache_namespace`, `key_type`, `layer` | Invalidation calls for the layers touched today | | `dialcache_coalesced_counter` | Counter | `cache_namespace`, `use_case`, `key_type`, `scope` | Coalesced requests split by `request_local` or `process` scope | | `dialcache_get_timer` | Histogram | `cache_namespace`, `use_case`, `key_type`, `layer` | Cache get latency in seconds | -| `dialcache_fallback_timer` | Histogram | `cache_namespace`, `use_case`, `key_type`, `layer` | Time spent in the underlying function | +| `dialcache_fallback_timer` | Histogram | `cache_namespace`, `use_case`, `key_type`, `layer` | Time DialCache waited for the underlying function, ending at its fallback deadline when applied | | `dialcache_serialization_timer` | Histogram | `cache_namespace`, `use_case`, `key_type`, `layer`, `operation` | Redis serializer dump/load latency | | `dialcache_size_histogram` | Histogram | `cache_namespace`, `use_case`, `key_type`, `layer` | Serialized Redis payload size in bytes | @@ -514,7 +581,7 @@ The Datadog adapter emits exact increments of `1` for counters and preserves sec | `dialcache.invalidation.count` | Count | `cache_namespace`, `key_type`, `layer` | Invalidation calls for the layers touched | | `dialcache.coalesced.count` | Count | `cache_namespace`, `use_case`, `key_type`, `scope` | Coalesced requests by sharing scope | | `dialcache.get.duration` | Distribution or histogram | `cache_namespace`, `use_case`, `key_type`, `layer` | Cache get latency in seconds | -| `dialcache.fallback.duration` | Distribution or histogram | `cache_namespace`, `use_case`, `key_type`, `layer` | Time spent in the underlying function in seconds | +| `dialcache.fallback.duration` | Distribution or histogram | `cache_namespace`, `use_case`, `key_type`, `layer` | Time DialCache waited for the underlying function, ending at its fallback deadline when applied | | `dialcache.serialization.duration` | Distribution or histogram | `cache_namespace`, `use_case`, `key_type`, `layer`, `operation` | Redis serializer dump/load latency in seconds | | `dialcache.serialization.size` | Distribution or histogram | `cache_namespace`, `use_case`, `key_type`, `layer` | Serialized Redis payload size in bytes | @@ -533,7 +600,7 @@ The `error` label reports where an operation failed rather than copying the thro | `serialization_load` | Deserializing a Redis payload failed | | `serialization_dump` | Serializing a value for Redis failed | | `invalidation` | Writing an invalidation watermark failed | -| `fallback` | The wrapped application function failed | +| `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 for existing dashboards. @@ -585,6 +652,8 @@ Included: - Dynamically extended watermark TTL with a configurable `DEFAULT_WATERMARK_TTL_SEC` floor - Future-buffer behavior that avoids cache writes during active invalidation windows - Request-scoped and instance-scoped coalescing for active cache work +- A 60-second enabled-fallback deadline with per-wrapper override and explicit disable +- Exact per-instance process-coalescing state inspection Not included yet: diff --git a/scripts/benchmark-request-local.mjs b/scripts/benchmark-request-local.mjs index b24d5aa..25c63a5 100644 --- a/scripts/benchmark-request-local.mjs +++ b/scripts/benchmark-request-local.mjs @@ -122,12 +122,21 @@ async function benchmarkProcessCoalescing(fanout) { ); await started.promise; await nextTurn(); + const activeState = dialcache.getCoalescingState().process; + assert.equal(activeState.activeLeaders, 1); + assert.equal(activeState.activeFollowers, fanout - 1); + assert.equal(typeof activeState.oldestLeaderAgeMs, "number"); gate.resolve(); const values = await valuesPromise; const elapsedMs = performance.now() - start; assert.deepEqual(new Set(values), new Set(["shared"])); assert.equal(fallbackCalls, 1, "process coalescing should execute the fallback once across enabled scopes"); + assert.deepEqual(dialcache.getCoalescingState().process, { + activeLeaders: 0, + activeFollowers: 0, + oldestLeaderAgeMs: null, + }); return { scenario: "process coalescing", operations: fanout, elapsedMs, fallbackCalls }; } diff --git a/scripts/test-package.mjs b/scripts/test-package.mjs index 5f94b94..d4f1b75 100644 --- a/scripts/test-package.mjs +++ b/scripts/test-package.mjs @@ -14,18 +14,21 @@ const rootConsumer = `import { DialCacheKey, DialCacheKeyConfig, DialCacheRedisProtocolError, + FallbackTimeoutError, JsonSerializer, type CacheMetricLabels, type CacheConfigProvider, type CachedOptions, type CoalescedMetricLabels, type CoalescingScope, + type CoalescingState, type DialCacheConfig, type DialCacheKeyInit, type DialCacheMetricsAdapter, type DialCacheRedisClient, type InvalidationMetricLabels, type MetricErrorKind, + type ProcessCoalescingState, type RedisConfig, type Serializer, } from "dialcache"; @@ -71,12 +74,22 @@ const datadogClassAdapter = new DatadogDialCacheMetrics(datadogOptions); 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 coalescingState: CoalescingState = cache.getCoalescingState(); +const processCoalescingState: ProcessCoalescingState = coalescingState.process; const load = cache.cached(async (id: string) => id, { keyType: "id", useCase: "Load", cacheKey: (id) => id, + fallbackTimeoutMs: 1_000, defaultConfig: DialCacheKeyConfig.enabled(60), }); +const loadWithoutFallbackDeadline = cache.cached(async (id: string) => id, { + keyType: "id", + useCase: "LoadWithoutFallbackDeadline", + cacheKey: (id) => id, + fallbackTimeoutMs: null, +}); interface JsonCompatibleRecord { readonly id: string; @@ -223,6 +236,7 @@ const rootHasNoPrometheusFactory: "createPrometheusDialCacheMetrics" extends key const rootHasNoDatadogFactory: "createDatadogDialCacheMetrics" extends keyof DialCacheRoot ? false : true = true; void load; +void loadWithoutFallbackDeadline; void loadJsonRecord; void loadEmptyObject; void loadUndefined; @@ -240,6 +254,9 @@ void keyInitHasNoUrnPrefix; void legacyKeyInit; void namespacedKey.namespace; void redisProtocolError.name; +void fallbackTimeoutError.timeoutMs; +void coalescingState.process; +void processCoalescingState.activeLeaders; void requestLocalCoalescingScope; void boundedErrorKind; void metricErrorKinds; @@ -381,6 +398,15 @@ try { const nodeRedis = await import("dialcache/node-redis"); await import("dialcache/datadog"); await import("dialcache/redis-protocol"); +const fallbackTimeoutError = new root.FallbackTimeoutError("PackageRuntime", 1000); +if (!(fallbackTimeoutError instanceof root.DialCacheError) || fallbackTimeoutError.timeoutMs !== 1000) { + throw new Error("The root ESM fallback-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)) { + throw new Error("The root ESM coalescing snapshot export is invalid"); +} try { nodeRedis.dialcacheRedisScripts.dialcacheWrite.transformReply(2); throw new Error("Expected an invalid node-redis script reply to fail"); @@ -400,6 +426,15 @@ try { const nodeRedis = require("dialcache/node-redis"); require("dialcache/datadog"); require("dialcache/redis-protocol"); +const fallbackTimeoutError = new root.FallbackTimeoutError("PackageRuntime", 1000); +if (!(fallbackTimeoutError instanceof root.DialCacheError) || fallbackTimeoutError.timeoutMs !== 1000) { + throw new Error("The root CommonJS fallback-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)) { + throw new Error("The root CommonJS coalescing snapshot export is invalid"); +} try { nodeRedis.dialcacheRedisScripts.dialcacheWrite.transformReply(2); throw new Error("Expected an invalid node-redis script reply to fail"); diff --git a/src/config.ts b/src/config.ts index 0f9fdb2..8ebf01d 100644 --- a/src/config.ts +++ b/src/config.ts @@ -16,6 +16,10 @@ export interface CacheRampSample { readonly ramp: number; } +/** + * Selects a rollout sample. Async implementations must settle within a finite + * application-defined deadline; DialCache does not add one. + */ export type CacheRampSampler = (sample: CacheRampSample) => Awaitable; export const deterministicRampSampler: CacheRampSampler = ({ key, layer }) => stablePercent(`${key.urn}:${layer}`); @@ -53,6 +57,10 @@ export class DialCacheKeyConfig { } } +/** + * Resolves runtime cache policy. Async implementations must settle within a + * finite application-defined deadline; DialCache does not add one. + */ export type CacheConfigProvider = (key: DialCacheKey) => Awaitable; export type Logger = Pick; diff --git a/src/dialcache.ts b/src/dialcache.ts index 8ee1f13..650db90 100644 --- a/src/dialcache.ts +++ b/src/dialcache.ts @@ -11,14 +11,13 @@ import { type Logger, } from "./config.js"; import { DialCacheContext, getOrCreateRequestLocalCache, type RequestLocalCache } from "./context.js"; -import { UseCaseIsAlreadyRegisteredError, UseCaseNameIsReservedError } from "./errors.js"; +import { FallbackTimeoutError, UseCaseIsAlreadyRegisteredError, UseCaseNameIsReservedError } from "./errors.js"; import { DialCacheKey, assertValidNamespace, normalizeArgs } from "./key.js"; import { NO_CACHE_LAYER, REQUEST_LOCAL_CACHE_LAYER, labelsFor, type CacheMetricLabels, - type CoalescingScope, type DialCacheMetricsAdapter, type MetricErrorKind, type MetricLayer, @@ -105,6 +104,16 @@ interface CachedOptionsBase { readonly useCase: string; readonly defaultConfig?: DialCacheKeyConfig | null; readonly trackForInvalidation?: boolean; + /** + * Maximum time to wait once an initially enabled invocation starts its + * fallback, in milliseconds. Must be at most 2,147,483,647. Defaults to 60 + * seconds. Set to `null` to disable the deadline. + * + * Concurrent same-key callers share the leader's remaining budget. Timing + * out rejects those callers and prevents the eventual result from being + * published by DialCache, but does not cancel the underlying operation. + */ + readonly fallbackTimeoutMs?: number | null; /** * Select every input dimension that can affect the returned value. Concurrent * enabled calls with the same cache key may share one in-flight execution. @@ -130,7 +139,27 @@ export type CachedOptions = CachedOptionsBase & Serializer */ export type CachedFn = (...args: Parameters) => Promise>; +/** Exact process-scoped single-flight state for one DialCache instance. */ +export interface ProcessCoalescingState { + readonly activeLeaders: number; + readonly activeFollowers: number; + readonly oldestLeaderAgeMs: number | null; +} + +/** A point-in-time snapshot of DialCache-owned coalescing state. */ +export interface CoalescingState { + readonly process: ProcessCoalescingState; +} + +interface ProcessFlight { + readonly promise: Promise; + readonly startedAtMs: number; + followers: number; +} + 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; @@ -144,7 +173,8 @@ export class DialCache { private readonly rampSampler: CacheRampSampler; private readonly redisCache: RedisCache | null; private readonly metrics: DialCacheMetricsAdapter | null; - private readonly inFlight = new Map>(); + private readonly processFlights = new Map(); + private activeProcessFollowers = 0; constructor(config: DialCacheConfig = {}) { if (Object.hasOwn(config, "urnPrefix")) { @@ -196,16 +226,30 @@ export class DialCache { return this.context.isEnabled(); } + /** Returns exact process-scoped single-flight state for this instance. */ + getCoalescingState(): CoalescingState { + const oldestFlight = this.processFlights.values().next().value as ProcessFlight | undefined; + return { + process: { + activeLeaders: this.processFlights.size, + activeFollowers: this.activeProcessFollowers, + oldestLeaderAgeMs: + oldestFlight === undefined ? null : Math.max(performance.now() - oldestFlight.startedAtMs, 0), + }, + }; + } + /** * Wraps a function with the configured cache chain. Returned in-memory * values are shared by reference and must be treated as immutable. */ cached(fn: Fn, options: CachedOptions): CachedFn { + const fallbackTimeoutMs = resolveFallbackTimeoutMs(options.fallbackTimeoutMs); this.registerUseCase(options.useCase); const run = async (...args: Parameters): Promise> => { // `fn`'s awaited result is the cached value by construction; the generic `Fn` erases it to `unknown`. - const fallback = async (): Promise> => (await fn(...args)) as CachedValue; + const rawFallback = async (): Promise> => (await fn(...args)) as CachedValue; const noLayerLabels = { cacheNamespace: this.namespace, useCase: options.useCase, @@ -215,9 +259,12 @@ export class DialCache { if (!this.isEnabled()) { this.metrics?.disabled({ ...noLayerLabels, reason: "context" }); - return await fallback(); + return await rawFallback(); } + const fallback = (): Promise> => + withFallbackTimeout(rawFallback, options.useCase, fallbackTimeoutMs); + let key: DialCacheKey; try { key = this.buildKey(options, options.cacheKey(...args)); @@ -323,7 +370,7 @@ export class DialCache { keyConfig: DialCacheKeyConfig, fallback: () => Promise, ): Promise { - return await this.singleFlight(requestLocalCache.inFlight, key, "request_local", async () => { + return await this.singleFlightRequestLocal(requestLocalCache.inFlight, key, async () => { const start = performance.now(); const result = requestLocalCache.read(key.urn); this.metrics?.request(labelsFor(key, REQUEST_LOCAL_CACHE_LAYER)); @@ -347,7 +394,7 @@ export class DialCache { ): Promise { const localLayer = await this.resolveLocalLayerConfig(key, keyConfig); if (localLayer.status === "enabled") { - return await this.singleFlight(this.inFlight, key, "process", async () => + return await this.singleFlightProcess(key, async () => await this.getThroughActiveLocal(key, keyConfig, localLayer.config, fallback), ); } @@ -363,7 +410,7 @@ export class DialCache { return await this.callFallback(labelsFor(key, fallbackLayer), fallback); } - return await this.singleFlight(this.inFlight, key, "process", async () => { + return await this.singleFlightProcess(key, async () => { const remote = await this.readRemoteWithResolvedConfig(redisCache, key, remoteLayer.config); return await this.finishRedisChain(redisCache, key, localLayer, remote, fallback, remoteLayer.config); }); @@ -562,10 +609,9 @@ export class DialCache { }); } - private singleFlight( + private singleFlightRequestLocal( inFlight: Map>, key: DialCacheKey, - scope: CoalescingScope, run: () => Promise, ): Promise { const existing = inFlight.get(key.urn); @@ -574,7 +620,7 @@ export class DialCache { cacheNamespace: key.namespace, useCase: key.useCase, keyType: key.keyType, - scope, + scope: "request_local", }); return existing as Promise; } @@ -589,6 +635,37 @@ export class DialCache { void promise.then(clear, clear); return promise; } + + private singleFlightProcess(key: DialCacheKey, run: () => Promise): Promise { + const existing = this.processFlights.get(key.urn); + if (existing !== undefined) { + existing.followers += 1; + this.activeProcessFollowers += 1; + this.metrics?.coalesced?.({ + cacheNamespace: key.namespace, + useCase: key.useCase, + keyType: key.keyType, + scope: "process", + }); + return existing.promise as Promise; + } + + const promise = run(); + const flight: ProcessFlight = { + promise, + startedAtMs: performance.now(), + followers: 0, + }; + this.processFlights.set(key.urn, flight); + const clear = (): void => { + if (this.processFlights.get(key.urn) === flight) { + this.activeProcessFollowers -= flight.followers; + this.processFlights.delete(key.urn); + } + }; + void promise.then(clear, clear); + return promise; + } } function elapsedSeconds(startMs: number): number { @@ -601,6 +678,81 @@ function assertValidFutureBufferMs(futureBufferMs: number): void { } } +function resolveFallbackTimeoutMs(value: number | null | undefined): number | null { + if (value === null) { + return null; + } + + const timeoutMs = value ?? DEFAULT_FALLBACK_TIMEOUT_MS; + if (!Number.isSafeInteger(timeoutMs) || timeoutMs <= 0 || timeoutMs > MAX_TIMER_DELAY_MS) { + throw new RangeError( + `DialCache fallbackTimeoutMs must be null or a positive safe integer no greater than ${MAX_TIMER_DELAY_MS}`, + ); + } + return timeoutMs; +} + +function withFallbackTimeout( + fallback: () => Promise, + useCase: string, + timeoutMs: number | null, +): Promise { + 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 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(rejectTimeout, remainingMs); + timer.unref(); + } + + void operation.then( + (value) => { + settleBeforeDeadline(() => resolve(value)); + }, + (error: unknown) => { + settleBeforeDeadline(() => reject(error)); + }, + ); + }); +} + function safeLogger(logger: Logger): Logger { return { debug: (...args: Parameters) => callLogger(() => logger.debug(...args)), diff --git a/src/errors.ts b/src/errors.ts index 2c422f7..ed748b7 100644 --- a/src/errors.ts +++ b/src/errors.ts @@ -5,6 +5,17 @@ export class DialCacheError extends Error { } } +export class FallbackTimeoutError extends DialCacheError { + readonly timeoutMs: number; + readonly useCase: string; + + constructor(useCase: string, timeoutMs: number) { + super(`DialCache fallback 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 c7bfe4a..f66473b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -16,12 +16,20 @@ export type { } from "./metrics.js"; export { DialCacheError, + FallbackTimeoutError, MissingKeyConfigError, UseCaseIsAlreadyRegisteredError, UseCaseNameIsReservedError, } from "./errors.js"; export { DialCache } from "./dialcache.js"; -export type { CacheKeySpec, CachedFn, CachedOptions, CachedValue } from "./dialcache.js"; +export type { + CacheKeySpec, + CachedFn, + CachedOptions, + CachedValue, + CoalescingState, + ProcessCoalescingState, +} from "./dialcache.js"; export { DialCacheKey, invalidationPrefix, normalizeArgs, redisClusterHashTag } from "./key.js"; export type { DialCacheKeyInit } from "./key.js"; export { diff --git a/src/internal/redis-cache.ts b/src/internal/redis-cache.ts index bcdb262..bb1bf72 100644 --- a/src/internal/redis-cache.ts +++ b/src/internal/redis-cache.ts @@ -11,7 +11,9 @@ import { fetchKeyConfig, resolveLayerConfigResult, type ResolvedLayerConfig } fr export interface RedisConfig { /** * Caller-created, connected, and lifecycle-owned semantic Redis client. - * DialCache borrows it and never drains, disposes, or closes it. + * 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. */ readonly client: DialCacheRedisClient; readonly serializer?: Serializer; diff --git a/src/node-redis.ts b/src/node-redis.ts index 605e3f7..b84d95f 100644 --- a/src/node-redis.ts +++ b/src/node-redis.ts @@ -151,7 +151,10 @@ interface NodeRedisScriptClient { /** * Create a resource-free semantic view over a caller-owned node-redis client. - * The caller remains responsible for draining work and closing the 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. */ export function createNodeRedisDialCacheClient(client: NodeRedisScriptClient): DialCacheRedisClient { return { diff --git a/src/prometheus.ts b/src/prometheus.ts index b6813e2..7114312 100644 --- a/src/prometheus.ts +++ b/src/prometheus.ts @@ -198,7 +198,7 @@ function collectorConfigs(prefix: string) { fallbackTimer: { type: "histogram", name: `${prefix}dialcache_fallback_timer`, - help: "Time spent in the underlying fallback function in seconds.", + help: "Time DialCache waited for the fallback function in seconds.", labelNames: ["cache_namespace", "use_case", "key_type", "layer"], buckets: TIMER_BUCKETS, }, diff --git a/src/redis-client.ts b/src/redis-client.ts index 4212e2a..aa565ed 100644 --- a/src/redis-client.ts +++ b/src/redis-client.ts @@ -75,6 +75,12 @@ 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. */ export interface DialCacheRedisClient { /** Atomically read and validate a value against its watermark when tracked. */ diff --git a/src/serializer.ts b/src/serializer.ts index 934bbf2..1be0827 100644 --- a/src/serializer.ts +++ b/src/serializer.ts @@ -1,7 +1,9 @@ import type { Awaitable } from "./config.js"; export interface Serializer { + /** Async implementations must settle within an application-defined deadline. */ dump(value: T): Awaitable; + /** Async implementations must settle within an application-defined deadline. */ load(value: string | Buffer): Awaitable; } diff --git a/src/valkey-glide.ts b/src/valkey-glide.ts index 965cdde..97fc6c4 100644 --- a/src/valkey-glide.ts +++ b/src/valkey-glide.ts @@ -38,7 +38,9 @@ export interface ValkeyGlideDialCacheClient extends DialCacheRedisClient { /** * Wrap a caller-owned GLIDE connection. The returned adapter owns only its - * Script handles; callers dispose those after draining work, then close GLIDE. + * Script handles and preserves the connection's `requestTimeout`; callers + * dispose the handles after draining work, then close GLIDE. A request timeout + * bounds client waiting but is not server-side command cancellation. */ export function createValkeyGlideDialCacheClient( client: SupportedValkeyGlideClient, diff --git a/test/dialcache-liveness.test.ts b/test/dialcache-liveness.test.ts new file mode 100644 index 0000000..21e9261 --- /dev/null +++ b/test/dialcache-liveness.test.ts @@ -0,0 +1,566 @@ +import { performance } from "node:perf_hooks"; + +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { + CacheLayer, + DialCache, + DialCacheError, + DialCacheKeyConfig, + FallbackTimeoutError, + type CachedOptions, + type DialCacheMetricsAdapter, +} from "../src/index.js"; +import { FakeRedis } from "./fake-redis.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 localConfig = new DialCacheKeyConfig({ + ttlSec: { [CacheLayer.LOCAL]: 60 }, + ramp: { [CacheLayer.LOCAL]: 100 }, +}); + +const remoteConfig = new DialCacheKeyConfig({ + ttlSec: { [CacheLayer.REMOTE]: 60 }, + ramp: { [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(), + }; +} + +describe("DialCache fallback liveness", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it("defaults to 60 seconds and shares the leader deadline with process followers", async () => { + let nowMs = 0; + vi.spyOn(performance, "now").mockImplementation(() => nowMs); + const gate = deferred<{ readonly id: string }>(); + const started = deferred(); + const error = vi.fn(); + const dialcache = new DialCache({ metrics: metricsWithError(error) }); + let calls = 0; + const getUser = dialcache.cached(async (id: string) => { + calls += 1; + started.resolve(); + return await gate.promise; + }, { + keyType: "user_id", + useCase: "DefaultFallbackDeadline", + cacheKey: (id) => id, + defaultConfig: localConfig, + }); + + const result = dialcache.enable(async () => + await Promise.allSettled([getUser("123"), getUser("123"), getUser("123")]), + ); + await started.promise; + + expect(calls).toBe(1); + expect(dialcache.getCoalescingState().process).toMatchObject({ + activeLeaders: 1, + activeFollowers: 2, + }); + expect(dialcache.getCoalescingState().process.oldestLeaderAgeMs).toBeGreaterThanOrEqual(0); + + nowMs = 59_999; + await vi.advanceTimersByTimeAsync(59_999); + expect(dialcache.getCoalescingState().process.activeLeaders).toBe(1); + + nowMs = 60_000; + await vi.advanceTimersByTimeAsync(1); + const settled = await result; + expect(settled).toHaveLength(3); + const reasons = settled.map((entry) => entry.status === "rejected" ? entry.reason : undefined); + expect(reasons[0]).toBeInstanceOf(FallbackTimeoutError); + expect(reasons[1]).toBe(reasons[0]); + expect(reasons[2]).toBe(reasons[0]); + expect(reasons[0]).toMatchObject({ + useCase: "DefaultFallbackDeadline", + timeoutMs: 60_000, + }); + expect(reasons[0]).toBeInstanceOf(DialCacheError); + expect(error).toHaveBeenCalledTimes(1); + expect(error).toHaveBeenCalledWith({ + cacheNamespace: "urn", + useCase: "DefaultFallbackDeadline", + keyType: "user_id", + layer: CacheLayer.LOCAL, + error: "fallback", + inFallback: true, + }); + expect(dialcache.getCoalescingState().process).toEqual({ + activeLeaders: 0, + activeFollowers: 0, + oldestLeaderAgeMs: null, + }); + }); + + it("uses a caller-selected timeout on enabled pass-through calls", async () => { + const started = deferred(); + const gate = deferred(); + const dialcache = new DialCache(); + const load = dialcache.cached(async () => { + started.resolve(); + return await gate.promise; + }, { + keyType: "id", + useCase: "PassThroughFallbackDeadline", + cacheKey: () => "123", + fallbackTimeoutMs: 25, + }); + + const result = Promise.allSettled([dialcache.enable(async () => await load())]); + await started.promise; + await vi.advanceTimersByTimeAsync(24); + expect(dialcache.getCoalescingState().process.activeLeaders).toBe(0); + + await vi.advanceTimersByTimeAsync(1); + const [settled] = await result; + expect(settled).toEqual({ status: "rejected", reason: expect.any(FallbackTimeoutError) }); + expect(settled?.status === "rejected" ? settled.reason.timeoutMs : undefined).toBe(25); + }); + + it("gives a late follower only the leader's remaining budget", async () => { + let nowMs = 0; + vi.spyOn(performance, "now").mockImplementation(() => nowMs); + const started = deferred(); + const dialcache = new DialCache(); + let calls = 0; + const load = dialcache.cached(async () => { + calls += 1; + started.resolve(); + return await new Promise(() => undefined); + }, { + keyType: "id", + useCase: "FollowerInheritsDeadline", + cacheKey: () => "123", + fallbackTimeoutMs: 1_000, + defaultConfig: localConfig, + }); + + const leader = dialcache.enable(async () => await load()); + await started.promise; + nowMs = 900; + await vi.advanceTimersByTimeAsync(900); + const follower = dialcache.enable(async () => await load()); + await vi.advanceTimersByTimeAsync(0); + const result = Promise.allSettled([leader, follower]); + + expect(calls).toBe(1); + expect(dialcache.getCoalescingState().process.activeFollowers).toBe(1); + nowMs = 999; + await vi.advanceTimersByTimeAsync(99); + expect(dialcache.getCoalescingState().process.activeLeaders).toBe(1); + + nowMs = 1_000; + await vi.advanceTimersByTimeAsync(1); + const settled = await result; + expect( + settled.every((entry) => entry.status === "rejected" && entry.reason instanceof FallbackTimeoutError), + ).toBe(true); + expect(dialcache.getCoalescingState().process.activeLeaders).toBe(0); + }); + + it("shares one deadline across request-local followers and permits a same-scope retry", async () => { + const firstStarted = deferred(); + const dialcache = new DialCache(); + let calls = 0; + const load = dialcache.cached(async () => { + calls += 1; + if (calls === 1) { + firstStarted.resolve(); + return await new Promise(() => undefined); + } + return "retry"; + }, { + keyType: "id", + useCase: "RequestLocalFallbackDeadline", + cacheKey: () => "123", + fallbackTimeoutMs: 10, + defaultConfig: new DialCacheKeyConfig({ requestLocal: true }), + }); + + const result = dialcache.enable(async () => { + const settled = await Promise.allSettled([load(), load(), load()]); + const retry = await load(); + return { settled, retry }; + }); + await firstStarted.promise; + expect(dialcache.getCoalescingState().process.activeLeaders).toBe(0); + + await vi.advanceTimersByTimeAsync(10); + const { settled, retry } = await result; + const reasons = settled.map((entry) => entry.status === "rejected" ? entry.reason : undefined); + expect(calls).toBe(2); + expect(reasons[0]).toBeInstanceOf(FallbackTimeoutError); + expect(reasons[1]).toBe(reasons[0]); + expect(reasons[2]).toBe(reasons[0]); + expect(retry).toBe("retry"); + }); + + it("counts synchronous pre-await fallback work against the deadline", async () => { + let nowMs = 1_000; + vi.spyOn(performance, "now").mockImplementation(() => nowMs); + const dialcache = new DialCache(); + const load = dialcache.cached(async () => { + nowMs += 11; + return "late"; + }, { + keyType: "id", + useCase: "SynchronousFallbackDeadline", + cacheKey: () => "123", + fallbackTimeoutMs: 10, + defaultConfig: localConfig, + }); + + await expect(dialcache.enable(async () => await load())).rejects.toMatchObject({ + useCase: "SynchronousFallbackDeadline", + timeoutMs: 10, + }); + expect(vi.getTimerCount()).toBe(0); + }); + + it("keeps initially disabled calls as true pass-through", async () => { + const started = deferred(); + const gate = deferred(); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const dialcache = new DialCache(); + const load = dialcache.cached(async () => { + started.resolve(); + return await gate.promise; + }, { + keyType: "id", + useCase: "DisabledContextHasNoDeadline", + cacheKey: () => { + throw new Error("cacheKey must not run"); + }, + fallbackTimeoutMs: 1, + defaultConfig: localConfig, + }); + + const result = load(); + await started.promise; + await vi.advanceTimersByTimeAsync(10_000); + + expect(setTimeoutSpy).not.toHaveBeenCalled(); + gate.resolve("value"); + await expect(result).resolves.toBe("value"); + }); + + it("allows the fallback deadline to be explicitly disabled", async () => { + const started = deferred(); + const gate = deferred(); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const dialcache = new DialCache(); + const load = dialcache.cached(async () => { + started.resolve(); + return await gate.promise; + }, { + keyType: "id", + useCase: "DisabledFallbackDeadline", + cacheKey: () => "123", + fallbackTimeoutMs: null, + defaultConfig: localConfig, + }); + + const result = dialcache.enable(async () => await load()); + await started.promise; + await vi.advanceTimersByTimeAsync(60_001); + + expect(setTimeoutSpy).not.toHaveBeenCalled(); + expect(dialcache.getCoalescingState().process.activeLeaders).toBe(1); + gate.resolve("value"); + await expect(result).resolves.toBe("value"); + expect(dialcache.getCoalescingState().process.activeLeaders).toBe(0); + }); + + it("clears a successful fallback timer and creates no timer on a cache hit", async () => { + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const dialcache = new DialCache(); + let calls = 0; + const load = dialcache.cached(async () => ++calls, { + keyType: "id", + useCase: "FallbackTimerCleanup", + cacheKey: () => "123", + fallbackTimeoutMs: 1_000, + defaultConfig: localConfig, + }); + + await expect(dialcache.enable(async () => await load())).resolves.toBe(1); + expect(setTimeoutSpy).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + + await expect(dialcache.enable(async () => await load())).resolves.toBe(1); + expect(setTimeoutSpy).toHaveBeenCalledTimes(1); + }); + + it("does not publish a fallback result that resolves after its deadline", async () => { + const firstGate = deferred<{ readonly id: string; readonly version: number }>(); + const firstStarted = deferred(); + const redis = new FakeRedis(); + const dialcache = new DialCache({ redis: { client: redis } }); + let calls = 0; + const getUser = dialcache.cached(async (id: string) => { + calls += 1; + if (calls === 1) { + firstStarted.resolve(); + return await firstGate.promise; + } + return { id, version: 2 }; + }, { + keyType: "user_id", + useCase: "IgnoreLateFallbackResult", + cacheKey: (id) => id, + fallbackTimeoutMs: 10, + defaultConfig: remoteConfig, + }); + + const first = Promise.allSettled([dialcache.enable(async () => await getUser("123"))]); + await firstStarted.promise; + await vi.advanceTimersByTimeAsync(10); + expect((await first)[0]).toEqual({ status: "rejected", reason: expect.any(FallbackTimeoutError) }); + expect(redis.setCalls).toBe(0); + + await expect(dialcache.enable(async () => await getUser("123"))).resolves.toEqual({ id: "123", version: 2 }); + expect(redis.setCalls).toBe(1); + + firstGate.resolve({ id: "123", version: 1 }); + await vi.advanceTimersByTimeAsync(0); + expect(redis.setCalls).toBe(1); + await expect(dialcache.enable(async () => await getUser("123"))).resolves.toEqual({ id: "123", version: 2 }); + expect(calls).toBe(2); + }); + + it("consumes a late fallback rejection without recording a second error", async () => { + const gate = deferred(); + const started = deferred(); + const error = vi.fn(); + const dialcache = new DialCache({ metrics: metricsWithError(error) }); + const load = dialcache.cached(async () => { + started.resolve(); + return await gate.promise; + }, { + keyType: "id", + useCase: "IgnoreLateFallbackError", + cacheKey: () => "123", + fallbackTimeoutMs: 10, + defaultConfig: localConfig, + }); + + const result = Promise.allSettled([dialcache.enable(async () => await load())]); + await started.promise; + await vi.advanceTimersByTimeAsync(10); + await result; + expect(error).toHaveBeenCalledTimes(1); + + gate.reject(new Error("late source failure")); + await vi.advanceTimersByTimeAsync(0); + expect(error).toHaveBeenCalledTimes(1); + }); + + it.each([0, -1, 1.5, Number.NaN, Number.POSITIVE_INFINITY, 2_147_483_648])( + "rejects invalid fallbackTimeoutMs value %s without reserving the use case", + (fallbackTimeoutMs) => { + const dialcache = new DialCache(); + const options: CachedOptions<() => Promise> = { + keyType: "id", + useCase: "InvalidFallbackDeadline", + cacheKey: () => "123", + fallbackTimeoutMs, + }; + + expect(() => dialcache.cached(async () => "value", options)).toThrow( + new RangeError( + "DialCache fallbackTimeoutMs must be null or a positive safe integer no greater than 2147483647", + ), + ); + expect(() => dialcache.cached(async () => "value", { ...options, fallbackTimeoutMs: 1 })).not.toThrow(); + }, + ); + + it("rejects nonnumeric fallbackTimeoutMs from untyped callers", () => { + const dialcache = new DialCache(); + const options = { + keyType: "id", + useCase: "UntypedFallbackDeadline", + cacheKey: () => "123", + fallbackTimeoutMs: "1000", + } as unknown as CachedOptions<() => Promise>; + + expect(() => dialcache.cached(async () => "value", options)).toThrow(RangeError); + }); +}); + +describe("DialCache coalescing state", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it("reports leaders, followers, oldest age, and settlement cleanup", async () => { + let nowMs = 1_000; + vi.spyOn(performance, "now").mockImplementation(() => nowMs); + const firstGate = deferred(); + const secondGate = deferred(); + const firstStarted = deferred(); + const secondStarted = deferred(); + const dialcache = new DialCache(); + const load = dialcache.cached(async (id: string) => { + if (id === "first") { + firstStarted.resolve(); + return await firstGate.promise; + } + secondStarted.resolve(); + return await secondGate.promise; + }, { + keyType: "id", + useCase: "InspectProcessFlights", + cacheKey: (id) => id, + defaultConfig: localConfig, + }); + + expect(dialcache.getCoalescingState()).toEqual({ + process: { activeLeaders: 0, activeFollowers: 0, oldestLeaderAgeMs: null }, + }); + + const firstLeader = dialcache.enable(async () => await load("first")); + const firstFollower = dialcache.enable(async () => await load("first")); + const secondFollower = dialcache.enable(async () => await load("first")); + await firstStarted.promise; + nowMs = 1_040; + const secondLeader = dialcache.enable(async () => await load("second")); + await secondStarted.promise; + nowMs = 1_100; + + expect(dialcache.getCoalescingState()).toEqual({ + process: { activeLeaders: 2, activeFollowers: 2, oldestLeaderAgeMs: 100 }, + }); + + firstGate.resolve("first-value"); + await expect(Promise.all([firstLeader, firstFollower, secondFollower])).resolves.toEqual([ + "first-value", + "first-value", + "first-value", + ]); + expect(dialcache.getCoalescingState()).toEqual({ + process: { activeLeaders: 1, activeFollowers: 0, oldestLeaderAgeMs: 60 }, + }); + + secondGate.reject(new Error("second failed")); + await expect(secondLeader).rejects.toThrow("second failed"); + expect(dialcache.getCoalescingState()).toEqual({ + process: { activeLeaders: 0, activeFollowers: 0, oldestLeaderAgeMs: null }, + }); + }); + + it("keeps process state isolated by instance and excludes request-local-only work", async () => { + const processGate = deferred(); + const processStarted = deferred(); + const requestGate = deferred(); + const requestStarted = deferred(); + const first = new DialCache(); + const second = new DialCache(); + const processLoad = first.cached(async () => { + processStarted.resolve(); + return await processGate.promise; + }, { + keyType: "id", + useCase: "InstanceProcessState", + cacheKey: () => "123", + defaultConfig: localConfig, + }); + const requestLoad = second.cached(async () => { + requestStarted.resolve(); + return await requestGate.promise; + }, { + keyType: "id", + useCase: "RequestLocalStateExcluded", + cacheKey: () => "123", + defaultConfig: new DialCacheKeyConfig({ requestLocal: true }), + }); + + const processResult = first.enable(async () => await processLoad()); + const requestResult = second.enable(async () => await Promise.all([requestLoad(), requestLoad()])); + await Promise.all([processStarted.promise, requestStarted.promise]); + + expect(first.getCoalescingState().process.activeLeaders).toBe(1); + expect(second.getCoalescingState()).toEqual({ + process: { activeLeaders: 0, activeFollowers: 0, oldestLeaderAgeMs: null }, + }); + + processGate.resolve("process"); + requestGate.resolve("request"); + await expect(processResult).resolves.toBe("process"); + await expect(requestResult).resolves.toEqual(["request", "request"]); + }); + + it("releases a burst of unique-key flights at their fallback deadlines", async () => { + const dialcache = new DialCache(); + let started = 0; + const allStarted = deferred(); + const load = dialcache.cached(async (id: string) => { + started += 1; + if (started === 32) { + allStarted.resolve(); + } + return await new Promise(() => undefined); + }, { + keyType: "id", + useCase: "ReleaseUniqueKeyBurst", + cacheKey: (id) => id, + fallbackTimeoutMs: 5, + defaultConfig: localConfig, + }); + + const result = Promise.allSettled( + Array.from({ length: 32 }, (_, index) => dialcache.enable(async () => await load(String(index)))), + ); + await allStarted.promise; + expect(dialcache.getCoalescingState().process.activeLeaders).toBe(32); + + await vi.advanceTimersByTimeAsync(5); + const settled = await result; + expect( + settled.every((entry) => entry.status === "rejected" && entry.reason instanceof FallbackTimeoutError), + ).toBe(true); + expect(dialcache.getCoalescingState()).toEqual({ + process: { activeLeaders: 0, activeFollowers: 0, oldestLeaderAgeMs: null }, + }); + }); +}); From a08f507a01153960ff2468e3e741821da31cbfd5 Mon Sep 17 00:00:00 2001 From: Lev Neiman Date: Sun, 19 Jul 2026 16:31:55 -0700 Subject: [PATCH 2/4] fix: harden fallback deadline scheduling --- README.md | 2 +- scripts/benchmark-request-local.mjs | 67 +++++++++ src/dialcache.ts | 19 ++- test/dialcache-liveness.test.ts | 220 ++++++++++++++++++++++++++-- 4 files changed, 294 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index 7953420..df3935e 100644 --- a/README.md +++ b/README.md @@ -320,7 +320,7 @@ DialCache coalesces in-flight work at the lifetime of the first active cache lay 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. -Coalescing only applies when at least one cache layer is active. Calls outside `enable()` are true pass-through, and calls where request-local, process-local, and Redis are all disabled are true pass-through. +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. Because coalescing is keyed by `cacheKey`, concurrent calls with the same key share the leader's execution. Any argument ignored by `cacheKey` must be safe to share this way; include inputs such as locale, auth context, or cancellation behavior in the key when they can change the returned value or whether the underlying function should run separately. diff --git a/scripts/benchmark-request-local.mjs b/scripts/benchmark-request-local.mjs index 25c63a5..f89be83 100644 --- a/scripts/benchmark-request-local.mjs +++ b/scripts/benchmark-request-local.mjs @@ -9,6 +9,8 @@ const coalescingFanout = readPositiveInteger("DIALCACHE_BENCH_FANOUT", 1_000); const results = [ await benchmarkSequentialRequestLocalHits(sequentialIterations), + await benchmarkSequentialProcessLocalHits(sequentialIterations), + await benchmarkEnabledFallbacks(sequentialIterations), await benchmarkRequestLocalCoalescing(coalescingFanout), await benchmarkProcessCoalescing(coalescingFanout), ]; @@ -57,6 +59,71 @@ async function benchmarkSequentialRequestLocalHits(iterations) { return { scenario: "request-local sequential hits", operations: iterations, elapsedMs, fallbackCalls }; } +async function benchmarkSequentialProcessLocalHits(iterations) { + const dialcache = new DialCache(); + let fallbackCalls = 0; + const getValue = dialcache.cached( + async (id) => { + fallbackCalls += 1; + return { id }; + }, + { + keyType: "benchmark_id", + useCase: "BenchmarkProcessLocalSequential", + cacheKey: (id) => id, + defaultConfig: new DialCacheKeyConfig({ + ttlSec: { [CacheLayer.LOCAL]: 60 }, + ramp: { [CacheLayer.LOCAL]: 100 }, + }), + }, + ); + + let elapsedMs = 0; + await dialcache.enable(async () => { + const expected = await getValue("shared"); + let actual = expected; + const start = performance.now(); + for (let index = 0; index < iterations; index += 1) { + actual = await getValue("shared"); + } + elapsedMs = performance.now() - start; + assert.strictEqual(actual, expected); + }); + + assert.equal(fallbackCalls, 1, "process-local hits should reuse the first fallback value"); + return { scenario: "process-local sequential hits", operations: iterations, elapsedMs, fallbackCalls }; +} + +async function benchmarkEnabledFallbacks(iterations) { + const dialcache = new DialCache(); + let fallbackCalls = 0; + const getValue = dialcache.cached( + async (id) => { + fallbackCalls += 1; + return id; + }, + { + keyType: "benchmark_id", + useCase: "BenchmarkEnabledFallbacks", + cacheKey: (id) => id, + }, + ); + + let elapsedMs = 0; + await dialcache.enable(async () => { + await getValue("warmup"); + fallbackCalls = 0; + const start = performance.now(); + for (let index = 0; index < iterations; index += 1) { + await getValue("shared"); + } + elapsedMs = performance.now() - start; + }); + + assert.equal(fallbackCalls, iterations, "enabled uncached calls should each run a bounded fallback"); + return { scenario: "enabled bounded fallbacks", operations: iterations, elapsedMs, fallbackCalls }; +} + async function benchmarkRequestLocalCoalescing(fanout) { const dialcache = new DialCache(); const gate = deferred(); diff --git a/src/dialcache.ts b/src/dialcache.ts index 650db90..e1e9552 100644 --- a/src/dialcache.ts +++ b/src/dialcache.ts @@ -650,10 +650,11 @@ export class DialCache { return existing.promise as Promise; } + const startedAtMs = performance.now(); const promise = run(); const flight: ProcessFlight = { promise, - startedAtMs: performance.now(), + startedAtMs, followers: 0, }; this.processFlights.set(key.urn, flight); @@ -721,6 +722,19 @@ function withFallbackTimeout( 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; @@ -738,8 +752,7 @@ function withFallbackTimeout( if (remainingMs <= 0) { rejectTimeout(); } else { - timer = setTimeout(rejectTimeout, remainingMs); - timer.unref(); + timer = setTimeout(onTimer, Math.ceil(remainingMs)); } void operation.then( diff --git a/test/dialcache-liveness.test.ts b/test/dialcache-liveness.test.ts index 21e9261..ba7ccd9 100644 --- a/test/dialcache-liveness.test.ts +++ b/test/dialcache-liveness.test.ts @@ -39,7 +39,10 @@ const remoteConfig = new DialCacheKeyConfig({ ramp: { [CacheLayer.REMOTE]: 100 }, }); -function metricsWithError(error: DialCacheMetricsAdapter["error"]): DialCacheMetricsAdapter { +function metricsWithError( + error: DialCacheMetricsAdapter["error"], + observeFallback: DialCacheMetricsAdapter["observeFallback"] = vi.fn(), +): DialCacheMetricsAdapter { return { request: vi.fn(), miss: vi.fn(), @@ -47,15 +50,21 @@ function metricsWithError(error: DialCacheMetricsAdapter["error"]): DialCacheMet error, invalidation: vi.fn(), observeGet: vi.fn(), - observeFallback: vi.fn(), + observeFallback, observeSerialization: vi.fn(), observeSize: vi.fn(), }; } +function useFakeTimersWithMonotonicClock(): void { + vi.useFakeTimers(); + const clockOriginMs = Date.now(); + vi.spyOn(performance, "now").mockImplementation(() => Date.now() - clockOriginMs); +} + describe("DialCache fallback liveness", () => { beforeEach(() => { - vi.useFakeTimers(); + useFakeTimersWithMonotonicClock(); }); afterEach(() => { @@ -69,7 +78,9 @@ describe("DialCache fallback liveness", () => { const gate = deferred<{ readonly id: string }>(); const started = deferred(); const error = vi.fn(); - const dialcache = new DialCache({ metrics: metricsWithError(error) }); + const observeFallback = vi.fn(); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const dialcache = new DialCache({ metrics: metricsWithError(error, observeFallback) }); let calls = 0; const getUser = dialcache.cached(async (id: string) => { calls += 1; @@ -93,6 +104,7 @@ describe("DialCache fallback liveness", () => { activeFollowers: 2, }); expect(dialcache.getCoalescingState().process.oldestLeaderAgeMs).toBeGreaterThanOrEqual(0); + expect(setTimeoutSpy).toHaveBeenCalledTimes(1); nowMs = 59_999; await vi.advanceTimersByTimeAsync(59_999); @@ -120,11 +132,48 @@ describe("DialCache fallback liveness", () => { error: "fallback", inFallback: true, }); + expect(observeFallback).toHaveBeenCalledTimes(1); + expect(observeFallback).toHaveBeenCalledWith({ + cacheNamespace: "urn", + useCase: "DefaultFallbackDeadline", + keyType: "user_id", + layer: CacheLayer.LOCAL, + }, 60); expect(dialcache.getCoalescingState().process).toEqual({ activeLeaders: 0, activeFollowers: 0, oldestLeaderAgeMs: null, }); + + gate.resolve({ id: "late" }); + await vi.advanceTimersByTimeAsync(0); + expect(observeFallback).toHaveBeenCalledTimes(1); + }); + + it("keeps a pending fallback deadline referenced", async () => { + const started = deferred(); + const gate = deferred(); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const dialcache = new DialCache(); + const load = dialcache.cached(async () => { + started.resolve(); + return await gate.promise; + }, { + keyType: "id", + useCase: "ReferencedFallbackDeadline", + cacheKey: () => "123", + fallbackTimeoutMs: 10_000, + defaultConfig: localConfig, + }); + + const result = dialcache.enable(async () => await load()); + await started.promise; + + const timer = setTimeoutSpy.mock.results[0]?.value as NodeJS.Timeout | undefined; + expect(timer?.hasRef()).toBe(true); + + gate.resolve("value"); + await expect(result).resolves.toBe("value"); }); it("uses a caller-selected timeout on enabled pass-through calls", async () => { @@ -252,6 +301,151 @@ describe("DialCache fallback liveness", () => { expect(vi.getTimerCount()).toBe(0); }); + it("rechecks the monotonic deadline when a timer fires early", async () => { + let nowMs = 0; + vi.spyOn(performance, "now").mockImplementation(() => nowMs); + const fakeSetTimeout = globalThis.setTimeout; + let scheduledTimers = 0; + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout").mockImplementation((( + callback: (...args: unknown[]) => void, + delay?: number, + ...args: unknown[] + ) => { + scheduledTimers += 1; + const adjustedDelay = scheduledTimers === 1 ? Math.max((delay ?? 0) - 1, 0) : delay; + return fakeSetTimeout(callback, adjustedDelay, ...args); + }) as typeof setTimeout); + const started = deferred(); + const dialcache = new DialCache(); + const load = dialcache.cached(async () => { + started.resolve(); + return await new Promise(() => undefined); + }, { + keyType: "id", + useCase: "EarlyFallbackTimer", + cacheKey: () => "123", + fallbackTimeoutMs: 10, + defaultConfig: localConfig, + }); + + const result = Promise.allSettled([dialcache.enable(async () => await load())]); + await started.promise; + nowMs = 9; + await vi.advanceTimersByTimeAsync(9); + + expect(setTimeoutSpy).toHaveBeenCalledTimes(2); + expect(dialcache.getCoalescingState().process.activeLeaders).toBe(1); + + nowMs = 10; + await vi.advanceTimersByTimeAsync(1); + expect((await result)[0]).toEqual({ status: "rejected", reason: expect.any(FallbackTimeoutError) }); + }); + + it("rejects an async result that settles after its monotonic deadline before the timer callback", async () => { + let nowMs = 0; + vi.spyOn(performance, "now").mockImplementation(() => nowMs); + const gate = deferred(); + const started = deferred(); + const dialcache = new DialCache(); + let calls = 0; + const load = dialcache.cached(async () => { + calls += 1; + if (calls === 1) { + started.resolve(); + return await gate.promise; + } + return "fresh"; + }, { + keyType: "id", + useCase: "DelayedFallbackTimer", + cacheKey: () => "123", + fallbackTimeoutMs: 10, + defaultConfig: localConfig, + }); + + const first = Promise.allSettled([dialcache.enable(async () => await load())]); + await started.promise; + nowMs = 10; + gate.resolve("overdue"); + await vi.advanceTimersByTimeAsync(0); + + expect((await first)[0]).toEqual({ status: "rejected", reason: expect.any(FallbackTimeoutError) }); + expect(vi.getTimerCount()).toBe(0); + await expect(dialcache.enable(async () => await load())).resolves.toBe("fresh"); + expect(calls).toBe(2); + }); + + it("times out fallback after cache-key construction fails", async () => { + const started = deferred(); + const dialcache = new DialCache({ logger: { debug: vi.fn(), error: vi.fn(), warn: vi.fn() } }); + const load = dialcache.cached(async () => { + started.resolve(); + return await new Promise(() => undefined); + }, { + keyType: "id", + useCase: "KeyFailureFallbackDeadline", + cacheKey: () => { + throw new Error("invalid key"); + }, + fallbackTimeoutMs: 5, + }); + + const result = Promise.allSettled([dialcache.enable(async () => await load())]); + await started.promise; + await vi.advanceTimersByTimeAsync(5); + expect((await result)[0]).toEqual({ status: "rejected", reason: expect.any(FallbackTimeoutError) }); + }); + + it("times out fallback after runtime config resolution fails", async () => { + const started = deferred(); + const dialcache = new DialCache({ + cacheConfigProvider: async () => { + throw new Error("config unavailable"); + }, + logger: { debug: vi.fn(), error: vi.fn(), warn: vi.fn() }, + }); + const load = dialcache.cached(async () => { + started.resolve(); + return await new Promise(() => undefined); + }, { + keyType: "id", + useCase: "ConfigFailureFallbackDeadline", + cacheKey: () => "123", + fallbackTimeoutMs: 5, + }); + + const result = Promise.allSettled([dialcache.enable(async () => await load())]); + await started.promise; + await vi.advanceTimersByTimeAsync(5); + expect((await result)[0]).toEqual({ status: "rejected", reason: expect.any(FallbackTimeoutError) }); + }); + + it("times out detached fallback after config resolves outside its closed enabled scope", async () => { + const configGate = deferred(); + const started = deferred(); + const dialcache = new DialCache({ cacheConfigProvider: async () => await configGate.promise }); + const load = dialcache.cached(async () => { + started.resolve(); + return await new Promise(() => undefined); + }, { + keyType: "id", + useCase: "DetachedFallbackDeadline", + cacheKey: () => "123", + fallbackTimeoutMs: 5, + }); + let detached!: Promise; + + await dialcache.enable(() => { + detached = load(); + }); + configGate.resolve(localConfig); + await started.promise; + const result = Promise.allSettled([detached]); + await vi.advanceTimersByTimeAsync(5); + + expect((await result)[0]).toEqual({ status: "rejected", reason: expect.any(FallbackTimeoutError) }); + }); + it("keeps initially disabled calls as true pass-through", async () => { const started = deferred(); const gate = deferred(); @@ -425,7 +619,7 @@ describe("DialCache fallback liveness", () => { describe("DialCache coalescing state", () => { beforeEach(() => { - vi.useFakeTimers(); + useFakeTimersWithMonotonicClock(); }); afterEach(() => { @@ -443,6 +637,7 @@ describe("DialCache coalescing state", () => { const dialcache = new DialCache(); const load = dialcache.cached(async (id: string) => { if (id === "first") { + nowMs += 25; firstStarted.resolve(); return await firstGate.promise; } @@ -461,29 +656,34 @@ describe("DialCache coalescing state", () => { const firstLeader = dialcache.enable(async () => await load("first")); const firstFollower = dialcache.enable(async () => await load("first")); - const secondFollower = dialcache.enable(async () => await load("first")); + const additionalFirstFollower = dialcache.enable(async () => await load("first")); await firstStarted.promise; nowMs = 1_040; const secondLeader = dialcache.enable(async () => await load("second")); + const secondFollower = dialcache.enable(async () => await load("second")); await secondStarted.promise; nowMs = 1_100; expect(dialcache.getCoalescingState()).toEqual({ - process: { activeLeaders: 2, activeFollowers: 2, oldestLeaderAgeMs: 100 }, + process: { activeLeaders: 2, activeFollowers: 3, oldestLeaderAgeMs: 100 }, }); firstGate.resolve("first-value"); - await expect(Promise.all([firstLeader, firstFollower, secondFollower])).resolves.toEqual([ + await expect(Promise.all([firstLeader, firstFollower, additionalFirstFollower])).resolves.toEqual([ "first-value", "first-value", "first-value", ]); expect(dialcache.getCoalescingState()).toEqual({ - process: { activeLeaders: 1, activeFollowers: 0, oldestLeaderAgeMs: 60 }, + process: { activeLeaders: 1, activeFollowers: 1, oldestLeaderAgeMs: 60 }, }); secondGate.reject(new Error("second failed")); - await expect(secondLeader).rejects.toThrow("second failed"); + const secondResults = await Promise.allSettled([secondLeader, secondFollower]); + expect(secondResults).toEqual([ + { status: "rejected", reason: expect.objectContaining({ message: "second failed" }) }, + { status: "rejected", reason: expect.objectContaining({ message: "second failed" }) }, + ]); expect(dialcache.getCoalescingState()).toEqual({ process: { activeLeaders: 0, activeFollowers: 0, oldestLeaderAgeMs: null }, }); From eea27dde365043fe505441337e5aaec62ee49e3e Mon Sep 17 00:00:00 2001 From: Lev Neiman Date: Sun, 19 Jul 2026 17:09:05 -0700 Subject: [PATCH 3/4] fix: tighten fallback liveness guarantees --- README.md | 8 +- scripts/test-package.mjs | 48 ++++++- src/dialcache.ts | 27 +++- test/dialcache-liveness.test.ts | 225 ++++++++++++++++++++++++++++++++ 4 files changed, 296 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index df3935e..008088a 100644 --- a/README.md +++ b/README.md @@ -338,7 +338,7 @@ const getUser = dialcache.cached( ### Fallback deadlines -Once an initially enabled invocation starts its fallback, DialCache waits at most 60 seconds by default. Set `fallbackTimeoutMs` once on a cached wrapper to choose a positive integer deadline in milliseconds, up to 2,147,483,647, or set it to `null` to preserve an intentionally unbounded fallback: +Once an initially enabled invocation starts its fallback, DialCache applies a 60-second monotonic deadline by default. Set `fallbackTimeoutMs` once on a cached wrapper to choose a positive integer deadline in milliseconds, up to 2,147,483,647, or set it to `null` to preserve an intentionally unbounded fallback: ```ts import { FallbackTimeoutError } from "dialcache"; @@ -368,6 +368,8 @@ 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`. +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. + Timing out rejects the DialCache chain and clears its flight normally. A later fallback resolution is ignored, so that timed-out invocation cannot proceed to serializer, Redis, or local-cache publication. The underlying function is not canceled and may continue its own I/O or side effects; give the source operation its own native timeout or `AbortSignal` whenever possible. `fallbackTimeoutMs: null` disables this guard and makes finite fallback settlement entirely application-owned. This default is a behavioral compatibility change for enabled calls whose fallback previously remained pending beyond 60 seconds. Use the `null` escape hatch only after intentionally accepting that liveness risk. Timeout failures retain the bounded metrics classification `error="fallback"` with `in_fallback="true"`; the typed error provides the timeout details without adding high-cardinality labels. @@ -661,7 +663,7 @@ Not included yet: - `cachedObject` - Expanded examples -## Request-local benchmark +## Cache-path benchmark From a repository checkout, run the semantic microbenchmark after installing dependencies: @@ -669,7 +671,7 @@ From a repository checkout, run the semantic microbenchmark after installing dep pnpm benchmark:request-local ``` -The command builds `dist` before reporting request-local sequential-hit throughput plus request-local and instance-scoped coalescing fan-out. The benchmark is a maintainer tool and is not included in the published package. It asserts fallback counts 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 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`. ## Releasing diff --git a/scripts/test-package.mjs b/scripts/test-package.mjs index d4f1b75..b69b03e 100644 --- a/scripts/test-package.mjs +++ b/scripts/test-package.mjs @@ -8,6 +8,7 @@ import { promisify } from "node:util"; const exec = promisify(execFile); const root = dirname(dirname(fileURLToPath(import.meta.url))); const workspace = await mkdtemp(join(tmpdir(), "dialcache-package-")); +const fallbackTimeoutMarker = "dialcache-fallback-timeout-delivered"; const rootConsumer = `import { CacheLayer, DialCache, @@ -389,7 +390,7 @@ try { ), ]); - await exec( + const { stdout: esmRootRuntimeOutput } = await exec( process.execPath, [ "--input-type=module", @@ -407,6 +408,23 @@ const idleCoalescingState = { process: { activeLeaders: 0, activeFollowers: 0, o if (JSON.stringify(coalescingState) !== JSON.stringify(idleCoalescingState)) { throw new Error("The root ESM coalescing snapshot export is invalid"); } +const timeoutCache = new root.DialCache(); +const neverSettles = timeoutCache.cached(async () => await new Promise(() => undefined), { + keyType: "id", + useCase: "PackageOnlyHandleTimeout", + cacheKey: () => "1", + defaultConfig: root.DialCacheKeyConfig.enabled(60), + fallbackTimeoutMs: 20, +}); +try { + await timeoutCache.enable(() => neverSettles()); + throw new Error("Expected the packaged ESM fallback to time out"); +} catch (error) { + if (!(error instanceof root.FallbackTimeoutError) || error.timeoutMs !== 20) { + throw new Error("The packaged ESM fallback timeout was not delivered"); + } + console.log("${fallbackTimeoutMarker}"); +} try { nodeRedis.dialcacheRedisScripts.dialcacheWrite.transformReply(2); throw new Error("Expected an invalid node-redis script reply to fail"); @@ -418,7 +436,11 @@ try { ], { cwd: workspace }, ); - await exec( + if (!esmRootRuntimeOutput.includes(fallbackTimeoutMarker)) { + throw new Error("The packaged ESM only-handle fallback timeout marker is missing"); + } + + const { stdout: cjsRootRuntimeOutput } = await exec( process.execPath, [ "--eval", @@ -435,6 +457,25 @@ const idleCoalescingState = { process: { activeLeaders: 0, activeFollowers: 0, o if (JSON.stringify(coalescingState) !== JSON.stringify(idleCoalescingState)) { throw new Error("The root CommonJS coalescing snapshot export is invalid"); } +const timeoutCache = new root.DialCache(); +const neverSettles = timeoutCache.cached(async () => await new Promise(() => undefined), { + keyType: "id", + useCase: "PackageOnlyHandleTimeout", + cacheKey: () => "1", + defaultConfig: root.DialCacheKeyConfig.enabled(60), + fallbackTimeoutMs: 20, +}); +void (async () => { + try { + await timeoutCache.enable(() => neverSettles()); + throw new Error("Expected the packaged CommonJS fallback to time out"); + } catch (error) { + if (!(error instanceof root.FallbackTimeoutError) || error.timeoutMs !== 20) { + throw new Error("The packaged CommonJS fallback timeout was not delivered"); + } + console.log("${fallbackTimeoutMarker}"); + } +})(); try { nodeRedis.dialcacheRedisScripts.dialcacheWrite.transformReply(2); throw new Error("Expected an invalid node-redis script reply to fail"); @@ -446,6 +487,9 @@ try { ], { cwd: workspace }, ); + if (!cjsRootRuntimeOutput.includes(fallbackTimeoutMarker)) { + throw new Error("The packaged CommonJS only-handle fallback timeout marker is missing"); + } await exec( join(workspace, "node_modules", ".bin", "tsc"), ["--project", join(workspace, "tsconfig.root.json")], diff --git a/src/dialcache.ts b/src/dialcache.ts index e1e9552..9edcfb4 100644 --- a/src/dialcache.ts +++ b/src/dialcache.ts @@ -105,9 +105,11 @@ interface CachedOptionsBase { readonly defaultConfig?: DialCacheKeyConfig | null; readonly trackForInvalidation?: boolean; /** - * Maximum time to wait once an initially enabled invocation starts its + * Monotonic deadline applied once an initially enabled invocation starts its * fallback, in milliseconds. Must be at most 2,147,483,647. Defaults to 60 - * seconds. Set to `null` to disable the deadline. + * seconds. Set to `null` to disable the deadline. Like every JavaScript + * timer, delivery requires event-loop progress and cannot preempt synchronous + * work. * * Concurrent same-key callers share the leader's remaining budget. Timing * out rejects those callers and prevents the eventual result from being @@ -152,7 +154,7 @@ export interface CoalescingState { } interface ProcessFlight { - readonly promise: Promise; + promise: Promise | null; readonly startedAtMs: number; followers: number; } @@ -639,6 +641,9 @@ export class DialCache { private singleFlightProcess(key: DialCacheKey, run: () => Promise): Promise { const existing = this.processFlights.get(key.urn); if (existing !== undefined) { + if (existing.promise === null) { + throw new Error("DialCache process flight was joined before initialization"); + } existing.followers += 1; this.activeProcessFollowers += 1; this.metrics?.coalesced?.({ @@ -650,14 +655,22 @@ export class DialCache { return existing.promise as Promise; } - const startedAtMs = performance.now(); - const promise = run(); const flight: ProcessFlight = { - promise, - startedAtMs, + promise: null, + startedAtMs: performance.now(), followers: 0, }; this.processFlights.set(key.urn, flight); + let promise: Promise; + try { + promise = run(); + } catch (error) { + if (this.processFlights.get(key.urn) === flight) { + this.processFlights.delete(key.urn); + } + throw error; + } + flight.promise = promise; const clear = (): void => { if (this.processFlights.get(key.urn) === flight) { this.activeProcessFollowers -= flight.followers; diff --git a/test/dialcache-liveness.test.ts b/test/dialcache-liveness.test.ts index ba7ccd9..dcabf58 100644 --- a/test/dialcache-liveness.test.ts +++ b/test/dialcache-liveness.test.ts @@ -10,6 +10,8 @@ import { FallbackTimeoutError, type CachedOptions, type DialCacheMetricsAdapter, + type DialCacheRedisClient, + type Serializer, } from "../src/index.js"; import { FakeRedis } from "./fake-redis.js"; @@ -446,6 +448,208 @@ describe("DialCache fallback liveness", () => { expect((await result)[0]).toEqual({ status: "rejected", reason: expect.any(FallbackTimeoutError) }); }); + it("does not start the fallback deadline while the config provider is pending", async () => { + const configGate = deferred(); + const configStarted = deferred(); + const fallback = vi.fn(async () => "value"); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const dialcache = new DialCache({ + cacheConfigProvider: async () => { + configStarted.resolve(); + return await configGate.promise; + }, + }); + const load = dialcache.cached(fallback, { + keyType: "id", + useCase: "PendingConfigHasCallerOwnedDeadline", + cacheKey: () => "123", + fallbackTimeoutMs: 5, + }); + + const result = dialcache.enable(async () => await load()); + await configStarted.promise; + await vi.advanceTimersByTimeAsync(100); + + expect(fallback).not.toHaveBeenCalled(); + expect(setTimeoutSpy).not.toHaveBeenCalled(); + expect(dialcache.getCoalescingState().process.activeLeaders).toBe(0); + + configGate.resolve(localConfig); + await expect(result).resolves.toBe("value"); + expect(fallback).toHaveBeenCalledTimes(1); + }); + + it("does not start the fallback deadline while the ramp sampler is pending", async () => { + const rampGate = deferred(); + const rampStarted = deferred(); + const fallback = vi.fn(async () => "value"); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const dialcache = new DialCache({ + rampSampler: async () => { + rampStarted.resolve(); + return await rampGate.promise; + }, + }); + const load = dialcache.cached(fallback, { + keyType: "id", + useCase: "PendingRampHasCallerOwnedDeadline", + cacheKey: () => "123", + fallbackTimeoutMs: 5, + defaultConfig: new DialCacheKeyConfig({ + ttlSec: { [CacheLayer.LOCAL]: 60 }, + ramp: { [CacheLayer.LOCAL]: 50 }, + }), + }); + + const result = dialcache.enable(async () => await load()); + await rampStarted.promise; + await vi.advanceTimersByTimeAsync(100); + + expect(fallback).not.toHaveBeenCalled(); + expect(setTimeoutSpy).not.toHaveBeenCalled(); + expect(dialcache.getCoalescingState().process.activeLeaders).toBe(0); + + rampGate.resolve(0); + await expect(result).resolves.toBe("value"); + expect(fallback).toHaveBeenCalledTimes(1); + }); + + it("does not apply the fallback deadline to a pending Redis read", async () => { + const readGate = deferred(); + 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; + }, + write: async () => true, + invalidate: async () => undefined, + }; + const dialcache = new DialCache({ redis: { client: redis } }); + const load = dialcache.cached(fallback, { + keyType: "id", + useCase: "PendingRedisReadHasCallerOwnedDeadline", + cacheKey: () => "123", + fallbackTimeoutMs: 5, + defaultConfig: remoteConfig, + }); + + const result = dialcache.enable(async () => await load()); + await readStarted.promise; + await vi.advanceTimersByTimeAsync(100); + + expect(fallback).not.toHaveBeenCalled(); + expect(setTimeoutSpy).not.toHaveBeenCalled(); + expect(dialcache.getCoalescingState().process.activeLeaders).toBe(1); + + readGate.resolve(null); + await expect(result).resolves.toBe("value"); + expect(fallback).toHaveBeenCalledTimes(1); + expect(dialcache.getCoalescingState().process.activeLeaders).toBe(0); + }); + + it("does not apply the fallback deadline to a pending serializer load", async () => { + const loadGate = deferred(); + const loadStarted = deferred(); + const fallback = vi.fn(async () => "fallback"); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const serializer: Serializer = { + dump: (value) => value, + load: async () => { + loadStarted.resolve(); + return await loadGate.promise; + }, + }; + const redis: DialCacheRedisClient = { + read: async () => "stored", + write: async () => true, + invalidate: async () => undefined, + }; + const dialcache = new DialCache({ redis: { client: redis } }); + const load = dialcache.cached(async () => await fallback(), { + keyType: "id", + useCase: "PendingSerializerLoadHasCallerOwnedDeadline", + cacheKey: () => "123", + fallbackTimeoutMs: 5, + defaultConfig: remoteConfig, + serializer, + }); + + const result = dialcache.enable(async () => await load()); + await loadStarted.promise; + await vi.advanceTimersByTimeAsync(100); + + expect(fallback).not.toHaveBeenCalled(); + expect(setTimeoutSpy).not.toHaveBeenCalled(); + expect(dialcache.getCoalescingState().process.activeLeaders).toBe(1); + + loadGate.resolve("cached"); + await expect(result).resolves.toBe("cached"); + expect(dialcache.getCoalescingState().process.activeLeaders).toBe(0); + }); + + it("does not apply a completed fallback's deadline to serializer dump or Redis write", async () => { + const dumpGate = deferred(); + const dumpStarted = deferred(); + const writeGate = deferred(); + const writeStarted = deferred(); + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const serializer: Serializer = { + dump: async () => { + dumpStarted.resolve(); + return await dumpGate.promise; + }, + load: (value) => value.toString(), + }; + const redis: DialCacheRedisClient = { + read: async () => null, + write: async () => { + writeStarted.resolve(); + return await writeGate.promise; + }, + invalidate: async () => undefined, + }; + const dialcache = new DialCache({ redis: { client: redis } }); + const load = dialcache.cached(async () => "value", { + keyType: "id", + useCase: "PendingPublicationHasCallerOwnedDeadline", + cacheKey: () => "123", + fallbackTimeoutMs: 5, + defaultConfig: remoteConfig, + serializer, + }); + + let settled = false; + const result = dialcache.enable(async () => await load()); + void result.then( + () => { + settled = true; + }, + () => { + settled = true; + }, + ); + await dumpStarted.promise; + expect(setTimeoutSpy).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + + await vi.advanceTimersByTimeAsync(100); + expect(settled).toBe(false); + expect(dialcache.getCoalescingState().process.activeLeaders).toBe(1); + + dumpGate.resolve("value"); + await writeStarted.promise; + await vi.advanceTimersByTimeAsync(100); + expect(settled).toBe(false); + expect(dialcache.getCoalescingState().process.activeLeaders).toBe(1); + + writeGate.resolve(true); + await expect(result).resolves.toBe("value"); + expect(dialcache.getCoalescingState().process.activeLeaders).toBe(0); + }); + it("keeps initially disabled calls as true pass-through", async () => { const started = deferred(); const gate = deferred(); @@ -627,6 +831,27 @@ describe("DialCache coalescing state", () => { vi.restoreAllMocks(); }); + it("registers a process leader before its synchronous fallback prefix runs", async () => { + const dialcache = new DialCache(); + let observedState = dialcache.getCoalescingState().process; + const load = dialcache.cached(async () => { + observedState = dialcache.getCoalescingState().process; + return "value"; + }, { + keyType: "id", + useCase: "ObserveSynchronousLeaderPrefix", + cacheKey: () => "123", + defaultConfig: localConfig, + }); + + await expect(dialcache.enable(async () => await load())).resolves.toBe("value"); + expect(observedState).toMatchObject({ activeLeaders: 1, activeFollowers: 0 }); + expect(observedState.oldestLeaderAgeMs).toBeGreaterThanOrEqual(0); + expect(dialcache.getCoalescingState()).toEqual({ + process: { activeLeaders: 0, activeFollowers: 0, oldestLeaderAgeMs: null }, + }); + }); + it("reports leaders, followers, oldest age, and settlement cleanup", async () => { let nowMs = 1_000; vi.spyOn(performance, "now").mockImplementation(() => nowMs); From 12be5f065ca209cd8d4b88f467564ad0fce5f5bf Mon Sep 17 00:00:00 2001 From: Lev Neiman Date: Sun, 19 Jul 2026 17:14:58 -0700 Subject: [PATCH 4/4] docs: clarify fallback duration timing --- README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 008088a..f793719 100644 --- a/README.md +++ b/README.md @@ -526,7 +526,7 @@ The Prometheus adapter emits: | `dialcache_invalidation_counter` | Counter | `cache_namespace`, `key_type`, `layer` | Invalidation calls for the layers touched today | | `dialcache_coalesced_counter` | Counter | `cache_namespace`, `use_case`, `key_type`, `scope` | Coalesced requests split by `request_local` or `process` scope | | `dialcache_get_timer` | Histogram | `cache_namespace`, `use_case`, `key_type`, `layer` | Cache get latency in seconds | -| `dialcache_fallback_timer` | Histogram | `cache_namespace`, `use_case`, `key_type`, `layer` | Time DialCache waited for the underlying function, ending at its fallback deadline when applied | +| `dialcache_fallback_timer` | Histogram | `cache_namespace`, `use_case`, `key_type`, `layer` | Elapsed time until the underlying function settles or timeout rejection is delivered | | `dialcache_serialization_timer` | Histogram | `cache_namespace`, `use_case`, `key_type`, `layer`, `operation` | Redis serializer dump/load latency | | `dialcache_size_histogram` | Histogram | `cache_namespace`, `use_case`, `key_type`, `layer` | Serialized Redis payload size in bytes | @@ -583,7 +583,7 @@ The Datadog adapter emits exact increments of `1` for counters and preserves sec | `dialcache.invalidation.count` | Count | `cache_namespace`, `key_type`, `layer` | Invalidation calls for the layers touched | | `dialcache.coalesced.count` | Count | `cache_namespace`, `use_case`, `key_type`, `scope` | Coalesced requests by sharing scope | | `dialcache.get.duration` | Distribution or histogram | `cache_namespace`, `use_case`, `key_type`, `layer` | Cache get latency in seconds | -| `dialcache.fallback.duration` | Distribution or histogram | `cache_namespace`, `use_case`, `key_type`, `layer` | Time DialCache waited for the underlying function, ending at its fallback deadline when applied | +| `dialcache.fallback.duration` | Distribution or histogram | `cache_namespace`, `use_case`, `key_type`, `layer` | Elapsed time until the underlying function settles or timeout rejection is delivered | | `dialcache.serialization.duration` | Distribution or histogram | `cache_namespace`, `use_case`, `key_type`, `layer`, `operation` | Redis serializer dump/load latency in seconds | | `dialcache.serialization.size` | Distribution or histogram | `cache_namespace`, `use_case`, `key_type`, `layer` | Serialized Redis payload size in bytes |