Feature/ccm 16073 rate limit it fix

lambdas/client-transform-filter-lambda/src/__tests__/index.component.test.ts

@@ -33,6 +33,7 @@ jest.mock("aws-embedded-metrics", () => ({
     Count: "Count",
     Milliseconds: "Milliseconds",
   },
+  StorageResolution: { High: 1, Standard: 60 },
 }));
 
 import { GetObjectCommand, NoSuchKey } from "@aws-sdk/client-s3";

lambdas/client-transform-filter-lambda/src/__tests__/services/metrics.test.ts

@@ -1,7 +1,15 @@
-import { Unit, createMetricsLogger } from "aws-embedded-metrics";
+import {
+  StorageResolution,
+  Unit,
+  createMetricsLogger,
+} from "aws-embedded-metrics";
 import { CallbackMetrics, createMetricLogger } from "services/metrics";
 
-jest.mock("aws-embedded-metrics");
+jest.mock("aws-embedded-metrics", () => ({
+  Unit: { Count: "Count" },
+  StorageResolution: { High: 1, Standard: 60 },
+  createMetricsLogger: jest.fn(),
+}));
 
 const mockPutMetric = jest.fn();
 const mockSetDimensions = jest.fn();
@@ -80,6 +88,7 @@ describe("CallbackMetrics", () => {
         "EventsReceived",
         1,
         Unit.Count,
+        StorageResolution.High,
       );
     });
   });
@@ -92,6 +101,7 @@ describe("CallbackMetrics", () => {
         "TransformationsSuccessful",
         1,
         Unit.Count,
+        StorageResolution.High,
       );
     });
   });
@@ -104,6 +114,7 @@ describe("CallbackMetrics", () => {
         "TransformationsFailed",
         1,
         Unit.Count,
+        StorageResolution.High,
       );
     });
   });
@@ -116,6 +127,7 @@ describe("CallbackMetrics", () => {
         "CallbacksInitiated",
         1,
         Unit.Count,
+        StorageResolution.High,
       );
     });
   });
@@ -128,6 +140,7 @@ describe("CallbackMetrics", () => {
         "ValidationErrors",
         1,
         Unit.Count,
+        StorageResolution.High,
       );
     });
   });
@@ -140,6 +153,7 @@ describe("CallbackMetrics", () => {
         "FilteringStarted",
         1,
         Unit.Count,
+        StorageResolution.High,
       );
     });
   });
@@ -152,6 +166,7 @@ describe("CallbackMetrics", () => {
         "FilteringMatched",
         1,
         Unit.Count,
+        StorageResolution.High,
       );
     });
   });

lambdas/client-transform-filter-lambda/src/services/metrics.ts

