feat(matches): backoff and circuit breaker for the external result feed

backend/src/common/circuit-breaker.spec.ts

@@ -0,0 +1,130 @@
+import {
+  CircuitBreaker,
+  CircuitOpenError,
+} from './circuit-breaker';
+
+/** Controllable clock, so no test has to wait for a real `openMs` to elapse. */
+function fakeClock(start = 1_000_000) {
+  let now = start;
+  return {
+    now: () => now,
+    advance: (ms: number) => {
+      now += ms;
+    },
+  };
+}
+
+function makeBreaker(clock: ReturnType<typeof fakeClock>) {
+  return new CircuitBreaker({
+    label: 'test upstream',
+    failureThreshold: 3,
+    openMs: 30_000,
+    now: clock.now,
+  });
+}
+
+const boom = () => Promise.reject(new Error('upstream down'));
+
+describe('CircuitBreaker', () => {
+  it('stays closed and passes results through while calls succeed', async () => {
+    const breaker = makeBreaker(fakeClock());
+
+    await expect(breaker.run(async () => 'ok')).resolves.toBe('ok');
+    expect(breaker.health().state).toBe('closed');
+    expect(breaker.health().consecutiveFailures).toBe(0);
+  });
+
+  it('opens after the failure threshold is reached', async () => {
+    const clock = fakeClock();
+    const breaker = makeBreaker(clock);
+
+    for (let i = 0; i < 2; i++) {
+      await expect(breaker.run(boom)).rejects.toThrow('upstream down');
+      expect(breaker.health().state).toBe('closed');
+    }
+
+    await expect(breaker.run(boom)).rejects.toThrow('upstream down');
+
+    const health = breaker.health();
+    expect(health.state).toBe('open');
+    expect(health.consecutiveFailures).toBe(3);
+    expect(health.lastError).toBe('upstream down');
+    expect(health.retryAfterMs).toBe(30_000);
+  });
+
+  it('rejects without calling upstream while open', async () => {
+    const clock = fakeClock();
+    const breaker = makeBreaker(clock);
+    for (let i = 0; i < 3; i++) {
+      await expect(breaker.run(boom)).rejects.toThrow('upstream down');
+    }
+
+    const call = jest.fn(() => Promise.resolve('ok'));
+    await expect(breaker.run(call)).rejects.toBeInstanceOf(CircuitOpenError);
+    expect(call).not.toHaveBeenCalled();
+  });
+
+  it('recovers when the half-open probe succeeds', async () => {
+    const clock = fakeClock();
+    const breaker = makeBreaker(clock);
+    for (let i = 0; i < 3; i++) {
+      await expect(breaker.run(boom)).rejects.toThrow('upstream down');
+    }
+
+    clock.advance(30_000);
+    expect(breaker.health().state).toBe('half-open');
+
+    await expect(breaker.run(async () => 'recovered')).resolves.toBe(
+      'recovered',
+    );
+
+    const health = breaker.health();
+    expect(health.state).toBe('closed');
+    expect(health.consecutiveFailures).toBe(0);
+    expect(health.lastError).toBeNull();
+  });
+
+  it('re-opens for a full window when the half-open probe fails', async () => {
+    const clock = fakeClock();
+    const breaker = makeBreaker(clock);
+    for (let i = 0; i < 3; i++) {
+      await expect(breaker.run(boom)).rejects.toThrow('upstream down');
+    }
+
+    clock.advance(30_000);
+    await expect(breaker.run(boom)).rejects.toThrow('upstream down');
+
+    expect(breaker.health().state).toBe('open');
+    expect(breaker.health().retryAfterMs).toBe(30_000);
+
+    clock.advance(29_999);
+    expect(breaker.health().state).toBe('open');
+    clock.advance(1);
+    expect(breaker.health().state).toBe('half-open');
+  });
+
+  it('admits only one probe while half-open', async () => {
+    const clock = fakeClock();
+    const breaker = makeBreaker(clock);
+    for (let i = 0; i < 3; i++) {
+      await expect(breaker.run(boom)).rejects.toThrow('upstream down');
+    }
+    clock.advance(30_000);
+
+    let releaseProbe: (value: string) => void = () => undefined;
+    const probe = breaker.run(
+      () =>
+        new Promise<string>((resolve) => {
+          releaseProbe = resolve;
+        }),
+    );
+
+    const second = jest.fn(() => Promise.resolve('ok'));
+    await expect(breaker.run(second)).rejects.toBeInstanceOf(CircuitOpenError);
+    expect(second).not.toHaveBeenCalled();
+
+    releaseProbe('ok');
+    await expect(probe).resolves.toBe('ok');
+    expect(breaker.health().state).toBe('closed');
+  });
+});

