feat(provider): add upstream circuit breaker

Author: AlisinaDevelo

CHANGELOG.md

@@ -22,6 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Docker healthcheck now honors the runtime `PORT` environment variable.
 - Versioned `/api/v1/*` endpoints with `{ data, meta }` response envelopes, cache metadata, and structured error bodies.
 - OpenAPI-backed response contract tests for the v1 success and structured error envelopes.
+- Provider circuit breaker for repeated GNews failures, including Prometheus circuit event metrics and `503` short-circuit responses.
 
 ### Changed
 

README.md

@@ -8,7 +8,7 @@ A search runs through a small, explicit pipeline — each stage is a separate, t
 
 1. **Validate** — query params are checked before any network call: `query` is required, `count` is clamped (≤100), `lang`/`country` must be ISO two-letter codes, `from`/`to` must parse as ISO 8601, and `sortBy` ∈ {`publishedAt`, `relevance`}. Bad input fails fast with `400` instead of wasting an upstream call or quota.
 2. **Cache** — parameters are normalized into a deterministic key and read through a pluggable store (in-memory by default, Redis when `REDIS_URL` is set). Cache failures are logged/metriced but do not fail article requests; identical in-flight misses in the same process share one upstream call.
-3. **Upstream** — on a miss, GNews is called with a hard timeout; transport/provider failures surface as `502` rather than a hung request.
+3. **Upstream** — on a miss, GNews is called with a hard timeout; transport/provider failures surface as `502`, and repeated failures open a short circuit that returns `503` without amplifying the outage.
 4. **Observe** — each step emits structured Pino logs (carrying `x-request-id`), Prometheus counters (cache hit/miss, upstream outcome, latency histogram), and optional OpenTelemetry spans.
 5. **Respond** — legacy endpoints return raw article arrays, while `/api/v1/*` returns `{ data, meta }` envelopes with request/cache metadata and structured error bodies.
 
