From 69a91706b50918a51cad1ca6846b9bff65d094cb Mon Sep 17 00:00:00 2001 From: Lev Neiman Date: Fri, 14 Aug 2026 13:38:34 -0700 Subject: [PATCH 1/2] feat(shadow)!: record validated value age on shadow match and mismatch Shadow validation now measures how stale a cached value was when a verdict was delivered. Every Redis frame already carried an 8-byte createdAtMs header (Redis server time for tracked writes, the writer's client clock for untracked ones), but both decoders discarded it. The decoders and DialCacheRedisClient.read() now return a DecodedRedisFrame ({ payload, createdAtMs }), the shadow flight retains the frame, and a match or mismatch verdict records now - createdAtMs in seconds, clamped at zero, through the new optional observeShadowValueAge adapter hook. The Prometheus adapter exposes dialcache_shadow_value_age_histogram with 1s..7d buckets and the Datadog adapter emits dialcache.shadow.value_age, both labeled by outcome. Outcomes that deliver no verdict on a retained value (superseded, filled, errors, timeout) record no age, and the hook does not gate shadow eligibility; only shadowValidation does. BREAKING CHANGE: decodeRedisFrame, decodeTrackedRedisFrame, and DialCacheRedisClient.read() return DecodedRedisFrame | null instead of the bare payload. Custom Redis clients must return the decoded frame; the bundled node-redis and Valkey GLIDE adapters inherit the change unchanged. Pre-1.0 policy releases this as a minor version. --- README.md | 12 ++- scripts/test-package.mjs | 23 +++-- src/datadog.ts | 21 ++-- src/dialcache.ts | 62 +++++++----- src/index.ts | 1 + src/internal/cache-result.ts | 4 +- src/internal/redis-cache.ts | 20 ++-- src/internal/redis-payload.ts | 41 +++++--- src/metrics.ts | 11 ++ src/prometheus.ts | 31 ++++-- src/redis-client.ts | 33 ++++-- src/redis-protocol.ts | 8 +- test/datadog.test.ts | 25 +++++ test/dialcache-liveness.test.ts | 2 +- test/dialcache-redis-read-deadline.test.ts | 10 +- test/dialcache-redis.test.ts | 6 +- test/dialcache-shadow-confirmation.test.ts | 51 +++++++++- test/dialcache-shadow-validation.test.ts | 111 ++++++++++++++++++++- test/fake-redis.ts | 12 ++- test/node-redis.test.ts | 11 +- test/prometheus.test.ts | 15 +++ test/redis-cluster.integration.test.ts | 16 +-- test/redis-payload.test.ts | 32 +++--- test/redis-real.integration.test.ts | 65 +++++++----- test/valkey-glide.test.ts | 9 +- 25 files changed, 475 insertions(+), 157 deletions(-) diff --git a/README.md b/README.md index 8d85510..87e5fc2 100644 --- a/README.md +++ b/README.md @@ -428,7 +428,7 @@ Writes, invalidations, async `cacheConfigProvider` calls, and custom serializer #### Serialization -The core Redis boundary is the client-agnostic `DialCacheRedisClient` interface. It exchanges serialized values as `string | Buffer` and does not expose client commands or wire encodings. The shared `encodeRedisFrame`, `encodeTrackedRedisPlaceholder`, `decodeRedisFrame`, and `decodeTrackedRedisFrame` helpers, the `resolveTrackedRedisWriteReply`, `validateRedisSetReply`, and `validateRedisScriptInvalidationReply` reply helpers, the `ceilSupportedCacheTtlMs` TTL guard, and the tracked stamp and invalidation Lua sources are available from `dialcache/redis-protocol`, so custom adapters can reuse the bundled adapters' exact framing, miss, watermark-fencing, TTL-domain, and reply rules. A custom tracked write must pass the stamp script `KEYS = [valueKey, watermarkKey]` and `ARGV = [cacheTtlMs, nonce]`, running `cacheTtlMs` through `ceilSupportedCacheTtlMs` and using the result for both the paired `SET`'s `PX` and `ARGV[1]` (the stamp script re-validates the same domain server-side as defense in depth), with the nonce from the same `encodeTrackedRedisPlaceholder` call; `resolveTrackedRedisWriteReply` maps the reply, failing the write with the root-exported `DialCacheRedisPlaceholderLostError` when the stamp replies `2`. Custom adapters can throw the root-exported `DialCacheRedisPayloadError`, `DialCacheRedisPayloadEncodingError`, `DialCacheRedisProtocolError`, and `DialCacheRedisPlaceholderLostError` classes to distinguish malformed replies, unsupported encodings, reply-domain violations, and lost placeholders in logs. DialCache records bounded `cache_read`, `cache_write`, or `invalidation` metrics by failure site. +The core Redis boundary is the client-agnostic `DialCacheRedisClient` interface. Writes accept serialized values as `string | Buffer`; reads return a `DecodedRedisFrame` — the decoded `string | Buffer` payload plus the frame header's `createdAtMs` (Redis server time for tracked frames, the writer's informational client clock for untracked ones) — and the interface does not expose client commands or wire encodings. The shared `encodeRedisFrame`, `encodeTrackedRedisPlaceholder`, `decodeRedisFrame`, and `decodeTrackedRedisFrame` helpers, the `resolveTrackedRedisWriteReply`, `validateRedisSetReply`, and `validateRedisScriptInvalidationReply` reply helpers, the `ceilSupportedCacheTtlMs` TTL guard, and the tracked stamp and invalidation Lua sources are available from `dialcache/redis-protocol`, so custom adapters can reuse the bundled adapters' exact framing, miss, watermark-fencing, TTL-domain, and reply rules. A custom tracked write must pass the stamp script `KEYS = [valueKey, watermarkKey]` and `ARGV = [cacheTtlMs, nonce]`, running `cacheTtlMs` through `ceilSupportedCacheTtlMs` and using the result for both the paired `SET`'s `PX` and `ARGV[1]` (the stamp script re-validates the same domain server-side as defense in depth), with the nonce from the same `encodeTrackedRedisPlaceholder` call; `resolveTrackedRedisWriteReply` maps the reply, failing the write with the root-exported `DialCacheRedisPlaceholderLostError` when the stamp replies `2`. Custom adapters can throw the root-exported `DialCacheRedisPayloadError`, `DialCacheRedisPayloadEncodingError`, `DialCacheRedisProtocolError`, and `DialCacheRedisPlaceholderLostError` classes to distinguish malformed replies, unsupported encodings, reply-domain violations, and lost placeholders in logs. DialCache records bounded `cache_read`, `cache_write`, or `invalidation` metrics by failure site. Redis values use a compact binary frame: @@ -580,12 +580,14 @@ A `match` means application-level value equality: - An optional typed `shadowComparator(cachedValue, sourceValue)` on `cached()` or `getOrLoad()` can define narrower domain equality, such as ignoring a volatile timestamp. It must synchronously return a boolean and must be deterministic, side-effect-free, non-mutating, and bounded. - A comparator throw or non-boolean return is `comparison_error`, never `mismatch`. An accidental promise is not accepted as a comparison result; DialCache consumes its settlement while retaining the shadow slot, subject to the same detached deadline. -DialCache retains the semantic `string | Buffer` returned by the Redis client but never exposes it to the comparator. After the source read completes, detached work calls the same effective serializer's `load` method to create an independent cached snapshot, then compares that snapshot with the raw value returned by the source loader. It does not reuse the cached object already returned to a served-hit caller, so caller mutation cannot contaminate validation. No payload copy, shadow deserialization, deep comparison, or hash is added to the served-hit request path. +DialCache retains the semantic frame returned by the Redis client — its `string | Buffer` payload and `createdAtMs` — but never exposes it to the comparator. After the source read completes, detached work calls the same effective serializer's `load` method to create an independent cached snapshot, then compares that snapshot with the raw value returned by the source loader. It does not reuse the cached object already returned to a served-hit caller, so caller mutation cannot contaminate validation. No payload copy, shadow deserialization, deep comparison, or hash is added to the served-hit request path. -The effective serializer's `load` method therefore runs a second time for a sampled served hit and once in detached work for a shadow-only hit. It must be repeatable, non-mutating, and return independently usable values. On a clean miss, its `dump` method may run after the caller has received `S`, so `S` must remain immutable through detached serialization. A custom `DialCacheRedisClient` must return an operation-owned payload whose string/Buffer contents remain stable after `read()` settles. Comparing the deserialized cached snapshot with the raw source value intentionally detects lossy serialization; use a custom comparator only when such normalization or ignored fields are valid use-case semantics. +The effective serializer's `load` method therefore runs a second time for a sampled served hit and once in detached work for a shadow-only hit. It must be repeatable, non-mutating, and return independently usable values. On a clean miss, its `dump` method may run after the caller has received `S`, so `S` must remain immutable through detached serialization. A custom `DialCacheRedisClient` must return an operation-owned frame whose string/Buffer payload contents remain stable after `read()` settles. Comparing the deserialized cached snapshot with the raw source value intentionally detects lossy serialization; use a custom comparator only when such normalization or ignored fields are valid use-case semantics. Shadow metrics use the bounded outcomes `match`, `mismatch`, `superseded`, `filled`, `fill_blocked`, `fill_error`, `redis_error`, `source_error`, `deserialization_error`, `comparison_error`, `confirmation_error`, `timeout`, and `dropped`. `redis_error` applies to the initial shadow-only `C0` read; `confirmation_error` applies to `C1`; `fill_blocked` applies only when a tracked watermark rejects the write. A clean `C0` miss is an ordinary `miss{layer="remote_shadow"}` and terminates with a fill, source, or timeout outcome rather than a second shadow outcome for the miss itself. Labels never contain cache ids, values, payloads, Redis keys, or raw exception text. +A `match` or `mismatch` verdict additionally records the validated value's age through the optional `observeShadowValueAge` adapter hook: the observing process's epoch clock minus the served frame's `createdAtMs`, in seconds, clamped at zero. A confirmed `mismatch` age therefore measures how long the stale value had been readable when validation caught it. Tracked frames are stamped with Redis server time and untracked frames with the writer's client clock, so the age mixes clocks and is coarse operational evidence rather than a precise measurement. Outcomes that deliver no verdict on a retained value — including `superseded`, `filled`, and every error or timeout outcome — record no age. The hook does not gate shadow eligibility; only `shadowValidation` does. + Confirmed-mismatch logging is separately opt-in through `shadow.logMismatches`; it never replaces the required `shadowValidation` metric or emits for a mismatch candidate that becomes `superseded`. The single warning contains `cacheNamespace`, `useCase`, `keyType`, `outcome: "mismatch"`, `cacheKey`, `cachedValueJson`, and `sourceValueJson`. `cacheKey` is the logical DialCache URN capped at 2 KiB, not the physical Redis storage key. DialCache independently applies native `JSON.stringify` to the deserialized cached snapshot and raw source value supplied to the comparator, then caps each resulting string at 8 KiB. A byte-clipped field ends in `...[truncated]`, counted inside its cap. If native JSON throws or returns `undefined`, the corresponding JSON field is `null`; the other side is still attempted. DialCache does not compute a textual diff or call the configured serializer again for logging. Mismatch logging is intentionally default-off. Logical URNs can contain ids and arguments, while cached and source values may contain secrets or personal data; truncation is not redaction. DialCache creates the JSON strings only after terminal mismatch confirmation and never passes the raw compared-value references to the logger. Native JSON semantics apply: getters and `toJSON` methods may run, unsupported values may be omitted or normalized, and cycles or `bigint` can make a field unavailable. Stringification is synchronous, and the 8 KiB caps apply only after `JSON.stringify` returns; they do not bound input traversal, hook execution, event-loop time, or the intermediate JSON string. Enable mismatch logging only for trusted, reasonably bounded values and with an approved logger, redaction, transport, access, and retention policy. @@ -793,6 +795,7 @@ The Prometheus adapter emits: | `dialcache_invalidation_counter` | Counter | `cache_namespace`, `key_type`, `layer` | Invalidation calls for the layers touched | | `dialcache_coalesced_counter` | Counter | `cache_namespace`, `use_case`, `key_type`, `scope` | Coalesced requests split by `request_local` or `process` scope | | `dialcache_shadow_validation_counter` | Counter | `cache_namespace`, `use_case`, `key_type`, `outcome` | Sampled Redis shadow-job outcomes | +| `dialcache_shadow_value_age_histogram` | Histogram | `cache_namespace`, `use_case`, `key_type`, `outcome` | Age in seconds of the validated Redis value at shadow comparison time, recorded for `match` and `mismatch` | | `dialcache_compression_counter` | Counter | `cache_namespace`, `use_case`, `key_type`, `layer`, `outcome` | Payload compression outcomes: writes record `compressed`, `below_threshold`, `not_smaller`, or `write_over_limit`; reads record `decompressed`, `fallback_raw`, or `read_over_limit` | | `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` | Elapsed time until the underlying function settles or timeout rejection is delivered | @@ -855,6 +858,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.shadow.count` | Count | `cache_namespace`, `use_case`, `key_type`, `outcome` | Sampled Redis shadow-job outcomes | +| `dialcache.shadow.value_age` | Distribution or histogram | `cache_namespace`, `use_case`, `key_type`, `outcome` | Age in seconds of the validated Redis value at shadow comparison time, recorded for `match` and `mismatch` | | `dialcache.compression.count` | Count | `cache_namespace`, `use_case`, `key_type`, `layer`, `outcome` | Payload compression outcomes: writes record `compressed`, `below_threshold`, `not_smaller`, or `write_over_limit`; reads record `decompressed`, `fallback_raw`, or `read_over_limit` | | `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` | Elapsed time until the underlying function settles or timeout rejection is delivered | @@ -888,7 +892,7 @@ These values are defined by the backend-neutral core and are identical for every ### Custom adapters -For other telemetry backends, implement `DialCacheMetricsAdapter` and pass the adapter through `new DialCache({ metrics })`. Every backend-neutral label object exposes the logical namespace as camel-case `cacheNamespace`; adapters should map it to their backend's `cache_namespace` label/tag. This field is present even when no key or cache layer was reached. Implement the optional `shadowValidation` method to enable shadow work as well as record its outcomes; omitting it leaves all shadow work disabled even when `shadow.ramp` is nonzero or mismatch logging is enabled. Every metrics callback is fire-and-forget: DialCache isolates synchronous throws and consumes rejections from returned promises or thenables, but never awaits or drains observer work. Omit `metrics` to disable metrics. +For other telemetry backends, implement `DialCacheMetricsAdapter` and pass the adapter through `new DialCache({ metrics })`. Every backend-neutral label object exposes the logical namespace as camel-case `cacheNamespace`; adapters should map it to their backend's `cache_namespace` label/tag. This field is present even when no key or cache layer was reached. Implement the optional `shadowValidation` method to enable shadow work as well as record its outcomes; omitting it leaves all shadow work disabled even when `shadow.ramp` is nonzero or mismatch logging is enabled. The optional `observeShadowValueAge` method records the validated value's age in seconds for `match` and `mismatch` outcomes; omitting it skips only that observation without affecting shadow eligibility. Every metrics callback is fire-and-forget: DialCache isolates synchronous throws and consumes rejections from returned promises or thenables, but never awaits or drains observer work. Omit `metrics` to disable metrics. ## Maintainers diff --git a/scripts/test-package.mjs b/scripts/test-package.mjs index 90f8702..03b00ed 100644 --- a/scripts/test-package.mjs +++ b/scripts/test-package.mjs @@ -64,6 +64,7 @@ import { validateRedisScriptInvalidationReply, validateRedisSetReply, WRITE_TRACKED_STAMP_SCRIPT, + type DecodedRedisFrame, type TrackedRedisPlaceholder, } from "dialcache/redis-protocol"; // @ts-expect-error The codec functions replaced the frame-version wire constant. @@ -170,8 +171,8 @@ const redisProtocolError = new DialCacheRedisProtocolError("Invalid DialCache Re const emptyRedisFrame = Buffer.alloc(10); emptyRedisFrame[0] = 1; emptyRedisFrame.writeBigUInt64BE(1n, 1); -const decodedEmptyRedisPayload: string | Buffer | null = decodeRedisFrame(emptyRedisFrame); -const decodedStaleRedisPayload: string | Buffer | null = decodeTrackedRedisFrame( +const decodedEmptyRedisFrame: DecodedRedisFrame | null = decodeRedisFrame(emptyRedisFrame); +const decodedStaleRedisFrame: DecodedRedisFrame | null = decodeTrackedRedisFrame( emptyRedisFrame, Buffer.from("1"), ); @@ -401,7 +402,7 @@ const unboundedCompressionOutcome: CompressionOutcome = "inflated"; const customRedisClient: DialCacheRedisClient = { // The optional second read argument preserves one-argument custom clients. - read: async () => Buffer.from([0, 255]), + read: async () => ({ payload: Buffer.from([0, 255]), createdAtMs: 1 }), write: async ({ value }) => typeof value === "string" || Buffer.isBuffer(value), invalidate: async () => undefined, }; @@ -527,8 +528,8 @@ void compressionOutcomes; void unboundedCompressionOutcome; void redisConfigAcceptsCompressionOptOut; void createNodeRedisDialCacheClient; -void decodedEmptyRedisPayload; -void decodedStaleRedisPayload; +void decodedEmptyRedisFrame; +void decodedStaleRedisFrame; // @ts-expect-error Native reads removed the legacy node-redis registration. void dialcacheRedisScripts.dialcacheRead; // @ts-expect-error Native tracked reads removed the legacy node-redis registration. @@ -802,7 +803,8 @@ if ( if (typeof redisProtocol.WRITE_TRACKED_STAMP_SCRIPT !== "string") { throw new Error("The packed ESM Redis protocol entry must export the tracked stamp script source"); } -if (redisProtocol.decodeRedisFrame(redisProtocol.encodeRedisFrame("value", 1)) !== "value") { +const esmRoundTrip = redisProtocol.decodeRedisFrame(redisProtocol.encodeRedisFrame("value", 1)); +if (esmRoundTrip?.payload !== "value" || esmRoundTrip.createdAtMs !== 1) { throw new Error("The packed ESM Redis protocol encoder did not round-trip through the decoder"); } if (redisProtocol.decodeTrackedRedisFrame(redisProtocol.encodeRedisFrame("pending", 0), Buffer.from("0")) !== null) { @@ -878,7 +880,7 @@ if (!(esmBrandedLost instanceof root.DialCacheRedisPlaceholderLostError)) { const esmEmptyFrame = Buffer.alloc(10); esmEmptyFrame[0] = 1; esmEmptyFrame.writeBigUInt64BE(1n, 1); -if (redisProtocol.decodeRedisFrame(esmEmptyFrame) !== "") { +if (redisProtocol.decodeRedisFrame(esmEmptyFrame)?.payload !== "") { throw new Error("The packed ESM Redis protocol decoder did not preserve an empty UTF-8 payload"); } if (redisProtocol.decodeTrackedRedisFrame(esmEmptyFrame, Buffer.from("1")) !== null) { @@ -1013,7 +1015,7 @@ console.log("${observerIsolationMarker}");`, let payload = Buffer.alloc(4 * 1024 * 1024, 1); const payloadReference = new WeakRef(payload); const redis = { - read: async () => payload, + read: async () => ({ payload, createdAtMs: 1 }), write: async () => true, invalidate: async () => undefined, }; @@ -1176,7 +1178,8 @@ if ( if (typeof redisProtocol.WRITE_TRACKED_STAMP_SCRIPT !== "string") { throw new Error("The packed CommonJS Redis protocol entry must export the tracked stamp script source"); } -if (redisProtocol.decodeRedisFrame(redisProtocol.encodeRedisFrame("value", 1)) !== "value") { +const cjsRoundTrip = redisProtocol.decodeRedisFrame(redisProtocol.encodeRedisFrame("value", 1)); +if (cjsRoundTrip?.payload !== "value" || cjsRoundTrip.createdAtMs !== 1) { throw new Error("The packed CommonJS Redis protocol encoder did not round-trip through the decoder"); } if (redisProtocol.decodeTrackedRedisFrame(redisProtocol.encodeRedisFrame("pending", 0), Buffer.from("0")) !== null) { @@ -1252,7 +1255,7 @@ if (!(cjsBrandedLost instanceof root.DialCacheRedisPlaceholderLostError)) { const cjsEmptyFrame = Buffer.alloc(10); cjsEmptyFrame[0] = 1; cjsEmptyFrame.writeBigUInt64BE(1n, 1); -if (redisProtocol.decodeRedisFrame(cjsEmptyFrame) !== "") { +if (redisProtocol.decodeRedisFrame(cjsEmptyFrame)?.payload !== "") { throw new Error("The packed CommonJS Redis protocol decoder did not preserve an empty UTF-8 payload"); } if (redisProtocol.decodeTrackedRedisFrame(cjsEmptyFrame, Buffer.from("1")) !== null) { diff --git a/src/datadog.ts b/src/datadog.ts index d3e7dee..1a28fda 100644 --- a/src/datadog.ts +++ b/src/datadog.ts @@ -48,6 +48,7 @@ const METRIC_SUFFIXES = { invalidation: "invalidation.count", coalesced: "coalesced.count", shadowValidation: "shadow.count", + shadowValueAge: "shadow.value_age", compression: "compression.count", get: "get.duration", fallback: "fallback.duration", @@ -120,12 +121,11 @@ export class DatadogDialCacheMetrics implements DialCacheMetricsAdapter { } shadowValidation(labels: ShadowValidationMetricLabels): void { - return this.increment(this.metricNames.shadowValidation, { - cache_namespace: labels.cacheNamespace, - use_case: labels.useCase, - key_type: labels.keyType, - outcome: labels.outcome, - }); + return this.increment(this.metricNames.shadowValidation, shadowValidationTags(labels)); + } + + observeShadowValueAge(labels: ShadowValidationMetricLabels, seconds: number): void { + this.observe(this.metricNames.shadowValueAge, seconds, shadowValidationTags(labels)); } compression(labels: CompressionMetricLabels): void { @@ -184,6 +184,15 @@ function cacheTags(labels: CacheMetricLabels): Record { }; } +function shadowValidationTags(labels: ShadowValidationMetricLabels): DatadogTags { + return { + cache_namespace: labels.cacheNamespace, + use_case: labels.useCase, + key_type: labels.keyType, + outcome: labels.outcome, + }; +} + function validateClient(client: DatadogDogStatsDClient): DatadogDogStatsDClient { if (client === null || typeof client !== "object") { throw new TypeError("Datadog metrics client must be an object."); diff --git a/src/dialcache.ts b/src/dialcache.ts index 62954b4..a9be5b5 100644 --- a/src/dialcache.ts +++ b/src/dialcache.ts @@ -23,7 +23,7 @@ import { type MetricLayer, type ShadowValidationOutcome, } from "./metrics.js"; -import type { RedisCachePayload } from "./redis-client.js"; +import type { DecodedRedisFrame, RedisCachePayload } 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"; @@ -211,7 +211,7 @@ interface ProcessFlight { } interface ShadowFlight { - cachedPayload: RedisCachePayload | null; + cachedFrame: DecodedRedisFrame | null; abandoned: boolean; readonly startedAtMs: number; } @@ -229,7 +229,7 @@ interface ShadowMismatchDetails { } type ShadowValidationStart = - | { readonly kind: "retained"; readonly payload: RedisCachePayload } + | { readonly kind: "retained"; readonly frame: DecodedRedisFrame } | { readonly kind: "redis"; /** The caller-owned, fallback-deadline-bounded SoT operation. */ @@ -764,7 +764,7 @@ export class DialCache { redisCache, key, keyConfig, - { kind: "retained", payload: remote.payload }, + { kind: "retained", frame: remote.frame }, shadowValidation, keyConfig?.remoteReadTimeoutMs ?? redisCache.readTimeoutMs, ); @@ -848,7 +848,7 @@ export class DialCache { const logMismatches = this.resolveShadowLogging(key, resolvedShadowConfig); const flight: ShadowFlight = { - cachedPayload: start.kind === "retained" ? start.payload : null, + cachedFrame: start.kind === "retained" ? start.frame : null, abandoned: false, startedAtMs: start.kind === "redis" && start.startedAtMs !== null ? start.startedAtMs @@ -927,7 +927,7 @@ export class DialCache { return; } released = true; - flight.cachedPayload = null; + flight.cachedFrame = null; if (this.shadowFlights.get(key.urn) === flight) { this.shadowFlights.delete(key.urn); } @@ -938,11 +938,11 @@ export class DialCache { } }; const finishOperation = (): void => { - flight.cachedPayload = null; + flight.cachedFrame = null; operationFinished = true; maybeRelease(); }; - const readShadowPayload = (): Promise => { + const readShadowFrame = (): Promise => { const read = redisCache.startPayloadReadForShadow(key, readTimeoutMs); pendingRedisReads.add(read.settled); void read.settled.then(() => { @@ -957,13 +957,14 @@ export class DialCache { const abandonIfExpired = (): boolean => { if (!flight.abandoned && performance.now() - deadlineStartedAtMs >= plan.timeoutMs) { flight.abandoned = true; - flight.cachedPayload = null; + flight.cachedFrame = null; } return flight.abandoned; }; const elapsedBeforeStartMs = Math.max(performance.now() - deadlineStartedAtMs, 0); const remainingTimeoutMs = Math.max(plan.timeoutMs - elapsedBeforeStartMs, 0); let mismatchDetails: ShadowMismatchDetails | undefined; + let validatedValueAgeSeconds: number | undefined; const validation = withMonotonicDeadline({ timeoutMs: remainingTimeoutMs, @@ -971,7 +972,7 @@ export class DialCache { timeoutError: () => new Error("DialCache shadow validation timed out"), onTimeout: () => { flight.abandoned = true; - flight.cachedPayload = null; + flight.cachedFrame = null; signalShadowTimeout(); }, operation: async (): Promise => { @@ -982,19 +983,19 @@ export class DialCache { let shadowFillConfig: ResolvedLayerConfig | null = null; if (start.kind === "redis") { - let payload: RedisCachePayload | null; + let frame: DecodedRedisFrame | null; try { - payload = await readShadowPayload(); + frame = await readShadowFrame(); } catch { return "redis_error"; } if (abandonIfExpired()) { return "timeout"; } - if (payload === null) { + if (frame === null) { shadowFillConfig = start.remoteConfig; } else { - flight.cachedPayload = payload; + flight.cachedFrame = frame; } } @@ -1044,14 +1045,14 @@ export class DialCache { } } - const retainedPayload = flight.cachedPayload; - if (retainedPayload === null) { + const retainedFrame = flight.cachedFrame; + if (retainedFrame === null) { return "timeout"; } let cachedValue: T; try { - cachedValue = await redisCache.deserializeForShadow(key, retainedPayload); + cachedValue = await redisCache.deserializeForShadow(key, retainedFrame.payload); } catch { return "deserialization_error"; } @@ -1077,12 +1078,13 @@ export class DialCache { return "timeout"; } if (matches) { + validatedValueAgeSeconds = shadowValueAgeSeconds(retainedFrame.createdAtMs); return "match"; } - let confirmationPayload: RedisCachePayload | null; + let confirmationFrame: DecodedRedisFrame | null; try { - confirmationPayload = await readShadowPayload(); + confirmationFrame = await readShadowFrame(); } catch { return "confirmation_error"; } @@ -1090,16 +1092,17 @@ export class DialCache { return "timeout"; } - const originalPayload = flight.cachedPayload; - if (originalPayload === null) { + const originalFrame = flight.cachedFrame; + if (originalFrame === null) { return "timeout"; } - if (confirmationPayload === null || !redisPayloadsEqual(originalPayload, confirmationPayload)) { + if (confirmationFrame === null || !redisPayloadsEqual(originalFrame.payload, confirmationFrame.payload)) { return "superseded"; } if (logMismatches) { mismatchDetails = { cachedValue, sourceValue }; } + validatedValueAgeSeconds = shadowValueAgeSeconds(originalFrame.createdAtMs); return "mismatch"; } finally { finishOperation(); @@ -1108,7 +1111,7 @@ export class DialCache { }); void validation.then( - (outcome) => this.recordShadowValidation(key, outcome, logMismatches, mismatchDetails), + (outcome) => this.recordShadowValidation(key, outcome, logMismatches, mismatchDetails, validatedValueAgeSeconds), () => this.recordShadowValidation(key, "timeout"), ); } @@ -1118,6 +1121,7 @@ export class DialCache { outcome: ShadowValidationOutcome, logMismatches = false, mismatchDetails?: ShadowMismatchDetails, + valueAgeSeconds?: number, ): void { const labels = { cacheNamespace: key.namespace, @@ -1126,6 +1130,9 @@ export class DialCache { outcome, } as const; this.metrics?.shadowValidation?.(labels); + if (valueAgeSeconds !== undefined) { + this.metrics?.observeShadowValueAge?.(labels, valueAgeSeconds); + } if (outcome !== "mismatch" || !logMismatches) { return; } @@ -1536,6 +1543,8 @@ function safeMetrics(metrics: DialCacheMetricsAdapter | null): DialCacheMetricsA callObserver(() => metrics.shadowValidation!(labels)), } : {}), + observeShadowValueAge: (labels, seconds) => + callObserver(() => metrics.observeShadowValueAge?.(labels, seconds)), observeGet: (labels, seconds) => callObserver(() => metrics.observeGet(labels, seconds)), observeFallback: (labels, seconds) => callObserver(() => metrics.observeFallback(labels, seconds)), observeSerialization: (labels, seconds) => callObserver(() => metrics.observeSerialization(labels, seconds)), @@ -1565,6 +1574,13 @@ function resolveShadowComparator( return comparator ?? isDeepStrictEqual; } +// Frame stamps are epoch-based (Redis server time for tracked writes, writer +// client clock for untracked), so the age uses the epoch clock and clamps +// negative cross-clock skew to zero. +function shadowValueAgeSeconds(createdAtMs: number): number { + return Math.max((Date.now() - createdAtMs) / 1000, 0); +} + function redisPayloadsEqual(left: RedisCachePayload, right: RedisCachePayload): boolean { if (typeof left === "string" && typeof right === "string") { return left === right; diff --git a/src/index.ts b/src/index.ts index 8f27a09..cc10a23 100644 --- a/src/index.ts +++ b/src/index.ts @@ -48,6 +48,7 @@ export { export type { CompressionConfig } from "./internal/compression.js"; export type { RedisConfig } from "./internal/redis-cache.js"; export type { + DecodedRedisFrame, DialCacheRedisClient, RedisCachePayload, RedisInvalidationRequest, diff --git a/src/internal/cache-result.ts b/src/internal/cache-result.ts index 85b8188..9abae50 100644 --- a/src/internal/cache-result.ts +++ b/src/internal/cache-result.ts @@ -1,6 +1,6 @@ import type { ResolvedLayerConfig } from "./runtime-config.js"; import type { DisabledReason } from "../metrics.js"; -import type { RedisCachePayload } from "../redis-client.js"; +import type { DecodedRedisFrame } from "../redis-client.js"; export type CacheGetResult = | { readonly status: "hit"; readonly value: T } @@ -12,7 +12,7 @@ export type CacheGetResult = }; export type RedisCacheGetResult = - | { readonly status: "hit"; readonly value: T; readonly payload: RedisCachePayload } + | { readonly status: "hit"; readonly value: T; readonly frame: DecodedRedisFrame } | Exclude, { readonly status: "hit" }>; export type RemoteCacheGetResult = diff --git a/src/internal/redis-cache.ts b/src/internal/redis-cache.ts index 41eabed..ad25fe8 100644 --- a/src/internal/redis-cache.ts +++ b/src/internal/redis-cache.ts @@ -10,7 +10,7 @@ import { type MetricErrorKind, type MetricLayer, } from "../metrics.js"; -import type { DialCacheRedisClient, RedisCachePayload } from "../redis-client.js"; +import type { DecodedRedisFrame, DialCacheRedisClient, RedisCachePayload } from "../redis-client.js"; import { JsonSerializer, type Serializer } from "../serializer.js"; import type { RedisCacheGetResult } from "./cache-result.js"; import { @@ -53,7 +53,7 @@ interface RedisCacheOptions { interface StartedRedisRead { /** Result bounded by the effective Redis read deadline. */ - readonly result: Promise; + readonly result: Promise; /** Fulfills only after the underlying semantic Redis read settles. */ readonly settled: Promise; } @@ -126,9 +126,9 @@ export class RedisCache { const start = performance.now(); this.recordMetric((metrics) => metrics.request(labelsFor(key, metricLayer))); try { - let payload: RedisCachePayload | null; + let frame: DecodedRedisFrame | null; try { - payload = await this.startPayloadRead(key, readTimeoutMs, false).result; + frame = await this.startPayloadRead(key, readTimeoutMs, false).result; } catch (error) { this.recordError( key, @@ -137,14 +137,14 @@ export class RedisCache { ); throw error; } - if (payload === null) { + if (frame === null) { this.recordMetric((metrics) => metrics.miss(labelsFor(key, metricLayer))); return { status: "miss", config: layerConfig }; } try { - const value = await this.deserializePayload(key, payload, metricLayer); - return { status: "hit", value, payload }; + const value = await this.deserializePayload(key, frame.payload, metricLayer); + return { status: "hit", value, frame }; } catch { this.recordMetric((metrics) => metrics.miss(labelsFor(key, metricLayer))); return { status: "miss", config: layerConfig }; @@ -345,11 +345,11 @@ export class RedisCache { this.recordMetric((metrics) => metrics.request(labelsFor(key, metricLayer))); const read = this.startPayloadRead(key, readTimeoutMs, unrefTimer); const result = read.result.then( - (payload) => { - if (payload === null) { + (frame) => { + if (frame === null) { this.recordMetric((metrics) => metrics.miss(labelsFor(key, metricLayer))); } - return payload; + return frame; }, (error: unknown) => { this.recordError( diff --git a/src/internal/redis-payload.ts b/src/internal/redis-payload.ts index 40d4e55..97fb93b 100644 --- a/src/internal/redis-payload.ts +++ b/src/internal/redis-payload.ts @@ -3,6 +3,7 @@ import { randomBytes } from "node:crypto"; import { DialCacheRedisPayloadEncodingError, DialCacheRedisPayloadError, + type DecodedRedisFrame, type RedisCachePayload, } from "../redis-client.js"; @@ -118,27 +119,34 @@ export function encodeTrackedRedisPlaceholder(payload: RedisCachePayload): Track } /** - * Decode an untracked DialCache frame returned as a Redis bulk string. - * Missing, short, and unsupported-version frames are cache misses. Invalid - * runtime reply types and unsupported payload encodings throw typed errors. + * Decode an untracked DialCache frame returned as a Redis bulk string into + * its serializer payload and header creation time (the writer's informational + * client clock). Missing, short, and unsupported-version frames are cache + * misses. Invalid runtime reply types and unsupported payload encodings throw + * typed errors. */ -export function decodeRedisFrame(raw: unknown): RedisCachePayload | null { +export function decodeRedisFrame(raw: unknown): DecodedRedisFrame | null { const frame = validateRedisBulkStringReply(raw); - return isSupportedRedisFrame(frame) - ? decodeRedisPayload(frame.subarray(REDIS_FRAME_HEADER_BYTES)) - : null; + if (!isSupportedRedisFrame(frame)) { + return null; + } + return { + payload: decodeRedisPayload(frame.subarray(REDIS_FRAME_HEADER_BYTES)), + createdAtMs: readFrameCreatedAtMs(frame), + }; } /** * Decode a tracked DialCache frame against a watermark from the same atomic, - * authoritative snapshot. Missing or malformed state and frames created at or - * before the watermark are cache misses. Invalid runtime reply types and - * unsupported payload encodings throw typed errors. + * authoritative snapshot into its serializer payload and header creation time + * (Redis server time written by the stamp script). Missing or malformed state + * and frames created at or before the watermark are cache misses. Invalid + * runtime reply types and unsupported payload encodings throw typed errors. */ export function decodeTrackedRedisFrame( raw: unknown, rawWatermark: unknown, -): RedisCachePayload | null { +): DecodedRedisFrame | null { const frame = validateRedisBulkStringReply(raw); const watermarkFrame = validateRedisBulkStringReply(rawWatermark); if (!isSupportedRedisFrame(frame)) { @@ -148,8 +156,15 @@ export function decodeTrackedRedisFrame( if (watermark === null) { return null; } - const createdAtMs = Number(frame.readBigUInt64BE(REDIS_FRAME_TIMESTAMP_OFFSET)); + const createdAtMs = readFrameCreatedAtMs(frame); return createdAtMs <= watermark ? null - : decodeRedisPayload(frame.subarray(REDIS_FRAME_HEADER_BYTES)); + : { + payload: decodeRedisPayload(frame.subarray(REDIS_FRAME_HEADER_BYTES)), + createdAtMs, + }; +} + +function readFrameCreatedAtMs(frame: Buffer): number { + return Number(frame.readBigUInt64BE(REDIS_FRAME_TIMESTAMP_OFFSET)); } diff --git a/src/metrics.ts b/src/metrics.ts index 58f8574..4727644 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -116,6 +116,17 @@ export interface DialCacheMetricsAdapter { coalesced?(labels: CoalescedMetricLabels): void; // Optional so existing custom adapters keep compiling without changes. shadowValidation?(labels: ShadowValidationMetricLabels): void; + /** + * Age in seconds of the validated Redis value at shadow comparison time: + * the observing process's epoch clock minus the frame header's + * `createdAtMs`, clamped at zero. Emitted only alongside terminal `match` + * and `mismatch` outcomes; other outcomes deliver no verdict on a retained + * value. Tracked frames are stamped with Redis server time and untracked + * frames with the writer's client clock, so the age mixes clocks and is + * coarse operational evidence, not a precise measurement. Optional so + * existing custom adapters keep compiling without changes. + */ + observeShadowValueAge?(labels: ShadowValidationMetricLabels, seconds: number): void; // Optional so existing custom adapters keep compiling without changes. compression?(labels: CompressionMetricLabels): void; observeGet(labels: CacheMetricLabels, seconds: number): void; diff --git a/src/prometheus.ts b/src/prometheus.ts index c379776..5b5ae36 100644 --- a/src/prometheus.ts +++ b/src/prometheus.ts @@ -58,6 +58,8 @@ interface CollectorShape { const TIMER_BUCKETS = [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10]; const SIZE_BUCKETS = [100, 1_000, 10_000, 100_000, 1_000_000, 10_000_000]; const RATIO_BUCKETS = [0.05, 0.1, 0.2, 0.3, 0.5, 0.7, 0.9, 1]; +// Value ages span seconds to the 365-day TTL ceiling: 1s..15m, then 1h, 3h, 12h, 1d, 3d, 7d. +const VALUE_AGE_BUCKETS = [1, 5, 15, 60, 300, 900, 3_600, 10_800, 43_200, 86_400, 259_200, 604_800]; export class PrometheusDialCacheMetrics implements DialCacheMetricsAdapter { private readonly requestCounter: Counter; @@ -67,6 +69,7 @@ export class PrometheusDialCacheMetrics implements DialCacheMetricsAdapter { private readonly invalidationCounter: Counter; private readonly coalescedCounter: Counter; private readonly shadowValidationCounter: Counter; + private readonly shadowValueAgeHistogram: Histogram; private readonly compressionCounter: Counter; private readonly getTimer: Histogram; private readonly fallbackTimer: Histogram; @@ -89,6 +92,7 @@ export class PrometheusDialCacheMetrics implements DialCacheMetricsAdapter { this.invalidationCounter = counter(registry, collectors.invalidationCounter); this.coalescedCounter = counter(registry, collectors.coalescedCounter); this.shadowValidationCounter = counter(registry, collectors.shadowValidationCounter); + this.shadowValueAgeHistogram = histogram(registry, collectors.shadowValueAgeHistogram); this.compressionCounter = counter(registry, collectors.compressionCounter); this.getTimer = histogram(registry, collectors.getTimer); this.fallbackTimer = histogram(registry, collectors.fallbackTimer); @@ -137,12 +141,11 @@ export class PrometheusDialCacheMetrics implements DialCacheMetricsAdapter { } shadowValidation(labels: ShadowValidationMetricLabels): void { - this.shadowValidationCounter.inc({ - cache_namespace: labels.cacheNamespace, - use_case: labels.useCase, - key_type: labels.keyType, - outcome: labels.outcome, - }); + this.shadowValidationCounter.inc(shadowValidationLabels(labels)); + } + + observeShadowValueAge(labels: ShadowValidationMetricLabels, seconds: number): void { + this.shadowValueAgeHistogram.observe(shadowValidationLabels(labels), seconds); } compression(labels: CompressionMetricLabels): void { @@ -191,6 +194,15 @@ function cacheLabels(labels: CacheMetricLabels): Record { }; } +function shadowValidationLabels(labels: ShadowValidationMetricLabels): Record { + return { + cache_namespace: labels.cacheNamespace, + use_case: labels.useCase, + key_type: labels.keyType, + outcome: labels.outcome, + }; +} + function collectorConfigs(prefix: string) { return { disabledCounter: { @@ -235,6 +247,13 @@ function collectorConfigs(prefix: string) { help: "Sampled DialCache Redis shadow-validation outcomes.", labelNames: ["cache_namespace", "use_case", "key_type", "outcome"], }, + shadowValueAgeHistogram: { + type: "histogram", + name: `${prefix}dialcache_shadow_value_age_histogram`, + help: "Age in seconds of the validated Redis value at DialCache shadow comparison time.", + labelNames: ["cache_namespace", "use_case", "key_type", "outcome"], + buckets: VALUE_AGE_BUCKETS, + }, compressionCounter: { type: "counter", name: `${prefix}dialcache_compression_counter`, diff --git a/src/redis-client.ts b/src/redis-client.ts index 93f0c1e..4fdcfc4 100644 --- a/src/redis-client.ts +++ b/src/redis-client.ts @@ -87,6 +87,20 @@ export class DialCacheRedisPlaceholderLostError extends Error { /** Serialized cache data, independent of any Redis client or wire framing. */ export type RedisCachePayload = string | Buffer; +/** + * A served Redis frame: the decoded serializer payload plus the creation time + * from the frame header. Tracked frames carry Redis server time written by the + * stamp script; untracked frames carry the writer's informational client + * clock. DialCache consumes `createdAtMs` only for observability (the shadow + * value-age observation) — tracked watermark fencing already happened inside + * the decoder — so it never affects serving decisions. + */ +export interface DecodedRedisFrame { + readonly payload: RedisCachePayload; + /** Epoch milliseconds copied from the frame header. */ + readonly createdAtMs: number; +} + interface RedisValueRequest { readonly valueKey: string; } @@ -144,9 +158,10 @@ export interface RedisInvalidationRequest { */ export interface DialCacheRedisClient { /** - * Read a DialCache Redis frame and return its decoded serializer payload. - * Implementations must use `decodeRedisFrame` / `decodeTrackedRedisFrame` - * from `dialcache/redis-protocol`, or preserve their exact behavior. + * Read a DialCache Redis frame and return its decoded serializer payload + * together with the frame header's creation time. Implementations must use + * `decodeRedisFrame` / `decodeTrackedRedisFrame` from + * `dialcache/redis-protocol`, or preserve their exact behavior. * * Raw values are Redis bulk strings (`Buffer`) or null. A missing value, a * frame shorter than the version/timestamp/encoding header, or an @@ -159,13 +174,13 @@ export interface DialCacheRedisClient { * Tracked implementations must read the value and watermark atomically from * one authoritative snapshot; replica lag must not hide an invalidation. * - * A non-null payload is transferred to DialCache. A returned Buffer must - * remain stable and must not be mutated, pooled, or reused after this method - * settles; DialCache may retain it beyond the request for best-effort shadow - * deserialization. Adapters that recycle response storage must return a - * dedicated Buffer. + * A non-null frame is transferred to DialCache. A returned Buffer payload + * must remain stable and must not be mutated, pooled, or reused after this + * method settles; DialCache may retain it beyond the request for + * best-effort shadow deserialization. Adapters that recycle response + * storage must return a dedicated Buffer. */ - read(request: RedisReadRequest, context?: RedisReadContext): Awaitable; + read(request: RedisReadRequest, context?: RedisReadContext): Awaitable; /** * Write a DialCache Redis frame using the `dialcache/redis-protocol` * encoders, or preserve their exact behavior. diff --git a/src/redis-protocol.ts b/src/redis-protocol.ts index 0afaeb6..10642c0 100644 --- a/src/redis-protocol.ts +++ b/src/redis-protocol.ts @@ -4,9 +4,10 @@ * These exports encode frames and mint tracked placeholders (use * `encodeRedisFrame` and `encodeTrackedRedisPlaceholder` rather than * reimplementing them — see the latter's JSDoc for the nonce contract), - * decode a frame into its payload bytes, resolve and validate mutation - * replies, guard the write-TTL acceptance domain, and carry the tracked - * stamp and invalidation Lua sources the bundled adapters dispatch. The + * decode a frame into its payload bytes and header creation time, resolve + * and validate mutation replies, guard the write-TTL acceptance domain, and + * carry the tracked stamp and invalidation Lua sources the bundled adapters + * dispatch. The * payload region past the header is opaque at this layer: entries written by * DialCache releases with payload compression may begin with a compression * envelope byte (0x00 escape, 0x01/0x02 zstd; see the README Compression @@ -25,6 +26,7 @@ export { encodeTrackedRedisPlaceholder, type TrackedRedisPlaceholder, } from "./internal/redis-payload.js"; +export type { DecodedRedisFrame } from "./redis-client.js"; export { resolveTrackedRedisWriteReply, validateRedisScriptInvalidationReply, diff --git a/test/datadog.test.ts b/test/datadog.test.ts index cfd083c..fbb6b93 100644 --- a/test/datadog.test.ts +++ b/test/datadog.test.ts @@ -160,6 +160,15 @@ describe("Datadog metrics adapter", () => { keyType: "user_id", outcome: "match", }); + metrics.observeShadowValueAge( + { + cacheNamespace: cacheLabels.cacheNamespace, + useCase: "LoadUser", + keyType: "user_id", + outcome: "mismatch", + }, + 42.5, + ); metrics.compression({ ...cacheLabels, outcome: "compressed" }); metrics.observeGet(cacheLabels, 0.125); metrics.observeFallback(cacheLabels, 0.5); @@ -203,6 +212,12 @@ describe("Datadog metrics adapter", () => { value: 1, tags: { cache_namespace: "users", use_case: "LoadUser", key_type: "user_id", outcome: "match" }, }, + { + method: "distribution", + name: "dialcache.shadow.value_age", + value: 42.5, + tags: { cache_namespace: "users", use_case: "LoadUser", key_type: "user_id", outcome: "mismatch" }, + }, { method: "increment", name: "dialcache.compression.count", @@ -247,6 +262,15 @@ describe("Datadog metrics adapter", () => { metrics.observeStoredSize(cacheLabels, 96); metrics.observeCompressionRatio(cacheLabels, 0.04); metrics.observeCompression({ ...cacheLabels, operation: "decompress" }, 0.05); + metrics.observeShadowValueAge( + { + cacheNamespace: cacheLabels.cacheNamespace, + useCase: cacheLabels.useCase, + keyType: cacheLabels.keyType, + outcome: "match", + }, + 60, + ); expect(client.calls.map(({ method, name, value }) => ({ method, name, value }))).toEqual([ { method: observationMetricType, name: "service.cache.get.duration", value: 0.01 }, @@ -256,6 +280,7 @@ describe("Datadog metrics adapter", () => { { method: observationMetricType, name: "service.cache.stored.size", value: 96 }, { method: observationMetricType, name: "service.cache.compression.ratio", value: 0.04 }, { method: observationMetricType, name: "service.cache.compression.duration", value: 0.05 }, + { method: observationMetricType, name: "service.cache.shadow.value_age", value: 60 }, ]); }); } diff --git a/test/dialcache-liveness.test.ts b/test/dialcache-liveness.test.ts index dc20218..8f31305 100644 --- a/test/dialcache-liveness.test.ts +++ b/test/dialcache-liveness.test.ts @@ -588,7 +588,7 @@ describe("DialCache fallback liveness", () => { }, }; const redis: DialCacheRedisClient = { - read: async () => "stored", + read: async () => ({ payload: "stored", createdAtMs: Date.now() }), write: async () => true, invalidate: async () => undefined, }; diff --git a/test/dialcache-redis-read-deadline.test.ts b/test/dialcache-redis-read-deadline.test.ts index acc57bd..211c56c 100644 --- a/test/dialcache-redis-read-deadline.test.ts +++ b/test/dialcache-redis-read-deadline.test.ts @@ -9,9 +9,9 @@ import { DialCacheKeyConfig, RedisReadTimeoutError, type CachedOptions, + type DecodedRedisFrame, type DialCacheMetricsAdapter, type DialCacheRedisClient, - type RedisCachePayload, type RedisConfig, type RedisReadContext, } from "../src/index.js"; @@ -327,7 +327,7 @@ describe("DialCache Redis read deadlines", () => { const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); const redis = redisClient( vi.fn() - .mockResolvedValueOnce(JSON.stringify({ source: "redis" })) + .mockResolvedValueOnce({ payload: JSON.stringify({ source: "redis" }), createdAtMs: Date.now() }) .mockResolvedValueOnce(null), ); const dialcache = new DialCache({ redis: { client: redis.client, readTimeoutMs: 100 } }); @@ -605,13 +605,13 @@ describe("DialCache Redis read deadlines", () => { it.each(["fulfillment", "rejection"] as const)( "consumes late read %s and lets a later invocation recover", async (settlement) => { - const firstRead = deferred(); + const firstRead = deferred(); let readCalls = 0; const redis = redisClient(async () => { readCalls += 1; return readCalls === 1 ? await firstRead.promise - : JSON.stringify({ source: "redis" }); + : { payload: JSON.stringify({ source: "redis" }), createdAtMs: Date.now() }; }); const error = vi.fn(); const logger = { debug: vi.fn(), warn: vi.fn(), error: vi.fn() }; @@ -634,7 +634,7 @@ describe("DialCache Redis read deadlines", () => { await expect(dialcache.enable(async () => await load())).resolves.toEqual({ source: "redis" }); if (settlement === "fulfillment") { - firstRead.resolve(JSON.stringify({ source: "late" })); + firstRead.resolve({ payload: JSON.stringify({ source: "late" }), createdAtMs: Date.now() }); } else { firstRead.reject(new Error("late Redis failure")); } diff --git a/test/dialcache-redis.test.ts b/test/dialcache-redis.test.ts index b59432d..3c9abfd 100644 --- a/test/dialcache-redis.test.ts +++ b/test/dialcache-redis.test.ts @@ -301,12 +301,12 @@ describe("DialCache Redis TTL layer", () => { await redis.write({ valueKey, cacheTtlMs: 60_000, value: payload }); const firstRead = await redis.read({ valueKey }); - if (!Buffer.isBuffer(firstRead)) { + if (!Buffer.isBuffer(firstRead?.payload)) { throw new Error("Expected a binary Redis payload"); } - firstRead[0] = 0xff; + firstRead.payload[0] = 0xff; - expect(await redis.read({ valueKey })).toEqual(payload); + expect((await redis.read({ valueKey }))?.payload).toEqual(payload); }); it("fails open when Redis serializer dump fails", async () => { diff --git a/test/dialcache-shadow-confirmation.test.ts b/test/dialcache-shadow-confirmation.test.ts index c024aef..8e32e57 100644 --- a/test/dialcache-shadow-confirmation.test.ts +++ b/test/dialcache-shadow-confirmation.test.ts @@ -10,6 +10,7 @@ import { FallbackTimeoutError, type CacheMetricLabels, type CoalescedMetricLabels, + type DecodedRedisFrame, type DialCacheConfig, type DialCacheMetricsAdapter, type DialCacheRedisClient, @@ -54,22 +55,26 @@ function deferred(): Deferred { type ReadStep = () => RedisCachePayload | null | Promise; +const SCRIPTED_FRAME_CREATED_AT_MS = 1_700_000_000_000; + class ScriptedRedis implements DialCacheRedisClient { readonly requests: RedisReadRequest[] = []; readonly contexts: Array = []; readonly write = vi.fn(async (_request: RedisWriteRequest): Promise => true); readonly invalidate = vi.fn(async (_request: RedisInvalidationRequest): Promise => undefined); + frameCreatedAtMs = SCRIPTED_FRAME_CREATED_AT_MS; constructor(private readonly steps: ReadStep[]) {} - async read(request: RedisReadRequest, context?: RedisReadContext): Promise { + async read(request: RedisReadRequest, context?: RedisReadContext): Promise { this.requests.push(request); this.contexts.push(context); const step = this.steps.shift(); if (step === undefined) { throw new Error("Unexpected Redis read"); } - return await step(); + const payload = await step(); + return payload === null ? null : { payload, createdAtMs: this.frameCreatedAtMs }; } } @@ -90,9 +95,15 @@ interface OrdinaryMetricEvent { readonly labels: Record; } +interface ShadowAgeEvent { + readonly labels: ShadowValidationMetricLabels; + readonly seconds: number; +} + class RecordingMetrics implements DialCacheMetricsAdapter { readonly ordinaryEvents: OrdinaryMetricEvent[] = []; readonly shadowEvents: ShadowValidationMetricLabels[] = []; + readonly shadowAgeEvents: ShadowAgeEvent[] = []; request(labels: CacheMetricLabels): void { this.record("request", labels); @@ -122,6 +133,10 @@ class RecordingMetrics implements DialCacheMetricsAdapter { this.shadowEvents.push({ ...labels }); } + observeShadowValueAge(labels: ShadowValidationMetricLabels, seconds: number): void { + this.shadowAgeEvents.push({ labels: { ...labels }, seconds }); + } + observeGet(labels: CacheMetricLabels, _seconds: number): void { this.record("get", labels); } @@ -529,11 +544,42 @@ describe("DialCache Redis shadow confirmation", () => { await waitForShadowEvents(metrics, 1); expect(metrics.shadowEvents.map(({ outcome }) => outcome)).toEqual(["superseded"]); + expect(metrics.shadowAgeEvents).toEqual([]); expect(serializer.load).toHaveBeenCalledTimes(2); expect(serializer.dump).not.toHaveBeenCalled(); expectTrackedReads(redis, 2); }); + it("records the validated value age only for a confirmed mismatch verdict", async () => { + const nowMs = 1_700_000_090_000; + const nowSpy = vi.spyOn(Date, "now").mockReturnValue(nowMs); + try { + const payload = JSON.stringify({ id: "123", version: 1 }); + const redis = new ScriptedRedis([() => payload, () => payload]); + redis.frameCreatedAtMs = nowMs - 90_000; + const metrics = new RecordingMetrics(); + const dialcache = createCache(redis, metrics); + const getUser = dialcache.cached(async () => ({ id: "123", version: 2 }), { + ...trackedOptions("ShadowMismatchValueAge", remoteConfig(100)), + cacheKey: () => "123", + }); + + await dialcache.enable(async () => await getUser()); + await waitForShadowEvents(metrics, 1); + + expect(metrics.shadowEvents.map(({ outcome }) => outcome)).toEqual(["mismatch"]); + expect(metrics.shadowAgeEvents).toHaveLength(1); + expect(metrics.shadowAgeEvents[0]?.seconds).toBe(90); + expect(metrics.shadowAgeEvents[0]?.labels).toMatchObject({ + useCase: "ShadowMismatchValueAge", + keyType: "user_id", + outcome: "mismatch", + }); + } finally { + nowSpy.mockRestore(); + } + }); + it("does not log a mismatch candidate when C1 is superseded", async () => { const original = JSON.stringify({ id: "123", version: 1 }); const confirmation = JSON.stringify({ id: "123", version: 3 }); @@ -944,6 +990,7 @@ describe("DialCache Redis shadow confirmation", () => { await waitForShadowEvents(metrics, 1); expect(metrics.shadowEvents.map(({ outcome }) => outcome)).toEqual(["filled"]); + expect(metrics.shadowAgeEvents).toEqual([]); expect(source).toHaveBeenCalledOnce(); if (tracked) { expectTrackedReads(redis, 1); diff --git a/test/dialcache-shadow-validation.test.ts b/test/dialcache-shadow-validation.test.ts index dafbf25..0eed650 100644 --- a/test/dialcache-shadow-validation.test.ts +++ b/test/dialcache-shadow-validation.test.ts @@ -20,8 +20,14 @@ import { import { deterministicShadowRampSample } from "../src/internal/ramp.js"; import { encodeFrame, FakeRedis } from "./fake-redis.js"; +interface ShadowAgeEvent { + readonly labels: ShadowValidationMetricLabels; + readonly seconds: number; +} + class RecordingMetrics implements DialCacheMetricsAdapter { readonly shadowEvents: ShadowValidationMetricLabels[] = []; + readonly shadowAgeEvents: ShadowAgeEvent[] = []; readonly errorEvents: ErrorMetricLabels[] = []; request(_labels: CacheMetricLabels): void {} @@ -38,6 +44,10 @@ class RecordingMetrics implements DialCacheMetricsAdapter { this.shadowEvents.push({ ...labels }); } + observeShadowValueAge(labels: ShadowValidationMetricLabels, seconds: number): void { + this.shadowAgeEvents.push({ labels: { ...labels }, seconds }); + } + observeGet(_labels: CacheMetricLabels, _seconds: number): void {} observeFallback(_labels: CacheMetricLabels, _seconds: number): void {} observeSerialization(_labels: SerializationMetricLabels, _seconds: number): void {} @@ -90,6 +100,7 @@ function seedRedis( readonly useCase: string; readonly payload: string | Buffer; readonly tracked?: boolean; + readonly createdAtMs?: number; }, ): DialCacheKey { const key = new DialCacheKey({ @@ -100,7 +111,7 @@ function seedRedis( }); redis.setRaw( `${key.urn}:dialcache-frame-v1`, - encodeFrame(options.payload, Date.now(), Buffer.isBuffer(options.payload) ? 1 : 0), + encodeFrame(options.payload, options.createdAtMs ?? Date.now(), Buffer.isBuffer(options.payload) ? 1 : 0), ); if (key.trackForInvalidation) { redis.setRaw(`${key.prefix}#watermark`, "0"); @@ -263,6 +274,104 @@ describe("DialCache Redis shadow validation", () => { expect(redis.setCalls).toBe(0); }); + it("records the validated value age alongside a match verdict", async () => { + const nowMs = 1_700_000_000_000; + const nowSpy = vi.spyOn(Date, "now").mockReturnValue(nowMs); + try { + const redis = new FakeRedis(); + const metrics = new RecordingMetrics(); + const useCase = "ShadowValueAgeMatch"; + const cachedValue = { id: "123", version: 1 }; + seedRedis(redis, { + id: "123", + useCase, + payload: JSON.stringify(cachedValue), + createdAtMs: nowMs - 45_000, + }); + const dialcache = createShadowCache(redis, metrics); + const getUser = dialcache.cached(async () => cachedValue, { + ...trackedRemoteDefaults(useCase), + cacheKey: () => "123", + }); + + expect(await dialcache.enable(async () => await getUser())).toEqual(cachedValue); + await waitForShadowEvents(metrics, 1); + + expect(metrics.shadowEvents[0]?.outcome).toBe("match"); + expect(metrics.shadowAgeEvents).toHaveLength(1); + expect(metrics.shadowAgeEvents[0]?.seconds).toBe(45); + expect(metrics.shadowAgeEvents[0]?.labels).toMatchObject({ + useCase, + keyType: "user_id", + outcome: "match", + }); + } finally { + nowSpy.mockRestore(); + } + }); + + it("records the mismatched value age from the retained frame's creation time", async () => { + const nowMs = 1_700_000_000_000; + const nowSpy = vi.spyOn(Date, "now").mockReturnValue(nowMs); + try { + const redis = new FakeRedis(); + const metrics = new RecordingMetrics(); + const useCase = "ShadowValueAgeMismatch"; + seedRedis(redis, { + id: "123", + useCase, + payload: JSON.stringify({ id: "123", version: 1 }), + createdAtMs: nowMs - 120_000, + }); + const dialcache = createShadowCache(redis, metrics); + const getUser = dialcache.cached(async () => ({ id: "123", version: 2 }), { + ...trackedRemoteDefaults(useCase), + cacheKey: () => "123", + }); + + expect(await dialcache.enable(async () => await getUser())).toEqual({ id: "123", version: 1 }); + await waitForShadowEvents(metrics, 1); + + expect(metrics.shadowEvents[0]?.outcome).toBe("mismatch"); + expect(metrics.shadowAgeEvents).toHaveLength(1); + expect(metrics.shadowAgeEvents[0]?.seconds).toBe(120); + expect(metrics.shadowAgeEvents[0]?.labels).toMatchObject({ useCase, outcome: "mismatch" }); + } finally { + nowSpy.mockRestore(); + } + }); + + it("clamps a future-stamped frame to a zero value age instead of a negative one", async () => { + const nowMs = 1_700_000_000_000; + const nowSpy = vi.spyOn(Date, "now").mockReturnValue(nowMs); + try { + const redis = new FakeRedis(); + const metrics = new RecordingMetrics(); + const useCase = "ShadowValueAgeClamped"; + const cachedValue = { id: "123" }; + seedRedis(redis, { + id: "123", + useCase, + payload: JSON.stringify(cachedValue), + createdAtMs: nowMs + 60_000, + }); + const dialcache = createShadowCache(redis, metrics); + const getUser = dialcache.cached(async () => cachedValue, { + ...trackedRemoteDefaults(useCase), + cacheKey: () => "123", + }); + + expect(await dialcache.enable(async () => await getUser())).toEqual(cachedValue); + await waitForShadowEvents(metrics, 1); + + expect(metrics.shadowEvents[0]?.outcome).toBe("match"); + expect(metrics.shadowAgeEvents).toHaveLength(1); + expect(metrics.shadowAgeEvents[0]?.seconds).toBe(0); + } finally { + nowSpy.mockRestore(); + } + }); + it("re-deserializes the retained payload instead of comparing a caller-mutated hit", async () => { const redis = new FakeRedis(); const metrics = new RecordingMetrics(); diff --git a/test/fake-redis.ts b/test/fake-redis.ts index 93f5102..4633b99 100644 --- a/test/fake-redis.ts +++ b/test/fake-redis.ts @@ -1,4 +1,5 @@ import type { + DecodedRedisFrame, DialCacheRedisClient, RedisCachePayload, RedisInvalidationRequest, @@ -27,7 +28,7 @@ export class FakeRedis implements DialCacheRedisClient { failWatermarkGet = false; getGate: Promise | null = null; - async read({ valueKey, watermarkKey }: RedisReadRequest): Promise { + async read({ valueKey, watermarkKey }: RedisReadRequest): Promise { if (watermarkKey === undefined) { this.getCalls += 1; } else { @@ -122,12 +123,13 @@ export class FakeRedis implements DialCacheRedisClient { } } - private readPayload(valueKey: string, watermarkKey: string | null): RedisCachePayload | null { + private readPayload(valueKey: string, watermarkKey: string | null): DecodedRedisFrame | null { const raw = this.readRaw(valueKey); if (raw === null || raw.length < PAYLOAD_OFFSET || raw[0] !== FRAME_VERSION) { return null; } + const createdAtMs = Number(readTimestamp(raw)); if (watermarkKey !== null) { let watermark: number | null; try { @@ -135,17 +137,17 @@ export class FakeRedis implements DialCacheRedisClient { } catch { return null; } - if (watermark === null || Number(readTimestamp(raw)) <= watermark) { + if (watermark === null || createdAtMs <= watermark) { return null; } } const encoding = raw[ENCODING_OFFSET]; if (encoding === 0) { - return raw.subarray(PAYLOAD_OFFSET).toString("utf8"); + return { payload: raw.subarray(PAYLOAD_OFFSET).toString("utf8"), createdAtMs }; } if (encoding === 1) { - return Buffer.from(raw.subarray(PAYLOAD_OFFSET)); + return { payload: Buffer.from(raw.subarray(PAYLOAD_OFFSET)), createdAtMs }; } throw new DialCacheRedisPayloadEncodingError("Invalid DialCache Redis payload encoding"); } diff --git a/test/node-redis.test.ts b/test/node-redis.test.ts index b91cc8e..258e3aa 100644 --- a/test/node-redis.test.ts +++ b/test/node-redis.test.ts @@ -136,10 +136,13 @@ describe("node-redis adapter", () => { }); const adapter = createNodeRedisDialCacheClient(client as never); - await expect(adapter.read({ valueKey: "plain:value" })).resolves.toBe("plain"); + await expect(adapter.read({ valueKey: "plain:value" })).resolves.toEqual({ + payload: "plain", + createdAtMs: 1, + }); await expect( adapter.read({ valueKey: "tracked:{id}:value", watermarkKey: "tracked:{id}:watermark" }), - ).resolves.toEqual(Buffer.from([0, 0xff])); + ).resolves.toEqual({ payload: Buffer.from([0, 0xff]), createdAtMs: 2 }); await expect( adapter.write({ valueKey: "plain:value", cacheTtlMs: 1_000, value: "plain" }), ).resolves.toBe(true); @@ -410,7 +413,7 @@ describe("node-redis adapter", () => { await expect(adapter.read( { valueKey: "tracked:{id}:value", watermarkKey: "tracked:{id}:watermark" }, { timeoutMs: 25, signal: controller.signal }, - )).resolves.toBe("tracked"); + )).resolves.toEqual({ payload: "tracked", createdAtMs: 2 }); expect(client.sendCommand).toHaveBeenCalledWith( "tracked:{id}:value", @@ -432,7 +435,7 @@ describe("node-redis adapter", () => { await expect(adapter.read({ valueKey: "tracked:{id}:value", watermarkKey: "tracked:{id}:watermark", - })).resolves.toBe("tracked"); + })).resolves.toEqual({ payload: "tracked", createdAtMs: 2 }); expect(client.sendCommand).toHaveBeenCalledWith( ["MGET", "tracked:{id}:value", "tracked:{id}:watermark"], diff --git a/test/prometheus.test.ts b/test/prometheus.test.ts index 89bbc62..7508886 100644 --- a/test/prometheus.test.ts +++ b/test/prometheus.test.ts @@ -182,6 +182,15 @@ describe("Prometheus metrics adapter", () => { keyType: labels.keyType, outcome: "match", }); + metrics.observeShadowValueAge( + { + cacheNamespace: labels.cacheNamespace, + useCase: labels.useCase, + keyType: labels.keyType, + outcome: "match", + }, + 42, + ); metrics.compression({ ...labels, outcome: "compressed" }); metrics.observeGet(labels, 0.05); metrics.observeFallback(labels, 0.05); @@ -237,6 +246,11 @@ describe("Prometheus metrics adapter", () => { "schema_dialcache_shadow_validation_counter", ["cache_namespace", "use_case", "key_type", "outcome"], ), + histogramSchema( + "schema_dialcache_shadow_value_age_histogram", + ["cache_namespace", "use_case", "key_type", "outcome"], + VALUE_AGE_BUCKETS, + ), histogramSchema("schema_dialcache_size_histogram", ["cache_namespace", "use_case", "key_type", "layer"], SIZE_BUCKETS), histogramSchema( "schema_dialcache_stored_size_histogram", @@ -646,6 +660,7 @@ describe("Prometheus metrics adapter", () => { const TIMER_BUCKETS = [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, "+Inf"]; const SIZE_BUCKETS = [100, 1_000, 10_000, 100_000, 1_000_000, 10_000_000, "+Inf"]; const RATIO_BUCKETS = [0.05, 0.1, 0.2, 0.3, 0.5, 0.7, 0.9, 1, "+Inf"]; +const VALUE_AGE_BUCKETS = [1, 5, 15, 60, 300, 900, 3_600, 10_800, 43_200, 86_400, 259_200, 604_800, "+Inf"]; interface MetricValue { readonly metricName?: string; diff --git a/test/redis-cluster.integration.test.ts b/test/redis-cluster.integration.test.ts index 32c1a09..bdffb16 100644 --- a/test/redis-cluster.integration.test.ts +++ b/test/redis-cluster.integration.test.ts @@ -258,7 +258,9 @@ describe("DialCache Redis protocol on Redis Cluster", () => { const payload = Buffer.from(Array.from({ length: 256 }, (_, index) => index)); expect(await scriptClient.write({ valueKey, cacheTtlMs: 60_000, value: payload })).toBe(true); - expect(await scriptClient.read({ valueKey })).toEqual(payload); + const untrackedRead = await scriptClient.read({ valueKey }); + expect(untrackedRead?.payload).toEqual(payload); + expect(untrackedRead?.createdAtMs).toBeGreaterThan(0); const stored = await cluster.get(commandOptions({ returnBuffers: true }), valueKey); expect(stored?.length).toBe(10 + payload.length); @@ -276,7 +278,9 @@ describe("DialCache Redis protocol on Redis Cluster", () => { value: trackedPayload, }), ).toBe(true); - expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toEqual(trackedPayload); + const trackedRead = await scriptClient.read({ valueKey: trackedValueKey, watermarkKey }); + expect(trackedRead?.payload).toEqual(trackedPayload); + expect(trackedRead?.createdAtMs).toBeGreaterThan(0); }); it("runs GLIDE tracked mutations against the real cluster", async (ctx) => { @@ -290,7 +294,7 @@ describe("DialCache Redis protocol on Redis Cluster", () => { expect( await adapter.write({ valueKey, watermarkKey, cacheTtlMs: 60_000, value: "glide" }), ).toBe(true); - expect(await adapter.read({ valueKey, watermarkKey })).toBe("glide"); + expect((await adapter.read({ valueKey, watermarkKey }))?.payload).toBe("glide"); await adapter.invalidate({ watermarkKey, futureBufferMs: 0 }); // The follow-up write's stamp is fenced unless server time passes the @@ -300,11 +304,11 @@ describe("DialCache Redis protocol on Redis Cluster", () => { expect( await adapter.write({ valueKey, watermarkKey, cacheTtlMs: 60_000, value: "glide-2" }), ).toBe(true); - expect(await adapter.read({ valueKey, watermarkKey })).toBe("glide-2"); + expect((await adapter.read({ valueKey, watermarkKey }))?.payload).toBe("glide-2"); const untrackedKey = "glide-cluster:{item:untracked}:value"; expect(await adapter.write({ valueKey: untrackedKey, cacheTtlMs: 60_000, value: "plain" })).toBe(true); - expect(await adapter.read({ valueKey: untrackedKey })).toBe("plain"); + expect((await adapter.read({ valueKey: untrackedKey }))?.payload).toBe("plain"); await expect(adapter.write({ valueKey: "{glide-a}:value", @@ -335,7 +339,7 @@ describe("DialCache Redis protocol on Redis Cluster", () => { expect( await adapter.write({ valueKey, watermarkKey, cacheTtlMs: 60_000, value: "recovered" }), ).toBe(true); - expect(await adapter.read({ valueKey, watermarkKey })).toBe("recovered"); + expect((await adapter.read({ valueKey, watermarkKey }))?.payload).toBe("recovered"); await flushAllMasters(); await expect(adapter.invalidate({ watermarkKey, futureBufferMs: 0 })).resolves.toBeUndefined(); diff --git a/test/redis-payload.test.ts b/test/redis-payload.test.ts index 144b172..3eb3d12 100644 --- a/test/redis-payload.test.ts +++ b/test/redis-payload.test.ts @@ -27,18 +27,19 @@ function encodeFrame( describe("Redis frame decoding", () => { it("decodes UTF-8 and binary payloads without copying binary data", () => { - expect(decodeRedisFrame(encodeFrame("cached"))).toBe("cached"); + expect(decodeRedisFrame(encodeFrame("cached"))).toEqual({ payload: "cached", createdAtMs: 1_000 }); - const frame = encodeFrame(Buffer.from([0, 0xff, 0x80]), 1); + const frame = encodeFrame(Buffer.from([0, 0xff, 0x80]), 1, 2_000); const decoded = decodeRedisFrame(frame); - expect(decoded).toEqual(Buffer.from([0, 0xff, 0x80])); - expect(Buffer.isBuffer(decoded)).toBe(true); - if (!Buffer.isBuffer(decoded)) { + expect(decoded).toEqual({ payload: Buffer.from([0, 0xff, 0x80]), createdAtMs: 2_000 }); + const payload = decoded?.payload; + expect(Buffer.isBuffer(payload)).toBe(true); + if (!Buffer.isBuffer(payload)) { throw new Error("Expected a binary Redis payload"); } - expect(decoded.buffer).toBe(frame.buffer); - expect(decoded.byteOffset).toBe(frame.byteOffset + 10); - expect(decoded.byteLength).toBe(frame.byteLength - 10); + expect(payload.buffer).toBe(frame.buffer); + expect(payload.byteOffset).toBe(frame.byteOffset + 10); + expect(payload.byteLength).toBe(frame.byteLength - 10); }); it("treats missing, short, and unsupported frames as misses", () => { @@ -65,9 +66,10 @@ describe("Redis frame decoding", () => { it("validates tracked frames against integer and fractional watermarks", () => { const frame = encodeFrame("cached", 0, 1_000); + const decoded = { payload: "cached", createdAtMs: 1_000 }; - expect(decodeTrackedRedisFrame(frame, Buffer.from("999"))).toBe("cached"); - expect(decodeTrackedRedisFrame(frame, Buffer.from("999.5"))).toBe("cached"); + expect(decodeTrackedRedisFrame(frame, Buffer.from("999"))).toEqual(decoded); + expect(decodeTrackedRedisFrame(frame, Buffer.from("999.5"))).toEqual(decoded); expect(decodeTrackedRedisFrame(frame, Buffer.from("1000"))).toBeNull(); expect(decodeTrackedRedisFrame(frame, Buffer.from("1000.5"))).toBeNull(); }); @@ -106,18 +108,18 @@ describe("Redis frame decoding", () => { expect(utf8[0]).toBe(1); expect(Number(utf8.readBigUInt64BE(1))).toBe(1_000); expect(utf8[9]).toBe(0); - expect(decodeRedisFrame(utf8)).toBe("cachéd ✓"); - expect(decodeTrackedRedisFrame(utf8, Buffer.from("999"))).toBe("cachéd ✓"); + expect(decodeRedisFrame(utf8)).toEqual({ payload: "cachéd ✓", createdAtMs: 1_000 }); + expect(decodeTrackedRedisFrame(utf8, Buffer.from("999"))).toEqual({ payload: "cachéd ✓", createdAtMs: 1_000 }); const binaryPayload = Buffer.from([0, 0xff, 0x80]); const binary = encodeRedisFrame(binaryPayload, 2_000); expect(binary[9]).toBe(1); expect(binary).toEqual(encodeFrame(binaryPayload, 1, 2_000)); - expect(decodeRedisFrame(binary)).toEqual(binaryPayload); + expect(decodeRedisFrame(binary)).toEqual({ payload: binaryPayload, createdAtMs: 2_000 }); const empty = encodeRedisFrame("", 1); expect(empty.byteLength).toBe(10); - expect(decodeRedisFrame(empty)).toBe(""); + expect(decodeRedisFrame(empty)).toEqual({ payload: "", createdAtMs: 1 }); }); it("keeps zero-stamped version-1 frames unreadable on the tracked path", () => { @@ -126,7 +128,7 @@ describe("Redis frame decoding", () => { expect(decodeTrackedRedisFrame(zeroStamped, null)).toBeNull(); expect(decodeTrackedRedisFrame(zeroStamped, Buffer.from("0"))).toBeNull(); expect(decodeTrackedRedisFrame(zeroStamped, Buffer.from("1"))).toBeNull(); - expect(decodeRedisFrame(zeroStamped)).toBe("pending"); + expect(decodeRedisFrame(zeroStamped)).toEqual({ payload: "pending", createdAtMs: 0 }); }); it("encodes tracked placeholders that no read path serves", () => { diff --git a/test/redis-real.integration.test.ts b/test/redis-real.integration.test.ts index 798e2b0..645f7d7 100644 --- a/test/redis-real.integration.test.ts +++ b/test/redis-real.integration.test.ts @@ -378,8 +378,9 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => { const roundTrip = await scriptClient.read({ valueKey }); const stored = await admin.get(commandOptions({ returnBuffers: true }), valueKey); - expect(Buffer.isBuffer(roundTrip)).toBe(true); - expect(roundTrip).toEqual(payload); + expect(Buffer.isBuffer(roundTrip?.payload)).toBe(true); + expect(roundTrip?.payload).toEqual(payload); + expect(roundTrip?.createdAtMs).toBeGreaterThan(0); expect(stored).not.toBeNull(); expect(stored?.length).toBe(10 + payload.length); expect(stored?.[0]).toBe(1); @@ -398,7 +399,9 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => { value: trackedPayload, }), ).toBe(true); - expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toEqual(trackedPayload); + const trackedRead = await scriptClient.read({ valueKey: trackedValueKey, watermarkKey }); + expect(trackedRead?.payload).toEqual(trackedPayload); + expect(trackedRead?.createdAtMs).toBeGreaterThan(0); }); it("shadow-validates the deserialized tracked value without repairing a mismatch", async () => { @@ -435,6 +438,7 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => { mismatched.resolve(); } }), + observeShadowValueAge: vi.fn(), observeGet: vi.fn(), observeFallback: vi.fn(), observeSerialization: vi.fn(), @@ -556,6 +560,19 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => { keyType: "item_id", outcome: "superseded", }); + // The stored frame was stamped with createdAtMs=1, so both verdicts see + // a huge positive age; superseded outcomes record none. + expect(metrics.observeShadowValueAge).toHaveBeenCalledTimes(2); + expect(metrics.observeShadowValueAge).toHaveBeenNthCalledWith( + 1, + { cacheNamespace: namespace, useCase, keyType: "item_id", outcome: "match" }, + expect.any(Number), + ); + expect(metrics.observeShadowValueAge).toHaveBeenNthCalledWith( + 2, + { cacheNamespace: namespace, useCase, keyType: "item_id", outcome: "mismatch" }, + expect.any(Number), + ); expect(read).toHaveBeenCalledTimes(9); expect(read.mock.calls.every(([request]) => request.valueKey === valueKey && request.watermarkKey === watermarkKey @@ -780,10 +797,10 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => { value: JSON.stringify(sourceValue), ...(tracked ? { watermarkKey } : {}), }); - expect(await client.adapter.read({ + expect((await client.adapter.read({ valueKey, ...(tracked ? { watermarkKey } : {}), - })).toBe(JSON.stringify(sourceValue)); + }))?.payload).toBe(JSON.stringify(sourceValue)); expect(await admin.pTTL(valueKey)).toBeGreaterThan(55_000); expect(await admin.pTTL(valueKey)).toBeLessThanOrEqual(60_000); if (tracked) { @@ -915,7 +932,7 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => { await admin.scriptFlush(); expect(await scriptClient.write({ valueKey, cacheTtlMs: 60_000, value: "untracked" })).toBe(true); - expect(await scriptClient.read({ valueKey })).toBe("untracked"); + expect((await scriptClient.read({ valueKey }))?.payload).toBe("untracked"); const trackedValueKey = "script-recovery:{item:tracked}:value"; const watermarkKey = "script-recovery:{item:tracked}:watermark"; @@ -928,7 +945,7 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => { value: "tracked", }), ).toBe(true); - expect(await scriptClient.read({ valueKey: trackedValueKey, watermarkKey })).toBe("tracked"); + expect((await scriptClient.read({ valueKey: trackedValueKey, watermarkKey }))?.payload).toBe("tracked"); // The recovered write must cache the stamp under sha1(source) — the // digest node-redis registers and the GLIDE batch dispatches — so later // writes take the single-round-trip path. (The unit suites pin each @@ -975,7 +992,7 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => { expect(await scriptClient.read({ valueKey, watermarkKey })).toBeNull(); await admin.set(watermarkKey, "999.5"); - expect(await scriptClient.read({ valueKey, watermarkKey })).toBe("tracked"); + expect((await scriptClient.read({ valueKey, watermarkKey }))?.payload).toBe("tracked"); }); it("records a stale tracked frame as a remote miss without a read error", async () => { @@ -1532,7 +1549,7 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => { expect(wrote).toBe(true); expect(await admin.get(watermarkKey)).toBe("1.75"); expect(await admin.pTTL(watermarkKey)).toBeGreaterThanOrEqual(61_000); - expect(await scriptClient.read({ valueKey, watermarkKey })).toBe("cached"); + expect((await scriptClient.read({ valueKey, watermarkKey }))?.payload).toBe("cached"); }); it("does not rewrite sufficient or persistent watermarks on tracked writes", async () => { @@ -1609,7 +1626,7 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => { await new Promise((resolve) => setTimeout(resolve, 110)); expect(await scriptClient.write({ ...writeRequest, value: "fresh" })).toBe(true); - expect(await scriptClient.read({ valueKey, watermarkKey })).toBe("fresh"); + expect((await scriptClient.read({ valueKey, watermarkKey }))?.payload).toBe("fresh"); }); it("documents that losing a watermark removes its publication fence", async () => { @@ -1633,7 +1650,7 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => { expect(await scriptClient.write(staleWrite)).toBe(true); expect(await admin.get(watermarkKey)).toBe("0"); - expect(await scriptClient.read({ valueKey, watermarkKey })).toBe("stale"); + expect((await scriptClient.read({ valueKey, watermarkKey }))?.payload).toBe("stale"); }); it("never serves an unstamped placeholder and refuses foreign stamps", async () => { @@ -1657,7 +1674,7 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => { // Only the paired nonce promotes it to a served, server-stamped frame. expect(await client.raw.stamp(valueKey, watermarkKey, 2_000, nonce)).toBe(1); - expect(await client.adapter.read({ valueKey, watermarkKey })).toBe("pending"); + expect((await client.adapter.read({ valueKey, watermarkKey }))?.payload).toBe("pending"); const stored = await admin.get(commandOptions({ returnBuffers: true }), valueKey); expect(stored?.[0]).toBe(1); expect(stored?.readBigUInt64BE(1) ?? 0n).toBeGreaterThan(0n); @@ -1706,10 +1723,10 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => { const binary = Buffer.from([0, 0xff, 0xc3, 0x28, 0x80]); await nodeRedis.write({ valueKey: "interop:node-to-glide", cacheTtlMs: 60_000, value: binary }); - await expect(valkeyGlide.read({ valueKey: "interop:node-to-glide" })).resolves.toEqual(binary); + expect((await valkeyGlide.read({ valueKey: "interop:node-to-glide" }))?.payload).toEqual(binary); await valkeyGlide.write({ valueKey: "interop:glide-to-node", cacheTtlMs: 60_000, value: "hello" }); - await expect(nodeRedis.read({ valueKey: "interop:glide-to-node" })).resolves.toBe("hello"); + expect((await nodeRedis.read({ valueKey: "interop:glide-to-node" }))?.payload).toBe("hello"); const nodeTrackedValueKey = "interop:{node-tracked}:value"; const nodeTrackedWatermarkKey = "interop:{node-tracked}:watermark"; @@ -1719,12 +1736,10 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => { cacheTtlMs: 60_000, value: binary, }); - await expect( - valkeyGlide.read({ - valueKey: nodeTrackedValueKey, - watermarkKey: nodeTrackedWatermarkKey, - }), - ).resolves.toEqual(binary); + expect((await valkeyGlide.read({ + valueKey: nodeTrackedValueKey, + watermarkKey: nodeTrackedWatermarkKey, + }))?.payload).toEqual(binary); const glideTrackedValueKey = "interop:{glide-tracked}:value"; const glideTrackedWatermarkKey = "interop:{glide-tracked}:watermark"; @@ -1734,11 +1749,9 @@ describe.each(engines)("DialCache Redis protocol on $name", ({ image }) => { cacheTtlMs: 60_000, value: "tracked", }); - await expect( - nodeRedis.read({ - valueKey: glideTrackedValueKey, - watermarkKey: glideTrackedWatermarkKey, - }), - ).resolves.toBe("tracked"); + expect((await nodeRedis.read({ + valueKey: glideTrackedValueKey, + watermarkKey: glideTrackedWatermarkKey, + }))?.payload).toBe("tracked"); }); }); diff --git a/test/valkey-glide.test.ts b/test/valkey-glide.test.ts index ae2426b..cb92929 100644 --- a/test/valkey-glide.test.ts +++ b/test/valkey-glide.test.ts @@ -153,10 +153,13 @@ describe("Valkey GLIDE adapter", () => { ); const adapter = createValkeyGlideDialCacheClient(client, mockGlide); - await expect(adapter.read({ valueKey: "plain:value" })).resolves.toBe("plain"); + await expect(adapter.read({ valueKey: "plain:value" })).resolves.toEqual({ + payload: "plain", + createdAtMs: 1_000, + }); await expect( adapter.read({ valueKey: "tracked:{id}:value", watermarkKey: "tracked:{id}:watermark" }), - ).resolves.toEqual(Buffer.from([0, 0xff])); + ).resolves.toEqual({ payload: Buffer.from([0, 0xff]), createdAtMs: 1_000 }); await expect(adapter.read({ valueKey: "missing:value" })).resolves.toBeNull(); expect(client.get).toHaveBeenNthCalledWith( @@ -195,7 +198,7 @@ describe("Valkey GLIDE adapter", () => { valueKey: "cluster:{id}:value", watermarkKey: "cluster:{id}:watermark", }), - ).resolves.toBe("tracked-cluster"); + ).resolves.toEqual({ payload: "tracked-cluster", createdAtMs: 1_000 }); expect(client.customCommand).toHaveBeenCalledWith( ["MGET", "cluster:{id}:value", "cluster:{id}:watermark"], From 44801f7b1437108519230367a2194af2e0fc468b Mon Sep 17 00:00:00 2001 From: Lev Neiman Date: Fri, 14 Aug 2026 13:38:34 -0700 Subject: [PATCH 2/2] fix(shadow): apply review findings to the value-age instrumentation Review-loop findings on cf087d0, all four addressed: - Skip the value-age observation when the computed age is non-finite. An out-of-contract custom client stamping NaN (or omitting createdAtMs in plain JS) previously flowed through Math.max(NaN, 0) into the metrics backend, permanently poisoning prom-client histogram sums with no diagnostic surface; the write side already RangeErrors the same field. - Correct the two stale "untracked reads never consult it" contract docs: the stamp is never consulted for serving or miss decisions, but it now surfaces on the decoded frame and feeds the shadow value-age observation, so untracked writers must stamp real client time. Also tighten DecodedRedisFrame payload wording (compression envelope) and the verdict-time phrasing in metrics docs, Prometheus help, and README. - Pin mismatch-age provenance: the confirmation read now republishes identical bytes with a fresh stamp while the test still expects the original frame's 90s age, so regressing to the confirmation frame's stamp fails instead of passing silently. - Assert the Prometheus shadow value-age histogram sum (42) and its exact label set, closing the one unpinned numeric pass-through. --- README.md | 6 ++--- src/dialcache.ts | 12 ++++++--- src/internal/redis-payload.ts | 10 +++++--- src/metrics.ts | 17 +++++++------ src/prometheus.ts | 2 +- src/redis-client.ts | 22 ++++++++++------ test/dialcache-shadow-confirmation.test.ts | 29 +++++++++++++++++++++- test/prometheus.test.ts | 13 ++++++++++ 8 files changed, 83 insertions(+), 28 deletions(-) diff --git a/README.md b/README.md index 87e5fc2..405f0e6 100644 --- a/README.md +++ b/README.md @@ -586,7 +586,7 @@ The effective serializer's `load` method therefore runs a second time for a samp Shadow metrics use the bounded outcomes `match`, `mismatch`, `superseded`, `filled`, `fill_blocked`, `fill_error`, `redis_error`, `source_error`, `deserialization_error`, `comparison_error`, `confirmation_error`, `timeout`, and `dropped`. `redis_error` applies to the initial shadow-only `C0` read; `confirmation_error` applies to `C1`; `fill_blocked` applies only when a tracked watermark rejects the write. A clean `C0` miss is an ordinary `miss{layer="remote_shadow"}` and terminates with a fill, source, or timeout outcome rather than a second shadow outcome for the miss itself. Labels never contain cache ids, values, payloads, Redis keys, or raw exception text. -A `match` or `mismatch` verdict additionally records the validated value's age through the optional `observeShadowValueAge` adapter hook: the observing process's epoch clock minus the served frame's `createdAtMs`, in seconds, clamped at zero. A confirmed `mismatch` age therefore measures how long the stale value had been readable when validation caught it. Tracked frames are stamped with Redis server time and untracked frames with the writer's client clock, so the age mixes clocks and is coarse operational evidence rather than a precise measurement. Outcomes that deliver no verdict on a retained value — including `superseded`, `filled`, and every error or timeout outcome — record no age. The hook does not gate shadow eligibility; only `shadowValidation` does. +A `match` or `mismatch` verdict additionally records the validated value's age through the optional `observeShadowValueAge` adapter hook: the observing process's epoch clock minus the validated frame's `createdAtMs` (the served `C0` frame, or the detached `C0` frame on a ramped-down path), in seconds, clamped at zero. The age is captured at verdict time, so a mismatch age lands one confirmation read after the comparison itself. A confirmed `mismatch` age therefore measures how long the stale value had been readable when validation caught it. Tracked frames are stamped with Redis server time and untracked frames with the writer's client clock, so the age mixes clocks and is coarse operational evidence rather than a precise measurement. Outcomes that deliver no verdict on a retained value — including `superseded`, `filled`, and every error or timeout outcome — record no age. The hook does not gate shadow eligibility; only `shadowValidation` does. Confirmed-mismatch logging is separately opt-in through `shadow.logMismatches`; it never replaces the required `shadowValidation` metric or emits for a mismatch candidate that becomes `superseded`. The single warning contains `cacheNamespace`, `useCase`, `keyType`, `outcome: "mismatch"`, `cacheKey`, `cachedValueJson`, and `sourceValueJson`. `cacheKey` is the logical DialCache URN capped at 2 KiB, not the physical Redis storage key. DialCache independently applies native `JSON.stringify` to the deserialized cached snapshot and raw source value supplied to the comparator, then caps each resulting string at 8 KiB. A byte-clipped field ends in `...[truncated]`, counted inside its cap. If native JSON throws or returns `undefined`, the corresponding JSON field is `null`; the other side is still attempted. DialCache does not compute a textual diff or call the configured serializer again for logging. @@ -795,7 +795,7 @@ The Prometheus adapter emits: | `dialcache_invalidation_counter` | Counter | `cache_namespace`, `key_type`, `layer` | Invalidation calls for the layers touched | | `dialcache_coalesced_counter` | Counter | `cache_namespace`, `use_case`, `key_type`, `scope` | Coalesced requests split by `request_local` or `process` scope | | `dialcache_shadow_validation_counter` | Counter | `cache_namespace`, `use_case`, `key_type`, `outcome` | Sampled Redis shadow-job outcomes | -| `dialcache_shadow_value_age_histogram` | Histogram | `cache_namespace`, `use_case`, `key_type`, `outcome` | Age in seconds of the validated Redis value at shadow comparison time, recorded for `match` and `mismatch` | +| `dialcache_shadow_value_age_histogram` | Histogram | `cache_namespace`, `use_case`, `key_type`, `outcome` | Age in seconds of the validated Redis value at shadow verdict time, recorded for `match` and `mismatch` | | `dialcache_compression_counter` | Counter | `cache_namespace`, `use_case`, `key_type`, `layer`, `outcome` | Payload compression outcomes: writes record `compressed`, `below_threshold`, `not_smaller`, or `write_over_limit`; reads record `decompressed`, `fallback_raw`, or `read_over_limit` | | `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` | Elapsed time until the underlying function settles or timeout rejection is delivered | @@ -858,7 +858,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.shadow.count` | Count | `cache_namespace`, `use_case`, `key_type`, `outcome` | Sampled Redis shadow-job outcomes | -| `dialcache.shadow.value_age` | Distribution or histogram | `cache_namespace`, `use_case`, `key_type`, `outcome` | Age in seconds of the validated Redis value at shadow comparison time, recorded for `match` and `mismatch` | +| `dialcache.shadow.value_age` | Distribution or histogram | `cache_namespace`, `use_case`, `key_type`, `outcome` | Age in seconds of the validated Redis value at shadow verdict time, recorded for `match` and `mismatch` | | `dialcache.compression.count` | Count | `cache_namespace`, `use_case`, `key_type`, `layer`, `outcome` | Payload compression outcomes: writes record `compressed`, `below_threshold`, `not_smaller`, or `write_over_limit`; reads record `decompressed`, `fallback_raw`, or `read_over_limit` | | `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` | Elapsed time until the underlying function settles or timeout rejection is delivered | diff --git a/src/dialcache.ts b/src/dialcache.ts index a9be5b5..afd8468 100644 --- a/src/dialcache.ts +++ b/src/dialcache.ts @@ -1576,9 +1576,15 @@ function resolveShadowComparator( // Frame stamps are epoch-based (Redis server time for tracked writes, writer // client clock for untracked), so the age uses the epoch clock and clamps -// negative cross-clock skew to zero. -function shadowValueAgeSeconds(createdAtMs: number): number { - return Math.max((Date.now() - createdAtMs) / 1000, 0); +// negative cross-clock skew to zero. A custom client that violates the decode +// contract can hand over a non-finite stamp; recording it would permanently +// poison backend histogram sums, so the observation is skipped instead. +function shadowValueAgeSeconds(createdAtMs: number): number | undefined { + const ageSeconds = (Date.now() - createdAtMs) / 1000; + if (!Number.isFinite(ageSeconds)) { + return undefined; + } + return Math.max(ageSeconds, 0); } function redisPayloadsEqual(left: RedisCachePayload, right: RedisCachePayload): boolean { diff --git a/src/internal/redis-payload.ts b/src/internal/redis-payload.ts index 97fb93b..fbc3f8e 100644 --- a/src/internal/redis-payload.ts +++ b/src/internal/redis-payload.ts @@ -79,10 +79,12 @@ function encodeFrameBytes(payload: RedisCachePayload, version: number, stampByte /** * Encode a serializer payload into a servable DialCache Redis frame. * - * Untracked writes stamp an informational client-clock `createdAtMs`; - * untracked reads never consult it. Tracked writes must not use this - * directly — they pair `encodeTrackedRedisPlaceholder` with - * `WRITE_TRACKED_STAMP_SCRIPT` instead. + * Untracked writes stamp a client-clock `createdAtMs`. Untracked reads never + * consult the stamp for serving or miss decisions, but they surface it as the + * decoded frame's `createdAtMs`, which feeds the shadow value-age + * observation — so stamp real client time, not a constant. Tracked writes + * must not use this directly — they pair `encodeTrackedRedisPlaceholder` + * with `WRITE_TRACKED_STAMP_SCRIPT` instead. */ export function encodeRedisFrame(payload: RedisCachePayload, createdAtMs: number): Buffer { if (!Number.isSafeInteger(createdAtMs) || createdAtMs < 0) { diff --git a/src/metrics.ts b/src/metrics.ts index 4727644..998469f 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -117,14 +117,15 @@ export interface DialCacheMetricsAdapter { // Optional so existing custom adapters keep compiling without changes. shadowValidation?(labels: ShadowValidationMetricLabels): void; /** - * Age in seconds of the validated Redis value at shadow comparison time: - * the observing process's epoch clock minus the frame header's - * `createdAtMs`, clamped at zero. Emitted only alongside terminal `match` - * and `mismatch` outcomes; other outcomes deliver no verdict on a retained - * value. Tracked frames are stamped with Redis server time and untracked - * frames with the writer's client clock, so the age mixes clocks and is - * coarse operational evidence, not a precise measurement. Optional so - * existing custom adapters keep compiling without changes. + * Age in seconds of the validated Redis value at shadow verdict time (for + * a mismatch, after the confirming re-read): the observing process's epoch + * clock minus the validated frame's `createdAtMs`, clamped at zero. + * Emitted only alongside terminal `match` and `mismatch` outcomes; other + * outcomes deliver no verdict on a retained value. Tracked frames are + * stamped with Redis server time and untracked frames with the writer's + * client clock, so the age mixes clocks and is coarse operational + * evidence, not a precise measurement. Optional so existing custom + * adapters keep compiling without changes. */ observeShadowValueAge?(labels: ShadowValidationMetricLabels, seconds: number): void; // Optional so existing custom adapters keep compiling without changes. diff --git a/src/prometheus.ts b/src/prometheus.ts index 5b5ae36..7674675 100644 --- a/src/prometheus.ts +++ b/src/prometheus.ts @@ -250,7 +250,7 @@ function collectorConfigs(prefix: string) { shadowValueAgeHistogram: { type: "histogram", name: `${prefix}dialcache_shadow_value_age_histogram`, - help: "Age in seconds of the validated Redis value at DialCache shadow comparison time.", + help: "Age in seconds of the validated Redis value at DialCache shadow verdict time.", labelNames: ["cache_namespace", "use_case", "key_type", "outcome"], buckets: VALUE_AGE_BUCKETS, }, diff --git a/src/redis-client.ts b/src/redis-client.ts index 4fdcfc4..1b58f3d 100644 --- a/src/redis-client.ts +++ b/src/redis-client.ts @@ -88,12 +88,15 @@ export class DialCacheRedisPlaceholderLostError extends Error { export type RedisCachePayload = string | Buffer; /** - * A served Redis frame: the decoded serializer payload plus the creation time - * from the frame header. Tracked frames carry Redis server time written by the - * stamp script; untracked frames carry the writer's informational client - * clock. DialCache consumes `createdAtMs` only for observability (the shadow - * value-age observation) — tracked watermark fencing already happened inside - * the decoder — so it never affects serving decisions. + * A served Redis frame: the payload bytes past the frame header plus the + * header's creation time. The payload is the serializer output, possibly + * still wrapped in a compression envelope that DialCache core interprets + * above the adapter (see the `dialcache/redis-protocol` module doc). Tracked + * frames carry Redis server time written by the stamp script; untracked + * frames carry the writer's client clock. DialCache consumes `createdAtMs` + * only for observability (the shadow value-age observation) — tracked + * watermark fencing already happened inside the decoder — so it never + * affects serving decisions. */ export interface DecodedRedisFrame { readonly payload: RedisCachePayload; @@ -186,8 +189,11 @@ export interface DialCacheRedisClient { * encoders, or preserve their exact behavior. * * Untracked writes are one native `SET valueKey frame PX cacheTtlMs` whose - * frame comes from `encodeRedisFrame` with an informational client-clock - * `createdAtMs`; untracked reads never consult it. + * frame comes from `encodeRedisFrame` with a client-clock `createdAtMs`. + * Untracked reads never consult that stamp for serving or miss decisions, + * but they do surface it as the decoded frame's `createdAtMs`, where it + * feeds the shadow value-age observation — so untracked writers must stamp + * real client time, not a constant. * * Tracked writes issue two commands ordered on one connection without a * transaction: a native `SET` of an `encodeTrackedRedisPlaceholder` frame, diff --git a/test/dialcache-shadow-confirmation.test.ts b/test/dialcache-shadow-confirmation.test.ts index 8e32e57..0bd33b7 100644 --- a/test/dialcache-shadow-confirmation.test.ts +++ b/test/dialcache-shadow-confirmation.test.ts @@ -555,7 +555,16 @@ describe("DialCache Redis shadow confirmation", () => { const nowSpy = vi.spyOn(Date, "now").mockReturnValue(nowMs); try { const payload = JSON.stringify({ id: "123", version: 1 }); - const redis = new ScriptedRedis([() => payload, () => payload]); + const redis = new ScriptedRedis([ + () => payload, + () => { + // A concurrent writer republished identical bytes with a fresh + // stamp; the confirmation still holds and the reported age must + // come from the original frame, not this one. + redis.frameCreatedAtMs = nowMs - 1_000; + return payload; + }, + ]); redis.frameCreatedAtMs = nowMs - 90_000; const metrics = new RecordingMetrics(); const dialcache = createCache(redis, metrics); @@ -580,6 +589,24 @@ describe("DialCache Redis shadow confirmation", () => { } }); + it("skips the value-age observation when an out-of-contract client stamps a non-finite time", async () => { + const payload = JSON.stringify({ id: "123", version: 1 }); + const redis = new ScriptedRedis([() => payload]); + redis.frameCreatedAtMs = Number.NaN; + const metrics = new RecordingMetrics(); + const dialcache = createCache(redis, metrics); + const getUser = dialcache.cached(async () => ({ id: "123", version: 1 }), { + ...trackedOptions("ShadowValueAgeNonFinite", remoteConfig(100)), + cacheKey: () => "123", + }); + + await dialcache.enable(async () => await getUser()); + await waitForShadowEvents(metrics, 1); + + expect(metrics.shadowEvents.map(({ outcome }) => outcome)).toEqual(["match"]); + expect(metrics.shadowAgeEvents).toEqual([]); + }); + it("does not log a mismatch candidate when C1 is superseded", async () => { const original = JSON.stringify({ id: "123", version: 1 }); const confirmation = JSON.stringify({ id: "123", version: 3 }); diff --git a/test/prometheus.test.ts b/test/prometheus.test.ts index 7508886..bcfebbc 100644 --- a/test/prometheus.test.ts +++ b/test/prometheus.test.ts @@ -259,6 +259,18 @@ describe("Prometheus metrics adapter", () => { ), ]); + const shadowValueAge = families.find(({ name }) => name === "schema_dialcache_shadow_value_age_histogram"); + const shadowValueAgeSum = shadowValueAge?.values.find( + ({ metricName }) => metricName === "schema_dialcache_shadow_value_age_histogram_sum", + ); + expect(shadowValueAgeSum?.value).toBe(42); + expect(shadowValueAgeSum?.labels).toEqual({ + cache_namespace: labels.cacheNamespace, + use_case: labels.useCase, + key_type: labels.keyType, + outcome: "match", + }); + const serialization = families.find(({ name }) => name === "schema_dialcache_serialization_timer"); const serializationLabels = serialization?.values .filter(({ metricName }) => metricName === `${serialization.name}_sum`) @@ -665,6 +677,7 @@ const VALUE_AGE_BUCKETS = [1, 5, 15, 60, 300, 900, 3_600, 10_800, 43_200, 86_400 interface MetricValue { readonly metricName?: string; readonly labels: Record; + readonly value?: number; } interface MetricFamily {