feat: enforce per-use-case Redis read deadlines

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) => {

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<Record<MetricErrorKind, true>> = {
 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<RedisConfig, "client"> ? false : true = true;
+const redisConfigRequiresReadTimeout: {} extends Pick<RedisConfig, "readTimeoutMs"> ? 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<unknown> = {
   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)) {