diff --git a/app/api/indexer/status/route.ts b/app/api/indexer/status/route.ts index 83b89ef1..8189e6e1 100644 --- a/app/api/indexer/status/route.ts +++ b/app/api/indexer/status/route.ts @@ -13,11 +13,19 @@ export const dynamic = "force-dynamic"; export const runtime = "nodejs"; +import { isCircuitBreakerOpen } from "@/app/lib/admin-guard"; + interface IndexerStatus { ledgerCursor: number; lagMs: number; queueDepth: number; syncedAt: string; + /** + * True when an admin has tripped the indexer circuit breaker via + * POST /api/admin/circuit-breaker. Ingestion is halted; cursor and lag + * are frozen at their last values and will not advance until reset. + */ + breakerOpen: boolean; } function getIndexerStatus(): IndexerStatus { @@ -27,6 +35,7 @@ function getIndexerStatus(): IndexerStatus { lagMs: Math.floor(Math.random() * 3000), queueDepth: Math.floor(Math.random() * 50), syncedAt: new Date().toISOString(), + breakerOpen: isCircuitBreakerOpen("indexer"), }; } diff --git a/app/lib/webhook-circuit-breaker.test.ts b/app/lib/webhook-circuit-breaker.test.ts new file mode 100644 index 00000000..1298c482 --- /dev/null +++ b/app/lib/webhook-circuit-breaker.test.ts @@ -0,0 +1,184 @@ +/** @jest-environment node */ +/** + * Enforcement tests for the GLOBAL admin circuit breaker. + * + * These cover the wiring between `POST /api/admin/circuit-breaker` (control + * plane) and the webhook dispatch path (data plane). The admin breaker is + * distinct from the per-endpoint failure breaker in WebhookDeliveryClient — + * see the webhook-delivery module docblock. The critical behavioural + * difference asserted here: the admin breaker DEFERS deliveries, it does not + * DLQ them. + */ +import { WebhookDeliveryClient, WebhookEndpoint, WebhookEvent } from "./webhook-delivery"; +import { WebhookDeliveryWorker } from "./webhook-delivery-worker"; +import { webhookDeliveryStore } from "./webhook-delivery-store"; +import { webhookOutboxStore } from "./webhook-outbox"; +import { + setCircuitBreaker, + isCircuitBreakerOpen, + _resetAdminStateForTesting, +} from "./admin-guard"; + +const ADMIN_ADDRESS = "GADMIN_TEST_ADDRESS_12345"; + +/** Admin-authenticated request used to drive the breaker in tests. */ +function adminRequest(): Request { + return new Request("http://localhost/api/admin/circuit-breaker", { + method: "POST", + headers: { "Actor-Wallet-Address": ADMIN_ADDRESS }, + }); +} + +/** Trip or reset the global breaker through the real admin-guard entry point. */ +function toggleBreaker(target: "indexer" | "webhook", open: boolean): void { + const result = setCircuitBreaker(adminRequest(), target, open); + if (!("circuitBreakers" in result)) { + throw new Error("Failed to toggle breaker — admin auth rejected in test setup"); + } +} + +const endpoint: WebhookEndpoint = { + id: "ep-breaker-test", + url: "https://example.test/hook", + maxRetries: 3, +}; + +const event: WebhookEvent = { + id: "evt-1", + eventType: "stream.created", + streamId: "stream-1", + data: { amount: "100" }, + timestamp: new Date().toISOString(), +}; + +const fetchMock = jest.fn(); + +beforeEach(() => { + _resetAdminStateForTesting(ADMIN_ADDRESS); + // Both stores are module-level singletons — clear them so DLQ and outbox + // assertions cannot be polluted by a previous test. + webhookDeliveryStore.clear(); + webhookOutboxStore.clear(); + fetchMock.mockReset(); + global.fetch = fetchMock as unknown as typeof fetch; +}); + +describe("admin breaker → attemptDelivery", () => { + it("does not contact the endpoint while the webhook breaker is open", async () => { + toggleBreaker("webhook", true); + const client = new WebhookDeliveryClient(); + + const result = await client.attemptDelivery(endpoint, event, "d-1", 1); + + expect(result.success).toBe(false); + expect(result.deferred).toBe(true); + expect(result.shouldRetry).toBe(false); + expect(result.error).toMatch(/admin circuit breaker/i); + // The decisive assertion: no outbound HTTP happened at all. + expect(fetchMock).not.toHaveBeenCalled(); + }); + + it("delivers normally once the breaker is reset", async () => { + toggleBreaker("webhook", true); + toggleBreaker("webhook", false); + fetchMock.mockResolvedValue({ status: 200, statusText: "OK" }); + const client = new WebhookDeliveryClient(); + + const result = await client.attemptDelivery(endpoint, event, "d-2", 1); + + expect(result.success).toBe(true); + expect(result.deferred).toBeUndefined(); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it("is unaffected by the indexer breaker (targets are independent)", async () => { + toggleBreaker("indexer", true); + fetchMock.mockResolvedValue({ status: 200, statusText: "OK" }); + const client = new WebhookDeliveryClient(); + + const result = await client.attemptDelivery(endpoint, event, "d-3", 1); + + expect(result.success).toBe(true); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it("does not record an endpoint failure when deferring", async () => { + toggleBreaker("webhook", true); + const client = new WebhookDeliveryClient(); + + // Defer many times — well past the per-endpoint threshold of 5. + for (let i = 0; i < 10; i++) { + await client.attemptDelivery(endpoint, event, `d-loop-${i}`, 1); + } + + // The endpoint is blameless; its own breaker must stay closed so that + // traffic resumes immediately when the admin resets the global breaker. + expect(client.isCircuitOpen(endpoint.id)).toBe(false); + }); +}); + +describe("admin breaker → processDelivery", () => { + it("defers instead of moving the delivery to the DLQ", async () => { + toggleBreaker("webhook", true); + const worker = new WebhookDeliveryWorker({}, () => Promise.resolve()); + + const result = await worker.processDelivery(endpoint, event, "d-4"); + + expect(result.success).toBe(false); + expect(result.deferred).toBe(true); + expect(result.dlqed).toBeUndefined(); + + // Regression guard: an operator-initiated pause must never discard events. + const dlqIds = webhookDeliveryStore.getAllDLQEntries().map((e) => e.deliveryId); + expect(dlqIds).not.toContain("d-4"); + }); + + it("still DLQs on a genuine non-retryable endpoint failure", async () => { + // Breaker closed — confirms the deferral path did not swallow real failures. + fetchMock.mockResolvedValue({ status: 400, statusText: "Bad Request" }); + const worker = new WebhookDeliveryWorker({}, () => Promise.resolve()); + + const result = await worker.processDelivery(endpoint, event, "d-5"); + + expect(result.success).toBe(false); + expect(result.dlqed).toBe(true); + expect(result.deferred).toBeUndefined(); + }); +}); + +describe("admin breaker → drainOutbox", () => { + it("skips the drain entirely while the breaker is open", async () => { + webhookOutboxStore.addToOutbox(endpoint, event); + toggleBreaker("webhook", true); + const worker = new WebhookDeliveryWorker({}, () => Promise.resolve()); + + await worker.drainOutbox(); + + expect(fetchMock).not.toHaveBeenCalled(); + }); + + it("leaves deferred outbox entries retryable rather than failed", async () => { + const entry = webhookOutboxStore.addToOutbox(endpoint, event); + toggleBreaker("webhook", true); + const worker = new WebhookDeliveryWorker({}, () => Promise.resolve()); + + await worker.processOutboxEntry(entry); + + // Back to pending, not "failed" or "dlq" — it must be picked up again + // by the next drain once the operator resets the breaker. + const entries = webhookOutboxStore.getPendingOutboxEntries(100); + expect(entries.map((e) => e.id)).toContain(entry.id); + }); +}); + +describe("isCircuitBreakerOpen", () => { + it("reports per-target state independently", () => { + expect(isCircuitBreakerOpen("indexer")).toBe(false); + expect(isCircuitBreakerOpen("webhook")).toBe(false); + + toggleBreaker("indexer", true); + + expect(isCircuitBreakerOpen("indexer")).toBe(true); + expect(isCircuitBreakerOpen("webhook")).toBe(false); + }); +}); diff --git a/app/lib/webhook-delivery-worker.ts b/app/lib/webhook-delivery-worker.ts index 114149e1..eb5603bd 100644 --- a/app/lib/webhook-delivery-worker.ts +++ b/app/lib/webhook-delivery-worker.ts @@ -11,6 +11,7 @@ import { } from "./webhook-delivery"; import { webhookDeliveryStore } from "./webhook-delivery-store"; import { webhookOutboxStore, WebhookOutboxEntry } from "./webhook-outbox"; +import { isCircuitBreakerOpen } from "./admin-guard"; export class WebhookDeliveryWorker { private client: WebhookDeliveryClient; @@ -49,7 +50,14 @@ export class WebhookDeliveryWorker { event: WebhookEvent, deliveryId: string, reissuedFrom?: string, - ): Promise<{ success: boolean; deliveryId: string; attempts: number; dlqed?: boolean }> { + ): Promise<{ + success: boolean; + deliveryId: string; + attempts: number; + dlqed?: boolean; + /** Held by the global admin breaker — retryable once the breaker resets. */ + deferred?: boolean; + }> { const ctx = getCorrelationContext(); withWebhookContext(deliveryId); const maxAttempts = this.maxRetries; @@ -69,6 +77,19 @@ export class WebhookDeliveryWorker { webhookDeliveryStore.getDelivery(deliveryId)?.attempts ?? [], ); + // ── Global admin breaker → hold, do NOT DLQ ────────────────────────── + // The operator paused dispatch deliberately. Leave the delivery pending + // so it can be retried once the breaker is reset; DLQing here would + // discard every in-flight event for the duration of an incident. + if (result.deferred) { + logger.warn("Webhook delivery deferred — admin circuit breaker open", { + delivery_id: deliveryId, endpoint_id: endpoint.id, + event_id: event.id, breaker: "admin.webhook", + correlation_id: ctx?.correlation_id, + }); + return { success: false, deliveryId, attempts: 0, deferred: true }; + } + if (result.error?.includes("Circuit breaker open")) { webhookDeliveryStore.moveToDLQ(deliveryId, result.error); return { success: false, deliveryId, attempts: 0, dlqed: true }; @@ -168,6 +189,13 @@ export class WebhookDeliveryWorker { status: "delivered", attempts: result.attempts, }); + } else if (result.deferred) { + // Admin breaker is open — return the entry to the pending pool untouched + // so the next drain (after the breaker resets) picks it up again. + webhookOutboxStore.updateOutboxEntry(entry.id, { + status: "pending", + attempts: result.attempts, + }); } else if (result.dlqed) { webhookOutboxStore.updateOutboxEntry(entry.id, { status: "dlq", @@ -187,6 +215,17 @@ export class WebhookDeliveryWorker { * Drain the outbox: process all pending entries */ async drainOutbox(limit: number = 100): Promise { + // Skip the whole drain while the admin breaker is open. Without this the + // worker would churn every pending entry through processDelivery once per + // tick, only to defer each one back to pending. + if (isCircuitBreakerOpen("webhook")) { + logger.warn("Outbox drain skipped — admin circuit breaker open", { + breaker: "admin.webhook", + correlation_id: getCorrelationContext()?.correlation_id, + }); + return; + } + const entries = webhookOutboxStore.getPendingOutboxEntries(limit); logger.info("Draining webhook outbox", { diff --git a/app/lib/webhook-delivery.ts b/app/lib/webhook-delivery.ts index c03925cf..d299b6d3 100644 --- a/app/lib/webhook-delivery.ts +++ b/app/lib/webhook-delivery.ts @@ -1,6 +1,7 @@ import crypto from 'crypto'; import { logger, withWebhookContext, getCorrelationContext } from './logger'; import { getActiveSigningSecrets } from '@/app/lib/webhook-secrets'; +import { isCircuitBreakerOpen } from '@/app/lib/admin-guard'; /** * Idempotent webhook delivery with exponential backoff + full jitter and DLQ support. @@ -20,6 +21,21 @@ import { getActiveSigningSecrets } from '@/app/lib/webhook-secrets'; * - 4xx (except 408, 429) → non-retryable client error, go straight to DLQ * - 408, 429, 5xx → retryable server/transient error * - network timeout → retryable + * + * ## Two distinct circuit breakers + * This module is subject to two independent breakers. They are NOT the same + * mechanism and must not be conflated: + * + * 1. **Global admin breaker** — `isCircuitBreakerOpen("webhook")` from + * admin-guard. Manually toggled subsystem-wide via + * `POST /api/admin/circuit-breaker`. Checked FIRST. When open, no endpoint + * is contacted at all. Deliveries are held (`deferred: true`), NOT DLQed — + * an operator pausing dispatch during an incident must not lose events. + * + * 2. **Per-endpoint failure breaker** — `WebhookDeliveryClient.isCircuitOpen`. + * Trips itself after N consecutive failures to one endpoint and auto-resets + * after 5 min. When open, that endpoint's delivery goes to DLQ, because the + * endpoint itself is demonstrably broken. */ // ── Types ───────────────────────────────────────────────────────────────────── @@ -390,10 +406,32 @@ export class WebhookDeliveryClient { error?: string; shouldRetry: boolean; nextRetryAt?: string; + /** + * Set when the global admin breaker held this delivery. The caller must + * leave the delivery pending rather than DLQing it — see the module + * docblock on the two breakers. + */ + deferred?: boolean; }> { const context = getCorrelationContext(); withWebhookContext(deliveryId); + // ── Global admin breaker (checked before anything endpoint-specific) ───── + // Operator-controlled kill switch. Hold the delivery; do not DLQ it and do + // not record a failure against the endpoint — the endpoint is not at fault. + if (isCircuitBreakerOpen('webhook')) { + const error = 'Webhook dispatch halted: admin circuit breaker is open'; + logger.warn('Webhook delivery deferred by admin circuit breaker', { + delivery_id: deliveryId, + endpoint_id: endpoint.id, + endpoint_url: endpoint.url, + event_id: event.id, + breaker: 'admin.webhook', + correlation_id: context?.correlation_id, + }); + return { success: false, error, shouldRetry: false, deferred: true }; + } + if (this.isCircuitOpen(endpoint.id, endpoint.circuitBreakerThreshold)) { const error = 'Circuit breaker open: endpoint experiencing repeated failures'; logger.warn('Webhook delivery blocked by circuit breaker', { diff --git a/docs/admin-circuit-breaker.md b/docs/admin-circuit-breaker.md new file mode 100644 index 00000000..9cb94b34 --- /dev/null +++ b/docs/admin-circuit-breaker.md @@ -0,0 +1,88 @@ +# Admin Circuit Breaker + +## Overview + +The admin circuit breaker is an **operator-controlled kill switch** for the indexer and webhook subsystems. It is toggled manually via `POST /api/admin/circuit-breaker` and is intended for incident response — when a downstream is melting down, an operator trips the breaker to stop dispatch without deploying code. + +## Two different things are called "circuit breaker" + +This is the most important distinction in this document. They are independent mechanisms with opposite failure semantics. + +| | Admin breaker | Per-endpoint breaker | +|---|---|---| +| Defined in | [`admin-guard.ts`](../app/lib/admin-guard.ts) | [`webhook-delivery.ts`](../app/lib/webhook-delivery.ts) | +| Trips | Manually, by an admin | Automatically, after 5 consecutive failures | +| Resets | Manually, by an admin | Automatically, after 5 minutes | +| Scope | Whole subsystem | One endpoint | +| Effect on events | **Held (deferred)** | **Sent to DLQ** | + +The reason the effects differ: when a *single endpoint* fails repeatedly, that endpoint is broken and its events belong in the DLQ for later inspection. When an *operator* pauses the whole subsystem, nothing is broken on the customer side — DLQing every in-flight event would turn a deliberate pause into mass event loss. So the admin breaker defers. + +The admin breaker is checked **first**, before any endpoint-specific logic, and deferral does **not** count as a failure against the endpoint. This matters: without that, ten deferrals during a pause would trip the per-endpoint breaker too, and traffic would not resume cleanly when the admin reset the global one. + +## API + +### Trip a breaker + +```http +POST /api/admin/circuit-breaker +Actor-Wallet-Address: G...ADMIN +Content-Type: application/json + +{ "target": "webhook", "open": true } +``` + +### Read current state + +```http +GET /api/admin/circuit-breaker +Actor-Wallet-Address: G...ADMIN +``` + +```json +{ + "data": { + "indexer": { "open": false, "updatedAt": null, "updatedBy": null }, + "webhook": { "open": true, "updatedAt": "2026-07-30T12:00:00.000Z", "updatedBy": "G...ADMIN" } + } +} +``` + +`target` must be `indexer` or `webhook`; `open` must be a boolean. Both handlers are gated by admin auth and every toggle is written to the privileged audit log. + +### Error codes + +| Code | HTTP | Meaning | +|---|---|---| +| `INVALID_REQUEST` | 400 | Body is not valid JSON | +| `VALIDATION_ERROR` | 422 | Missing/invalid `target` or `open` | +| `Unauthorized` | 403 | Caller is not the admin | + +## What each target actually does + +### `webhook` + +Enforced in the dispatch path. While open: + +- `attemptDelivery` returns `{ deferred: true }` without making any outbound HTTP request. +- `processDelivery` returns `{ deferred: true }` and does **not** move the delivery to the DLQ. +- `processOutboxEntry` returns the entry to `pending` status, so it is retried later. +- `drainOutbox` skips the drain entirely, rather than churning every pending entry through a deferral each tick. + +Events accumulate in the outbox and flow normally once the breaker is reset. + +### `indexer` + +Surfaced on `GET /api/indexer/status` as the `breakerOpen` field, so operators and dashboards can see that ingestion is paused. + +> **Note:** this repo does not currently contain a real indexer ingestion loop — [`app/api/indexer/status/route.ts`](../app/api/indexer/status/route.ts) returns mocked cursor/lag values. When a real ingestion worker lands, it should call `isCircuitBreakerOpen("indexer")` at the top of its poll loop, following the webhook pattern above. + +## Operational notes + +**State is in-memory.** Breaker state lives in the `_state` singleton in `admin-guard.ts`. It resets to closed on process restart and is **not shared across instances** — in a multi-instance deployment, tripping the breaker affects only the instance that served the request. Moving this to shared storage (Redis or Postgres) is required before relying on it in a horizontally-scaled production environment. + +**Reset is not automatic.** Unlike the per-endpoint breaker, nothing re-closes the admin breaker on a timer. It stays open until an admin explicitly sets `open: false`. Pair tripping it with an alert so a pause is never forgotten. + +## Tests + +Enforcement is covered by [`app/lib/webhook-circuit-breaker.test.ts`](../app/lib/webhook-circuit-breaker.test.ts) — no outbound HTTP while open, no DLQ on defer, no endpoint-failure recording on defer, outbox entries stay retryable, drain skipping, target independence, and normal delivery after reset. Endpoint auth/validation is covered by [`app/api/admin/circuit-breaker/route.test.ts`](../app/api/admin/circuit-breaker/route.test.ts). diff --git a/openapi.json b/openapi.json index 2d639ac9..16c5b8af 100644 --- a/openapi.json +++ b/openapi.json @@ -26,6 +26,46 @@ } }, "schemas": { + "CircuitBreakerState": { + "type": "object", + "required": [ + "open", + "updatedAt", + "updatedBy" + ], + "properties": { + "open": { + "type": "boolean", + "description": "True when the breaker is tripped and dispatch is halted." + }, + "updatedAt": { + "type": "string", + "format": "date-time", + "nullable": true, + "description": "ISO-8601 timestamp of the last toggle; null if never toggled." + }, + "updatedBy": { + "type": "string", + "nullable": true, + "description": "Admin address that last toggled this breaker; null if never toggled." + } + } + }, + "CircuitBreakers": { + "type": "object", + "required": [ + "indexer", + "webhook" + ], + "properties": { + "indexer": { + "$ref": "#/components/schemas/CircuitBreakerState" + }, + "webhook": { + "$ref": "#/components/schemas/CircuitBreakerState" + } + } + }, "ErrorEnvelope": { "type": "object", "required": [ @@ -3772,6 +3812,112 @@ } } }, + "/api/admin/circuit-breaker": { + "get": { + "operationId": "adminGetCircuitBreakers", + "summary": "Read circuit breaker state", + "description": "Returns the current state of every named circuit breaker.\n\n**Auth:** Admin wallet address via `Actor-Wallet-Address` header or JWT.", + "tags": [ + "Admin" + ], + "responses": { + "200": { + "description": "Current breaker state.", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "data" + ], + "properties": { + "data": { + "$ref": "#/components/schemas/CircuitBreakers" + } + } + } + } + } + }, + "403": { + "$ref": "#/components/responses/403" + }, + "500": { + "$ref": "#/components/responses/500" + } + } + }, + "post": { + "operationId": "adminSetCircuitBreaker", + "summary": "Toggle an indexer or webhook circuit breaker", + "description": "Trips or resets a named subsystem circuit breaker. Gated by admin auth — only the configured admin address may call this.\n\nWhen `open` is true the named subsystem halts dispatch. For the `webhook` target this means no outbound delivery is attempted and the outbox drain is skipped; pending events are **held, not discarded**, and resume when the breaker is reset. For the `indexer` target the state is surfaced on `GET /api/indexer/status` as `breakerOpen`.\n\nThis global operator-controlled breaker is distinct from the automatic per-endpoint failure breaker, which trips itself after repeated delivery failures to a single endpoint and sends that endpoint's deliveries to the DLQ.\n\n**Auth:** Admin wallet address via `Actor-Wallet-Address` header or JWT.\n\n**Idempotency:** Naturally idempotent — setting the same state twice is a no-op beyond the updated timestamp.", + "tags": [ + "Admin" + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "target", + "open" + ], + "properties": { + "target": { + "type": "string", + "enum": [ + "indexer", + "webhook" + ], + "description": "Which subsystem breaker to toggle.", + "example": "webhook" + }, + "open": { + "type": "boolean", + "description": "true trips the breaker (halt dispatch); false resets it (resume).", + "example": true + } + } + } + } + } + }, + "responses": { + "200": { + "description": "Breaker updated.", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "data" + ], + "properties": { + "data": { + "$ref": "#/components/schemas/CircuitBreakers" + } + } + } + } + } + }, + "400": { + "$ref": "#/components/responses/400" + }, + "403": { + "$ref": "#/components/responses/403" + }, + "422": { + "$ref": "#/components/responses/422" + }, + "500": { + "$ref": "#/components/responses/500" + } + } + } + }, "/api/debug/kms-sign": { "post": { "operationId": "debugKmsSign",