@@ -17,8 +17,8 @@ Full diagram and component notes live in [docs/ARCHITECTURE.md](docs/ARCHITECTUR
 ## Production-oriented features
 
 - **Security:** [Helmet](https://helmetjs.github.io/) headers, configurable rate limiting, optional `TRUST_PROXY` for correct client IPs behind a load balancer, optional **`CLIENT_API_KEYS`** + `X-API-Key` on `/api/*`.
-- **Reliability:** Upstream HTTP timeouts, response validation, `502` for provider/transport failures, graceful shutdown on `SIGTERM` / `SIGINT`.
-- **Observability:** JSON logs via [Pino](https://getpino.io/), `x-request-id`, **`GET /metrics`** ([Prometheus](https://prometheus.io/) text format), cache hit/miss/error/coalescing + upstream latency metrics, and optional **OpenTelemetry** traces to OTLP (`OTEL_EXPORTER_OTLP_*`).
+- **Reliability:** Upstream HTTP timeouts, response validation, `502` for provider/transport failures, provider circuit breaker with `503` short-circuiting, graceful shutdown on `SIGTERM` / `SIGINT`.
+- **Observability:** JSON logs via [Pino](https://getpino.io/), `x-request-id`, **`GET /metrics`** ([Prometheus](https://prometheus.io/) text format), cache hit/miss/error/coalescing + upstream latency/circuit metrics, and optional **OpenTelemetry** traces to OTLP (`OTEL_EXPORTER_OTLP_*`).
 - **Kubernetes-style probes:** `GET /health` (liveness), `GET /ready` (readiness when the API key is configured).
 - **Supply chain:** `npm audit` in CI; **SPDX SBOM** artifacts; Docker builds with **SBOM + provenance**; **dependency review** on PRs; optional **SLSA-style lockfile attestation** on `main`; lockfile-only installs.
 - **Contract:** OpenAPI at **`GET /openapi.yaml`** (also on disk as [docs/openapi.yaml](docs/openapi.yaml)).

docs/ARCHITECTURE.md

@@ -17,8 +17,9 @@ flowchart LR
 2. **Express** (`src/app.ts`) applies middleware in order: trust-proxy (optional), **Pino** request logging, **metrics** observer, **Helmet**, JSON body parser, **rate limiting** (skips `/health`, `/ready`, `/openapi.yaml`, `/metrics`), then mounts `/api` routes.
 3. **Controllers** validate query parameters and map domain results to HTTP status codes.
 4. **News service** builds cache keys from normalized search parameters (`query`, `count`, `lang`, `country`, `from`, `to`, `sortBy`), reads through `getCacheStore()` (in-memory or **Redis** when `REDIS_URL` is set), coalesces identical in-flight misses per process, and delegates upstream fetches to the GNews provider adapter. Cache backend errors are logged and metriced without failing the article request.
-5. **Response mapping** keeps legacy `/api/articles*` endpoints backward compatible with raw arrays, while `/api/v1/*` returns `{ data, meta }` envelopes with `requestId`, normalized filters, and cache status.
-6. **Title** and **source** endpoints reuse the search call, then narrow results in memory (exact title match; case-insensitive source name match).
+5. **Provider adapter** maps domain search options to GNews parameters, validates provider payloads, records upstream metrics, and opens a short circuit after repeated provider failures so outages are shed locally instead of amplified.
+6. **Response mapping** keeps legacy `/api/articles*` endpoints backward compatible with raw arrays, while `/api/v1/*` returns `{ data, meta }` envelopes with `requestId`, normalized filters, and cache status.
+7. **Title** and **source** endpoints reuse the search call, then narrow results in memory (exact title match; case-insensitive source name match).
 
 ### A search, end to end
 
@@ -85,6 +86,10 @@ Article arrays are stored per normalized search key with a **600-second** TTL (`
 
 The service treats the cache as a quota and latency optimization, not as a hard dependency. Read failures fall through to the upstream provider, write failures return the upstream response without caching it, and both paths emit warning logs plus cache error metrics. Within a single process, concurrent misses for the same normalized key are coalesced so only the first request calls the provider.
 
+## Provider Circuit Breaker
+
+The GNews provider adapter tracks consecutive provider failures. After `UPSTREAM_CIRCUIT_FAILURE_THRESHOLD` failures, it opens a cooldown window (`UPSTREAM_CIRCUIT_COOLDOWN_MS`) and returns `503` without making another upstream call. After the cooldown, one request is allowed through; success closes the circuit, while another failure opens it again.
+
 ## Metrics
 
-`src/metrics/register.ts` exports a single Prometheus registry used by `/metrics`. HTTP middleware records response counts, while `newsService` records cache hits/misses/errors/coalesced misses, upstream request outcomes, and upstream latency buckets.
+`src/metrics/register.ts` exports a single Prometheus registry used by `/metrics`. HTTP middleware records response counts, while `newsService` and the provider adapter record cache hits/misses/errors/coalesced misses, upstream request outcomes, upstream latency buckets, and circuit breaker events.

docs/OPERATIONS.md

@@ -10,6 +10,8 @@
 | `LOG_LEVEL` | `info` (non-test) | Pino level (`trace`–`fatal`, or `silent`). |
 | `GNEWS_BASE_URL` | `https://gnews.io/api/v4` | Upstream provider base URL. Override only for local integration tests, benchmarks, or compatible provider mocks. |
 | `HTTP_TIMEOUT_MS` | `15000` | Outbound GNews request timeout (max `60000`). |
+| `UPSTREAM_CIRCUIT_FAILURE_THRESHOLD` | `3` | Consecutive provider failures before the circuit opens. |
+| `UPSTREAM_CIRCUIT_COOLDOWN_MS` | `30000` | How long to short-circuit provider calls after the circuit opens (max `300000`). |
 | `SHUTDOWN_TIMEOUT_MS` | `10000` | Force-exit if `server.close` does not finish. |
 | `RATE_LIMIT_MAX` | `120` | Max requests per IP per window. |
 | `RATE_LIMIT_WINDOW_MS` | `60000` | Rate-limit window. |
@@ -52,8 +54,9 @@ On shutdown the server closes the Redis connection when that backend was used.
 | `news_cache_errors_total` | `operation=get|set` | Cache backend errors that were tolerated by falling through to upstream or returning an uncached upstream response. |
 | `news_upstream_requests_total` | `outcome=success|error|invalid_payload` | GNews provider request outcomes. |
 | `news_upstream_request_duration_seconds` | `outcome=success|error|invalid_payload` | GNews provider request latency histogram. |
+| `news_upstream_circuit_events_total` | `event=opened|short_circuit|half_open|closed` | Provider circuit breaker state transitions and short-circuited requests. |
 
-Use cache hit rate and coalesced miss counts to understand quota protection, cache error metrics to detect Redis/backend trouble, and upstream latency/error metrics to separate provider trouble from local API trouble.
+Use cache hit rate and coalesced miss counts to understand quota protection, cache error metrics to detect Redis/backend trouble, upstream latency/error metrics to separate provider trouble from local API trouble, and circuit events to see when repeated provider failures are being shed locally.
 
 ## Docker
 

docs/openapi.yaml

@@ -103,6 +103,8 @@ paths:
           description: Rate limit exceeded
         "502":
           $ref: "#/components/responses/V1BadGateway"
+        "503":
+          $ref: "#/components/responses/V1ServiceUnavailable"
         "500":
           $ref: "#/components/responses/V1InternalError"
 
@@ -137,6 +139,8 @@ paths:
           description: Rate limit exceeded
         "502":
           $ref: "#/components/responses/V1BadGateway"
+        "503":
+          $ref: "#/components/responses/V1ServiceUnavailable"
         "500":
           $ref: "#/components/responses/V1InternalError"
 
@@ -173,6 +177,8 @@ paths:
           description: Rate limit exceeded
         "502":
           $ref: "#/components/responses/V1BadGateway"
+        "503":
+          $ref: "#/components/responses/V1ServiceUnavailable"
         "500":
           $ref: "#/components/responses/V1InternalError"
 
@@ -211,6 +217,8 @@ paths:
           description: Rate limit exceeded
         "502":
           $ref: "#/components/responses/V1BadGateway"
+        "503":
+          $ref: "#/components/responses/V1ServiceUnavailable"
         "500":
           $ref: "#/components/responses/V1InternalError"
 
@@ -293,6 +301,8 @@ paths:
           description: Rate limit exceeded
         "502":
           $ref: "#/components/responses/BadGateway"
+        "503":
+          description: Upstream provider circuit breaker is open
         "500":
           $ref: "#/components/responses/InternalError"
 
@@ -334,6 +344,8 @@ paths:
           description: Rate limit exceeded
         "502":
           $ref: "#/components/responses/BadGateway"
+        "503":
+          description: Upstream provider circuit breaker is open
         "500":
           $ref: "#/components/responses/InternalError"
 
@@ -416,6 +428,8 @@ paths:
           description: Rate limit exceeded
         "502":
           $ref: "#/components/responses/BadGateway"
+        "503":
+          description: Upstream provider circuit breaker is open
         "500":
           $ref: "#/components/responses/InternalError"
 
@@ -540,6 +554,12 @@ components:
         application/json:
           schema:
             $ref: "#/components/schemas/ErrorEnvelope"
+    V1ServiceUnavailable:
+      description: Upstream provider circuit breaker is open
+      content:
+        application/json:
+          schema:
+            $ref: "#/components/schemas/ErrorEnvelope"
 
   schemas:
     ErrorBody:

src/metrics/register.ts

@@ -39,3 +39,10 @@ export const upstreamRequestDurationSeconds = new Histogram({
   buckets: [0.05, 0.1, 0.25, 0.5, 1, 2, 5, 10, 15],
   registers: [register],
 });
+
+export const upstreamCircuitEventsTotal = new Counter({
+  name: "news_upstream_circuit_events_total",
+  help: "Total upstream provider circuit breaker events",
+  labelNames: ["event"],
+  registers: [register],
+});

src/providers/gnewsProvider.ts

@@ -2,6 +2,7 @@ import axios from "axios";
 import { UPSTREAM_BASE_URL, UPSTREAM_TIMEOUT_MS } from "../config/upstream";
 import { HttpError } from "../errors/HttpError";
 import {
+  upstreamCircuitEventsTotal,
   upstreamRequestDurationSeconds,
   upstreamRequestsTotal,
 } from "../metrics/register";
@@ -12,6 +13,66 @@ export interface NewsProvider {
   search(options: ArticleSearchOptions): Promise<Article[]>;
 }
 
+interface CircuitState {
+  failures: number;
+  openedAt: number | undefined;
+}
+
+const circuit: CircuitState = {
+  failures: 0,
+  openedAt: undefined,
+};
+
+function failureThreshold(): number {
+  const raw = Number(process.env.UPSTREAM_CIRCUIT_FAILURE_THRESHOLD ?? 3);
+  return Number.isFinite(raw) && raw > 0 ? Math.floor(raw) : 3;
+}
+
+function cooldownMs(): number {
+  const raw = Number(process.env.UPSTREAM_CIRCUIT_COOLDOWN_MS ?? 30_000);
+  return Number.isFinite(raw) && raw > 0 ? Math.min(raw, 300_000) : 30_000;
+}
+
+function assertCircuitAllowsRequest(now = Date.now()): void {
+  if (circuit.openedAt === undefined) {
+    return;
+  }
+
+  if (now - circuit.openedAt < cooldownMs()) {
+    upstreamCircuitEventsTotal.inc({ event: "short_circuit" });
+    throw new HttpError(
+      503,
+      "Upstream news service temporarily unavailable",
+      "upstream_circuit_open"
+    );
+  }
+
+  circuit.openedAt = undefined;
+  upstreamCircuitEventsTotal.inc({ event: "half_open" });
+}
+
+function recordProviderSuccess(): void {
+  if (circuit.failures > 0 || circuit.openedAt !== undefined) {
+    upstreamCircuitEventsTotal.inc({ event: "closed" });
+  }
+  circuit.failures = 0;
+  circuit.openedAt = undefined;
+}
+
+function recordProviderFailure(): void {
+  circuit.failures += 1;
+  if (circuit.failures >= failureThreshold() && circuit.openedAt === undefined) {
+    circuit.openedAt = Date.now();
+    upstreamCircuitEventsTotal.inc({ event: "opened" });
+  }
+}
+
+/** @internal tests */
+export function resetGNewsCircuitForTests(): void {
+  circuit.failures = 0;
+  circuit.openedAt = undefined;
+}
+
 function normalizeArticles(data: unknown): Article[] {
   if (
     data === null ||
@@ -39,6 +100,7 @@ function toProviderParams(options: ArticleSearchOptions): Record<string, string
 
 export class GNewsProvider implements NewsProvider {
   async search(options: ArticleSearchOptions): Promise<Article[]> {
+    assertCircuitAllowsRequest();
     const stopUpstreamTimer = upstreamRequestDurationSeconds.startTimer();
 
     try {
@@ -51,12 +113,14 @@ export class GNewsProvider implements NewsProvider {
       const articles = normalizeArticles(response.data);
       upstreamRequestsTotal.inc({ outcome: "success" });
       stopUpstreamTimer({ outcome: "success" });
+      recordProviderSuccess();
       return articles;
     } catch (err) {
       const outcome =
         err instanceof HttpError && err.statusCode === 502 ? "invalid_payload" : "error";
       upstreamRequestsTotal.inc({ outcome });
       stopUpstreamTimer({ outcome });
+      recordProviderFailure();
       if (axios.isAxiosError(err)) {
         throw new HttpError(502, "Upstream news service unavailable", "upstream_unavailable");
       }

test/app.test.ts

@@ -20,11 +20,13 @@ import axios from "axios";
 import app from "../src/app";
 import { sampleArticles } from "./fixtures/articles";
 import { resetCacheStoreForTests, setCacheStoreForTests } from "../src/cache/store";
+import { resetGNewsCircuitForTests } from "../src/providers/gnewsProvider";
 
 describe("app", () => {
   beforeEach(() => {
     mockGet.mockReset();
     resetCacheStoreForTests();
+    resetGNewsCircuitForTests();
   });
 
   it("GET /health returns ok", async () => {
@@ -336,10 +338,30 @@ describe("app", () => {
     });
     expect(typeof res.body.error.requestId).toBe("string");
   });
+
+  it("opens the upstream circuit after repeated provider failures", async () => {
+    mockGet.mockRejectedValue(new axios.AxiosError("timeout"));
+    const q = `circuit-${Math.random().toString(36).slice(2)}`;
+
+    await request(app).get(`/api/v1/articles?query=${encodeURIComponent(q)}&count=1`);
+    await request(app).get(`/api/v1/articles?query=${encodeURIComponent(q)}&count=1`);
+    await request(app).get(`/api/v1/articles?query=${encodeURIComponent(q)}&count=1`);
+    const res = await request(app).get(`/api/v1/articles?query=${encodeURIComponent(q)}&count=1`);
+
+    expect(res.status).toBe(503);
+    expect(res.body.error).toMatchObject({
+      code: "upstream_circuit_open",
+      message: "Upstream news service temporarily unavailable",
+    });
+    expect(mockGet).toHaveBeenCalledTimes(3);
+  });
 });
 
 describe("CLIENT_API_KEYS gate", () => {
   beforeEach(() => {
+    mockGet.mockReset();
+    resetCacheStoreForTests();
+    resetGNewsCircuitForTests();
     vi.stubEnv("CLIENT_API_KEYS", "secret-one");
   });