feat: add cluster-aware batch invalidation

package.json

@@ -76,6 +76,7 @@
     "LICENSE"
   ],
   "scripts": {
+    "benchmark:batch-invalidation": "pnpm build && node scripts/benchmark-batch-invalidation.mjs",
     "benchmark:request-local": "pnpm build && node scripts/benchmark-request-local.mjs",
     "build": "tsup src/index.ts src/datadog.ts src/node-redis.ts src/prometheus.ts src/redis-protocol.ts src/valkey-glide.ts --format esm,cjs --dts --clean",
     "check": "pnpm typecheck && pnpm test && pnpm build && pnpm test:package",

scripts/benchmark-batch-invalidation.mjs

@@ -0,0 +1,179 @@
+import assert from "node:assert/strict";
+import { performance } from "node:perf_hooks";
+
+import * as valkeyGlide from "@valkey/valkey-glide";
+import { createClient } from "redis";
+
+import { DialCache, invalidationPrefix, redisClusterHashTag } from "../dist/index.js";
+import {
+  createNodeRedisDialCacheClient,
+  dialcacheRedisScripts,
+} from "../dist/node-redis.js";
+import { createValkeyGlideDialCacheClient } from "../dist/valkey-glide.js";
+
+const redisUrl = process.env.DIALCACHE_BENCH_REDIS_URL ?? "redis://127.0.0.1:6379";
+const sizes = [10, 100, 1_000];
+const repetitions = positiveIntegerFromEnvironment("DIALCACHE_BENCH_REPETITIONS", 5);
+const runId = `${process.pid}-${Date.now()}`;
+const redisClient = createClient({
+  url: redisUrl,
+  scripts: dialcacheRedisScripts,
+  disableOfflineQueue: true,
+  commandsQueueMaxLength: 10_000,
+  socket: { connectTimeout: 2_000 },
+});
+redisClient.on("error", () => undefined);
+
+await redisClient.connect();
+let glideClient;
+let glideAdapter;
+
+try {
+  glideClient = await valkeyGlide.GlideClient.createClient(glideConfiguration(redisUrl));
+  glideAdapter = createValkeyGlideDialCacheClient(glideClient, valkeyGlide);
+  const adapters = [
+    { name: "node-redis", client: createNodeRedisDialCacheClient(redisClient) },
+    { name: "Valkey GLIDE", client: glideAdapter },
+  ];
+  const results = [];
+
+  for (const { name, client } of adapters) {
+    const namespace = `dialcache-batch-invalidation-benchmark-${name}-${runId}`;
+    const dialcache = new DialCache({
+      namespace,
+      redis: { client },
+    });
+
+    // Pay one-time connection and Lua loading costs before the measured runs.
+    await dialcache.invalidateRemote("benchmark_warmup", runId);
+
+    for (const size of sizes) {
+      const scalarSamples = [];
+      const batchSamples = [];
+      for (let repetition = 0; repetition < repetitions; repetition += 1) {
+        const scalarTargets = targetsFor(`${name}-scalar-${size}-${repetition}`, size);
+        const batchTargets = targetsFor(`${name}-batch-${size}-${repetition}`, size);
+        const runScalar = async () => {
+          const startedAt = performance.now();
+          const operations = scalarTargets.map(({ keyType, id }) =>
+            dialcache.invalidateRemote(keyType, id));
+          try {
+            await Promise.all(operations);
+          } catch (error) {
+            await Promise.allSettled(operations);
+            throw error;
+          }
+          scalarSamples.push(performance.now() - startedAt);
+          assert.equal(await countWatermarks(namespace, scalarTargets), size);
+        };
+        const runBatch = async () => {
+          const startedAt = performance.now();
+          await dialcache.invalidateRemoteMany(batchTargets);
+          batchSamples.push(performance.now() - startedAt);
+          assert.equal(await countWatermarks(namespace, batchTargets), size);
+        };
+
+        if (repetition % 2 === 0) {
+          await runScalar();
+          await runBatch();
+        } else {
+          await runBatch();
+          await runScalar();
+        }
+      }
+      const scalarMedianMs = median(scalarSamples);
+      const batchMedianMs = median(batchSamples);
+
+      results.push({
+        adapter: name,
+        targets: size,
+        repetitions,
+        "Promise.all scalar median (ms)": scalarMedianMs.toFixed(2),
+        "batch median (ms)": batchMedianMs.toFixed(2),
+        "median scalar / batch": (scalarMedianMs / batchMedianMs).toFixed(2),
+      });
+    }
+  }
+
+  console.table(results);
+  console.log(
+    "Directional maintainer benchmark only: results depend on Redis topology, client configuration, and network conditions; no timing threshold is asserted.",
+  );
+} finally {
+  try {
+    glideAdapter?.dispose();
+  } finally {
+    try {
+      glideClient?.close();
+    } finally {
+      await redisClient.quit();
+    }
+  }
+}
+
+function glideConfiguration(value) {
+  const url = new URL(value);
+  if (url.protocol !== "redis:" && url.protocol !== "rediss:") {
+    throw new Error("DIALCACHE_BENCH_REDIS_URL must use redis: or rediss:");
+  }
+  if (url.username !== "" && url.password === "") {
+    throw new Error("DIALCACHE_BENCH_REDIS_URL cannot specify a username without a password");
+  }
+
+  const databasePath = url.pathname.replace(/^\//, "");
+  const databaseId = databasePath === "" ? 0 : Number(databasePath);
+  if (!Number.isSafeInteger(databaseId) || databaseId < 0) {
+    throw new Error("DIALCACHE_BENCH_REDIS_URL must contain a nonnegative database id");
+  }
+
+  return {
+    addresses: [{ host: url.hostname, port: url.port === "" ? 6379 : Number(url.port) }],
+    databaseId,
+    useTLS: url.protocol === "rediss:",
+    requestTimeout: 10_000,
+    inflightRequestsLimit: 10_000,
+    advancedConfiguration: { connectionTimeout: 2_000 },
+    ...(url.password === ""
+      ? {}
+      : {
+          credentials: {
+            ...(url.username === "" ? {} : { username: decodeURIComponent(url.username) }),
+            password: decodeURIComponent(url.password),
+          },
+        }),
+  };
+}
+
+function targetsFor(prefix, count) {
+  return Array.from({ length: count }, (_, index) => ({
+    keyType: "benchmark_id",
+    id: `${runId}-${prefix}-${index}`,
+  }));
+}
+
+async function countWatermarks(namespace, targets) {
+  const keys = targets.map(({ keyType, id }) =>
+    `${redisClusterHashTag(invalidationPrefix(namespace, keyType, String(id)))}#watermark`);
+  return await redisClient.exists(keys);
+}
+
+function median(values) {
+  assert.ok(values.length > 0);
+  const sorted = [...values].sort((left, right) => left - right);
+  const middle = Math.floor(sorted.length / 2);
+  return sorted.length % 2 === 0
+    ? (sorted[middle - 1] + sorted[middle]) / 2
+    : sorted[middle];
+}
+
+function positiveIntegerFromEnvironment(name, fallback) {
+  const raw = process.env[name];
+  if (raw === undefined) {
+    return fallback;
+  }
+  const parsed = Number(raw);
+  if (!Number.isSafeInteger(parsed) || parsed <= 0) {
+    throw new Error(`${name} must be a positive safe integer`);
+  }
+  return parsed;
+}

scripts/test-package.mjs

@@ -40,6 +40,7 @@ const rootConsumer = `import {
   type RedisInvalidationRequest,
   type RedisReadContext,
   type RedisWriteRequest,
+  type RemoteInvalidationTarget,
   type Serializer,
   type ShadowComparator,
   type ShadowConfig,
@@ -136,6 +137,12 @@ const datadogClassAdapter = new DatadogDialCacheMetrics(datadogOptions);
 // @ts-expect-error The observation type is an explicit, required choice.
 const missingObservationType: DatadogMetricsOptions = { client: dogStatsDClient };
 const cache = new DialCache({ namespace: "consumer-cache", metrics });
+const remoteInvalidationTargets: readonly RemoteInvalidationTarget[] = [
+  { keyType: "id", id: "123" },
+  { keyType: "tenant_id", id: 456 },
+  { keyType: "organization_id", id: 789n },
+];
+const batchInvalidation: Promise<void> = cache.invalidateRemoteMany(remoteInvalidationTargets, 1_000);
 const redisProtocolError = new DialCacheRedisProtocolError("Invalid DialCache Redis write reply");
 const fallbackTimeoutError = new FallbackTimeoutError("Load", 1_000);
 const redisReadTimeoutError = new RedisReadTimeoutError("Load", 100);
@@ -324,12 +331,30 @@ const customRedisClient: DialCacheRedisClient = {
   write: async ({ value }) => typeof value === "string" || Buffer.isBuffer(value),
   invalidate: async () => undefined,
 };
+const narrowNodeRedisScriptClient = {
+  dialcacheRead: async () => null,
+  dialcacheReadTracked: async () => null,
+  dialcacheWrite: async () => 1,
+  dialcacheWriteTracked: async () => 1,
+  dialcacheInvalidate: async () => 1,
+};
+const nodeRedisAdapterWithoutMulti: DialCacheRedisClient = createNodeRedisDialCacheClient(
+  narrowNodeRedisScriptClient,
+);
+const cacheWithScalarOnlyInvalidation = new DialCache({ redis: { client: customRedisClient } });
+const scalarOnlyFallbackBatch: Promise<void> = cacheWithScalarOnlyInvalidation.invalidateRemoteMany(
+  remoteInvalidationTargets,
+);
 const redisClientMethods: Readonly<Record<keyof DialCacheRedisClient, true>> = {
   read: true,
   write: true,
   invalidate: true,
+  invalidateMany: true,
 };
 void redisClientMethods;
+const clientAllowsScalarOnlyInvalidation: {} extends Pick<DialCacheRedisClient, "invalidateMany">
+  ? true
+  : false = true;
 const cacheHasNoFlushAll: "flushAll" extends keyof DialCache ? false : true = true;
 const cacheHasNoClose: "close" extends keyof DialCache ? false : true = true;
 const clientHasNoFlushAll: "flushAll" extends keyof DialCacheRedisClient ? false : true = true;
@@ -434,8 +459,15 @@ void disabledOverlay;
 void metricErrorKinds;
 void unboundedErrorKind;
 void createNodeRedisDialCacheClient;
+void narrowNodeRedisScriptClient;
+void nodeRedisAdapterWithoutMulti;
 void READ_CACHE_SCRIPT;
 void customRedisClient;
+void cacheWithScalarOnlyInvalidation;
+void scalarOnlyFallbackBatch;
+void remoteInvalidationTargets;
+void batchInvalidation;
+void clientAllowsScalarOnlyInvalidation;
 const globalSerializer: Serializer<unknown> = {
   dump: () => "global",
   load: () => ({ source: "global" }),

src/dialcache.ts

@@ -49,6 +49,17 @@ type Id = string | number | bigint;
 /** A cache-key spec: a bare id, or an id plus extra (secondary) key dimensions. */
 export type CacheKeySpec = Id | { readonly id: Id; readonly args?: CacheKeyArgs };
 
+/** One remote invalidation identity. */
+export interface RemoteInvalidationTarget {
+  readonly keyType: string;
+  readonly id: Id;
+}
+
+interface NormalizedRemoteInvalidationTarget {
+  readonly keyType: string;
+  readonly id: string;
+}
+
 // "Any function" without using `any`, so Parameters/ReturnType still apply.
 type AnyFn = (...args: never[]) => unknown;
 /** The cached value type, derived from the wrapped function's return. */
@@ -543,6 +554,56 @@ export class DialCache {
     }
   }
 
+  /**
+   * Writes remote invalidation watermarks for multiple Redis-tracked identities.
+   *
+   * Canonically duplicate targets are coalesced. Each watermark update is
+   * atomic, but the batch is not atomic as a whole and can partially complete.
+   * Retrying the full batch is safe because watermarks advance monotonically.
+   * This has the same remote-only and future-buffer contract as
+   * {@link invalidateRemote}.
+   */
+  async invalidateRemoteMany(
+    targets: readonly RemoteInvalidationTarget[],
+    futureBufferMs = 0,
+  ): Promise<void> {
+    assertSupportedFutureBufferMs(futureBufferMs);
+
+    if (this.redisCache === null) {
+      return;
+    }
+
+    let normalizedTargets: readonly NormalizedRemoteInvalidationTarget[] = [];
+    try {
+      normalizedTargets = normalizeRemoteInvalidationTargets(targets);
+      if (normalizedTargets.length === 0) {
+        return;
+      }
+
+      for (const { keyType } of normalizedTargets) {
+        this.metrics?.invalidation({
+          cacheNamespace: this.namespace,
+          keyType,
+          layer: CacheLayer.REMOTE,
+        });
+      }
+      await this.redisCache.invalidateMany(normalizedTargets, futureBufferMs, this.namespace);
+    } catch (error) {
+      this.logger.warn("Error writing DialCache invalidation watermarks", error);
+      for (const keyType of new Set(normalizedTargets.map(({ keyType }) => keyType))) {
+        this.metrics?.error({
+          cacheNamespace: this.namespace,
+          useCase: "watermark",
+          keyType,
+          layer: CacheLayer.REMOTE,
+          error: "invalidation",
+          inFallback: false,
+        });
+      }
+      throw error;
+    }
+  }
+
   private async getThroughRequestLocal<T>(
     requestLocalCache: RequestLocalCache,
     key: DialCacheKey,
@@ -1495,6 +1556,29 @@ function withFallbackTimeout<T>(
   });
 }
 
+function normalizeRemoteInvalidationTargets(
+  targets: readonly RemoteInvalidationTarget[],
+): NormalizedRemoteInvalidationTarget[] {
+  const idsByKeyType = new Map<string, Set<string>>();
+  const normalized: NormalizedRemoteInvalidationTarget[] = [];
+
+  for (const target of targets) {
+    const id = String(target.id);
+    let ids = idsByKeyType.get(target.keyType);
+    if (ids === undefined) {
+      ids = new Set<string>();
+      idsByKeyType.set(target.keyType, ids);
+    }
+    if (ids.has(id)) {
+      continue;
+    }
+    ids.add(id);
+    normalized.push({ keyType: target.keyType, id });
+  }
+
+  return normalized;
+}
+
 function safeLogger(logger: Logger): Logger {
   return {
     debug: (...args: Parameters<Logger["debug"]>) => callObserver(() => logger.debug(...args)),

src/index.ts

@@ -32,6 +32,7 @@ export type {
   CoalescingState,
   GetOrLoadOptions,
   ProcessCoalescingState,
+  RemoteInvalidationTarget,
   ShadowComparator,
 } from "./dialcache.js";
 export { DialCacheKey, invalidationPrefix, normalizeArgs, redisClusterHashTag } from "./key.js";

src/internal/await-all.ts

@@ -0,0 +1,21 @@
+/** Wait for every launched operation, preserving one error or aggregating many. */
+export async function awaitAll(
+  operations: readonly Promise<unknown>[],
+  aggregateMessage: string,
+): Promise<void> {
+  const results = await Promise.allSettled(operations);
+  const errors: unknown[] = [];
+
+  for (const result of results) {
+    if (result.status === "rejected") {
+      errors.push(result.reason);
+    }
+  }
+
+  if (errors.length === 1) {
+    throw errors[0];
+  }
+  if (errors.length > 1) {
+    throw new AggregateError(errors, aggregateMessage);
+  }
+}