fix(webhooks): isolate half-open probe state per consumer URL

src/redis/webhookCircuitBreakerStore.ts

@@ -230,10 +230,38 @@ export class RedisWebhookCircuitBreakerStore implements WebhookCircuitBreakerSto
   }
 }
 
+/**
+ * In-memory implementation of the {@link WebhookCircuitBreakerStore}.
+ *
+ * Enforces half-open probe semantics and ensures the exactly-one-probe invariant:
+ * - When the circuit is 'open' and the reset period has passed, exactly one probe attempt
+ *   is allowed to be in-flight to check the service health.
+ * - This behavior is isolated per consumer URL. Concurrent probe requests on different
+ *   URLs do not interfere with one another.
+ * - A probe attempt is registered with a Time-To-Live (TTL) matching the Redis implementation's
+ *   lock duration (maximum 60 seconds) to ensure that if a dispatcher crashes or abandons
+ *   the probe, the state does not remain stale indefinitely.
+ * - The probe state is reset after a successful probe, after a failed probe, or after any completed
+ *   probe attempt (either manually via success/failure records, or when the TTL expires).
+ */
 export class InMemoryWebhookCircuitBreakerStore implements WebhookCircuitBreakerStore {
   private readonly states = new Map<string, WebhookCircuitBreakerRecord>();
-  private readonly probes = new Set<string>();
-
+  /** Maps hashed consumer URL key -> probe lock expiration timestamp (ms since epoch) */
+  private readonly probes = new Map<string, number>();
+
+  /**
+   * Checks the circuit breaker state and attempts to claim a half-open probe.
+   *
+   * @param consumerUrl The target webhook endpoint.
+   * @param policy The circuit breaker configuration policy.
+   * @param now The current timestamp.
+   * @returns A check result indicating if the attempt is allowed, the state, failures, and reset duration.
+   *
+   * @remarks
+   * Evaluates if a probe is already in-flight by verifying if a probe exists in the local map and
+   * its expiration time is in the future. Ensures exactly one probe is admitted in flight per consumer URL
+   * under concurrent calls.
+   */
   async checkAndClaimAttempt(
     consumerUrl: string,
     policy: CircuitBreakerPolicy,
@@ -250,10 +278,13 @@ export class InMemoryWebhookCircuitBreakerStore implements WebhookCircuitBreaker
       if (now < record.resetAt) {
         return { allowed: false, state: 'open', consecutiveFailures: record.consecutiveFailures, resetAt: record.resetAt };
       }
-      if (this.probes.has(key)) {
+      const probeExpiry = this.probes.get(key);
+      if (probeExpiry !== undefined && now < probeExpiry) {
         return { allowed: false, state: 'half-open', consecutiveFailures: record.consecutiveFailures, resetAt: record.resetAt };
       }
-      this.probes.add(key);
+      const resetMs = policy.circuitBreakerResetMs ?? 300_000;
+      const expiry = now + Math.min(resetMs, 60_000);
+      this.probes.set(key, expiry);
       this.states.set(key, { state: 'half-open', consecutiveFailures: record.consecutiveFailures, resetAt: 0 });
       emit('open', 'half-open');
       return { allowed: true, state: 'half-open', consecutiveFailures: record.consecutiveFailures, resetAt: null };
@@ -266,6 +297,10 @@ export class InMemoryWebhookCircuitBreakerStore implements WebhookCircuitBreaker
     };
   }
 
+  /**
+   * Records a successful probe outcome, resetting the consecutive failures
+   * count and closing the circuit. Clears the in-flight probe state.
+   */
   async recordSuccess(consumerUrl: string, _policy: CircuitBreakerPolicy): Promise<WebhookCircuitBreakerRecord> {
     const key = hashConsumerUrl(consumerUrl);
     const previous = this.states.get(key) ?? closed();
@@ -276,6 +311,10 @@ export class InMemoryWebhookCircuitBreakerStore implements WebhookCircuitBreaker
     return next;
   }
 
+  /**
+   * Records a failed probe outcome, re-opening the circuit breaker with a backoff delay.
+   * Clears the in-flight probe state.
+   */
   async recordFailure(consumerUrl: string, policy: CircuitBreakerPolicy, now = Date.now()): Promise<WebhookCircuitBreakerRecord> {
     const threshold = policy.circuitBreakerThreshold ?? 0;
     if (threshold <= 0) return closed();

tests/webhooks/circuitBreaker.store.test.ts

@@ -8,6 +8,7 @@ import {
   hashConsumerUrl,
   WEBHOOK_CIRCUIT_BREAKER_KEY_PREFIX,
   WEBHOOK_CIRCUIT_BREAKER_PROBE_PREFIX,
+  type WebhookCircuitBreakerStore,
 } from '../../src/redis/webhookCircuitBreakerStore.js';
 import {
   attemptWebhookDeliveryWithRateLimit,
@@ -200,6 +201,230 @@ describe('InMemoryWebhookCircuitBreakerStore', () => {
   });
 });
 
+describe('Circuit Breaker Store Concurrency and Lifecycle (Shared Semantics)', () => {
+  const urls = [
+    'https://consumer-a.example/webhooks',
+    'https://consumer-b.example/webhooks',
+  ];
+
+  const runSharedSuite = (getStore: () => WebhookCircuitBreakerStore, isRedis: boolean) => {
+    let store: WebhookCircuitBreakerStore;
+
+    beforeEach(() => {
+      store = getStore();
+    });
+
+    afterEach(async () => {
+      await store.close();
+    });
+
+    it('enforces exactly one probe admitted per URL under concurrency', async () => {
+      const now = 10_000_000;
+      const targetUrl = urls[0];
+
+      // Move URL to open state
+      for (let i = 0; i < 3; i++) {
+        await store.recordFailure(targetUrl, policy, now + i * 1000);
+      }
+
+      const openState = await store.getState(targetUrl);
+      expect(openState?.state).toBe('open');
+
+      // Attempt to claim probe concurrently
+      const results = await Promise.all([
+        store.checkAndClaimAttempt(targetUrl, policy, openState!.resetAt),
+        store.checkAndClaimAttempt(targetUrl, policy, openState!.resetAt),
+        store.checkAndClaimAttempt(targetUrl, policy, openState!.resetAt),
+      ]);
+
+      const allowedCount = results.filter((r) => r.allowed).length;
+      expect(allowedCount).toBe(1);
+
+      const allowedResult = results.find((r) => r.allowed);
+      expect(allowedResult?.state).toBe('half-open');
+
+      const deniedResults = results.filter((r) => !r.allowed);
+      expect(deniedResults).toHaveLength(2);
+      for (const res of deniedResults) {
+        expect(res.state).toBe('half-open');
+      }
+    });
+
+    it('ensures concurrent probes for two different URLs do not interfere', async () => {
+      const now = 20_000_000;
+      const urlA = urls[0];
+      const urlB = urls[1];
+
+      // Open both URL A and URL B
+      for (let i = 0; i < 3; i++) {
+        await store.recordFailure(urlA, policy, now + i * 1000);
+        await store.recordFailure(urlB, policy, now + i * 1000);
+      }
+
+      const stateA = await store.getState(urlA);
+      const stateB = await store.getState(urlB);
+
+      expect(stateA?.state).toBe('open');
+      expect(stateB?.state).toBe('open');
+
+      // Claim concurrently for different URLs
+      const [resA, resB] = await Promise.all([
+        store.checkAndClaimAttempt(urlA, policy, stateA!.resetAt),
+        store.checkAndClaimAttempt(urlB, policy, stateB!.resetAt),
+      ]);
+
+      expect(resA.allowed).toBe(true);
+      expect(resA.state).toBe('half-open');
+      expect(resB.allowed).toBe(true);
+      expect(resB.state).toBe('half-open');
+    });
+
+    it('resets probe state correctly after successful probe', async () => {
+      const now = 30_000_000;
+      const targetUrl = urls[0];
+
+      // Open
+      for (let i = 0; i < 3; i++) {
+        await store.recordFailure(targetUrl, policy, now + i * 1000);
+      }
+
+      const openState = await store.getState(targetUrl);
+      const probeRes = await store.checkAndClaimAttempt(targetUrl, policy, openState!.resetAt);
+      expect(probeRes.allowed).toBe(true);
+
+      // Record success
+      await store.recordSuccess(targetUrl, policy);
+
+      const stateAfterSuccess = await store.getState(targetUrl);
+      expect(stateAfterSuccess?.state).toBe('closed');
+      expect(stateAfterSuccess?.consecutiveFailures).toBe(0);
+
+      // Verify a new probe is not active and the circuit is closed (allowed = true, state = closed)
+      const checkRes = await store.checkAndClaimAttempt(targetUrl, policy, now);
+      expect(checkRes.allowed).toBe(true);
+      expect(checkRes.state).toBe('closed');
+    });
+
+    it('resets probe state correctly after failed probe', async () => {
+      const now = 40_000_000;
+      const targetUrl = urls[0];
+
+      // Open
+      for (let i = 0; i < 3; i++) {
+        await store.recordFailure(targetUrl, policy, now + i * 1000);
+      }
+
+      const openState1 = await store.getState(targetUrl);
+      const probeRes1 = await store.checkAndClaimAttempt(targetUrl, policy, openState1!.resetAt);
+      expect(probeRes1.allowed).toBe(true);
+
+      // Record failure (re-opens circuit)
+      const recordFailTime = openState1!.resetAt + 1000;
+      const reopenState = await store.recordFailure(targetUrl, policy, recordFailTime);
+      expect(reopenState.state).toBe('open');
+
+      // Now that it has failed, the probe state is cleared.
+      // After resetAt is reached again, we should be able to claim a new probe.
+      const probeRes2 = await store.checkAndClaimAttempt(targetUrl, policy, reopenState.resetAt);
+      expect(probeRes2.allowed).toBe(true);
+      expect(probeRes2.state).toBe('half-open');
+    });
+
+    it('repeated concurrent probes for the same URL behave correctly', async () => {
+      const now = 50_000_000;
+      const targetUrl = urls[0];
+
+      // 1. Open
+      for (let i = 0; i < 3; i++) {
+        await store.recordFailure(targetUrl, policy, now + i * 1000);
+      }
+
+      let openState = await store.getState(targetUrl);
+
+      // 2. Claim concurrently (Cycle 1)
+      const results1 = await Promise.all([
+        store.checkAndClaimAttempt(targetUrl, policy, openState!.resetAt),
+        store.checkAndClaimAttempt(targetUrl, policy, openState!.resetAt),
+      ]);
+      expect(results1.filter(r => r.allowed)).toHaveLength(1);
+
+      // 3. Complete probe with failure
+      const failTime = openState!.resetAt + 1000;
+      openState = await store.recordFailure(targetUrl, policy, failTime);
+      expect(openState.state).toBe('open');
+
+      // 4. Claim concurrently again (Cycle 2)
+      const results2 = await Promise.all([
+        store.checkAndClaimAttempt(targetUrl, policy, openState.resetAt),
+        store.checkAndClaimAttempt(targetUrl, policy, openState.resetAt),
+      ]);
+      expect(results2.filter(r => r.allowed)).toHaveLength(1);
+    });
+
+    it('probe expires and allows another probe', async () => {
+      const now = 60_000_000;
+      const targetUrl = urls[0];
+
+      // Open
+      for (let i = 0; i < 3; i++) {
+        await store.recordFailure(targetUrl, policy, now + i * 1000);
+      }
+
+      const openState = await store.getState(targetUrl);
+
+      // Claim first probe
+      const firstProbe = await store.checkAndClaimAttempt(targetUrl, policy, openState!.resetAt);
+      expect(firstProbe.allowed).toBe(true);
+
+      // Try to claim another probe immediately (should be denied)
+      const immediateSecond = await store.checkAndClaimAttempt(targetUrl, policy, openState!.resetAt);
+      expect(immediateSecond.allowed).toBe(false);
+
+      // Reset the state to 'open' to simulate that the first probe did not complete
+      // and we want to retry. (If state remains 'half-open', subsequent checks are blocked by state).
+      if (isRedis) {
+        const stateKey = `${WEBHOOK_CIRCUIT_BREAKER_KEY_PREFIX}${hashConsumerUrl(targetUrl)}`;
+        const record = { state: 'open' as const, consecutiveFailures: 3, resetAt: openState!.resetAt };
+        await (store as any).client.set(stateKey, JSON.stringify(record));
+      } else {
+        (store as any).states.set(hashConsumerUrl(targetUrl), { state: 'open', consecutiveFailures: 3, resetAt: openState!.resetAt });
+      }
+
+      // If we check now, it should still be blocked because the probe lock/key is still active
+      const stillBlocked = await store.checkAndClaimAttempt(targetUrl, policy, openState!.resetAt);
+      expect(stillBlocked.allowed).toBe(false);
+
+      // Now simulate expiration of the probe lock:
+      if (isRedis) {
+        const probeKey = `${WEBHOOK_CIRCUIT_BREAKER_PROBE_PREFIX}${hashConsumerUrl(targetUrl)}`;
+        await (store as any).client.del(probeKey);
+      }
+
+      const checkTime = isRedis ? openState!.resetAt : openState!.resetAt + 65_000;
+      const afterExpiry = await store.checkAndClaimAttempt(targetUrl, policy, checkTime);
+
+      // It should be allowed now because the probe lock expired/was cleared!
+      expect(afterExpiry.allowed).toBe(true);
+      expect(afterExpiry.state).toBe('half-open');
+    });
+  };
+
+  describe('InMemoryWebhookCircuitBreakerStore', () => {
+    runSharedSuite(() => new InMemoryWebhookCircuitBreakerStore(), false);
+  });
+
+  describe('RedisWebhookCircuitBreakerStore', () => {
+    let redis: FakeRedisClient;
+    beforeEach(() => {
+      redis = new FakeRedisClient();
+    });
+    afterEach(() => {
+      redis.reset();
+    });
+    runSharedSuite(() => new RedisWebhookCircuitBreakerStore(redis), true);
+  });
+});
+
 describe('checkWebhookDeliveryGate / attemptWebhookDeliveryWithRateLimit', () => {
   let redis: FakeRedisClient;
   let store: RedisWebhookCircuitBreakerStore;
@@ -307,7 +532,6 @@ describe('checkWebhookDeliveryGate / attemptWebhookDeliveryWithRateLimit', () =>
       attemptNumber: 1,
       policy,
       now,
-      lastAttempt: { attemptNumber: 1, timestamp: now, statusCode: 500 },
     });
 
     expect(plan.shouldRetry).toBe(true);