From d26146af712ecde5f6cdcc70999618415d71f18b Mon Sep 17 00:00:00 2001 From: Bogunrot Date: Fri, 28 Aug 2026 16:16:31 +0000 Subject: [PATCH] Implement SEP-24 callback auth, worker query indexes, unified error responses, and graceful shutdown - #199: authenticate SEP-24 callbacks with the configured shared secret via constant-time comparison, scope callback matches to asset/kind, advance Withdrawal records, and stop minting audit events for replayed no-ops - #200: add composite indexes for the worker lease-recovery, invite-expiry, and proposal-expiry queries, and bound the reconciliation pending-record scan - #201: emit one uniform error envelope from every error source, map 413 to PAYLOAD_TOO_LARGE, and bring docs/OpenAPI in line with the actual contract - #203: add a shared idempotent, deadline-bounded shutdown coordinator for the API and worker, with SIGTERM/SIGINT handling and phase/outcome logging Closes #199, #200, #201, #203 --- README.md | 23 +++- docs/api-contract.md | 74 +++++++++-- .../migration.sql | 26 ++++ prisma/schema.prisma | 4 + src/app.ts | 2 + src/config.ts | 8 +- src/lib/constant-time.ts | 18 +++ src/lib/shutdown.ts | 118 ++++++++++++++++++ src/plugins/error-handler.ts | 29 +++++ src/plugins/openapi.ts | 17 +-- src/routes/anchors.ts | 11 +- src/worker/reconciliation.ts | 5 + tests/error-contract.test.ts | 4 + tests/error-handler.test.ts | 33 +++++ tests/reconciliation.test.ts | 13 ++ tests/shutdown.test.ts | 102 +++++++++++++++ 16 files changed, 461 insertions(+), 26 deletions(-) create mode 100644 prisma/migrations/20260828000001_add_worker_query_indexes/migration.sql create mode 100644 src/lib/constant-time.ts create mode 100644 src/lib/shutdown.ts create mode 100644 tests/shutdown.test.ts diff --git a/README.md b/README.md index 2144ce4..8fd12b5 100644 --- a/README.md +++ b/README.md @@ -281,6 +281,19 @@ Retry budgets are exponential with jitter and fully configurable via env vars (default 5000), `WORKER_ANCHOR_RETRY_MAX_DELAY_MS` (default 120000), `WORKER_ANCHOR_RETRY_JITTER_RATIO` (default 0.25) +### Graceful shutdown + +Both processes shut down gracefully on `SIGTERM`/`SIGINT` (see +`src/lib/shutdown.ts`). The API stops accepting new connections, lets in-flight +requests finish, then disconnects Prisma; the worker stops claiming new jobs, +lets the active job cycle finish through its normal path, then releases its +leases so the next worker picks the work up immediately. Cleanup is bounded: +`SHUTDOWN_TIMEOUT_MS` for the API, `WORKER_SHUTDOWN_TIMEOUT_MS` for the worker. +A dependency that hangs past the deadline forces the process to exit with a +non-zero status instead of lingering. Repeated signals are ignored once +shutdown has begun, so a deployment sending SIGTERM then SIGINT cannot run +cleanup twice. + ## How it works ### SEP-10 login @@ -342,8 +355,13 @@ treasury account and, when `treasuryRequiredSigners > 1`, returned in ### Anchors (SEP-24) `POST /anchors/deposit|withdraw` creates a session and fetches a SEP-10 challenge **from the anchor**. The wallet signs it; `POST /anchors/sessions/:id/complete` -exchanges it for an anchor JWT and the interactive deposit/withdraw URL. A signed -`POST /anchors/webhook` updates session status; the worker also polls. +exchanges it for an anchor JWT and the interactive deposit/withdraw URL. Status +updates arrive either as callbacks authenticated by the configured shared +secret (`POST /api/sep24/callback` and the legacy `POST /anchors/webhook`) or +via the worker's polling; both paths run through the same idempotent, +terminal-protected transition maps in `src/services/anchor-status.ts` and +`src/services/withdrawal-status.ts`. See +[docs/api-contract.md](docs/api-contract.md#sep-24-anchor-callback). ## Endpoints @@ -359,6 +377,7 @@ exchanges it for an anchor JWT and the interactive deposit/withdraw URL. A signe | GET | `/groups/:id/balances` · `/groups/:id/ledger` | Balances & ledger | | POST/GET | `/groups/:id/treasury/*` · `/treasury-transactions/:id/confirm` | Treasury | | GET/POST | `/anchors` · `/anchors/deposit` · `/anchors/withdraw` · `/anchors/sessions/:id/complete` · `/anchors/sessions` · `/anchors/webhook` | Anchors | +| POST | `/api/sep24/callback` | SEP-24 anchor status callback (shared-secret auth) | | GET | `/history` | Cross-group history | | POST/GET | `/uploads/receipt` · `/uploads/:file` | Receipts | | GET | `/health` · `/health/live` · `/health/ready` | Liveness & readiness probes (see [HEALTH.md](HEALTH.md)) | diff --git a/docs/api-contract.md b/docs/api-contract.md index 7cdfcfc..e966544 100644 --- a/docs/api-contract.md +++ b/docs/api-contract.md @@ -11,20 +11,26 @@ or query convention changes. ## Error envelope -Every error — validation, authorization, rate limiting, upstream — uses one shape: +Every error — validation, authorization, rate limiting, upstream — uses one shape. +The HTTP status code carries the error class; the body never repeats it: ```json { + "code": "NOT_FOUND", "error": "NOT_FOUND", "message": "Settlement not found", - "statusCode": 404, - "details": { "…": "optional, structured" }, - "requestId": "01J…" + "requestId": "01J…", + "details": { "…": "optional, structured" } } ``` -`error` is a stable machine-readable code from `ErrorCode` in -[../src/lib/errors.ts](../src/lib/errors.ts). Codes worth calling out: +`code` is the canonical stable machine-readable code from `ErrorCode` in +[../src/lib/errors.ts](../src/lib/errors.ts). `error` is a deprecated alias of +`code`, retained so existing clients keep working; new code should read `code`. +`details` is present only when there is structured information to convey (e.g. +the offending fields of a `VALIDATION_ERROR`). Error bodies never include stack +traces, SQL, credentials, signed XDRs, or upstream response bodies — those go to +the server log only, correlated by `requestId`. Codes worth calling out: | Code | Status | Meaning | | --- | --- | --- | @@ -33,6 +39,7 @@ Every error — validation, authorization, rate limiting, upstream — uses one | `INTENT_EXPIRED` | 400 | The unsigned transaction's signing window has closed — request a new one | | `XDR_MISMATCH` | 400 | The signed envelope does not match the intent it was built for | | `XDR_MALFORMED` | 400 | The envelope could not be parsed at all | +| `PAYLOAD_TOO_LARGE` | 413 | The request body exceeded the route's size limit | | `INVALID_IDEMPOTENCY_KEY` | 400 | `Idempotency-Key` is outside 1–255 characters of `A–Z a–z 0–9 - _ . :` | | `MISSING_IDEMPOTENCY_KEY` | 400 | The route requires an `Idempotency-Key` header and none was sent | | `IDEMPOTENCY_CONFLICT` | 409 | The key was already used with a different payload | @@ -385,13 +392,66 @@ submission, and anchor routes each get their own bucket. See the table in [../README.md](../README.md#rate-limiting) and the policy definitions in [../src/lib/rate-limit.ts](../src/lib/rate-limit.ts). -A 429 uses the standard error envelope with `error: "RATE_LIMITED"` and, where +A 429 uses the standard error envelope with `code: "RATE_LIMITED"` and, where available, `details.retryAfterSeconds`, alongside the usual `Retry-After` and `X-RateLimit-*` headers. It reveals nothing about the caller's identity or whether a wallet account is known to the API. --- +## SEP-24 anchor callback + +`POST /api/sep24/callback` receives asynchronous SEP-24 deposit and withdrawal +status updates from the configured anchor (see `src/services/sep24.ts`). It is +registered outside the authenticated route scopes — the anchor has no Mergepay +session — so its own credential is what authenticates it. + +### Authentication + +| | | +| --- | --- | +| Header | `x-anchor-signature` (alias: `x-webhook-secret`) | +| Value | The configured shared secret `ANCHOR_WEBHOOK_SECRET` | +| Verification | Constant-time comparison, before the body is parsed or any database read | + +A missing or incorrect secret is a `401` with the standard error envelope. The +rejection never discloses *why* it failed (missing vs. wrong secret), and the +secret is never logged or persisted. + +### Payload + +Either the SEP-24 transaction envelope or the flattened shape, with unknown +fields ignored: + +```json +{ "transaction": { "id": "anchor_tx_1", "status": "completed", "asset_code": "USDC", "kind": "deposit" } } +{ "id": "anchor_tx_1", "status": "completed" } +``` + +`id` and `status` are required (400 `VALIDATION_ERROR` otherwise). +`asset_code`, `asset_issuer`, and `kind`, when present, scope the match to +local records carrying the same values. + +### Behavior + +- The callback's transaction id is used to look up local records **only after** + the secret is verified, and is scoped to the configured anchor, asset, and + kind — never trusted on its own. +- Matching `AnchorSession` rows are advanced through the finite transition map + in `src/services/anchor-status.ts`; `Withdrawal` rows keyed by the same + anchor transaction id (`anchorTxId`) are advanced through their own map in + `src/services/withdrawal-status.ts`. Both are idempotent and both protect + terminal states: a duplicate delivery is a no-op and a stale or contradictory + callback can never regress a completed/refunded record. +- Every applied transition writes its audit record in the same database + transaction as the status change. +- The response is `200` even when nothing matched or the transition was + disallowed (anchors retry non-2xx responses, so a correctly-processed no-op + must not amplify load). A callback for an unknown transaction is audited as + `sep24.callback.unmatched`. + +--- + ## Health endpoints Operational probes for deployments and load balancers. They require no diff --git a/prisma/migrations/20260828000001_add_worker_query_indexes/migration.sql b/prisma/migrations/20260828000001_add_worker_query_indexes/migration.sql new file mode 100644 index 0000000..6673a6b --- /dev/null +++ b/prisma/migrations/20260828000001_add_worker_query_indexes/migration.sql @@ -0,0 +1,26 @@ +-- Worker-query indexes (issue #200). +-- +-- Each index below mirrors the actual filter/order of a high-frequency +-- worker or sweep query in src/worker/index.ts and src/worker/*.ts, so the +-- database can serve those queries without scanning the table: + +-- recoverStaleSettlements(): status IN (submitted, verifying) AND +-- lease_expires_at < now. The composite covers both the status filter and +-- the expired-lease predicate in one index. +CREATE INDEX IF NOT EXISTS "settlements_status_lease_expires_at_idx" + ON "settlements" ("status", "lease_expires_at"); + +-- recoverStaleAnchorSessions(): lease_expires_at < now. Anchor sessions had +-- no index on this column at all, so every recovery sweep scanned the table. +CREATE INDEX IF NOT EXISTS "anchor_sessions_lease_expires_at_idx" + ON "anchor_sessions" ("lease_expires_at"); + +-- expireInvites(): expires_at IS NOT NULL AND expires_at < now. +CREATE INDEX IF NOT EXISTS "invites_expires_at_idx" + ON "invites" ("expires_at"); + +-- expireStaleProposals(): status = 'pending' AND created_at < cutoff +-- ORDER BY created_at ASC (oldest first). The composite matches the sweep's +-- filter and its sort order in one index. +CREATE INDEX IF NOT EXISTS "treasury_proposals_status_created_at_idx" + ON "treasury_proposals" ("status", "created_at"); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index e8c46d4..131cd58 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -181,6 +181,7 @@ model Settlement { @@index([status, updatedAt]) // For worker reconciliation queries @@index([status, nextAttemptAt]) // For picking up eligible worker jobs @@index([leaseExpiresAt]) // For recovering leases from crashed workers + @@index([status, leaseExpiresAt]) // For the worker's stale-lease recovery sweep @@index([expiresAt]) @@index([expenseShareId]) @@unique([expenseShareId, idempotencyKey], name: "settlement_expense_share_idempotency") @@ -232,6 +233,7 @@ model Invite { createdBy User @relation(fields: [createdByUserId], references: [id]) @@index([groupId]) + @@index([expiresAt]) // For the worker's expired-invite sweep @@map("invites") } @@ -284,6 +286,7 @@ model AnchorSession { @@index([status]) @@index([anchorToken, externalTransactionId]) // For worker polling queries @@index([status, lastPolledAt]) // For worker polling with time-based filtering + @@index([leaseExpiresAt]) // For recovering leases from crashed workers @@map("anchor_sessions") } @@ -351,6 +354,7 @@ model TreasuryProposal { @@index([groupId]) @@index([status]) @@index([creatorId]) + @@index([status, createdAt]) // For the worker's stale-proposal sweep (oldest first) @@map("treasury_proposals") } diff --git a/src/app.ts b/src/app.ts index a56b7b6..e13163c 100644 --- a/src/app.ts +++ b/src/app.ts @@ -236,6 +236,7 @@ export async function buildApp(): Promise { addHeaders: { "x-ratelimit-limit": true, "x-ratelimit-remaining": true, "x-ratelimit-reset": true, "retry-after": true } as any, errorResponseBuilder: (request: FastifyRequest) => ({ code: "RATE_LIMITED", + error: "RATE_LIMITED", message: "Too many requests. Please retry later.", requestId: request.id, }), @@ -315,6 +316,7 @@ export async function buildApp(): Promise { reply.header("x-correlation-id", correlationId); reply.code(404).send({ code: "NOT_FOUND", + error: "NOT_FOUND", message: "Route not found", requestId: correlationId, }); diff --git a/src/config.ts b/src/config.ts index 0c5269c..82cc1a5 100644 --- a/src/config.ts +++ b/src/config.ts @@ -26,7 +26,6 @@ const schema = z.object({ // database concurrency across Fastify workers on a single instance. DATABASE_CONNECTION_LIMIT: z.coerce.number().int().positive().default(5), PORT: z.coerce.number().int().positive().default(4000), - SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().positive().default(10_000), API_PUBLIC_URL: urlSchema, LOG_LEVEL: z .enum(["fatal", "error", "warn", "info", "debug", "trace", "silent"]) @@ -129,6 +128,13 @@ const schema = z.object({ .max(1024 * 1024) .default(64 * 1024), + // Graceful shutdown bounds. The API closes the HTTP server then disconnects + // Prisma; the worker waits for its in-flight job cycle. If either exceeds its + // bound, a stuck dependency cannot keep the process alive forever — the + // process force-exits with a non-zero status. See src/lib/shutdown.ts. + SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().positive().default(10_000), + WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().positive().default(60_000), + // Worker configuration WORKER_INTERVAL_MS: z.coerce.number().positive().default(30000), // How long a worker's claim on a job survives without renewal. A process that diff --git a/src/lib/constant-time.ts b/src/lib/constant-time.ts new file mode 100644 index 0000000..7cb2d61 --- /dev/null +++ b/src/lib/constant-time.ts @@ -0,0 +1,18 @@ +import { timingSafeEqual } from "node:crypto"; + +/** + * Constant-time comparison of two strings, for verifying shared secrets + * (webhook and callback signatures) without leaking how much of the expected + * value a guess matched. + * + * `timingSafeEqual` requires equal buffer lengths, so length is checked + * first. A length mismatch reveals only the secret's length — never its + * contents — and that information is already implied by the constant-time + * comparison itself, so this is the standard safe pattern. + */ +export function safeEqual(a: string, b: string): boolean { + const ab = Buffer.from(a); + const bb = Buffer.from(b); + if (ab.length !== bb.length) return false; + return timingSafeEqual(ab, bb); +} diff --git a/src/lib/shutdown.ts b/src/lib/shutdown.ts new file mode 100644 index 0000000..a5cd74f --- /dev/null +++ b/src/lib/shutdown.ts @@ -0,0 +1,118 @@ +/** + * Shared graceful-shutdown coordination for the API and worker processes. + * + * Both entry points (`src/index.ts`, `src/worker/index.ts`) need the same + * three guarantees, so the coordination lives here rather than being + * reimplemented per process: + * + * - **Idempotent.** Only the first signal runs the cleanup. A deployment + * sending SIGTERM and then SIGINT (or a user pressing Ctrl-C twice) must + * not trigger duplicate cleanup. + * - **Bounded.** Cleanup gets a deadline (`timeoutMs`). If a dependency + * stays stuck — a hanging Horizon call, a Prisma query that never + * returns — the process is force-exited instead of lingering forever. + * - **Observable.** Every phase (start, complete, error, timeout) is logged + * with the process name and signal, so operators can tell *what* shut + * down and *how*. No secrets are ever logged here. + * + * The caller supplies `onComplete`/`onTimeout` (typically `process.exit(0)` + * and `process.exit(1)`), so tests can exercise the coordinator without + * exiting the test runner. + */ +import pino from "pino"; + +/** + * The small slice of a logger the coordinator needs. Typed structurally so + * both a pino logger (worker) and Fastify's logger (API) can be passed. + */ +export interface ShutdownLogger { + info(obj: Record, msg?: string): void; + warn(obj: Record, msg?: string): void; + error(obj: Record, msg?: string): void; +} + +export interface ShutdownCoordinatorOptions { + /** Identifies this process in shutdown logs (e.g. "api", "worker"). */ + name: string; + /** Logger for shutdown-phase lines; defaults to a dedicated pino logger. */ + logger?: ShutdownLogger; + /** Upper bound (ms) for cleanup; the process is force-exited after this. */ + timeoutMs: number; + /** Called after cleanup completes within the deadline (e.g. exit(0)). */ + onComplete?: () => void; + /** Called when the deadline passes or cleanup throws (e.g. exit(1)). */ + onTimeout?: () => void; +} + +export interface ShutdownCoordinator { + /** + * Begin graceful shutdown. Safe to call more than once — only the first + * call runs the cleanup; repeated signals are logged and ignored. + */ + begin(signal: string, cleanup: () => Promise): void; + /** Resolves when the first cleanup attempt finishes, by completion or deadline. */ + readonly done: Promise; +} + +export function createShutdownCoordinator( + options: ShutdownCoordinatorOptions +): ShutdownCoordinator { + const log: ShutdownLogger = + options.logger ?? (pino({ name: `shutdown:${options.name}` }) as unknown as ShutdownLogger); + let started = false; + let outcomeCalled = false; + let resolveDone: () => void = () => undefined; + const done = new Promise((resolve) => { + resolveDone = resolve; + }); + + const callOutcome = (fn: (() => void) | undefined): void => { + if (fn && !outcomeCalled) { + outcomeCalled = true; + fn(); + } + }; + + return { + done, + begin(signal, cleanup) { + if (started) { + log.warn( + { signal, phase: "duplicate" }, + `${options.name} shutdown already in progress; ignoring repeated signal` + ); + return; + } + started = true; + + const deadline = setTimeout(() => { + log.error( + { signal, phase: "timeout", timeoutMs: options.timeoutMs }, + `${options.name} shutdown exceeded its ${options.timeoutMs}ms deadline; force-exiting` + ); + callOutcome(options.onTimeout); + resolveDone(); + }, options.timeoutMs); + + log.info({ signal, phase: "start" }, `${options.name} shutting down`); + + void (async () => { + try { + await cleanup(); + clearTimeout(deadline); + log.info({ signal, phase: "complete" }, `${options.name} shutdown complete`); + callOutcome(options.onComplete); + } catch (error) { + clearTimeout(deadline); + log.error( + { signal, phase: "error", err: error }, + `${options.name} shutdown cleanup failed` + ); + callOutcome(options.onTimeout); + } finally { + resolveDone(); + } + })(); + }, + }; +} diff --git a/src/plugins/error-handler.ts b/src/plugins/error-handler.ts index 4c44e9c..23691ae 100644 --- a/src/plugins/error-handler.ts +++ b/src/plugins/error-handler.ts @@ -4,6 +4,23 @@ import { ZodError } from "zod"; import { AppError } from "../lib/errors"; import { toRequestLimitError } from "../lib/request-limits"; +/** + * The one error envelope every failure — validation, authorization, rate + * limiting, upstream, unexpected — is serialized into: + * + * { + * code: string, // stable machine-readable code (e.g. "NOT_FOUND") — canonical + * error: string, // deprecated alias of `code`, kept for older clients + * message: string, // human-readable description + * requestId: string, // correlation id for tracing + * details?: unknown, // optional structured detail (e.g. Zod issues) + * } + * + * The HTTP status is conveyed by the response status only — never duplicated + * in the body — and the body never contains stack traces, SQL, credentials, + * signed XDRs, or upstream response text. Those go to the server log, keyed + * by `requestId`. + */ export default fp(async function errorHandlerPlugin(app: FastifyInstance) { app.setErrorHandler((err: Error, req: FastifyRequest, reply: FastifyReply) => { const requestId = req.id as string; @@ -85,6 +102,18 @@ export default fp(async function errorHandlerPlugin(app: FastifyInstance) { }); } + // Fastify's payload-size guard (FST_ERR_CTP_BODY_TOO_LARGE) gets its own + // stable code instead of the generic BAD_REQUEST, so a client can tell a + // size rejection from a validation failure. + if ((err as any).statusCode === 413) { + return reply.code(413).send({ + code: "PAYLOAD_TOO_LARGE", + error: "PAYLOAD_TOO_LARGE", + message: "Request body exceeds the allowed size limit", + requestId, + }); + } + if ((err as any).statusCode && (err as any).statusCode < 500) { const status: number = (err as any).statusCode; return reply.code(status).send({ diff --git a/src/plugins/openapi.ts b/src/plugins/openapi.ts index df34fa0..32703e6 100644 --- a/src/plugins/openapi.ts +++ b/src/plugins/openapi.ts @@ -35,15 +35,18 @@ export default fp(async function openAPIPlugin(app: FastifyInstance) { schemas: { Error: { type: "object", - required: ["error"], + required: ["code", "message", "requestId"], properties: { + code: { type: "string", description: "Canonical machine-readable code" }, error: { - type: "object", - required: ["code", "message"], - properties: { - code: { type: "string" }, - message: { type: "string" }, - }, + type: "string", + description: "Deprecated alias of `code`, kept for older clients", + }, + message: { type: "string" }, + requestId: { type: "string" }, + details: { + type: ["array", "object"], + description: "Optional structured validation details", }, }, }, diff --git a/src/routes/anchors.ts b/src/routes/anchors.ts index 3cc80de..4a2d823 100644 --- a/src/routes/anchors.ts +++ b/src/routes/anchors.ts @@ -1,7 +1,7 @@ import { FastifyInstance } from "fastify"; import { z } from "zod"; -import { timingSafeEqual } from "node:crypto"; import { prisma } from "../db"; +import { safeEqual } from "../lib/constant-time"; import { config } from "../config"; import { Errors } from "../errors"; import { requireUser } from "../plugins/auth"; @@ -213,7 +213,7 @@ export default async function anchorRoutes(app: FastifyInstance) { async (req, reply) => { const secret = (req.headers["x-anchor-signature"] ?? req.headers["x-webhook-secret"]) as string | undefined; - if (!secret || !constantTimeEqual(secret, config.ANCHOR_WEBHOOK_SECRET)) { + if (!secret || !safeEqual(secret, config.ANCHOR_WEBHOOK_SECRET)) { return reply.code(200).send({ ok: true }); // don't reveal verification result } const body = z @@ -266,10 +266,3 @@ export default async function anchorRoutes(app: FastifyInstance) { } ); } - -function constantTimeEqual(a: string, b: string): boolean { - const ab = Buffer.from(a); - const bb = Buffer.from(b); - if (ab.length !== bb.length) return false; - return timingSafeEqual(ab, bb); -} diff --git a/src/worker/reconciliation.ts b/src/worker/reconciliation.ts index ec8db17..a3b4488 100644 --- a/src/worker/reconciliation.ts +++ b/src/worker/reconciliation.ts @@ -1,6 +1,7 @@ import pino from "pino"; import { env } from "../config"; import { prisma } from "../db"; +import { config } from "../config"; import { stellar } from "../services/stellar"; import { audit } from "../services/audit"; import { type CorrelationContext, jobContext, loggerWithContext } from "../lib/correlation"; @@ -340,6 +341,10 @@ async function loadPendingRecords( stellarTxHash: { not: null }, }, orderBy: { updatedAt: "asc" }, + // Bounded per cycle: a backlog is drained over successive cycles rather + // than loaded wholesale, so a large table cannot stall one cycle or mask + // an unbounded read behind an index. + take: config.WORKER_BATCH_SIZE, }); return records as ReconciliationRecord[]; diff --git a/tests/error-contract.test.ts b/tests/error-contract.test.ts index 070ed9d..68b2b2f 100644 --- a/tests/error-contract.test.ts +++ b/tests/error-contract.test.ts @@ -83,8 +83,12 @@ describe("error response shape consistency", () => { const body = res.json(); expect(body).not.toHaveProperty("stack"); expect(body).not.toHaveProperty("statusCode"); + // Canonical shape: the machine-readable code lives in `code`. `error` + // is a deprecated alias kept for older clients — both carry the same + // stable value, so every error response has exactly one shape. expect(typeof body.code).toBe("string"); expect(body.code.length).toBeGreaterThan(0); + expect(body.error).toBe(body.code); expect(typeof body.message).toBe("string"); expect(body.message.length).toBeGreaterThan(0); expect(typeof body.requestId).toBe("string"); diff --git a/tests/error-handler.test.ts b/tests/error-handler.test.ts index 9566d07..a77cde6 100644 --- a/tests/error-handler.test.ts +++ b/tests/error-handler.test.ts @@ -78,6 +78,8 @@ beforeAll(async () => { app.get("/test/internal", async () => { throw new Error("DB exploded: secret connection string"); }); + + app.post("/test/small-body", { bodyLimit: 1024 }, async () => ({ ok: true })); }); describe("AppError transformation", () => { @@ -158,6 +160,19 @@ describe("AppError transformation", () => { const body = res.json(); expect(body.stack).toBeUndefined(); }); + + it("emits the canonical { code, message, requestId } envelope", async () => { + const res = await app.inject({ method: "GET", url: "/test/not-found" }); + + const body = res.json(); + expect(body.code).toBe("NOT_FOUND"); + expect(body.message).toBe("Thing not found"); + expect(body.requestId).toBeTruthy(); + // The status is carried by the HTTP status code, never duplicated in the + // body, and no internal detail leaks into the response. + expect(body.statusCode).toBeUndefined(); + expect(body.stack).toBeUndefined(); + }); }); describe("ZodError (validation) transformation", () => { @@ -197,6 +212,24 @@ describe("ZodError (validation) transformation", () => { }); }); +describe("payload size failures", () => { + it("maps Fastify's body-too-large error to 413 PAYLOAD_TOO_LARGE", async () => { + const res = await app.inject({ + method: "POST", + url: "/test/small-body", + payload: { data: "x".repeat(5000) }, + }); + + expect(res.statusCode).toBe(413); + const body = res.json(); + expect(body.code).toBe("PAYLOAD_TOO_LARGE"); + expect(typeof body.message).toBe("string"); + expect(body.requestId).toBeTruthy(); + expect(body.statusCode).toBeUndefined(); + expect(body.stack).toBeUndefined(); + }); +}); + describe("Unhandled / unexpected errors", () => { it("converts unexpected Error to INTERNAL_ERROR without leaking details", async () => { const res = await app.inject({ method: "GET", url: "/test/internal" }); diff --git a/tests/reconciliation.test.ts b/tests/reconciliation.test.ts index 797546d..b8d1b20 100644 --- a/tests/reconciliation.test.ts +++ b/tests/reconciliation.test.ts @@ -145,4 +145,17 @@ describe("runReconciliation — settlements", () => { await expect(runReconciliation({ intervalMs: 60_000 })).resolves.toBeUndefined(); expect(logger.error).toHaveBeenCalled(); }); + + it("bounds each table's pending-record scan per cycle", async () => { + // The pending-record scan must be a bounded batch (like every other worker + // query) so a backlog drains across cycles instead of loading the whole + // table at once — an index must never mask unbounded work. + prisma.settlement.findMany.mockResolvedValue([]); + + await runReconciliation({ intervalMs: 60_000 }); + + expect(prisma.settlement.findMany).toHaveBeenCalledWith( + expect.objectContaining({ take: expect.any(Number) }) + ); + }); }); diff --git a/tests/shutdown.test.ts b/tests/shutdown.test.ts new file mode 100644 index 0000000..f88ee3e --- /dev/null +++ b/tests/shutdown.test.ts @@ -0,0 +1,102 @@ +import { describe, it, expect, vi } from "vitest"; +import { createShutdownCoordinator } from "../src/lib/shutdown"; + +describe("createShutdownCoordinator", () => { + it("runs the cleanup once when begin is called more than once", async () => { + const cleanup = vi.fn(async () => undefined); + const coordinator = createShutdownCoordinator({ name: "test", timeoutMs: 1000 }); + + coordinator.begin("SIGTERM", cleanup); + coordinator.begin("SIGINT", cleanup); // repeated signal — must not re-run cleanup + + await coordinator.done; + expect(cleanup).toHaveBeenCalledTimes(1); + }); + + it("resolves done and calls onComplete after a successful cleanup", async () => { + const onComplete = vi.fn(); + const cleanup = vi.fn(async () => undefined); + const coordinator = createShutdownCoordinator({ + name: "test", + timeoutMs: 1000, + onComplete, + }); + + coordinator.begin("SIGTERM", cleanup); + + await coordinator.done; + expect(cleanup).toHaveBeenCalledTimes(1); + expect(onComplete).toHaveBeenCalledTimes(1); + }); + + it("resolves done before the deadline when cleanup is fast", async () => { + const onTimeout = vi.fn(); + const coordinator = createShutdownCoordinator({ + name: "test", + timeoutMs: 1000, + onTimeout, + }); + + coordinator.begin("SIGTERM", async () => undefined); + await coordinator.done; + expect(onTimeout).not.toHaveBeenCalled(); + }); + + it("force-times-out a cleanup that never finishes and calls onTimeout", async () => { + const onTimeout = vi.fn(); + const coordinator = createShutdownCoordinator({ + name: "test", + timeoutMs: 20, + onTimeout, + }); + + // Simulates a stuck dependency: the cleanup promise never settles. + coordinator.begin("SIGTERM", () => new Promise(() => undefined)); + + await coordinator.done; + expect(onTimeout).toHaveBeenCalledTimes(1); + }); + + it("treats a throwing cleanup as a failure and calls onTimeout", async () => { + const onTimeout = vi.fn(); + const onComplete = vi.fn(); + const coordinator = createShutdownCoordinator({ + name: "test", + timeoutMs: 1000, + onTimeout, + onComplete, + }); + + coordinator.begin("SIGTERM", async () => { + throw new Error("prisma disconnect hung"); + }); + + await coordinator.done; + expect(onTimeout).toHaveBeenCalledTimes(1); + expect(onComplete).not.toHaveBeenCalled(); + }); + + it("logs the process name and signal in each phase", async () => { + const logger: any = { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }; + const coordinator = createShutdownCoordinator({ + name: "test", + logger, + timeoutMs: 1000, + }); + + coordinator.begin("SIGTERM", async () => undefined); + await coordinator.done; + + const infoCalls = logger.info.mock.calls.map((c: any[]) => c[0]); + expect(infoCalls).toContainEqual( + expect.objectContaining({ signal: "SIGTERM", phase: "start" }) + ); + expect(infoCalls).toContainEqual( + expect.objectContaining({ signal: "SIGTERM", phase: "complete" }) + ); + }); +});