@@ -1,4 +1,8 @@
-import { Unit, createMetricsLogger } from "aws-embedded-metrics";
+import {
+  StorageResolution,
+  Unit,
+  createMetricsLogger,
+} from "aws-embedded-metrics";
 import type { MetricsLogger } from "aws-embedded-metrics";
 
 export const createMetricLogger = (): MetricsLogger => {
@@ -21,30 +25,65 @@ export class CallbackMetrics {
   constructor(private readonly metrics: MetricsLogger) {}
 
   emitEventReceived(): void {
-    this.metrics.putMetric("EventsReceived", 1, Unit.Count);
+    this.metrics.putMetric(
+      "EventsReceived",
+      1,
+      Unit.Count,
+      StorageResolution.High,
+    );
   }
 
   emitTransformationSuccess(): void {
-    this.metrics.putMetric("TransformationsSuccessful", 1, Unit.Count);
+    this.metrics.putMetric(
+      "TransformationsSuccessful",
+      1,
+      Unit.Count,
+      StorageResolution.High,
+    );
   }
 
   emitTransformationFailure(): void {
-    this.metrics.putMetric("TransformationsFailed", 1, Unit.Count);
+    this.metrics.putMetric(
+      "TransformationsFailed",
+      1,
+      Unit.Count,
+      StorageResolution.High,
+    );
   }
 
   emitDeliveryInitiated(): void {
-    this.metrics.putMetric("CallbacksInitiated", 1, Unit.Count);
+    this.metrics.putMetric(
+      "CallbacksInitiated",
+      1,
+      Unit.Count,
+      StorageResolution.High,
+    );
   }
 
   emitValidationError(): void {
-    this.metrics.putMetric("ValidationErrors", 1, Unit.Count);
+    this.metrics.putMetric(
+      "ValidationErrors",
+      1,
+      Unit.Count,
+      StorageResolution.High,
+    );
   }
 
   emitFilteringStarted(): void {
-    this.metrics.putMetric("FilteringStarted", 1, Unit.Count);
+    this.metrics.putMetric(
+      "FilteringStarted",
+      1,
+      Unit.Count,
+      StorageResolution.High,
+    );
   }
 
   emitFilteringMatched(): void {
-    this.metrics.putMetric("FilteringMatched", 1, Unit.Count);
+    this.metrics.putMetric(
+      "FilteringMatched",
+      1,
+      Unit.Count,
+      StorageResolution.High,
+    );
   }
 }

lambdas/https-client-lambda/src/__tests__/admit-lua.test.ts

@@ -1,7 +1,7 @@
 import admitLuaSrc from "services/admit.lua";
 import { createRedisStore, evalLua } from "__tests__/helpers/lua-redis-mock";
 
-// ARGV: [now, capacity, targetRateLimit, cooldownMs, recoveryPeriodMs, probeRateLimit, targetBatchSize]
+// ARGV: [now, capacity, targetRateLimit, cooldownMs, recoveryPeriodMs, probeRateLimit, targetBatchSize, cbEnabled]
 // KEYS: [epKey]
 // Returns: [consumedTokens, reason, retryAfterMs, effectiveRate]
 
@@ -13,6 +13,7 @@ type AdmitArgs = {
   recoveryPeriodMs: number;
   probeRateLimit: number;
   targetBatchSize: number;
+  cbEnabled: boolean;
 };
 
 const defaultArgs: AdmitArgs = {
@@ -23,6 +24,7 @@ const defaultArgs: AdmitArgs = {
   recoveryPeriodMs: 600_000,
   probeRateLimit: 1 / 60,
   targetBatchSize: 1,
+  cbEnabled: true,
 };
 
 type AdmitResult = {
@@ -49,6 +51,7 @@ function runAdmit(
       merged.recoveryPeriodMs.toString(),
       merged.probeRateLimit.toString(),
       merged.targetBatchSize.toString(),
+      merged.cbEnabled ? "1" : "0",
     ],
     store,
   ) as [number, string, number, number];
@@ -62,7 +65,7 @@ function runAdmit(
 
 describe("admit.lua", () => {
   describe("rate limiting", () => {
-    it("enters half-open probe on a fresh endpoint with no prior state", () => {
+    it("rate-limits on a fresh endpoint with no prior state", () => {
       const store = createRedisStore();
       const now = 1_000_000;
 
@@ -76,6 +79,21 @@ describe("admit.lua", () => {
       expect(effectiveRate).toBeCloseTo(1 / 60, 5);
     });
 
+    it("generates a probe token on the second call after enough elapsed time", () => {
+      const store = createRedisStore();
+
+      runAdmit(store, { now: 1_000_000, targetRateLimit: 10 });
+
+      const { consumedTokens, effectiveRate, reason } = runAdmit(store, {
+        now: 1_060_001,
+        targetRateLimit: 10,
+      });
+
+      expect(effectiveRate).toBeCloseTo(1 / 60, 5);
+      expect(consumedTokens).toBe(1);
+      expect(reason).toBe("some_allowed");
+    });
+
     it("does not persist circuit state on first contact", () => {
       const store = createRedisStore();
       const now = 1_000_000;
@@ -109,7 +127,7 @@ describe("admit.lua", () => {
       });
 
       expect(consumedTokens).toBeGreaterThanOrEqual(1);
-      expect(reason).toBe("allowed");
+      expect(reason).toBe("some_allowed");
     });
 
     it("allows a single request when bucket has tokens from refill", () => {
@@ -131,7 +149,7 @@ describe("admit.lua", () => {
       });
 
       expect(consumedTokens).toBe(1);
-      expect(reason).toBe("allowed");
+      expect(reason).toBe("some_allowed");
       expect(retryAfterMs).toBe(0);
     });
 
@@ -511,4 +529,121 @@ describe("admit.lua", () => {
       expect(store.has("ep:target-b")).toBe(true);
     });
   });
+
+  describe("circuit breaker disabled (cbEnabled = false)", () => {
+    it("uses full targetRateLimit on a fresh endpoint with no prior state", () => {
+      const store = createRedisStore();
+      const now = 1_000_000;
+
+      const { effectiveRate } = runAdmit(store, {
+        now,
+        targetRateLimit: 10,
+        cbEnabled: false,
+      });
+
+      expect(effectiveRate).toBe(10);
+    });
+
+    it("applies initial values on fresh endpoint so first call has no tokens", () => {
+      const store = createRedisStore();
+      const now = 1_000_000;
+
+      const { consumedTokens, effectiveRate, reason } = runAdmit(store, {
+        now,
+        targetRateLimit: 10,
+        cbEnabled: false,
+      });
+
+      expect(effectiveRate).toBe(10);
+      expect(consumedTokens).toBe(0);
+      expect(reason).toBe("rate_limited");
+    });
+
+    it("generates tokens at full rate after initial contact", () => {
+      const store = createRedisStore();
+
+      runAdmit(store, {
+        now: 1_000_000,
+        targetRateLimit: 10,
+        cbEnabled: false,
+      });
+      const { consumedTokens, reason } = runAdmit(store, {
+        now: 1_000_100,
+        targetRateLimit: 10,
+        cbEnabled: false,
+      });
+
+      expect(consumedTokens).toBe(1);
+      expect(reason).toBe("some_allowed");
+    });
+
+    it("ignores is_open state when CB is disabled", () => {
+      const store = createRedisStore();
+      const now = 1_000_000;
+      store.set(
+        "ep:t1",
+        new Map([
+          ["is_open", "1"],
+          ["switched_at", now.toString()],
+          ["bucket_tokens", "5"],
+          ["bucket_refilled_at", now.toString()],
+        ]),
+      );
+
+      const { consumedTokens, effectiveRate, reason } = runAdmit(store, {
+        now,
+        targetRateLimit: 10,
+        cbEnabled: false,
+      });
+
+      expect(effectiveRate).toBe(10);
+      expect(consumedTokens).toBe(1);
+      expect(reason).toBe("some_allowed");
+    });
+
+    it("does not zero bucket tokens when is_open and CB disabled", () => {
+      const store = createRedisStore();
+      const now = 1_000_000;
+      store.set(
+        "ep:t1",
+        new Map([
+          ["is_open", "1"],
+          ["switched_at", now.toString()],
+          ["bucket_tokens", "5"],
+          ["bucket_refilled_at", now.toString()],
+        ]),
+      );
+
+      const { consumedTokens } = runAdmit(store, {
+        now,
+        targetRateLimit: 10,
+        cbEnabled: false,
+        targetBatchSize: 3,
+      });
+
+      expect(consumedTokens).toBe(3);
+    });
+
+    it("never returns circuit_open when CB is disabled", () => {
+      const store = createRedisStore();
+      const now = 1_000_000;
+      store.set(
+        "ep:t1",
+        new Map([
+          ["is_open", "1"],
+          ["switched_at", (now - 10_000).toString()],
+          ["bucket_tokens", "0"],
+          ["bucket_refilled_at", now.toString()],
+        ]),
+      );
+
+      const { reason } = runAdmit(store, {
+        now,
+        cooldownMs: 120_000,
+        cbEnabled: false,
+      });
+
+      expect(reason).not.toBe("circuit_open");
+    });
+  });
 });