backend/src/common/circuit-breaker.ts

@@ -0,0 +1,138 @@
+/**
+ * Minimal circuit breaker for outbound calls to a single upstream.
+ *
+ * Complements {@link withRetry} in `retry.util.ts` rather than replacing it:
+ * retry handles one bad call, the breaker handles a bad *upstream*. Without a
+ * breaker, every scheduled poll still pays the full retry budget against a
+ * service that is already down, which is exactly how a client earns a
+ * rate-limit ban while it is failing anyway.
+ *
+ * States:
+ *   closed    — calls pass through; consecutive failures are counted.
+ *   open      — calls are rejected immediately for `openMs`.
+ *   half-open — one probe is allowed through. Success closes the breaker,
+ *               failure re-opens it for another `openMs`.
+ *
+ * Deliberately not generalised into a decorator or a module: one upstream, one
+ * instance, constructed by whoever owns the call.
+ */
+
+export type BreakerState = 'closed' | 'open' | 'half-open';
+
+export class CircuitOpenError extends Error {
+  constructor(label: string, retryAfterMs: number) {
+    super(
+      `${label} circuit is open; not calling upstream for another ${retryAfterMs}ms`,
+    );
+    this.name = 'CircuitOpenError';
+  }
+}
+
+export interface CircuitBreakerOptions {
+  /** Human-readable name for the upstream, used in errors and logs. */
+  label: string;
+  /** Consecutive failures that trip the breaker. Default 5. */
+  failureThreshold?: number;
+  /** How long the breaker stays open before allowing a probe. Default 30s. */
+  openMs?: number;
+  /**
+   * Current time in ms. Injectable so tests can advance the clock without
+   * sleeping — a breaker tested with real timers is a slow, flaky test.
+   */
+  now?: () => number;
+}
+
+export interface BreakerHealth {
+  state: BreakerState;
+  consecutiveFailures: number;
+  /** ms until the next probe is allowed. 0 unless the state is `open`. */
+  retryAfterMs: number;
+  lastError: string | null;
+}
+
+export class CircuitBreaker {
+  private readonly label: string;
+  private readonly failureThreshold: number;
+  private readonly openMs: number;
+  private readonly now: () => number;
+
+  private consecutiveFailures = 0;
+  private openedAt: number | null = null;
+  private probeInFlight = false;
+  private lastError: string | null = null;
+
+  constructor(options: CircuitBreakerOptions) {
+    this.label = options.label;
+    this.failureThreshold = options.failureThreshold ?? 5;
+    this.openMs = options.openMs ?? 30_000;
+    this.now = options.now ?? (() => Date.now());
+  }
+
+  get state(): BreakerState {
+    if (this.openedAt === null) return 'closed';
+    return this.now() - this.openedAt >= this.openMs ? 'half-open' : 'open';
+  }
+
+  health(): BreakerHealth {
+    const state = this.state;
+    return {
+      state,
+      consecutiveFailures: this.consecutiveFailures,
+      retryAfterMs:
+        state === 'open' && this.openedAt !== null
+          ? Math.max(0, this.openMs - (this.now() - this.openedAt))
+          : 0,
+      lastError: this.lastError,
+    };
+  }
+
+  /**
+   * Runs `fn` unless the breaker is open.
+   *
+   * While half-open, only one probe is admitted; concurrent callers are
+   * rejected as if the breaker were still open. Without that guard a burst of
+   * queued callers would all stampede the recovering upstream at once.
+   */
+  async run<T>(fn: () => Promise<T>): Promise<T> {
+    const state = this.state;
+
+    if (state === 'open') {
+      throw new CircuitOpenError(this.label, this.health().retryAfterMs);
+    }
+
+    if (state === 'half-open') {
+      if (this.probeInFlight) {
+        throw new CircuitOpenError(this.label, 0);
+      }
+      this.probeInFlight = true;
+    }
+
+    try {
+      const result = await fn();
+      this.reset();
+      return result;
+    } catch (error) {
+      this.recordFailure(error);
+      throw error;
+    } finally {
+      this.probeInFlight = false;
+    }
+  }
+
+  private reset(): void {
+    this.consecutiveFailures = 0;
+    this.openedAt = null;
+    this.lastError = null;
+  }
+
+  private recordFailure(error: unknown): void {
+    this.consecutiveFailures += 1;
+    this.lastError = error instanceof Error ? error.message : String(error);
+
+    // A failed probe re-opens the breaker immediately, without waiting for the
+    // threshold again — the upstream just told us it is still unhealthy.
+    if (this.openedAt !== null || this.consecutiveFailures >= this.failureThreshold) {
+      this.openedAt = this.now();
+    }
+  }
+}

