diff --git a/backend/SEP0001_GENERATOR_GUIDE.md b/backend/SEP0001_GENERATOR_GUIDE.md index 12fb370..6393e13 100644 --- a/backend/SEP0001_GENERATOR_GUIDE.md +++ b/backend/SEP0001_GENERATOR_GUIDE.md @@ -316,6 +316,33 @@ console.log('Accounts:', toml.ACCOUNTS); - Wait 1 hour for changes to propagate - Clear CDN cache if using a CDN +## Retry with Exponential Backoff (#1459) + +The only I/O behind `GET /.well-known/stellar.toml` is the merchant lookup. +It is retried when it fails transiently (`src/lib/sep0001-retry.js`): + +- Retried: network errors, 408/429/5xx, connection-class Postgres errors. +- Not retried: unknown or invalid merchant ids, PostgREST client errors. +- Delay: full jitter, `random(0, min(max, base * 2^n))`. + +| Variable | Default | Clamp | +|---|---|---| +| `SEP0001_RETRY_MAX_ATTEMPTS` | 3 | 1-5 | +| `SEP0001_RETRY_BASE_DELAY_MS` | 100 | 0-5000 | +| `SEP0001_RETRY_MAX_DELAY_MS` | 1000 | base-5000 | + +If the lookup is still failing transiently after the last attempt, the +endpoint answers **503** with `Retry-After: 5` instead of 500, so wallets +retry later instead of treating the domain as having no stellar.toml. +Deterministic lookup errors still return 500. Every error response carries +`Cache-Control: no-store`, so a CDN never caches a failure for the hour that +successful responses are cached. + +Retry outcomes are exported as +`sep0001_merchant_lookup_retries_total{result="scheduled|recovered|exhausted"}` +on `GET /metrics`. Logs carry the error code, attempt count and message, never +the merchant record. + ## Security Considerations 1. **Public Information**: stellar.toml is publicly accessible diff --git a/backend/docs/PAYMENT_PROCESSOR.md b/backend/docs/PAYMENT_PROCESSOR.md index 876d360..6f848f2 100644 --- a/backend/docs/PAYMENT_PROCESSOR.md +++ b/backend/docs/PAYMENT_PROCESSOR.md @@ -4,7 +4,9 @@ Architecture, granular metrics, and testing guide for the Payment Processor module (`src/routes/payments.js` + `src/services/paymentService.js`). Covers issues **#1086** (end-to-end testing), **#1087** (legacy refactor), -**#1088** (granular metrics) and **#1089** (load testing). +**#1088** (granular metrics), **#1089** (load testing), and Payment Processor +Telemetry: **#1477** (payload sanitization and strict validation), **#1478** +(alert metrics and health) and **#1479** (retry with exponential backoff). --- diff --git a/backend/docs/alerts/payment-processor.rules.yml b/backend/docs/alerts/payment-processor.rules.yml new file mode 100644 index 0000000..207e208 --- /dev/null +++ b/backend/docs/alerts/payment-processor.rules.yml @@ -0,0 +1,82 @@ +# Prometheus alerting rules for Payment Processor Telemetry (issues #1477-#1479). +# +# Load with: rule_files: ["docs/alerts/payment-processor.rules.yml"] +# Validate: promtool check rules docs/alerts/payment-processor.rules.yml +# +# Metric reference: docs/PAYMENT_PROCESSOR.md (section 6) + +groups: + - name: payment-processor + rules: + - alert: PaymentProcessorUnhealthy + expr: max(payment_processor_health_state) >= 2 + for: 5m + labels: + severity: critical + annotations: + summary: Payment processor reports unhealthy + description: > + Session persistence, verification or refund errors are above the + unhealthy ratio over the rolling window. See + GET /health/payment-processor for the counts and reasons. + + - alert: PaymentProcessorDegraded + expr: max(payment_processor_health_state) == 1 + for: 15m + labels: + severity: warning + annotations: + summary: Payment processor degraded for 15 minutes + description: > + An error ratio is above the degraded threshold, telemetry payloads + are being rejected, or read retries are being exhausted. + + - alert: PaymentProcessorSessionPersistenceFailures + expr: | + sum(rate(payment_processor_sessions_total{outcome="persistence_failed"}[10m])) + / + clamp_min(sum(rate(payment_processor_sessions_total{outcome=~"created|persistence_failed"}[10m])), 1e-9) + > 0.05 + for: 10m + labels: + severity: critical + annotations: + summary: More than 5% of payment sessions fail to persist + description: > + Session inserts are failing after their own retries. Check + database health and the "Payment session operation failed" logs. + + - alert: PaymentProcessorReadRetriesExhausted + expr: sum(increase(payment_processor_retries_total{result="exhausted"}[10m])) > 0 + labels: + severity: warning + annotations: + summary: Payment processor reads failed after all retries + description: > + Status or verification lookups kept failing with transient errors + until PAYMENT_PROCESSOR_RETRY_MAX_ATTEMPTS ran out. Customers saw + 5xx on payment-status or verify-payment. + + - alert: PaymentProcessorTelemetryRejections + expr: sum(increase(payment_processor_telemetry_rejected_total[15m])) > 0 + for: 15m + labels: + severity: warning + annotations: + summary: Payment processor telemetry payloads are being rejected + description: > + A code path is recording an outcome, stage or duration outside the + allow-list. The business series are not affected (rejected + payloads are dropped) but the gap hides real traffic. Check the + "Payment processor telemetry payload rejected" log for the metric. + + - alert: PaymentProcessorAssetLabelOverflow + expr: sum(increase(payment_processor_sessions_total{asset=~"other|invalid"}[1h])) > 100 + labels: + severity: info + annotations: + summary: Many sessions recorded under asset "other" or "invalid" + description: > + Either the distinct asset label cap + (PAYMENT_PROCESSOR_TELEMETRY_MAX_ASSET_LABELS) is too low for real + traffic, or clients are sending malformed asset codes. diff --git a/backend/src/lib/backoff-retry.js b/backend/src/lib/backoff-retry.js new file mode 100644 index 0000000..54e639c --- /dev/null +++ b/backend/src/lib/backoff-retry.js @@ -0,0 +1,189 @@ +/** + * Shared full-jitter exponential backoff for idempotent reads (issues #1459, + * #1479). + * + * Used by the SEP-0001 Stellar Info Generator (merchant lookup) and the + * Payment Processor (status / verification lookups). Both retry ONLY + * transient failures, classified by the same rules as the payment session + * writer (`isRetryableSessionError`): network errors, 408/429/5xx and + * connection-class Postgres errors. Validation, auth, not-found and + * constraint errors are never retried, so a retry can never turn a rejected + * request into an accepted one. + * + * delay(n) = random(0, min(maxDelayMs, baseDelayMs * 2^n)) + * + * Attempts and delays are clamped to per-caller ceilings so env/config can + * never make an HTTP request hang. + */ + +import { logger as defaultLogger } from "./logger.js"; +import { computeBackoffDelay, isRetryableSessionError } from "./payment-session-retry.js"; + +function clampInt(value, fallback, min, max) { + if (value === undefined || value === null || value === "") return fallback; + const n = Number(value); + if (!Number.isFinite(n)) return fallback; + return Math.min(Math.max(Math.trunc(n), min), max); +} + +/** + * Resolve retry options from explicit options, then `_MAX_ATTEMPTS` + * / `_BASE_DELAY_MS` / `_MAX_DELAY_MS`, then defaults, clamped to ceilings. + * + * @param {object} options + * @param {Record} env + * @param {{envPrefix:string, defaults:{maxAttempts:number, baseDelayMs:number, maxDelayMs:number}, maxAttemptsCeiling:number, maxDelayCeilingMs:number}} spec + */ +export function resolveBackoffOptions(options = {}, env = process.env, spec) { + const { envPrefix, defaults, maxAttemptsCeiling, maxDelayCeilingMs } = spec; + const maxAttempts = clampInt( + options.maxAttempts ?? env[`${envPrefix}_MAX_ATTEMPTS`], + defaults.maxAttempts, + 1, + maxAttemptsCeiling, + ); + const baseDelayMs = clampInt( + options.baseDelayMs ?? env[`${envPrefix}_BASE_DELAY_MS`], + defaults.baseDelayMs, + 0, + maxDelayCeilingMs, + ); + const maxDelayMs = clampInt( + options.maxDelayMs ?? env[`${envPrefix}_MAX_DELAY_MS`], + defaults.maxDelayMs, + baseDelayMs, + maxDelayCeilingMs, + ); + return { maxAttempts, baseDelayMs, maxDelayMs }; +} + +const defaultSleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); + +/** + * Run `operation`, retrying transient failures with full-jitter backoff. + * The final error is rethrown annotated with `retryAttempts`. + * + * @template T + * @param {(ctx:{attempt:number}) => Promise} operation + * @param {object} options + * @param {{maxAttempts:number, baseDelayMs:number, maxDelayMs:number}} options.resolved + * @param {string} [options.label] Fixed, server-side operation name + * @param {(err:unknown) => boolean} [options.shouldRetry] + * @param {(ms:number) => Promise} [options.sleep] + * @param {() => number} [options.random] + * @param {(info:{attempt:number, delayMs:number, error:unknown}) => void} [options.onRetry] + * @param {(info:{attempts:number}) => void} [options.onRecovered] + * @param {(info:{attempts:number, error:unknown}) => void} [options.onExhausted] + * @param {object} [options.context] Extra, non-sensitive log fields + * @param {object} [options.logger] + * @returns {Promise} + */ +export async function retryWithBackoff(operation, options) { + const { + resolved, + label = "operation", + shouldRetry = isRetryableSessionError, + sleep = defaultSleep, + random = Math.random, + onRetry, + onRecovered, + onExhausted, + context = {}, + logger = defaultLogger, + } = options; + + let lastError; + for (let attempt = 1; attempt <= resolved.maxAttempts; attempt += 1) { + try { + const result = await operation({ attempt }); + if (attempt > 1) onRecovered?.({ attempts: attempt }); + return result; + } catch (err) { + lastError = err; + const retryable = shouldRetry(err); + const exhausted = attempt >= resolved.maxAttempts; + + if (!retryable || exhausted) { + if (err && typeof err === "object") err.retryAttempts = attempt; + if (retryable && exhausted) { + onExhausted?.({ attempts: attempt, error: err }); + logger.error( + { ...context, label, attempts: attempt, err: err?.message, code: err?.code }, + "Operation failed after exhausting retries", + ); + } + throw err; + } + + const delayMs = computeBackoffDelay(attempt - 1, resolved, random); + logger.warn( + { + ...context, + label, + attempt, + maxAttempts: resolved.maxAttempts, + delayMs, + err: err?.message, + code: err?.code, + }, + "Operation failed, retrying with backoff", + ); + onRetry?.({ attempt, delayMs, error: err }); + await sleep(delayMs); + } + } + + throw lastError; +} + +/** Internal carrier so a Supabase `{ error }` result can drive the retry loop. */ +class RetryableResultError extends Error { + constructor(result) { + super(result?.error?.message ?? "retryable result error"); + this.name = "RetryableResultError"; + this.result = result; + this.code = result?.error?.code; + } +} + +/** + * Retry a Supabase-style read that reports failures as `{ data, error }` + * instead of throwing. `runQuery` must build a NEW query on every call + * (PostgREST builders are single-use). + * + * Transient `error`s are retried; when retries run out, or the error is not + * transient, the last `{ data, error }` result is returned unchanged so the + * caller's existing error handling and status codes still apply. Thrown + * transient errors are retried and rethrown when exhausted. + * + * @template R + * @param {() => Promise} runQuery + * @param {Parameters[1]} options + * @returns {Promise} + */ +export async function retrySupabaseRead(runQuery, options) { + const shouldRetry = options.shouldRetry ?? isRetryableSessionError; + try { + return await retryWithBackoff( + async (ctx) => { + const result = await runQuery(ctx); + if (result?.error && shouldRetry(result.error)) { + throw new RetryableResultError(result); + } + return result; + }, + { + ...options, + shouldRetry: (err) => err instanceof RetryableResultError || shouldRetry(err), + }, + ); + } catch (err) { + if (err instanceof RetryableResultError) { + if (err.result?.error && typeof err.result.error === "object") { + err.result.error.retryAttempts = err.retryAttempts; + } + return err.result; + } + throw err; + } +} diff --git a/backend/src/lib/backoff-retry.test.js b/backend/src/lib/backoff-retry.test.js new file mode 100644 index 0000000..502dce2 --- /dev/null +++ b/backend/src/lib/backoff-retry.test.js @@ -0,0 +1,156 @@ +import { describe, it, expect, vi } from "vitest"; + +vi.mock("./logger.js", () => ({ + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, +})); + +import { resolveBackoffOptions, retryWithBackoff, retrySupabaseRead } from "./backoff-retry.js"; + +const SPEC = { + envPrefix: "TEST_RETRY", + defaults: { maxAttempts: 3, baseDelayMs: 100, maxDelayMs: 1000 }, + maxAttemptsCeiling: 5, + maxDelayCeilingMs: 5000, +}; +const resolved = { maxAttempts: 3, baseDelayMs: 100, maxDelayMs: 1000 }; +const transient = (code = "ECONNRESET") => Object.assign(new Error(`transient ${code}`), { code }); +const sleep = vi.fn(async () => {}); + +describe("resolveBackoffOptions", () => { + it("falls back to defaults", () => { + expect(resolveBackoffOptions({}, {}, SPEC)).toEqual(SPEC.defaults); + }); + + it("reads prefixed env vars and lets options win", () => { + const env = { + TEST_RETRY_MAX_ATTEMPTS: "4", + TEST_RETRY_BASE_DELAY_MS: "20", + TEST_RETRY_MAX_DELAY_MS: "200", + }; + expect(resolveBackoffOptions({}, env, SPEC)).toEqual({ + maxAttempts: 4, + baseDelayMs: 20, + maxDelayMs: 200, + }); + expect(resolveBackoffOptions({ maxAttempts: 2 }, env, SPEC).maxAttempts).toBe(2); + }); + + it("clamps to the ceilings and never lets maxDelay drop below baseDelay", () => { + expect( + resolveBackoffOptions({ maxAttempts: 99, baseDelayMs: -1, maxDelayMs: 1e9 }, {}, SPEC), + ).toEqual({ maxAttempts: 5, baseDelayMs: 0, maxDelayMs: 5000 }); + expect(resolveBackoffOptions({ baseDelayMs: 900, maxDelayMs: 10 }, {}, SPEC).maxDelayMs).toBe( + 900, + ); + expect(resolveBackoffOptions({ maxAttempts: 0 }, {}, SPEC).maxAttempts).toBe(1); + }); + + it("ignores garbage values", () => { + expect(resolveBackoffOptions({}, { TEST_RETRY_MAX_ATTEMPTS: "many" }, SPEC).maxAttempts).toBe(3); + }); +}); + +describe("retryWithBackoff", () => { + it("returns the first success without sleeping", async () => { + sleep.mockClear(); + const op = vi.fn(async () => 7); + await expect(retryWithBackoff(op, { resolved, sleep })).resolves.toBe(7); + expect(sleep).not.toHaveBeenCalled(); + }); + + it("retries transient errors with full-jitter delays and reports recovery", async () => { + sleep.mockClear(); + const onRetry = vi.fn(); + const onRecovered = vi.fn(); + const op = vi + .fn() + .mockRejectedValueOnce(transient()) + .mockRejectedValueOnce(transient("08006")) + .mockResolvedValueOnce("ok"); + + await expect( + retryWithBackoff(op, { resolved, sleep, random: () => 0.5, onRetry, onRecovered }), + ).resolves.toBe("ok"); + expect(sleep.mock.calls.map(([ms]) => ms)).toEqual([50, 100]); + expect(onRetry).toHaveBeenCalledTimes(2); + expect(onRecovered).toHaveBeenCalledWith({ attempts: 3 }); + }); + + it("throws non-transient errors immediately", async () => { + const err = Object.assign(new Error("bad id"), { code: "22P02" }); + const op = vi.fn().mockRejectedValue(err); + await expect(retryWithBackoff(op, { resolved, sleep })).rejects.toBe(err); + expect(op).toHaveBeenCalledTimes(1); + expect(err.retryAttempts).toBe(1); + }); + + it("gives up after maxAttempts and reports exhaustion", async () => { + const onExhausted = vi.fn(); + const op = vi.fn().mockRejectedValue(transient()); + await expect(retryWithBackoff(op, { resolved, sleep, onExhausted })).rejects.toMatchObject({ + retryAttempts: 3, + }); + expect(op).toHaveBeenCalledTimes(3); + expect(onExhausted).toHaveBeenCalledTimes(1); + }); +}); + +describe("retrySupabaseRead", () => { + it("returns a successful result untouched", async () => { + const run = vi.fn(async () => ({ data: { id: 1 }, error: null })); + await expect(retrySupabaseRead(run, { resolved, sleep })).resolves.toEqual({ + data: { id: 1 }, + error: null, + }); + }); + + it("retries a transient { error } result until it clears", async () => { + const onRecovered = vi.fn(); + const run = vi + .fn() + .mockResolvedValueOnce({ data: null, error: { message: "TypeError: fetch failed", code: "" } }) + .mockResolvedValueOnce({ data: { id: 2 }, error: null }); + + await expect(retrySupabaseRead(run, { resolved, sleep, onRecovered })).resolves.toEqual({ + data: { id: 2 }, + error: null, + }); + expect(run).toHaveBeenCalledTimes(2); + expect(onRecovered).toHaveBeenCalledTimes(1); + }); + + it("returns a non-transient { error } result on the first attempt", async () => { + const result = { data: null, error: { message: "invalid input syntax for type uuid", code: "22P02" } }; + const run = vi.fn(async () => result); + await expect(retrySupabaseRead(run, { resolved, sleep })).resolves.toBe(result); + expect(run).toHaveBeenCalledTimes(1); + }); + + it("returns the last { error } result when retries run out, annotated", async () => { + const onExhausted = vi.fn(); + const run = vi.fn(async () => ({ data: null, error: { message: "upstream", code: "57P01" } })); + + const result = await retrySupabaseRead(run, { resolved, sleep, onExhausted }); + expect(result.error).toMatchObject({ code: "57P01", retryAttempts: 3 }); + expect(run).toHaveBeenCalledTimes(3); + expect(onExhausted).toHaveBeenCalledTimes(1); + }); + + it("retries a thrown transient error and rethrows it when exhausted", async () => { + const run = vi.fn().mockRejectedValue(transient("ETIMEDOUT")); + await expect(retrySupabaseRead(run, { resolved, sleep })).rejects.toMatchObject({ + code: "ETIMEDOUT", + }); + expect(run).toHaveBeenCalledTimes(3); + }); + + it("uses a custom shouldRetry for result errors", async () => { + const run = vi + .fn() + .mockResolvedValueOnce({ data: null, error: { message: "custom" } }) + .mockResolvedValueOnce({ data: 1, error: null }); + await expect( + retrySupabaseRead(run, { resolved, sleep, shouldRetry: () => true }), + ).resolves.toEqual({ data: 1, error: null }); + }); +}); diff --git a/backend/src/lib/payment-processor-retry.js b/backend/src/lib/payment-processor-retry.js new file mode 100644 index 0000000..5dacf61 --- /dev/null +++ b/backend/src/lib/payment-processor-retry.js @@ -0,0 +1,69 @@ +/** + * Automated retry with exponential backoff for Payment Processor reads + * (issue #1479), recorded through Payment Processor Telemetry. + * + * The status poll (`GET /api/payment-status/:id`) and the verification + * lookup (`POST /api/verify-payment/:id`) start with a single Supabase read. + * A dropped connection or a 5xx used to fail the request outright, and the + * failure was invisible in the processor metrics. Both reads are idempotent, + * so they are now retried: + * + * - Retried: network errors, 408/429/5xx, connection-class Postgres errors + * (shared classifier `isRetryableSessionError`). + * - Never retried: not-found, invalid ids, PostgREST/client errors, and + * every write. Status updates and tx claims are NOT wrapped: retrying a + * write is the job of the code that owns its idempotency. + * - Delay: full jitter, random(0, min(maxDelay, base * 2^n)). + * - Budget: PAYMENT_PROCESSOR_RETRY_MAX_ATTEMPTS (default 3, max 5), + * PAYMENT_PROCESSOR_RETRY_BASE_DELAY_MS (default 50), + * PAYMENT_PROCESSOR_RETRY_MAX_DELAY_MS (default 500, max 3000). + * Small on purpose: these are latency-sensitive request paths. + * + * Every retry outcome goes to `payment_processor_retries_total{operation, + * result}`; exhausted retries also degrade processor health. + */ + +import { resolveBackoffOptions, retrySupabaseRead } from "./backoff-retry.js"; +import { recordProcessorRetryTelemetry } from "./payment-processor-telemetry.js"; + +export const DEFAULT_PROCESSOR_RETRY_OPTIONS = Object.freeze({ + maxAttempts: 3, + baseDelayMs: 50, + maxDelayMs: 500, +}); + +const SPEC = Object.freeze({ + envPrefix: "PAYMENT_PROCESSOR_RETRY", + defaults: DEFAULT_PROCESSOR_RETRY_OPTIONS, + maxAttemptsCeiling: 5, + maxDelayCeilingMs: 3000, +}); + +/** @param {object} [options] @param {Record} [env] */ +export function resolveProcessorRetryOptions(options = {}, env = process.env) { + return resolveBackoffOptions(options, env, SPEC); +} + +/** + * Retry an idempotent Supabase read made by the payment processor. + * + * @template R + * @param {"status_lookup"|"verify_lookup"} operation Fixed telemetry label + * @param {() => Promise} runQuery Must build a NEW query on each call + * @param {object} [options] maxAttempts, baseDelayMs, maxDelayMs, sleep, random, shouldRetry + * @returns {Promise} the final `{ data, error }` result + */ +export function retryProcessorRead(operation, runQuery, options = {}) { + const { sleep, random, shouldRetry } = options; + const record = (result) => recordProcessorRetryTelemetry({ operation, result }); + return retrySupabaseRead(runQuery, { + resolved: resolveProcessorRetryOptions(options), + label: `payment-processor-${operation}`, + sleep, + random, + ...(shouldRetry ? { shouldRetry } : {}), + onRetry: () => record("scheduled"), + onRecovered: () => record("recovered"), + onExhausted: () => record("exhausted"), + }); +} diff --git a/backend/src/lib/payment-processor-retry.test.js b/backend/src/lib/payment-processor-retry.test.js new file mode 100644 index 0000000..c39d31c --- /dev/null +++ b/backend/src/lib/payment-processor-retry.test.js @@ -0,0 +1,109 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; + +vi.mock("./logger.js", () => ({ + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, +})); + +import { + DEFAULT_PROCESSOR_RETRY_OPTIONS, + resolveProcessorRetryOptions, + retryProcessorRead, +} from "./payment-processor-retry.js"; +import { + getPaymentProcessorHealth, + paymentProcessorRetriesTotal, + resetPaymentProcessorTelemetry, +} from "./payment-processor-telemetry.js"; + +const fast = { sleep: async () => {}, baseDelayMs: 0 }; +const ok = { data: { id: "p1", status: "pending" }, error: null }; +const outage = { data: null, error: { message: "TypeError: fetch failed", code: "" } }; + +async function retries(operation, result) { + const { values } = await paymentProcessorRetriesTotal.get(); + return ( + values.find((v) => v.labels.operation === operation && v.labels.result === result)?.value ?? 0 + ); +} + +beforeEach(() => { + resetPaymentProcessorTelemetry(); + paymentProcessorRetriesTotal.reset(); +}); + +describe("retryProcessorRead (#1479)", () => { + it("runs the query once when it succeeds", async () => { + const run = vi.fn(async () => ok); + await expect(retryProcessorRead("status_lookup", run, fast)).resolves.toBe(ok); + expect(run).toHaveBeenCalledTimes(1); + expect(await retries("status_lookup", "scheduled")).toBe(0); + }); + + it("retries a transient failure and records scheduled + recovered", async () => { + const run = vi.fn().mockResolvedValueOnce(outage).mockResolvedValueOnce(ok); + + await expect(retryProcessorRead("verify_lookup", run, fast)).resolves.toBe(ok); + expect(run).toHaveBeenCalledTimes(2); + expect(await retries("verify_lookup", "scheduled")).toBe(1); + expect(await retries("verify_lookup", "recovered")).toBe(1); + expect(getPaymentProcessorHealth().status).toBe("healthy"); + }); + + it("returns the last error result when retries run out and degrades health", async () => { + const run = vi.fn(async () => outage); + + const result = await retryProcessorRead("status_lookup", run, { ...fast, maxAttempts: 3 }); + expect(result.error).toMatchObject({ retryAttempts: 3 }); + expect(run).toHaveBeenCalledTimes(3); + expect(await retries("status_lookup", "exhausted")).toBe(1); + expect(getPaymentProcessorHealth()).toMatchObject({ + status: "degraded", + reasons: ["retries_exhausted"], + }); + }); + + it.each([ + ["not found / PostgREST", { message: "JSON object requested", code: "PGRST116" }], + ["invalid uuid", { message: "invalid input syntax for type uuid", code: "22P02" }], + ["missing column", { message: "column deleted_at does not exist", code: "42703" }], + ["permission", { message: "permission denied", code: "42501" }], + ])("never retries a deterministic error (%s)", async (_name, error) => { + const run = vi.fn(async () => ({ data: null, error })); + const result = await retryProcessorRead("status_lookup", run, fast); + expect(result.error).toBe(error); + expect(run).toHaveBeenCalledTimes(1); + }); + + it("retries thrown network errors and rethrows when exhausted", async () => { + const err = Object.assign(new Error("socket hang up"), { code: "ECONNRESET" }); + const run = vi.fn().mockRejectedValue(err); + await expect(retryProcessorRead("verify_lookup", run, { ...fast, maxAttempts: 2 })).rejects.toBe(err); + expect(run).toHaveBeenCalledTimes(2); + }); + + it("drops an unknown operation label instead of recording it", async () => { + const run = vi.fn().mockResolvedValueOnce(outage).mockResolvedValueOnce(ok); + await retryProcessorRead("drop_table", run, fast); + const { values } = await paymentProcessorRetriesTotal.get(); + expect(values.every((v) => ["status_lookup", "verify_lookup"].includes(v.labels.operation))).toBe(true); + }); +}); + +describe("resolveProcessorRetryOptions", () => { + it("defaults to a small latency budget", () => { + expect(resolveProcessorRetryOptions({}, {})).toEqual(DEFAULT_PROCESSOR_RETRY_OPTIONS); + }); + + it("reads PAYMENT_PROCESSOR_RETRY_* and clamps", () => { + expect( + resolveProcessorRetryOptions( + {}, + { + PAYMENT_PROCESSOR_RETRY_MAX_ATTEMPTS: "10", + PAYMENT_PROCESSOR_RETRY_BASE_DELAY_MS: "20", + PAYMENT_PROCESSOR_RETRY_MAX_DELAY_MS: "100000", + }, + ), + ).toEqual({ maxAttempts: 5, baseDelayMs: 20, maxDelayMs: 3000 }); + }); +}); diff --git a/backend/src/lib/payment-processor-telemetry.js b/backend/src/lib/payment-processor-telemetry.js new file mode 100644 index 0000000..27f9a01 --- /dev/null +++ b/backend/src/lib/payment-processor-telemetry.js @@ -0,0 +1,481 @@ +/** + * Payment Processor Telemetry: payload sanitization, strict validation + * (issue #1477), alert metrics and health (issue #1478) and retry outcome + * recording (issue #1479). + * + * Why sanitization matters + * ------------------------ + * The processor used to pass `req.body.asset` straight into Prometheus + * labels. A client could send a new random asset string on every request and + * create an unbounded number of series (memory growth in this process and in + * Prometheus), or inject arbitrary text into dashboards. Every recorder here: + * + * - accepts only a plain object payload + * - checks `outcome` / `stage` / `operation` against a fixed allow-list + * - normalises `asset` to an upper-case Stellar asset code + * (^[A-Z0-9]{1,12}$); anything else becomes "invalid" + * - caps distinct asset labels (PAYMENT_PROCESSOR_TELEMETRY_MAX_ASSET_LABELS, + * default 50); later new codes are recorded as "other" + * - accepts durations only if finite, >= 0 and <= 1 hour + * + * A rejected payload is NOT written to the business series; it increments + * `payment_processor_telemetry_rejected_total{metric, reason}` instead. + * Recorders never throw: telemetry must never fail a payment request. + * + * Health + * ------ + * A rolling window (default 5 min) of session, verification, refund and + * retry outcomes yields healthy | degraded | unhealthy, exposed as gauges + * (refreshed at scrape time) and via GET /health/payment-processor. Only + * counts and ratios are kept: no payment ids, amounts, addresses or assets. + */ + +import client from "prom-client"; +import { logger } from "./logger.js"; +import { + paymentProcessorRegister, + paymentProcessorSessionsTotal, + paymentProcessorSessionDuration, + paymentProcessorVerificationsTotal, + paymentProcessorVerificationDuration, + paymentProcessorStatusCacheHits, + paymentProcessorStatusCacheMisses, + paymentProcessorRefundsTotal, + paymentProcessorListRequestsTotal, +} from "./payment-processor-metrics.js"; + +// ── Allow-lists (#1477) ────────────────────────────────────────────────────── + +export const SESSION_OUTCOMES = Object.freeze([ + "created", + "validation_failed", + "persistence_failed", + "sandbox_skipped", +]); + +export const VERIFICATION_OUTCOMES = Object.freeze([ + "confirmed", + "already_confirmed", + "already_confirmed_concurrent", + "pending_no_match", + "signature_invalid", + "underpayment", + "overpayment", + "tx_claim_conflict", + "error", +]); + +export const REFUND_STAGES = Object.freeze(["generate", "confirm"]); +export const REFUND_OUTCOMES = Object.freeze([ + "success", + "rejected", + "not_found", + "failed_on_chain", + "error", +]); + +export const LIST_OUTCOMES = Object.freeze(["success", "pool_fallback", "error"]); +export const RETRY_OPERATIONS = Object.freeze(["status_lookup", "verify_lookup"]); +export const RETRY_RESULTS = Object.freeze(["scheduled", "recovered", "exhausted"]); + +const REJECT_REASONS = Object.freeze([ + "invalid_payload", + "invalid_outcome", + "invalid_stage", + "invalid_operation", + "invalid_duration", +]); + +const ASSET_CODE_RE = /^[A-Z0-9]{1,12}$/; +export const MAX_DURATION_SECONDS = 3600; + +function readNumberEnv(name, fallback, env = process.env) { + const raw = Number.parseFloat(env[name] ?? ""); + return Number.isFinite(raw) && raw >= 0 ? raw : fallback; +} + +let maxAssetLabels = Math.max( + 1, + Math.trunc(readNumberEnv("PAYMENT_PROCESSOR_TELEMETRY_MAX_ASSET_LABELS", 50)), +); +const seenAssets = new Set(); + +// ── New series (#1477, #1478, #1479) ──────────────────────────────────────── + +export const paymentProcessorTelemetryRejectedTotal = new client.Counter({ + name: "payment_processor_telemetry_rejected_total", + help: "Payment processor telemetry payloads rejected by strict validation, by metric and reason", + labelNames: ["metric", "reason"], + registers: [], +}); + +export const paymentProcessorRetriesTotal = new client.Counter({ + name: "payment_processor_retries_total", + help: "Payment processor read retries with exponential backoff, by operation and result", + labelNames: ["operation", "result"], + registers: [], +}); + +/** 0 = healthy, 1 = degraded, 2 = unhealthy. Refreshed at scrape time. */ +export const paymentProcessorHealthState = new client.Gauge({ + name: "payment_processor_health_state", + help: "Payment processor health (0 = healthy, 1 = degraded, 2 = unhealthy)", + registers: [], +}); + +export const paymentProcessorSessionFailureRatio = new client.Gauge({ + name: "payment_processor_session_failure_ratio", + help: "Share of session creations that failed to persist over the rolling window", + registers: [], +}); + +export const paymentProcessorVerificationErrorRatio = new client.Gauge({ + name: "payment_processor_verification_error_ratio", + help: "Share of verification attempts that ended in an internal error over the rolling window", + registers: [], +}); + +export const paymentProcessorRefundErrorRatio = new client.Gauge({ + name: "payment_processor_refund_error_ratio", + help: "Share of refund operations that errored or failed on chain over the rolling window", + registers: [], +}); + +export const paymentProcessorTelemetryRejectionRatio = new client.Gauge({ + name: "payment_processor_telemetry_rejection_ratio", + help: "Share of telemetry payloads rejected by validation over the rolling window", + registers: [], +}); + +export const HEALTH_STATE_VALUES = Object.freeze({ healthy: 0, degraded: 1, unhealthy: 2 }); + +// ── Health monitor (#1478) ─────────────────────────────────────────────────── + +export function resolveProcessorHealthOptions(options = {}, env = process.env) { + return { + windowMs: options.windowMs ?? readNumberEnv("PAYMENT_PROCESSOR_HEALTH_WINDOW_MS", 300_000, env), + bucketMs: options.bucketMs ?? 5_000, + minSamples: options.minSamples ?? readNumberEnv("PAYMENT_PROCESSOR_HEALTH_MIN_SAMPLES", 20, env), + degradedRatio: + options.degradedRatio ?? readNumberEnv("PAYMENT_PROCESSOR_DEGRADED_ERROR_RATIO", 0.05, env), + unhealthyRatio: + options.unhealthyRatio ?? readNumberEnv("PAYMENT_PROCESSOR_UNHEALTHY_ERROR_RATIO", 0.2, env), + }; +} + +function emptyBucket() { + return { + slot: -1, + sessions: 0, + sessionFailures: 0, + verifications: 0, + verificationErrors: 0, + refunds: 0, + refundErrors: 0, + events: 0, + rejected: 0, + retriesExhausted: 0, + }; +} + +/** Fixed ring of time buckets: memory is constant regardless of traffic. */ +export class PaymentProcessorHealthMonitor { + constructor(options = {}, now = () => Date.now()) { + this.options = resolveProcessorHealthOptions(options); + this.now = now; + this.bucketCount = Math.max(1, Math.ceil(this.options.windowMs / this.options.bucketMs)); + this.reset(); + } + + reset() { + this.buckets = Array.from({ length: this.bucketCount }, () => emptyBucket()); + } + + _bucket() { + const slot = Math.floor(this.now() / this.options.bucketMs); + const bucket = this.buckets[slot % this.bucketCount]; + if (bucket.slot !== slot) Object.assign(bucket, emptyBucket(), { slot }); + return bucket; + } + + recordSession(outcome) { + const b = this._bucket(); + b.events += 1; + if (outcome === "validation_failed" || outcome === "sandbox_skipped") return; + b.sessions += 1; + if (outcome === "persistence_failed") b.sessionFailures += 1; + } + + recordVerification(outcome) { + const b = this._bucket(); + b.events += 1; + b.verifications += 1; + if (outcome === "error") b.verificationErrors += 1; + } + + recordRefund(outcome) { + const b = this._bucket(); + b.events += 1; + b.refunds += 1; + if (outcome === "error" || outcome === "failed_on_chain") b.refundErrors += 1; + } + + recordEvent() { + this._bucket().events += 1; + } + + recordRejected() { + const b = this._bucket(); + b.events += 1; + b.rejected += 1; + } + + recordRetryExhausted() { + this._bucket().retriesExhausted += 1; + } + + snapshot() { + const currentSlot = Math.floor(this.now() / this.options.bucketMs); + const oldestSlot = currentSlot - this.bucketCount + 1; + const c = emptyBucket(); + for (const b of this.buckets) { + if (b.slot < oldestSlot || b.slot > currentSlot) continue; + for (const key of Object.keys(c)) if (key !== "slot") c[key] += b[key]; + } + + const ratio = (num, den) => (den > 0 ? num / den : 0); + const sessionFailureRatio = ratio(c.sessionFailures, c.sessions); + const verificationErrorRatio = ratio(c.verificationErrors, c.verifications); + const refundErrorRatio = ratio(c.refundErrors, c.refunds); + const rejectionRatio = ratio(c.rejected, c.events); + const { minSamples, degradedRatio, unhealthyRatio } = this.options; + + const reasons = []; + let status = "healthy"; + const check = (value, samples, reason) => { + if (samples < minSamples) return; + if (value >= unhealthyRatio) { + status = "unhealthy"; + reasons.push(`${reason}_critical`); + } else if (value >= degradedRatio) { + if (status === "healthy") status = "degraded"; + reasons.push(reason); + } + }; + check(sessionFailureRatio, c.sessions, "session_failure_ratio"); + check(verificationErrorRatio, c.verifications, "verification_error_ratio"); + check(refundErrorRatio, c.refunds, "refund_error_ratio"); + if (c.events >= minSamples && rejectionRatio >= degradedRatio) { + if (status === "healthy") status = "degraded"; + reasons.push("telemetry_rejections"); + } + if (c.retriesExhausted > 0) { + if (status === "healthy") status = "degraded"; + reasons.push("retries_exhausted"); + } + + const r4 = (n) => Number(n.toFixed(4)); + return { + status, + reasons, + window_ms: this.bucketCount * this.options.bucketMs, + sessions: c.sessions, + session_failures: c.sessionFailures, + verifications: c.verifications, + verification_errors: c.verificationErrors, + refunds: c.refunds, + refund_errors: c.refundErrors, + telemetry_events: c.events, + telemetry_rejected: c.rejected, + retries_exhausted: c.retriesExhausted, + session_failure_ratio: r4(sessionFailureRatio), + verification_error_ratio: r4(verificationErrorRatio), + refund_error_ratio: r4(refundErrorRatio), + telemetry_rejection_ratio: r4(rejectionRatio), + thresholds: { + min_samples: minSamples, + degraded_ratio: degradedRatio, + unhealthy_ratio: unhealthyRatio, + }, + }; + } +} + +const healthMonitor = new PaymentProcessorHealthMonitor(); + +paymentProcessorHealthState.collect = function collectPaymentProcessorHealth() { + const snap = healthMonitor.snapshot(); + this.set(HEALTH_STATE_VALUES[snap.status] ?? 0); + paymentProcessorSessionFailureRatio.set(snap.session_failure_ratio); + paymentProcessorVerificationErrorRatio.set(snap.verification_error_ratio); + paymentProcessorRefundErrorRatio.set(snap.refund_error_ratio); + paymentProcessorTelemetryRejectionRatio.set(snap.telemetry_rejection_ratio); +}; + +/** Current processor health for /health endpoints. Counts and ratios only. */ +export function getPaymentProcessorHealth() { + return healthMonitor.snapshot(); +} + +/** Test-only: clear the rolling window and the asset label set. Counters are kept. */ +export function resetPaymentProcessorTelemetry({ maxAssets } = {}) { + healthMonitor.reset(); + seenAssets.clear(); + if (maxAssets !== undefined) maxAssetLabels = Math.max(1, Math.trunc(maxAssets)); +} + +// ── Sanitization (#1477) ───────────────────────────────────────────────────── + +/** + * Normalise a client-supplied asset into a bounded label value. + * @param {unknown} raw + * @returns {string} an asset code, "unknown", "invalid" or "other" + */ +export function sanitizeAssetLabel(raw) { + if (raw === undefined || raw === null || raw === "") return "unknown"; + if (typeof raw !== "string") return "invalid"; + // Callers use the literal "unknown" when no asset context exists. + if (raw === "unknown") return "unknown"; + const code = raw.trim().toUpperCase(); + if (code === "NATIVE") return "XLM"; + if (!ASSET_CODE_RE.test(code)) return "invalid"; + if (seenAssets.has(code)) return code; + if (seenAssets.size >= maxAssetLabels) return "other"; + seenAssets.add(code); + return code; +} + +/** + * @param {unknown} value seconds + * @returns {number | null} null when the value must not be observed + */ +export function sanitizeDurationSeconds(value) { + if (typeof value !== "number" || !Number.isFinite(value)) return null; + if (value < 0 || value > MAX_DURATION_SECONDS) return null; + return value; +} + +function isPlainObject(value) { + return value !== null && typeof value === "object" && !Array.isArray(value); +} + +function reject(metric, reason) { + const safeReason = REJECT_REASONS.includes(reason) ? reason : "invalid_payload"; + safeRecord(() => { + paymentProcessorTelemetryRejectedTotal.inc({ metric, reason: safeReason }); + healthMonitor.recordRejected(); + }); + logger.warn({ metric, reason: safeReason }, "Payment processor telemetry payload rejected"); + return false; +} + +function safeRecord(fn) { + try { + fn(); + } catch (err) { + logger.error({ err: err?.message }, "Failed to record payment processor telemetry"); + } +} + +// ── Recorders ──────────────────────────────────────────────────────────────── + +/** + * @param {{asset?: unknown, outcome: string, durationSeconds?: number}} payload + * @returns {boolean} whether the payload was accepted + */ +export function recordSessionTelemetry(payload) { + if (!isPlainObject(payload)) return reject("sessions", "invalid_payload"); + if (!SESSION_OUTCOMES.includes(payload.outcome)) return reject("sessions", "invalid_outcome"); + const labels = { asset: sanitizeAssetLabel(payload.asset), outcome: payload.outcome }; + const duration = sanitizeDurationSeconds(payload.durationSeconds); + safeRecord(() => { + paymentProcessorSessionsTotal.inc(labels); + if (duration !== null) paymentProcessorSessionDuration.observe(labels, duration); + healthMonitor.recordSession(labels.outcome); + }); + if (payload.durationSeconds !== undefined && duration === null) { + reject("session_duration", "invalid_duration"); + } + return true; +} + +/** + * @param {{asset?: unknown, outcome: string, durationSeconds?: number}} payload + * @returns {boolean} + */ +export function recordVerificationTelemetry(payload) { + if (!isPlainObject(payload)) return reject("verifications", "invalid_payload"); + if (!VERIFICATION_OUTCOMES.includes(payload.outcome)) { + return reject("verifications", "invalid_outcome"); + } + const labels = { asset: sanitizeAssetLabel(payload.asset), outcome: payload.outcome }; + const duration = sanitizeDurationSeconds(payload.durationSeconds); + safeRecord(() => { + paymentProcessorVerificationsTotal.inc(labels); + if (duration !== null) paymentProcessorVerificationDuration.observe(labels, duration); + healthMonitor.recordVerification(labels.outcome); + }); + if (payload.durationSeconds !== undefined && duration === null) { + reject("verification_duration", "invalid_duration"); + } + return true; +} + +/** + * @param {{stage: string, outcome: string}} payload + * @returns {boolean} + */ +export function recordRefundTelemetry(payload) { + if (!isPlainObject(payload)) return reject("refunds", "invalid_payload"); + if (!REFUND_STAGES.includes(payload.stage)) return reject("refunds", "invalid_stage"); + if (!REFUND_OUTCOMES.includes(payload.outcome)) return reject("refunds", "invalid_outcome"); + safeRecord(() => { + paymentProcessorRefundsTotal.inc({ stage: payload.stage, outcome: payload.outcome }); + healthMonitor.recordRefund(payload.outcome); + }); + return true; +} + +/** @param {{outcome: string}} payload @returns {boolean} */ +export function recordListRequestTelemetry(payload) { + if (!isPlainObject(payload)) return reject("list_requests", "invalid_payload"); + if (!LIST_OUTCOMES.includes(payload.outcome)) return reject("list_requests", "invalid_outcome"); + safeRecord(() => { + paymentProcessorListRequestsTotal.inc({ outcome: payload.outcome }); + healthMonitor.recordEvent(); + }); + return true; +} + +/** @param {boolean} hit @returns {boolean} */ +export function recordStatusCacheTelemetry(hit) { + if (typeof hit !== "boolean") return reject("status_cache", "invalid_payload"); + safeRecord(() => { + (hit ? paymentProcessorStatusCacheHits : paymentProcessorStatusCacheMisses).inc(); + healthMonitor.recordEvent(); + }); + return true; +} + +/** + * @param {{operation: string, result: string}} payload + * @returns {boolean} + */ +export function recordProcessorRetryTelemetry(payload) { + if (!isPlainObject(payload)) return reject("retries", "invalid_payload"); + if (!RETRY_OPERATIONS.includes(payload.operation)) return reject("retries", "invalid_operation"); + if (!RETRY_RESULTS.includes(payload.result)) return reject("retries", "invalid_outcome"); + safeRecord(() => { + paymentProcessorRetriesTotal.inc({ operation: payload.operation, result: payload.result }); + if (payload.result === "exhausted") healthMonitor.recordRetryExhausted(); + }); + return true; +} + +paymentProcessorRegister.registerMetric(paymentProcessorTelemetryRejectedTotal); +paymentProcessorRegister.registerMetric(paymentProcessorRetriesTotal); +paymentProcessorRegister.registerMetric(paymentProcessorHealthState); +paymentProcessorRegister.registerMetric(paymentProcessorSessionFailureRatio); +paymentProcessorRegister.registerMetric(paymentProcessorVerificationErrorRatio); +paymentProcessorRegister.registerMetric(paymentProcessorRefundErrorRatio); +paymentProcessorRegister.registerMetric(paymentProcessorTelemetryRejectionRatio); diff --git a/backend/src/lib/payment-processor-telemetry.test.js b/backend/src/lib/payment-processor-telemetry.test.js new file mode 100644 index 0000000..cbeaf27 --- /dev/null +++ b/backend/src/lib/payment-processor-telemetry.test.js @@ -0,0 +1,313 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; + +vi.mock("./logger.js", () => ({ + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, +})); + +import { + HEALTH_STATE_VALUES, + MAX_DURATION_SECONDS, + PaymentProcessorHealthMonitor, + getPaymentProcessorHealth, + paymentProcessorHealthState, + paymentProcessorRetriesTotal, + paymentProcessorTelemetryRejectedTotal, + recordListRequestTelemetry, + recordProcessorRetryTelemetry, + recordRefundTelemetry, + recordSessionTelemetry, + recordStatusCacheTelemetry, + recordVerificationTelemetry, + resetPaymentProcessorTelemetry, + sanitizeAssetLabel, + sanitizeDurationSeconds, +} from "./payment-processor-telemetry.js"; +import { + paymentProcessorRegister, + paymentProcessorSessionsTotal, + paymentProcessorSessionDuration, + paymentProcessorVerificationsTotal, + paymentProcessorRefundsTotal, + paymentProcessorListRequestsTotal, + paymentProcessorStatusCacheHits, + paymentProcessorStatusCacheMisses, +} from "./payment-processor-metrics.js"; +import { logger } from "./logger.js"; + +async function value(metric, labels = {}) { + const { values } = await metric.get(); + const hit = values.find( + (v) => + !v.metricName?.endsWith("_bucket") && + Object.entries(labels).every(([k, val]) => v.labels[k] === val), + ); + return hit?.value ?? 0; +} + +async function labelValues(metric, label) { + const { values } = await metric.get(); + return [...new Set(values.map((v) => v.labels[label]).filter(Boolean))]; +} + +beforeEach(() => { + resetPaymentProcessorTelemetry({ maxAssets: 50 }); + for (const m of [ + paymentProcessorSessionsTotal, + paymentProcessorSessionDuration, + paymentProcessorVerificationsTotal, + paymentProcessorRefundsTotal, + paymentProcessorListRequestsTotal, + paymentProcessorStatusCacheHits, + paymentProcessorStatusCacheMisses, + paymentProcessorTelemetryRejectedTotal, + paymentProcessorRetriesTotal, + ]) { + m.reset(); + } + vi.mocked(logger.warn).mockClear(); +}); + +describe("sanitizeAssetLabel (#1477)", () => { + it.each([ + ["XLM", "XLM"], + [" usdc ", "USDC"], + ["native", "XLM"], + ["ABCDEFGHIJKL", "ABCDEFGHIJKL"], + [undefined, "unknown"], + [null, "unknown"], + ["", "unknown"], + ["unknown", "unknown"], + ["ABCDEFGHIJKLM", "invalid"], + ['XLM"} evil{x="', "invalid"], + ["US DC", "invalid"], + ["usd-c", "invalid"], + [42, "invalid"], + [{ code: "XLM" }, "invalid"], + ])("%p -> %s", (raw, expected) => { + expect(sanitizeAssetLabel(raw)).toBe(expected); + }); + + it("caps distinct asset labels and maps the overflow to other", () => { + resetPaymentProcessorTelemetry({ maxAssets: 2 }); + expect(sanitizeAssetLabel("AAA")).toBe("AAA"); + expect(sanitizeAssetLabel("BBB")).toBe("BBB"); + expect(sanitizeAssetLabel("CCC")).toBe("other"); + // Already-seen codes keep their own label. + expect(sanitizeAssetLabel("aaa")).toBe("AAA"); + }); +}); + +describe("sanitizeDurationSeconds", () => { + it.each([ + [0, 0], + [1.5, 1.5], + [MAX_DURATION_SECONDS, MAX_DURATION_SECONDS], + [-1, null], + [MAX_DURATION_SECONDS + 1, null], + [Number.NaN, null], + [Number.POSITIVE_INFINITY, null], + ["1", null], + [undefined, null], + ])("%p -> %p", (raw, expected) => { + expect(sanitizeDurationSeconds(raw)).toBe(expected); + }); +}); + +describe("recorders (#1477)", () => { + it("records a valid session with a sanitised asset", async () => { + expect(recordSessionTelemetry({ asset: "usdc", outcome: "created", durationSeconds: 0.2 })).toBe(true); + expect(await value(paymentProcessorSessionsTotal, { asset: "USDC", outcome: "created" })).toBe(1); + const text = await paymentProcessorRegister.metrics(); + expect(text).toMatch( + /payment_processor_session_duration_seconds_count\{[^}]*asset="USDC"[^}]*outcome="created"[^}]*\} 1/, + ); + }); + + it("stops a client from minting a new series per request", async () => { + resetPaymentProcessorTelemetry({ maxAssets: 3 }); + for (let i = 0; i < 500; i += 1) { + recordSessionTelemetry({ asset: `A${i}`, outcome: "validation_failed", durationSeconds: 0.01 }); + } + const assets = await labelValues(paymentProcessorSessionsTotal, "asset"); + expect(assets.length).toBeLessThanOrEqual(4); + expect(assets).toContain("other"); + }); + + it("rejects unknown outcomes without touching the business series", async () => { + expect(recordSessionTelemetry({ asset: "XLM", outcome: "pwned" })).toBe(false); + expect(recordVerificationTelemetry({ asset: "XLM", outcome: "made_up" })).toBe(false); + expect(await labelValues(paymentProcessorSessionsTotal, "outcome")).toEqual([]); + expect(await labelValues(paymentProcessorVerificationsTotal, "outcome")).toEqual([]); + expect( + await value(paymentProcessorTelemetryRejectedTotal, { metric: "sessions", reason: "invalid_outcome" }), + ).toBe(1); + expect(logger.warn).toHaveBeenCalledWith( + { metric: "sessions", reason: "invalid_outcome" }, + expect.any(String), + ); + }); + + it.each([null, undefined, "created", 42, ["created"]])("rejects a non-object payload %p", async (payload) => { + expect(recordSessionTelemetry(payload)).toBe(false); + expect( + await value(paymentProcessorTelemetryRejectedTotal, { metric: "sessions", reason: "invalid_payload" }), + ).toBe(1); + }); + + it("counts the event but drops an invalid duration", async () => { + expect( + recordVerificationTelemetry({ asset: "XLM", outcome: "confirmed", durationSeconds: -5 }), + ).toBe(true); + expect(await value(paymentProcessorVerificationsTotal, { asset: "XLM", outcome: "confirmed" })).toBe(1); + expect( + await value(paymentProcessorTelemetryRejectedTotal, { + metric: "verification_duration", + reason: "invalid_duration", + }), + ).toBe(1); + const text = await paymentProcessorRegister.metrics(); + expect(text).not.toMatch(/payment_processor_verification_duration_seconds_count\{[^}]*\} 1/); + }); + + it("validates refund stage and outcome", async () => { + expect(recordRefundTelemetry({ stage: "confirm", outcome: "failed_on_chain" })).toBe(true); + expect(recordRefundTelemetry({ stage: "steal", outcome: "success" })).toBe(false); + expect(recordRefundTelemetry({ stage: "generate", outcome: "weird" })).toBe(false); + expect(await value(paymentProcessorRefundsTotal, { stage: "confirm", outcome: "failed_on_chain" })).toBe(1); + expect(await labelValues(paymentProcessorRefundsTotal, "stage")).toEqual(["confirm"]); + }); + + it("validates list request outcomes", async () => { + expect(recordListRequestTelemetry({ outcome: "pool_fallback" })).toBe(true); + expect(recordListRequestTelemetry({ outcome: "nope" })).toBe(false); + expect(await value(paymentProcessorListRequestsTotal, { outcome: "pool_fallback" })).toBe(1); + }); + + it("records status cache hits and misses, rejecting non-booleans", async () => { + recordStatusCacheTelemetry(true); + recordStatusCacheTelemetry(false); + expect(recordStatusCacheTelemetry("yes")).toBe(false); + expect(await value(paymentProcessorStatusCacheHits)).toBe(1); + expect(await value(paymentProcessorStatusCacheMisses)).toBe(1); + }); + + it("validates retry operation and result", async () => { + expect(recordProcessorRetryTelemetry({ operation: "status_lookup", result: "scheduled" })).toBe(true); + expect(recordProcessorRetryTelemetry({ operation: "drop_table", result: "scheduled" })).toBe(false); + expect(recordProcessorRetryTelemetry({ operation: "verify_lookup", result: "maybe" })).toBe(false); + expect(await value(paymentProcessorRetriesTotal, { operation: "status_lookup", result: "scheduled" })).toBe(1); + expect(await labelValues(paymentProcessorRetriesTotal, "operation")).toEqual(["status_lookup"]); + }); + + it("never throws even if a metric write fails", () => { + const spy = vi.spyOn(paymentProcessorSessionsTotal, "inc").mockImplementation(() => { + throw new Error("boom"); + }); + expect(() => recordSessionTelemetry({ asset: "XLM", outcome: "created" })).not.toThrow(); + spy.mockRestore(); + }); +}); + +describe("PaymentProcessorHealthMonitor (#1478)", () => { + function monitor() { + let t = 1_700_000_000_000; + const m = new PaymentProcessorHealthMonitor( + { windowMs: 60_000, bucketMs: 1_000, minSamples: 10, degradedRatio: 0.05, unhealthyRatio: 0.2 }, + () => t, + ); + return { m, advance: (ms) => (t += ms) }; + } + + it("is healthy with no traffic and below min samples", () => { + const { m } = monitor(); + expect(m.snapshot().status).toBe("healthy"); + for (let i = 0; i < 5; i += 1) m.recordSession("persistence_failed"); + expect(m.snapshot().status).toBe("healthy"); + }); + + it("degrades and then becomes unhealthy as session persistence fails", () => { + const { m } = monitor(); + for (let i = 0; i < 18; i += 1) m.recordSession("created"); + for (let i = 0; i < 2; i += 1) m.recordSession("persistence_failed"); + expect(m.snapshot()).toMatchObject({ status: "degraded", reasons: ["session_failure_ratio"] }); + for (let i = 0; i < 10; i += 1) m.recordSession("persistence_failed"); + expect(m.snapshot()).toMatchObject({ + status: "unhealthy", + reasons: ["session_failure_ratio_critical"], + }); + }); + + it("does not count client validation failures as processor errors", () => { + const { m } = monitor(); + for (let i = 0; i < 50; i += 1) m.recordSession("validation_failed"); + m.recordSession("created"); + expect(m.snapshot()).toMatchObject({ status: "healthy", sessions: 1, session_failures: 0 }); + }); + + it("tracks verification and refund error ratios", () => { + const { m } = monitor(); + for (let i = 0; i < 10; i += 1) m.recordVerification(i < 3 ? "error" : "confirmed"); + for (let i = 0; i < 10; i += 1) m.recordRefund(i < 1 ? "failed_on_chain" : "success"); + const snap = m.snapshot(); + expect(snap.verification_error_ratio).toBe(0.3); + expect(snap.refund_error_ratio).toBe(0.1); + expect(snap.status).toBe("unhealthy"); + expect(snap.reasons).toEqual(["verification_error_ratio_critical", "refund_error_ratio"]); + }); + + it("degrades on telemetry rejections and exhausted retries", () => { + const { m } = monitor(); + for (let i = 0; i < 9; i += 1) m.recordEvent(); + m.recordRejected(); + expect(m.snapshot().reasons).toContain("telemetry_rejections"); + const { m: m2 } = monitor(); + m2.recordRetryExhausted(); + expect(m2.snapshot()).toMatchObject({ status: "degraded", reasons: ["retries_exhausted"] }); + }); + + it("forgets old buckets once they leave the window", () => { + const { m, advance } = monitor(); + for (let i = 0; i < 20; i += 1) m.recordSession("persistence_failed"); + expect(m.snapshot().status).toBe("unhealthy"); + advance(61_000); + expect(m.snapshot()).toMatchObject({ status: "healthy", sessions: 0 }); + }); + + it("exposes only counts and ratios", () => { + const { m } = monitor(); + m.recordSession("created"); + const snap = m.snapshot(); + expect(JSON.stringify(snap)).not.toMatch(/XLM|USDC|G[A-Z2-7]{55}/); + }); +}); + +describe("shared health + scrape-time gauges (#1478)", () => { + it("feeds getPaymentProcessorHealth from the recorders", () => { + recordVerificationTelemetry({ asset: "XLM", outcome: "error" }); + recordRefundTelemetry({ stage: "confirm", outcome: "success" }); + expect(getPaymentProcessorHealth()).toMatchObject({ verifications: 1, verification_errors: 1, refunds: 1 }); + }); + + it("refreshes health gauges when /metrics is scraped", async () => { + for (let i = 0; i < 25; i += 1) recordSessionTelemetry({ asset: "XLM", outcome: "persistence_failed" }); + const text = await paymentProcessorRegister.metrics(); + expect(text).toMatch(/payment_processor_health_state\{[^}]*\} 2/); + expect(text).toMatch(/payment_processor_session_failure_ratio\{[^}]*\} 1/); + expect(await value(paymentProcessorHealthState)).toBe(HEALTH_STATE_VALUES.unhealthy); + }); + + it("registers the new series on the processor registry", async () => { + const text = await paymentProcessorRegister.metrics(); + for (const name of [ + "payment_processor_telemetry_rejected_total", + "payment_processor_retries_total", + "payment_processor_health_state", + "payment_processor_session_failure_ratio", + "payment_processor_verification_error_ratio", + "payment_processor_refund_error_ratio", + "payment_processor_telemetry_rejection_ratio", + ]) { + expect(text).toContain(`# TYPE ${name}`); + } + }); +}); diff --git a/backend/src/lib/sep0001-retry.js b/backend/src/lib/sep0001-retry.js new file mode 100644 index 0000000..4a335e0 --- /dev/null +++ b/backend/src/lib/sep0001-retry.js @@ -0,0 +1,99 @@ +/** + * Automated retry with exponential backoff for the SEP-0001 Stellar Info + * Generator (issue #1459). + * + * `GET /.well-known/stellar.toml` is polled by wallets, anchors and + * validators. Its only I/O is the merchant lookup. A single dropped + * connection or 5xx from Supabase used to turn into a 500, which wallets + * cache as "no stellar.toml". The lookup is a read, so retrying it is safe. + * + * - Retried: network errors, 408/429/5xx, connection-class Postgres errors + * (shared classifier `isRetryableSessionError`). + * - Never retried: invalid ids (22P02), PostgREST/client errors, "no row". + * - Delay: full jitter, random(0, min(maxDelay, base * 2^n)). + * - Budget: SEP0001_RETRY_MAX_ATTEMPTS (default 3, max 5), + * SEP0001_RETRY_BASE_DELAY_MS (default 100), SEP0001_RETRY_MAX_DELAY_MS + * (default 1000, max 5000). Small on purpose: this is a request path. + * + * Retry outcomes are exported as `sep0001_merchant_lookup_retries_total`. + * Labels come from a fixed set; merchant ids never become labels or log + * fields beyond the request's own structured context. + */ + +import client from "prom-client"; +import { register } from "./metrics.js"; +import { resolveBackoffOptions, retrySupabaseRead } from "./backoff-retry.js"; +import { isRetryableSessionError } from "./payment-session-retry.js"; + +export const DEFAULT_SEP0001_RETRY_OPTIONS = Object.freeze({ + maxAttempts: 3, + baseDelayMs: 100, + maxDelayMs: 1000, +}); + +const SPEC = Object.freeze({ + envPrefix: "SEP0001_RETRY", + defaults: DEFAULT_SEP0001_RETRY_OPTIONS, + maxAttemptsCeiling: 5, + maxDelayCeilingMs: 5000, +}); + +/** result: scheduled | recovered | exhausted */ +export const sep0001LookupRetriesTotal = new client.Counter({ + name: "sep0001_merchant_lookup_retries_total", + help: "SEP-0001 stellar.toml merchant lookup retries with exponential backoff, by result", + labelNames: ["result"], + registers: [], +}); +register.registerMetric(sep0001LookupRetriesTotal); + +export const MERCHANT_TOML_FIELDS = + "id, business_name, email, notification_email, recipient, branding_config"; + +/** @param {object} [options] @param {Record} [env] */ +export function resolveSep0001RetryOptions(options = {}, env = process.env) { + return resolveBackoffOptions(options, env, SPEC); +} + +/** + * True when a lookup error is transient (worth a 503 + retry) rather than a + * deterministic failure. + * + * @param {unknown} err + */ +export function isTransientSep0001Error(err) { + return isRetryableSessionError(err); +} + +/** + * Fetch the merchant row used to build stellar.toml, retrying transient + * failures. Returns the final `{ data, error }` exactly like the plain + * Supabase call, so callers keep their own 404/500 handling. + * + * @param {object} supabase Supabase client + * @param {string} merchantId + * @param {object} [options] maxAttempts, baseDelayMs, maxDelayMs, sleep, random + * @returns {Promise<{data: object | null, error: object | null}>} + */ +export async function fetchMerchantForStellarToml(supabase, merchantId, options = {}) { + const { sleep, random } = options; + return retrySupabaseRead( + () => + supabase + .from("merchants") + .select(MERCHANT_TOML_FIELDS) + .eq("id", merchantId) + .is("deleted_at", null) + .maybeSingle(), + { + resolved: resolveSep0001RetryOptions(options), + label: "sep0001-merchant-lookup", + shouldRetry: isTransientSep0001Error, + sleep, + random, + onRetry: () => sep0001LookupRetriesTotal.inc({ result: "scheduled" }), + onRecovered: () => sep0001LookupRetriesTotal.inc({ result: "recovered" }), + onExhausted: () => sep0001LookupRetriesTotal.inc({ result: "exhausted" }), + }, + ); +} diff --git a/backend/src/lib/sep0001-retry.test.js b/backend/src/lib/sep0001-retry.test.js new file mode 100644 index 0000000..5d271cf --- /dev/null +++ b/backend/src/lib/sep0001-retry.test.js @@ -0,0 +1,151 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; + +vi.mock("./logger.js", () => ({ + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, +})); + +import { + DEFAULT_SEP0001_RETRY_OPTIONS, + MERCHANT_TOML_FIELDS, + fetchMerchantForStellarToml, + isTransientSep0001Error, + resolveSep0001RetryOptions, + sep0001LookupRetriesTotal, +} from "./sep0001-retry.js"; +import { register } from "./metrics.js"; + +const MERCHANT = { id: "m-1", business_name: "Acme", recipient: "GABC" }; +const fast = { sleep: async () => {}, baseDelayMs: 0 }; + +/** Supabase double: each from() call starts a new builder and takes the next result. */ +function fakeSupabase(results) { + const queue = [...results]; + const calls = []; + const supabase = { + from: vi.fn((table) => { + const call = { table, ops: [] }; + calls.push(call); + const builder = { + select: (fields) => (call.ops.push(["select", fields]), builder), + eq: (col, val) => (call.ops.push(["eq", col, val]), builder), + is: (col, val) => (call.ops.push(["is", col, val]), builder), + maybeSingle: async () => { + const next = queue.shift(); + if (next instanceof Error) throw next; + return next; + }, + }; + return builder; + }), + }; + return { supabase, calls }; +} + +async function retries(result) { + const { values } = await sep0001LookupRetriesTotal.get(); + return values.find((v) => v.labels.result === result)?.value ?? 0; +} + +describe("SEP-0001 merchant lookup retry (#1459)", () => { + beforeEach(() => vi.clearAllMocks()); + + it("queries the merchant row once on success", async () => { + const { supabase, calls } = fakeSupabase([{ data: MERCHANT, error: null }]); + + await expect(fetchMerchantForStellarToml(supabase, "m-1", fast)).resolves.toEqual({ + data: MERCHANT, + error: null, + }); + expect(calls).toHaveLength(1); + expect(calls[0]).toEqual({ + table: "merchants", + ops: [ + ["select", MERCHANT_TOML_FIELDS], + ["eq", "id", "m-1"], + ["is", "deleted_at", null], + ], + }); + }); + + it("retries a transient failure with a fresh query and records recovery", async () => { + const before = await retries("recovered"); + const { supabase, calls } = fakeSupabase([ + { data: null, error: { message: "TypeError: fetch failed", code: "" } }, + Object.assign(new Error("socket hang up"), { code: "ECONNRESET" }), + { data: MERCHANT, error: null }, + ]); + + const result = await fetchMerchantForStellarToml(supabase, "m-1", fast); + expect(result.data).toEqual(MERCHANT); + expect(calls).toHaveLength(3); + expect(await retries("recovered")).toBe(before + 1); + }); + + it("does not retry a deterministic error", async () => { + const { supabase, calls } = fakeSupabase([ + { data: null, error: { message: "invalid input syntax for type uuid", code: "22P02" } }, + ]); + + const result = await fetchMerchantForStellarToml(supabase, "not-a-uuid", fast); + expect(result.error.code).toBe("22P02"); + expect(calls).toHaveLength(1); + }); + + it("does not retry a missing merchant", async () => { + const { supabase, calls } = fakeSupabase([{ data: null, error: null }]); + await expect(fetchMerchantForStellarToml(supabase, "m-x", fast)).resolves.toEqual({ + data: null, + error: null, + }); + expect(calls).toHaveLength(1); + }); + + it("returns the last transient error after exhausting retries", async () => { + const before = await retries("exhausted"); + const outage = { data: null, error: { message: "connection failure", code: "08006" } }; + const { supabase, calls } = fakeSupabase([outage, outage, outage]); + + const result = await fetchMerchantForStellarToml(supabase, "m-1", { ...fast, maxAttempts: 3 }); + expect(result.error).toMatchObject({ code: "08006", retryAttempts: 3 }); + expect(calls).toHaveLength(3); + expect(await retries("exhausted")).toBe(before + 1); + }); + + it("exposes its counter on the main /metrics registry", async () => { + expect(await register.metrics()).toContain("sep0001_merchant_lookup_retries_total"); + }); +}); + +describe("resolveSep0001RetryOptions", () => { + it("defaults to a small request-path budget", () => { + expect(resolveSep0001RetryOptions({}, {})).toEqual(DEFAULT_SEP0001_RETRY_OPTIONS); + }); + + it("reads SEP0001_RETRY_* and clamps to 5 attempts / 5s", () => { + expect( + resolveSep0001RetryOptions( + {}, + { + SEP0001_RETRY_MAX_ATTEMPTS: "50", + SEP0001_RETRY_BASE_DELAY_MS: "10", + SEP0001_RETRY_MAX_DELAY_MS: "999999", + }, + ), + ).toEqual({ maxAttempts: 5, baseDelayMs: 10, maxDelayMs: 5000 }); + }); +}); + +describe("isTransientSep0001Error", () => { + it.each([ + [{ code: "08006" }, true], + [{ code: "57P01" }, true], + [{ message: "TypeError: fetch failed" }, true], + [{ status: 503 }, true], + [{ code: "22P02" }, false], + [{ code: "PGRST116" }, false], + [{ status: 404 }, false], + [null, false], + ])("%j -> %s", (err, expected) => { + expect(isTransientSep0001Error(err)).toBe(expected); + }); +}); diff --git a/backend/src/routes/payments.js b/backend/src/routes/payments.js index 53c4101..b4754d9 100644 --- a/backend/src/routes/payments.js +++ b/backend/src/routes/payments.js @@ -49,15 +49,14 @@ import { PAYMENT_SESSION_IN_PROGRESS, } from "../lib/payment-session-lock.js"; import { lookupIdempotentResponse } from "../lib/idempotency.js"; +// Payment Processor Telemetry: every label is sanitised and validated before +// it reaches Prometheus (issue #1477); reads retry with backoff (#1479). import { - paymentProcessorSessionsTotal, - paymentProcessorSessionDuration, - paymentProcessorVerificationsTotal, - paymentProcessorVerificationDuration, - paymentProcessorStatusCacheHits, - paymentProcessorStatusCacheMisses, - paymentProcessorRefundsTotal, -} from "../lib/payment-processor-metrics.js"; + recordSessionTelemetry, + recordVerificationTelemetry, + recordStatusCacheTelemetry, +} from "../lib/payment-processor-telemetry.js"; +import { retryProcessorRead } from "../lib/payment-processor-retry.js"; import { getExchangeRateQuote, NoPathFoundError, @@ -323,11 +322,7 @@ function createPaymentsRouter({ asset: assetLabel, reason: rejection.reason === "issuer_not_allowed" ? "invalid_issuer" : rejection.reason, }); - paymentProcessorSessionsTotal.inc({ asset: assetLabel, outcome: "validation_failed" }); - paymentProcessorSessionDuration.observe( - { asset: assetLabel, outcome: "validation_failed" }, - (Date.now() - sessionStart) / 1000, - ); + recordSessionTelemetry({ asset: assetLabel, outcome: "validation_failed", durationSeconds: (Date.now() - sessionStart) / 1000 }); return res.status(400).json({ error: rejection.message, ...(rejection.rule === "limits" ? rejection.details : {}), @@ -379,22 +374,14 @@ function createPaymentsRouter({ await insertPaymentSessionWithRetry(supabase, payload); } catch (insertError) { insertError.status = 500; - paymentProcessorSessionsTotal.inc({ asset: body.asset, outcome: "persistence_failed" }); - paymentProcessorSessionDuration.observe( - { asset: body.asset, outcome: "persistence_failed" }, - (Date.now() - sessionStart) / 1000, - ); + recordSessionTelemetry({ asset: body.asset, outcome: "persistence_failed", durationSeconds: (Date.now() - sessionStart) / 1000 }); throw insertError; } // Only record production metrics for non-sandbox payments. if (!isSandbox) { paymentCreatedCounter.inc({ asset: body.asset }); - paymentProcessorSessionsTotal.inc({ asset: body.asset, outcome: "created" }); - paymentProcessorSessionDuration.observe( - { asset: body.asset, outcome: "created" }, - (Date.now() - sessionStart) / 1000, - ); + recordSessionTelemetry({ asset: body.asset, outcome: "created", durationSeconds: (Date.now() - sessionStart) / 1000 }); } logger.info({ paymentId: paymentId }, "DEBUG: createSession success"); @@ -454,10 +441,10 @@ function createPaymentsRouter({ const redis = await connectRedisClient(); const cached = await getCachedPayment(redis, req.params.id, req.merchant?.id); if (cached) { - paymentProcessorStatusCacheHits.inc(); + recordStatusCacheTelemetry(true); return res.json({ payment: cached }); } - paymentProcessorStatusCacheMisses.inc(); + recordStatusCacheTelemetry(false); const baseFields = "id, merchant_id, amount, asset, asset_issuer, recipient, description, memo, memo_type, status, tx_id, metadata, created_at"; @@ -497,7 +484,9 @@ function createPaymentsRouter({ let queryError = null; let usedJoin = false; for (const attempt of attempts) { - const result = await runPaymentQuery(attempt); + // Transient read failures are retried with backoff before falling + // back to the next schema shape (issue #1479). + const result = await retryProcessorRead("status_lookup", () => runPaymentQuery(attempt)); if (!result.error) { data = result.data; usedJoin = result.includeJoin; @@ -612,29 +601,27 @@ function createPaymentsRouter({ const recordVerificationOutcome = (outcome, asset) => { if (verifyOutcomeRecorded) return; verifyOutcomeRecorded = true; - paymentProcessorVerificationsTotal.inc({ asset: asset || "unknown", outcome }); - paymentProcessorVerificationDuration.observe( - { asset: asset || "unknown", outcome }, - (Date.now() - verifyStart) / 1000, - ); + recordVerificationTelemetry({ asset: asset || "unknown", outcome, durationSeconds: (Date.now() - verifyStart) / 1000 }); }; try { const supabase = await getSupabaseClient(); - let query = supabase - .from("payments") - .select( - "id, merchant_id, amount, asset, asset_issuer, recipient, status, tx_id, memo, memo_type, webhook_url, metadata, created_at, merchants(webhook_secret, webhook_version, webhook_custom_headers, notification_email, email, business_name, subscribed_events)" - ); + // A fresh query per attempt: PostgREST builders are single-use. The + // lookup is a read, so transient failures are retried (issue #1479). + const buildVerifyQuery = () => { + let query = supabase + .from("payments") + .select( + "id, merchant_id, amount, asset, asset_issuer, recipient, status, tx_id, memo, memo_type, webhook_url, metadata, created_at, merchants(webhook_secret, webhook_version, webhook_custom_headers, notification_email, email, business_name, subscribed_events)" + ); - if (req.merchant?.id) { - query = query.eq("merchant_id", req.merchant.id); - } + if (req.merchant?.id) { + query = query.eq("merchant_id", req.merchant.id); + } - const { data, error } = await query + return query.eq("id", req.params.id).is("deleted_at", null).maybeSingle(); + }; - .eq("id", req.params.id) - .is("deleted_at", null) - .maybeSingle(); + const { data, error } = await retryProcessorRead("verify_lookup", buildVerifyQuery); console.log("DEBUG: Supabase query finished. data exists:", !!data, "error:", !!error); diff --git a/backend/src/routes/sep0001.js b/backend/src/routes/sep0001.js index 2343766..045698f 100644 --- a/backend/src/routes/sep0001.js +++ b/backend/src/routes/sep0001.js @@ -61,6 +61,11 @@ export function resolveMerchantId(req) { * description: Merchant not found * 500: * description: Failed to generate stellar.toml + * 503: + * description: > + * Merchant lookup kept failing with a transient error after + * automatic retries with exponential backoff (issue #1459). + * Includes a Retry-After header; never cached. */ router.get("/.well-known/stellar.toml", async (req, res, next) => { const startedAt = process.hrtime.bigint(); @@ -82,13 +87,9 @@ router.get("/.well-known/stellar.toml", async (req, res, next) => { } merchantId = resolved.merchantId; - // Fetch merchant data - const { data: merchant, error } = await supabase - .from("merchants") - .select("id, business_name, email, notification_email, recipient, branding_config") - .eq("id", merchantId) - .is("deleted_at", null) - .maybeSingle(); + // Fetch merchant data, retrying transient failures with exponential + // backoff (issue #1459). Deterministic errors come back on the first try. + const { data: merchant, error } = await fetchMerchantForStellarToml(supabase, merchantId); if (error) { finish("lookup_error"); diff --git a/backend/src/services/paymentService.js b/backend/src/services/paymentService.js index cb03438..192931c 100644 --- a/backend/src/services/paymentService.js +++ b/backend/src/services/paymentService.js @@ -31,6 +31,8 @@ import { logger } from "../lib/logger.js"; import { validatePaymentSession } from "../lib/payment-session-validator.js"; import { getSupabaseClient } from "../lib/supabase-client.js"; import { insertPaymentSessionWithRetry } from "../lib/payment-session-retry.js"; +// Payment Processor Telemetry: sanitised, validated recorders (issue #1477) +// and retried idempotent reads (#1479). import { paymentProcessorSessionsTotal, paymentProcessorSessionDuration, @@ -583,25 +585,25 @@ export const paymentService = { const redis = await connectRedisClient(); const cached = await getCachedPayment(redis, paymentId, merchantId); if (cached) { - paymentProcessorStatusCacheHits.inc(); + recordStatusCacheTelemetry(true); return { payment: cached }; } - paymentProcessorStatusCacheMisses.inc(); - - let query = supabase - .from("payments") - .select( - "id, amount, asset, asset_issuer, recipient, description, memo, memo_type, status, tx_id, metadata, created_at, merchants(branding_config)" - ); + recordStatusCacheTelemetry(false); + + // Fresh query per attempt; transient read failures retried (issue #1479). + const { data, error } = await retryProcessorRead("status_lookup", () => { + let query = supabase + .from("payments") + .select( + "id, amount, asset, asset_issuer, recipient, description, memo, memo_type, status, tx_id, metadata, created_at, merchants(branding_config)" + ); - if (merchantId) { - query = query.eq("merchant_id", merchantId); - } + if (merchantId) { + query = query.eq("merchant_id", merchantId); + } - const { data, error } = await query - .eq("id", paymentId) - .is("deleted_at", null) - .maybeSingle(); + return query.eq("id", paymentId).is("deleted_at", null).maybeSingle(); + }); if (error) { error.status = 500; @@ -633,27 +635,23 @@ export const paymentService = { async verifyPayment(paymentId, merchantId = null, io = null) { const verifyStart = Date.now(); const recordVerificationOutcome = (outcome) => { - paymentProcessorVerificationsTotal.inc({ asset: "unknown", outcome }); - paymentProcessorVerificationDuration.observe( - { asset: "unknown", outcome }, - (Date.now() - verifyStart) / 1000, - ); + recordVerificationTelemetry({ asset: "unknown", outcome, durationSeconds: (Date.now() - verifyStart) / 1000 }); }; const supabase = await getSupabaseClient(); - let query = supabase - .from("payments") - .select( - "id, merchant_id, amount, asset, asset_issuer, recipient, status, tx_id, memo, memo_type, webhook_url, created_at, merchants(webhook_secret, webhook_version, notification_email, email)" - ); + // Fresh query per attempt; transient read failures retried (issue #1479). + const { data, error } = await retryProcessorRead("verify_lookup", () => { + let query = supabase + .from("payments") + .select( + "id, merchant_id, amount, asset, asset_issuer, recipient, status, tx_id, memo, memo_type, webhook_url, created_at, merchants(webhook_secret, webhook_version, notification_email, email)" + ); - if (merchantId) { - query = query.eq("merchant_id", merchantId); - } + if (merchantId) { + query = query.eq("merchant_id", merchantId); + } - const { data, error } = await query - .eq("id", paymentId) - .is("deleted_at", null) - .maybeSingle(); + return query.eq("id", paymentId).is("deleted_at", null).maybeSingle(); + }); if (error) { error.status = 500; @@ -831,10 +829,10 @@ export const paymentService = { try { const result = await getMerchantPaymentsViaPool(merchantId, queryParams, page, limit, offset); - paymentProcessorListRequestsTotal.inc({ outcome: "success" }); + recordListRequestTelemetry({ outcome: "success" }); return result; } catch { - paymentProcessorListRequestsTotal.inc({ outcome: "pool_fallback" }); + recordListRequestTelemetry({ outcome: "pool_fallback" }); return getMerchantPaymentsViaSupabase(merchantId, queryParams, page, limit, offset); } }, @@ -868,14 +866,14 @@ export const paymentService = { } if (payment.status !== "confirmed") { - paymentProcessorRefundsTotal.inc({ stage: "generate", outcome: "rejected" }); + recordRefundTelemetry({ stage: "generate", outcome: "rejected" }); const err = new Error("Only confirmed payments can be refunded"); err.status = 400; throw err; } if (payment.metadata?.refund_status === "refunded") { - paymentProcessorRefundsTotal.inc({ stage: "generate", outcome: "rejected" }); + recordRefundTelemetry({ stage: "generate", outcome: "rejected" }); const err = new Error("Payment already refunded"); err.status = 400; throw err; @@ -921,7 +919,7 @@ export const paymentService = { }) .eq("id", payment.id); - paymentProcessorRefundsTotal.inc({ stage: "generate", outcome: "success" }); + recordRefundTelemetry({ stage: "generate", outcome: "success" }); return { xdr: refundTx.xdr, @@ -969,7 +967,7 @@ export const paymentService = { // generateRefundTx was never called for this payment (or predates // this field) — there is nothing to verify the submitted hash // against, so refuse rather than accept it on faith. - paymentProcessorRefundsTotal.inc({ stage: "confirm", outcome: "rejected" }); + recordRefundTelemetry({ stage: "confirm", outcome: "rejected" }); const err = new Error( "No pending refund found for this payment. Call the refund endpoint first.", ); @@ -978,7 +976,7 @@ export const paymentService = { } if (txHash.toLowerCase() !== expectedHash.toLowerCase()) { - paymentProcessorRefundsTotal.inc({ stage: "confirm", outcome: "rejected" }); + recordRefundTelemetry({ stage: "confirm", outcome: "rejected" }); const err = new Error( "Transaction hash does not match the refund transaction generated for this payment", ); @@ -1001,7 +999,7 @@ export const paymentService = { try { onChainTx = await server.transactions().transaction(txHash).call(); } catch (horizonErr) { - paymentProcessorRefundsTotal.inc({ stage: "confirm", outcome: "not_found" }); + recordRefundTelemetry({ stage: "confirm", outcome: "not_found" }); const err = new Error( "Transaction not found on Stellar network. Submit it before confirming.", ); @@ -1011,7 +1009,7 @@ export const paymentService = { } if (!onChainTx.successful) { - paymentProcessorRefundsTotal.inc({ stage: "confirm", outcome: "failed_on_chain" }); + recordRefundTelemetry({ stage: "confirm", outcome: "failed_on_chain" }); const err = new Error("Refund transaction failed on the Stellar network"); err.status = 400; throw err; @@ -1029,7 +1027,7 @@ export const paymentService = { }) .eq("id", payment.id); - paymentProcessorRefundsTotal.inc({ stage: "confirm", outcome: "success" }); + recordRefundTelemetry({ stage: "confirm", outcome: "success" }); return { message: "Refund confirmed successfully" }; },