diff --git a/backend/src/common/circuit-breaker.spec.ts b/backend/src/common/circuit-breaker.spec.ts new file mode 100644 index 00000000..3671c764 --- /dev/null +++ b/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) { + 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((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'); + }); +}); diff --git a/backend/src/common/circuit-breaker.ts b/backend/src/common/circuit-breaker.ts new file mode 100644 index 00000000..9a2740ab --- /dev/null +++ b/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(fn: () => Promise): Promise { + 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(); + } + } +} diff --git a/backend/src/matches/external-result-feed.client.spec.ts b/backend/src/matches/external-result-feed.client.spec.ts index cbbdeabd..169c18ec 100644 --- a/backend/src/matches/external-result-feed.client.spec.ts +++ b/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); + }); + }); }); diff --git a/backend/src/matches/external-result-feed.client.ts b/backend/src/matches/external-result-feed.client.ts index c3004bae..22538b84 100644 --- a/backend/src/matches/external-result-feed.client.ts +++ b/backend/src/matches/external-result-feed.client.ts @@ -1,8 +1,14 @@ -import { Injectable } from '@nestjs/common'; +import { Injectable, Logger } from '@nestjs/common'; import { HttpService } from '@nestjs/axios'; import { ConfigService } from '@nestjs/config'; import { firstValueFrom } from 'rxjs'; import { WinningTeam } from './entities/match.entity'; +import { withRetry } from '../common/retry.util'; +import { + BreakerHealth, + CircuitBreaker, + CircuitOpenError, +} from '../common/circuit-breaker'; export interface ExternalMatchResultPayload { externalId: string; @@ -19,23 +25,124 @@ export interface ExternalResultFeedClient { fetchResults(): Promise; } +/** Attempts per poll, including the first. Kept low: the poller runs again soon. */ +const MAX_ATTEMPTS = 3; +const BASE_DELAY_MS = 500; +/** Consecutive failed polls (after retries) before the breaker opens. */ +const FAILURE_THRESHOLD = 5; +const OPEN_MS = 60_000; + +/** Status codes worth another attempt: the upstream is busy, not wrong. */ +const TRANSIENT_STATUS = new Set([408, 425, 429, 500, 502, 503, 504]); + +const TRANSIENT_ERRNO = new Set([ + 'ECONNRESET', + 'ECONNREFUSED', + 'ETIMEDOUT', + 'ECONNABORTED', + 'EAI_AGAIN', + 'EPIPE', + 'ENETUNREACH', + 'ENOTFOUND', +]); + +/** + * A response that arrived but was rejected (4xx other than the codes above) is + * permanent: the URL or the credential is wrong, and retrying just repeats a + * request the upstream has already refused. + */ +export function isTransientFeedError(error: unknown): boolean { + if (!(error instanceof Error)) return false; + if (error.name === 'AbortError') return true; + + const axiosLike = error as { + response?: { status?: number }; + code?: string; + }; + + const status = axiosLike.response?.status; + if (typeof status === 'number') return TRANSIENT_STATUS.has(status); + + // No response at all — the request never completed, so it is worth retrying. + return typeof axiosLike.code === 'string' + ? TRANSIENT_ERRNO.has(axiosLike.code) + : true; +} + @Injectable() export class HttpExternalResultFeedClient implements ExternalResultFeedClient { + private readonly logger = new Logger(HttpExternalResultFeedClient.name); + + private readonly breaker = new CircuitBreaker({ + label: 'external result feed', + failureThreshold: FAILURE_THRESHOLD, + openMs: OPEN_MS, + }); + constructor( private readonly httpService: HttpService, private readonly configService: ConfigService, ) {} + /** + * Current state of the upstream as this client sees it, for monitoring. + * + * Exposed as a plain getter rather than wired into `HealthService`: that + * service composes injected Terminus indicators, and adding the feed to it + * changes which dependencies `/health` reports on. Left for a follow-up so + * that decision is made deliberately rather than as a side effect here. + */ + getHealth(): BreakerHealth { + return this.breaker.health(); + } + async fetchResults(): Promise { const url = this.configService.getOrThrow('MATCH_RESULTS_FEED_URL'); const credential = this.configService.get( 'MATCH_RESULTS_FEED_CREDENTIAL', ); - const response = await firstValueFrom( - this.httpService.get(url, { - headers: credential ? { Authorization: `Bearer ${credential}` } : {}, - }), - ); - return response.data; + + // Retry sits inside the breaker, not around it: one poll gets one retry + // budget, and the breaker counts polls rather than individual attempts. + return this.breaker.run(() => + withRetry( + async () => { + const response = await firstValueFrom( + this.httpService.get(url, { + headers: credential + ? { Authorization: `Bearer ${credential}` } + : {}, + }), + ); + return response.data; + }, + { + maxAttempts: MAX_ATTEMPTS, + baseDelayMs: BASE_DELAY_MS, + isTransient: isTransientFeedError, + onRetry: (error, attempt, delayMs) => { + this.logger.warn( + `result feed attempt ${attempt + 1} failed (${ + error instanceof Error ? error.message : String(error) + }); retrying in ${delayMs}ms`, + ); + }, + }, + ), + ).catch((error: unknown) => { + // A rejection from an open breaker is a local decision, not an upstream + // failure — log it differently so it is not mistaken for a new outage. + if (error instanceof CircuitOpenError) { + this.logger.warn(error.message); + } else { + const { state, consecutiveFailures } = this.breaker.health(); + this.logger.error( + `result feed poll failed (${ + error instanceof Error ? error.message : String(error) + }); breaker=${state} consecutiveFailures=${consecutiveFailures}`, + ); + } + throw error; + }); } }