fix: bound Redis usage idempotency cache

.changeset/metering-idempotency-cache-pruning.md

@@ -0,0 +1,5 @@
+---
+"@croco/metering-core": patch
+---
+
+Bound the Redis usage storage local idempotency cache and prune expired entries during later writes so long-running processes do not retain stale record keys indefinitely.

packages/metering-core/src/libs/RedisUsageStorage.ts

@@ -16,6 +16,10 @@ export class RedisUsageStorage implements UsageStorage {
   private static readonly USAGE_KEY_PREFIX = "usage";
   private static readonly IDEM_KEY_PREFIX = "idem";
   private static readonly RECORD_IDEMPOTENCY_TTL_SECONDS = 86400;
+  private static readonly RECORD_IDEMPOTENCY_TTL_MILLISECONDS =
+    RedisUsageStorage.RECORD_IDEMPOTENCY_TTL_SECONDS * 1000;
+  private static readonly RECORD_IDEMPOTENCY_CACHE_MAX_ENTRIES = 10_000;
+  private static readonly RECORD_IDEMPOTENCY_CACHE_PRUNE_INTERVAL_MILLISECONDS = 60_000;
   private static readonly RESET_SCAN_BATCH_SIZE = 500;
   private static readonly SCAN_AND_DELETE_USAGE_KEYS_SCRIPT = `
 local cursor = ARGV[1]
@@ -31,7 +35,9 @@ end
 
 return { nextCursor, #keys }
 `;
+  // Redis remains the idempotency source of truth; this cache only short-circuits duplicate quota writes.
   private readonly recordedRecordKeys = new Map<string, number>();
+  private nextRecordIdempotencyCachePruneAt = 0;
 
   private static buildCheckAndRecordWithinQuotaScript(dedupeKey: string): string {
     const dedupeKeyLiteral = RedisUsageStorage.toLuaLongString(dedupeKey);
@@ -271,13 +277,17 @@ return { exceeded and 1 or 0, newUsage }
   }
 
   private hasRecordedRecordKey(dedupeKey: string): boolean {
+    const now = Date.now();
+
+    this.pruneRecordedRecordKeysIfNeeded(now);
+
     const expiresAt = this.recordedRecordKeys.get(dedupeKey);
 
     if (expiresAt === undefined) {
       return false;
     }
 
-    if (expiresAt > Date.now()) {
+    if (expiresAt > now) {
       return true;
     }
 
@@ -286,9 +296,59 @@ return { exceeded and 1 or 0, newUsage }
   }
 
   private rememberRecordIdempotencyKey(dedupeKey: string): void {
-    const ttlMilliseconds = RedisUsageStorage.RECORD_IDEMPOTENCY_TTL_SECONDS * 1000;
-    this.recordedRecordKeys.set(dedupeKey, Date.now() + ttlMilliseconds);
+    const now = Date.now();
+
+    this.pruneRecordedRecordKeysIfNeeded(now);
+
+    if (
+      !this.recordedRecordKeys.has(dedupeKey) &&
+      this.recordedRecordKeys.size >= RedisUsageStorage.RECORD_IDEMPOTENCY_CACHE_MAX_ENTRIES
+    ) {
+      this.evictOldestRecordedRecordKeys(
+        this.recordedRecordKeys.size - RedisUsageStorage.RECORD_IDEMPOTENCY_CACHE_MAX_ENTRIES + 1,
+      );
+    }
+
+    this.recordedRecordKeys.set(
+      dedupeKey,
+      now + RedisUsageStorage.RECORD_IDEMPOTENCY_TTL_MILLISECONDS,
+    );
+  }
+
+  private pruneRecordedRecordKeysIfNeeded(now: number): void {
+    if (
+      now < this.nextRecordIdempotencyCachePruneAt &&
+      this.recordedRecordKeys.size < RedisUsageStorage.RECORD_IDEMPOTENCY_CACHE_MAX_ENTRIES
+    ) {
+      return;
+    }
+
+    this.pruneExpiredRecordedRecordKeys(now);
+    this.nextRecordIdempotencyCachePruneAt =
+      now + RedisUsageStorage.RECORD_IDEMPOTENCY_CACHE_PRUNE_INTERVAL_MILLISECONDS;
   }
+
+  private pruneExpiredRecordedRecordKeys(now: number): void {
+    for (const [dedupeKey, expiresAt] of this.recordedRecordKeys) {
+      if (expiresAt <= now) {
+        this.recordedRecordKeys.delete(dedupeKey);
+      }
+    }
+  }
+
+  private evictOldestRecordedRecordKeys(count: number): void {
+    let evicted = 0;
+
+    for (const dedupeKey of this.recordedRecordKeys.keys()) {
+      this.recordedRecordKeys.delete(dedupeKey);
+      evicted += 1;
+
+      if (evicted >= count) {
+        return;
+      }
+    }
+  }
+
   private serializeUsageMember(usage: Pick<UsageRecord, "id" | "value" | "metadata">): string {
     const base = `${usage.id}:${usage.value}`;
 

packages/metering-core/src/tests/RedisUsageStorage.spec.ts

@@ -8,6 +8,18 @@ describe("RedisUsageStorage", () => {
   let storage!: RedisUsageStorage;
   let mockRedis!: RedisClient;
 
+  const getRecordedRecordKeys = (): Map<string, number> =>
+    Reflect.get(storage, "recordedRecordKeys") as Map<string, number>;
+
+  const createUsageRecord = (idempotencyKey: string): UsageRecord => ({
+    id: `usage-${idempotencyKey}`,
+    tenantId: "tenant-1",
+    meterId: "api_calls",
+    value: 5,
+    timestamp: new Date("2024-01-15T10:30:00Z"),
+    idempotencyKey,
+  });
+
   beforeEach(() => {
     mockRedis = {
       zadd: vi.fn().mockResolvedValue(1),
@@ -502,6 +514,61 @@ describe("RedisUsageStorage", () => {
       expect(second).toEqual({ exceeded: false, newUsage: 8 });
     });
 
+    it("should prune expired local record idempotency keys during later quota writes", async () => {
+      vi.useFakeTimers();
+      vi.setSystemTime(new Date("2024-01-01T00:00:00.000Z"));
+      vi.mocked(mockRedis.eval).mockResolvedValue([0, 5]);
+
+      const firstRecord = createUsageRecord("key-1");
+      await storage.checkAndRecordWithinQuota({
+        tenantId: firstRecord.tenantId,
+        meterId: firstRecord.meterId,
+        value: firstRecord.value,
+        quota: 10,
+        allowOverQuota: false,
+        usageRecord: firstRecord,
+      });
+
+      expect(getRecordedRecordKeys().has("idem:tenant-1:api_calls:key-1")).toBe(true);
+
+      vi.setSystemTime(new Date("2024-01-02T00:00:01.000Z"));
+
+      const secondRecord = createUsageRecord("key-2");
+      await storage.checkAndRecordWithinQuota({
+        tenantId: secondRecord.tenantId,
+        meterId: secondRecord.meterId,
+        value: secondRecord.value,
+        quota: 10,
+        allowOverQuota: false,
+        usageRecord: secondRecord,
+      });
+
+      expect(getRecordedRecordKeys().has("idem:tenant-1:api_calls:key-1")).toBe(false);
+      expect(getRecordedRecordKeys().has("idem:tenant-1:api_calls:key-2")).toBe(true);
+      expect(getRecordedRecordKeys().size).toBe(1);
+    });
+
+    it("should cap the local record idempotency cache size", async () => {
+      vi.mocked(mockRedis.eval).mockResolvedValue([0, 5]);
+
+      for (let index = 0; index <= 10_000; index += 1) {
+        const usageRecord = createUsageRecord(`key-${index}`);
+
+        await storage.checkAndRecordWithinQuota({
+          tenantId: usageRecord.tenantId,
+          meterId: usageRecord.meterId,
+          value: usageRecord.value,
+          quota: 10,
+          allowOverQuota: false,
+          usageRecord,
+        });
+      }
+
+      expect(getRecordedRecordKeys().size).toBe(10_000);
+      expect(getRecordedRecordKeys().has("idem:tenant-1:api_calls:key-0")).toBe(false);
+      expect(getRecordedRecordKeys().has("idem:tenant-1:api_calls:key-10000")).toBe(true);
+    });
+
     it("should throw RedisProblem on eval error", async () => {
       vi.mocked(mockRedis.eval).mockRejectedValue(new Error("Script failed"));