backend/src/matches/external-result-feed.client.spec.ts

@@ -1,7 +1,31 @@
 import { HttpService } from '@nestjs/axios';
 import { ConfigService } from '@nestjs/config';
-import { of } from 'rxjs';
-import { HttpExternalResultFeedClient } from './external-result-feed.client';
+import { of, throwError } from 'rxjs';
+import {
+  HttpExternalResultFeedClient,
+  isTransientFeedError,
+} from './external-result-feed.client';
+import { CircuitOpenError } from '../common/circuit-breaker';
+
+/** Matches the shape axios rejects with, which is what the client classifies. */
+function httpError(status: number): Error {
+  return Object.assign(new Error(`Request failed with status code ${status}`), {
+    response: { status },
+  });
+}
+
+function networkError(code: string): Error {
+  return Object.assign(new Error(code), { code });
+}
+
+function makeClient(get: jest.Mock) {
+  const config = {
+    getOrThrow: jest.fn(() => 'https://feed.test/results'),
+    get: jest.fn(() => undefined),
+  } as unknown as ConfigService;
+  const http = { get } as unknown as HttpService;
+  return new HttpExternalResultFeedClient(http, config);
+}
 
 describe('HttpExternalResultFeedClient', () => {
   it('uses the configured feed URL and credential and responds to config changes', async () => {
@@ -30,4 +54,89 @@ describe('HttpExternalResultFeedClient', () => {
       { headers: { Authorization: 'Bearer secret' } },
     );
   });
+
+  describe('error classification', () => {
+    it.each([408, 425, 429, 500, 502, 503, 504])(
+      'treats %i as transient',
+      (status) => {
+        expect(isTransientFeedError(httpError(status))).toBe(true);
+      },
+    );
+
+    it.each([400, 401, 403, 404, 422])(
+      'treats %i as permanent',
+      (status) => {
+        expect(isTransientFeedError(httpError(status))).toBe(false);
+      },
+    );
+
+    it('treats a request that never got a response as transient', () => {
+      expect(isTransientFeedError(networkError('ECONNRESET'))).toBe(true);
+    });
+  });
+
+  describe('backoff', () => {
+    it('retries a transient failure and returns the eventual result', async () => {
+      const payload = [{ externalId: 'm-1' }];
+      const get: jest.Mock = jest
+        .fn()
+        .mockReturnValueOnce(throwError(() => httpError(503)))
+        .mockReturnValueOnce(of({ data: payload }));
+
+      await expect(makeClient(get).fetchResults()).resolves.toBe(payload);
+      expect(get).toHaveBeenCalledTimes(2);
+    });
+
+    it('does not retry a rejected credential', async () => {
+      const get: jest.Mock = jest.fn(() => throwError(() => httpError(401)));
+
+      await expect(makeClient(get).fetchResults()).rejects.toThrow(
+        'status code 401',
+      );
+      expect(get).toHaveBeenCalledTimes(1);
+    });
+  });
+
+  describe('circuit breaker', () => {
+    afterEach(() => {
+      jest.restoreAllMocks();
+    });
+
+    it('opens after repeated failures and stops calling the feed', async () => {
+      const get: jest.Mock = jest.fn(() => throwError(() => httpError(401)));
+      const client = makeClient(get);
+
+      for (let i = 0; i < 5; i++) {
+        await expect(client.fetchResults()).rejects.toThrow('status code 401');
+      }
+      expect(client.getHealth().state).toBe('open');
+
+      await expect(client.fetchResults()).rejects.toBeInstanceOf(
+        CircuitOpenError,
+      );
+      // Still five: the sixth poll was rejected locally.
+      expect(get).toHaveBeenCalledTimes(5);
+    });
+
+    it('closes again when the half-open probe succeeds', async () => {
+      const payload = [{ externalId: 'm-2' }];
+      const get: jest.Mock = jest.fn(() => throwError(() => httpError(401)));
+      const client = makeClient(get);
+
+      for (let i = 0; i < 5; i++) {
+        await expect(client.fetchResults()).rejects.toThrow('status code 401');
+      }
+      expect(client.getHealth().state).toBe('open');
+
+      // Move past the open window rather than sleeping through it.
+      const realNow = Date.now();
+      jest.spyOn(Date, 'now').mockReturnValue(realNow + 60_000);
+      expect(client.getHealth().state).toBe('half-open');
+
+      get.mockReturnValue(of({ data: payload }));
+      await expect(client.fetchResults()).resolves.toBe(payload);
+      expect(client.getHealth().state).toBe('closed');
+      expect(client.getHealth().consecutiveFailures).toBe(0);
+    });
+  });
 });