perf(redis): skip refills behind observed watermarks

scripts/test-package.mjs

@@ -149,6 +149,8 @@ const rootConsumer = `import {
   type RedisConfig,
   type RedisInvalidationRequest,
   type RedisReadContext,
+  type RedisReadResult,
+  type RedisWatermarkMiss,
   type RedisWriteRequest,
   type Serializer,
   type ShadowComparator,
@@ -172,6 +174,7 @@ import {
   ceilSupportedCacheTtlMs,
   decodeRedisFrame,
   decodeTrackedRedisFrame,
+  decodeTrackedRedisReadResult,
   encodeRedisFrame,
   validateRedisScriptInvalidationReply,
   validateRedisSetReply,
@@ -257,6 +260,7 @@ const shadowOutcomes: Readonly<Record<ShadowValidationOutcome, true>> = {
   mismatch: true,
   superseded: true,
   filled: true,
+  fill_fenced: true,
   fill_error: true,
   redis_error: true,
   source_error: true,
@@ -339,6 +343,18 @@ const decodedStaleRedisFrame: DecodedRedisFrame | null = decodeTrackedRedisFrame
   emptyRedisFrame,
   Buffer.from("1"),
 );
+const decodedTrackedRedisReadResult: RedisReadResult = decodeTrackedRedisReadResult(
+  emptyRedisFrame,
+  Buffer.from("1"),
+);
+if (
+  decodedTrackedRedisReadResult !== null
+  && "observedWatermarkMs" in decodedTrackedRedisReadResult
+) {
+  const typedWatermarkMiss: RedisWatermarkMiss = decodedTrackedRedisReadResult;
+  const observedWatermarkMs: number = typedWatermarkMiss.observedWatermarkMs;
+  void observedWatermarkMs;
+}
 const zeroTimestampRedisFrame: Buffer = encodeRedisFrame("pending", 0);
 const setReplyValidation: void = validateRedisSetReply("OK");
 const invalidationReplyValidation: 1 = validateRedisScriptInvalidationReply(1);
@@ -557,8 +573,11 @@ const compressionOperationMetricLabels: CompressionOperationMetricLabels = {
 const unboundedCompressionOutcome: CompressionOutcome = "inflated";
 
 const customRedisClient: DialCacheRedisClient = {
-  // The optional second read argument preserves one-argument custom clients.
-  read: async () => ({ payload: Buffer.from([0, 255]), createdAtMs: 1 }),
+  // The optional second argument and widened result preserve legacy frame-or-null clients.
+  read: async (): Promise<DecodedRedisFrame | null> => ({
+    payload: Buffer.from([0, 255]),
+    createdAtMs: 1,
+  }),
   write: async ({ value }) => {
     void (typeof value === "string" || Buffer.isBuffer(value));
   },
@@ -581,9 +600,9 @@ const writeHasNoWatermark: "watermarkKey" extends keyof RedisWriteRequest ? fals
 const trackedWriteHasNoWatermarkTtlFloor: "watermarkTtlFloorMs" extends keyof RedisWriteRequest
   ? false
   : true = true;
-const trackedWriteHasNoCreatedAt: "createdAtMs" extends keyof RedisWriteRequest
-  ? false
-  : true = true;
+const writeAcceptsOptionalCreatedAt: {} extends Pick<RedisWriteRequest, "createdAtMs">
+  ? true
+  : false = true;
 const invalidationHasNoWatermarkTtlFloor: "watermarkTtlFloorMs" extends keyof RedisInvalidationRequest
   ? false
   : true = true;
@@ -597,6 +616,17 @@ const legacyTrackedWriteRequest: RedisWriteRequest = {
   cacheTtlMs: 1_000,
   value: "tracked",
 };
+const timestampedWriteRequest: RedisWriteRequest = {
+  valueKey: "tracked:{id}:value",
+  cacheTtlMs: 1_000,
+  value: "tracked",
+  createdAtMs: 1,
+};
+const omittedTimestampWriteRequest: RedisWriteRequest = {
+  valueKey: "legacy:{id}:value",
+  cacheTtlMs: 1_000,
+  value: "legacy",
+};
 const legacyInvalidationRequest: RedisInvalidationRequest = {
   watermarkKey: "tracked:{id}:watermark",
   futureBufferMs: 0,
@@ -731,10 +761,12 @@ void cacheHasNoClose;
 void clientHasNoFlushAll;
 void writeHasNoWatermark;
 void trackedWriteHasNoWatermarkTtlFloor;
-void trackedWriteHasNoCreatedAt;
+void writeAcceptsOptionalCreatedAt;
 void invalidationHasNoWatermarkTtlFloor;
 void invalidationHasNoInvalidatedAt;
 void legacyTrackedWriteRequest;
+void timestampedWriteRequest;
+void omittedTimestampWriteRequest;
 void legacyInvalidationRequest;
 void configHasNoMetricsRegistry;
 void configHasNoMetricsPrefix;
@@ -974,9 +1006,19 @@ if (esmRoundTrip?.payload !== "value" || esmRoundTrip.createdAtMs !== 1) {
 if (redisProtocol.decodeTrackedRedisFrame(redisProtocol.encodeRedisFrame("pending", 0), Buffer.from("0")) !== null) {
   throw new Error("The packed ESM tracked decoder did not fence an equal timestamp");
 }
+const esmWatermarkMiss = redisProtocol.decodeTrackedRedisReadResult(
+  redisProtocol.encodeRedisFrame("pending", 0),
+  Buffer.from("0"),
+);
+if (esmWatermarkMiss?.observedWatermarkMs !== 0 || "payload" in esmWatermarkMiss) {
+  throw new Error("The packed ESM tracked result decoder did not preserve the observed watermark miss");
+}
 if (redisProtocol.decodeTrackedRedisFrame(redisProtocol.encodeRedisFrame("value", 1), null)?.payload !== "value") {
   throw new Error("The packed ESM tracked decoder did not use zero for a missing watermark");
 }
+if (redisProtocol.decodeTrackedRedisReadResult(null, null) !== null) {
+  throw new Error("The packed ESM tracked result decoder did not preserve a generic miss without a watermark");
+}
 if (
   "REDIS_FRAME_VERSION" in redisProtocol
   || "REDIS_ENCODING_UTF8" in redisProtocol
@@ -1349,9 +1391,19 @@ if (cjsRoundTrip?.payload !== "value" || cjsRoundTrip.createdAtMs !== 1) {
 if (redisProtocol.decodeTrackedRedisFrame(redisProtocol.encodeRedisFrame("pending", 0), Buffer.from("0")) !== null) {
   throw new Error("The packed CommonJS tracked decoder did not fence an equal timestamp");
 }
+const cjsWatermarkMiss = redisProtocol.decodeTrackedRedisReadResult(
+  redisProtocol.encodeRedisFrame("pending", 0),
+  Buffer.from("0"),
+);
+if (cjsWatermarkMiss?.observedWatermarkMs !== 0 || "payload" in cjsWatermarkMiss) {
+  throw new Error("The packed CommonJS tracked result decoder did not preserve the observed watermark miss");
+}
 if (redisProtocol.decodeTrackedRedisFrame(redisProtocol.encodeRedisFrame("value", 1), null)?.payload !== "value") {
   throw new Error("The packed CommonJS tracked decoder did not use zero for a missing watermark");
 }
+if (redisProtocol.decodeTrackedRedisReadResult(null, null) !== null) {
+  throw new Error("The packed CommonJS tracked result decoder did not preserve a generic miss without a watermark");
+}
 if (
   "REDIS_FRAME_VERSION" in redisProtocol
   || "REDIS_ENCODING_UTF8" in redisProtocol
@@ -1520,6 +1572,7 @@ const redisProtocol = await import("dialcache/redis-protocol");
 await import("dialcache/node-redis");
 ${packedInvalidationCheckSource}
 const esmCreatedAtMs = 1700000000123;
+const esmAdapterClockMs = esmCreatedAtMs + 999;
 if (appGlide.Script === otherGlide.Script) {
   throw new Error("The package test requires two distinct GLIDE module instances");
 }
@@ -1552,12 +1605,13 @@ const esmGlideRuntime = {
 };
 const adapter = glide.createValkeyGlideDialCacheClient(esmFakeGlideClient, esmGlideRuntime);
 const esmNativeDateNow = Date.now;
-Date.now = () => esmCreatedAtMs;
+Date.now = () => esmAdapterClockMs;
 try {
   await adapter.write({
     valueKey: "tracked:{id}:value",
     cacheTtlMs: 1_000,
     value: "payload",
+    createdAtMs: esmCreatedAtMs,
   });
   if (
     esmWriteCommand[0] !== "SET"
@@ -1567,7 +1621,7 @@ try {
     || esmWriteCommand[2][0] !== 1
     || esmWriteCommand[2].readBigUInt64BE(1) !== BigInt(esmCreatedAtMs)
   ) {
-    throw new Error("The packed ESM GLIDE write did not send one complete client-stamped frame");
+    throw new Error("The packed ESM GLIDE write did not preserve the supplied frame timestamp exactly");
   }
   const trackedRead = await adapter.read({
     valueKey: "tracked:{id}:value",
@@ -1649,7 +1703,7 @@ void (async () => {
       || cjsWriteCommand[2][0] !== 1
       || cjsWriteCommand[2].readBigUInt64BE(1) !== BigInt(cjsCreatedAtMs)
     ) {
-      throw new Error("The packed CommonJS GLIDE write did not send one complete client-stamped frame");
+      throw new Error("The packed CommonJS GLIDE write did not stamp an omitted timestamp from its client clock");
     }
     const trackedRead = await adapter.read({
       valueKey: "tracked:{id}:value",

src/dialcache.ts

@@ -24,7 +24,12 @@ import {
   type MetricLayer,
   type ShadowValidationOutcome,
 } from "./metrics.js";
-import type { DecodedRedisFrame, RedisCachePayload } from "./redis-client.js";
+import type {
+  DecodedRedisFrame,
+  RedisCachePayload,
+  RedisReadResult,
+  RedisWatermarkMiss,
+} from "./redis-client.js";
 import type { Serializer } from "./serializer.js";
 import type { CacheGetResult, RemoteCacheGetResult } from "./internal/cache-result.js";
 import { MAX_TIMER_DELAY_MS, withMonotonicDeadline } from "./internal/deadline.js";
@@ -866,7 +871,12 @@ export class DialCache {
       || (remoteWriteConfig !== undefined && key.trackForInvalidation);
     if (remoteWriteConfig !== undefined) {
       try {
-        await redisCache.put(key, value, remoteWriteConfig);
+        await redisCache.put(
+          key,
+          value,
+          remoteWriteConfig,
+          remote.status === "miss" ? remote.watermarkMiss : undefined,
+        );
       } catch (error) {
         this.logger.warn("Error putting value in Redis cache", error);
       }
@@ -1044,7 +1054,7 @@ export class DialCache {
     const readShadowFrame = (
       maxAgeSec: number | null,
       futureFramePolicy: FutureFramePolicy,
-    ): Promise<DecodedRedisFrame | null> => {
+    ): Promise<RedisReadResult> => {
       const read = redisCache.startPayloadReadForShadow(
         key,
         maxAgeSec,
@@ -1089,20 +1099,24 @@ export class DialCache {
           }
 
           let shadowFillConfig: ResolvedRemoteLayerConfig | null = null;
+          let shadowFillWatermarkMiss: RedisWatermarkMiss | undefined;
           if (start.kind === "redis") {
-            let frame: DecodedRedisFrame | null;
+            let readResult: RedisReadResult;
             try {
-              frame = await readShadowFrame(start.remoteConfig.ttlSec, "reject");
+              readResult = await readShadowFrame(start.remoteConfig.ttlSec, "reject");
             } catch {
               return "redis_error";
             }
             if (abandonIfExpired()) {
               return "timeout";
             }
-            if (frame === null) {
+            if (isRedisWatermarkMiss(readResult)) {
+              shadowFillConfig = start.remoteConfig;
+              shadowFillWatermarkMiss = readResult;
+            } else if (readResult === null) {
               shadowFillConfig = start.remoteConfig;
             } else {
-              flight.cachedFrame = frame;
+              flight.cachedFrame = readResult;
             }
           }
 
@@ -1134,18 +1148,19 @@ export class DialCache {
 
           if (shadowFillConfig !== null) {
             try {
-              await redisCache.putForShadow(
+              const filled = await redisCache.putForShadow(
                 key,
                 sourceValue,
                 shadowFillConfig,
                 () => !abandonIfExpired(),
+                shadowFillWatermarkMiss,
               );
               // A late result remains the already-emitted whole-job timeout:
               // dispatch success does not retroactively change its outcome.
               if (abandonIfExpired()) {
                 return "timeout";
               }
-              return "filled";
+              return filled ? "filled" : "fill_fenced";
             } catch (error) {
               this.logger.warn("Error populating Redis from DialCache shadow work", error);
               return "fill_error";
@@ -1189,9 +1204,9 @@ export class DialCache {
             return "match";
           }
 
-          let confirmationFrame: DecodedRedisFrame | null;
+          let confirmationResult: RedisReadResult;
           try {
-            confirmationFrame = await readShadowFrame(null, "retain");
+            confirmationResult = await readShadowFrame(null, "retain");
           } catch {
             return "confirmation_error";
           }
@@ -1203,6 +1218,9 @@ export class DialCache {
           if (originalFrame === null) {
             return "timeout";
           }
+          const confirmationFrame = isRedisWatermarkMiss(confirmationResult)
+            ? null
+            : confirmationResult;
           if (confirmationFrame === null || !redisPayloadsEqual(originalFrame.payload, confirmationFrame.payload)) {
             return "superseded";
           }
@@ -1755,6 +1773,14 @@ function redisPayloadsEqual(left: RedisCachePayload, right: RedisCachePayload):
   return Buffer.isBuffer(right) && right.equals(Buffer.from(left, "utf8"));
 }
 
+function isRedisWatermarkMiss(result: RedisReadResult): result is RedisWatermarkMiss {
+  return typeof result === "object"
+    && result !== null
+    && "observedWatermarkMs" in result
+    && !("payload" in result)
+    && !("createdAtMs" in result);
+}
+
 async function settleUnexpectedThenable(value: unknown): Promise<void> {
   if (value === null || (typeof value !== "object" && typeof value !== "function")) {
     return;

src/index.ts

@@ -62,6 +62,8 @@ export type {
   RedisInvalidationRequest,
   RedisReadContext,
   RedisReadRequest,
+  RedisReadResult,
+  RedisWatermarkMiss,
   RedisWriteRequest,
 } from "./redis-client.js";
 export { JsonSerializer } from "./serializer.js";

src/internal/cache-result.ts

@@ -1,6 +1,6 @@
 import type { ResolvedLayerConfig, ResolvedRemoteLayerConfig } from "./runtime-config.js";
 import type { DisabledReason } from "../metrics.js";
-import type { DecodedRedisFrame } from "../redis-client.js";
+import type { DecodedRedisFrame, RedisWatermarkMiss } from "../redis-client.js";
 
 export type CacheGetResult<T> =
   | { readonly status: "hit"; readonly value: T }
@@ -25,6 +25,7 @@ export type RedisCacheGetResult<T> =
       readonly status: "miss";
       readonly config: ResolvedRemoteLayerConfig;
       readonly reason: RedisCacheMissReason;
+      readonly watermarkMiss?: RedisWatermarkMiss;
     };
 
 export type RemoteCacheGetResult<T> =