diff --git a/backend/.env.example b/backend/.env.example index 6295ebb..ce745be 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -97,3 +97,18 @@ RESEND_API_KEY=your_resend_api_key # Sender address for receipt emails (must be a verified domain in Resend) # Defaults to: receipts@notifications.dripsnetwork.com RECEIPT_FROM_EMAIL=receipts@yourdomain.com + +# SEP-0001 stellar.toml merchant lookup retry (#1459; clamped 1-5 attempts, <=5000ms) +# SEP0001_RETRY_MAX_ATTEMPTS=3 +# SEP0001_RETRY_BASE_DELAY_MS=100 +# SEP0001_RETRY_MAX_DELAY_MS=1000 + +# Payment Processor Telemetry (#1477-#1479, docs/PAYMENT_PROCESSOR.md section 6) +# PAYMENT_PROCESSOR_TELEMETRY_MAX_ASSET_LABELS=50 +# PAYMENT_PROCESSOR_HEALTH_WINDOW_MS=300000 +# PAYMENT_PROCESSOR_HEALTH_MIN_SAMPLES=20 +# PAYMENT_PROCESSOR_DEGRADED_ERROR_RATIO=0.05 +# PAYMENT_PROCESSOR_UNHEALTHY_ERROR_RATIO=0.2 +# PAYMENT_PROCESSOR_RETRY_MAX_ATTEMPTS=3 +# PAYMENT_PROCESSOR_RETRY_BASE_DELAY_MS=50 +# PAYMENT_PROCESSOR_RETRY_MAX_DELAY_MS=500 diff --git a/backend/SEP0001_GENERATOR_GUIDE.md b/backend/SEP0001_GENERATOR_GUIDE.md index 42407dd..a73897b 100644 --- a/backend/SEP0001_GENERATOR_GUIDE.md +++ b/backend/SEP0001_GENERATOR_GUIDE.md @@ -262,6 +262,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 d39b214..f68b583 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). --- @@ -118,3 +120,105 @@ Include the printed percentile tables in PRs touching the processor. - Verification uses a DB-level conditional update so a `tx_hash` can only be claimed once even under concurrent confirmation attempts - Signature verification failure keeps payments pending (fail-closed) + +--- + +## 6. Payment Processor Telemetry (#1477, #1478, #1479) + +| File | Role | +|---|---| +| `src/lib/payment-processor-telemetry.js` | Sanitising, validating recorders; rolling-window health; new series | +| `src/lib/payment-processor-retry.js` | Retried idempotent reads (status / verification lookup) | +| `src/lib/backoff-retry.js` | Shared full-jitter backoff used by the processor and SEP-0001 | +| `docs/alerts/payment-processor.rules.yml` | Prometheus alert rules | + +### 6.1 Payload sanitization and strict validation (#1477) + +`routes/payments.js` and `services/paymentService.js` no longer call the +`payment_processor_*` metric objects directly. They call the recorders +(`recordSessionTelemetry`, `recordVerificationTelemetry`, +`recordRefundTelemetry`, `recordListRequestTelemetry`, +`recordStatusCacheTelemetry`, `recordProcessorRetryTelemetry`), which: + +| Input | Rule | Otherwise | +|---|---|---| +| payload | plain object | rejected (`invalid_payload`) | +| `outcome` / `stage` / `operation` | fixed allow-list per series | rejected (`invalid_outcome` / `invalid_stage` / `invalid_operation`) | +| `asset` | trimmed, upper-cased, `^[A-Z0-9]{1,12}$`; `native` becomes `XLM` | label `invalid`; missing becomes `unknown` | +| distinct assets | at most `PAYMENT_PROCESSOR_TELEMETRY_MAX_ASSET_LABELS` (default 50) | label `other` | +| duration | finite, `0` to `3600` seconds | count still recorded, observation dropped, `invalid_duration` | + +Before this, `req.body.asset` went straight into a label, so a client could +create a new Prometheus series on every request. Rejected payloads are never +written to the business series; they increment +`payment_processor_telemetry_rejected_total{metric, reason}` and log one +warning with the metric and reason only. Recorders never throw. + +Allow-lists: + +- sessions: created, validation_failed, persistence_failed, sandbox_skipped +- verifications: confirmed, already_confirmed, already_confirmed_concurrent, + pending_no_match, signature_invalid, underpayment, overpayment, + tx_claim_conflict, error +- refunds: stage generate or confirm; outcome success, rejected, not_found, + failed_on_chain, error +- list requests: success, pool_fallback, error + +### 6.2 Alert metrics and health (#1478) + +| Series | Type | Notes | +|---|---|---| +| `payment_processor_health_state` | gauge | 0 healthy, 1 degraded, 2 unhealthy; refreshed at scrape | +| `payment_processor_session_failure_ratio` | gauge | persistence_failed / (created + persistence_failed) | +| `payment_processor_verification_error_ratio` | gauge | error / all verifications | +| `payment_processor_refund_error_ratio` | gauge | (error + failed_on_chain) / all refund operations | +| `payment_processor_telemetry_rejection_ratio` | gauge | rejected / all telemetry events | +| `payment_processor_telemetry_rejected_total` | counter | `metric`, `reason` | +| `payment_processor_retries_total` | counter | `operation`, `result` (see 6.3) | + +Health uses a rolling window (`PAYMENT_PROCESSOR_HEALTH_WINDOW_MS`, default +5 min) and needs `PAYMENT_PROCESSOR_HEALTH_MIN_SAMPLES` (default 20) before a +ratio counts: + +- `unhealthy`: any error ratio >= `PAYMENT_PROCESSOR_UNHEALTHY_ERROR_RATIO` (0.2) +- `degraded`: any error ratio >= `PAYMENT_PROCESSOR_DEGRADED_ERROR_RATIO` + (0.05), telemetry rejections at that ratio, or any exhausted read retry +- `healthy`: otherwise + +Validation failures (bad client input) do not count as processor errors. + +`GET /health/payment-processor` returns the snapshot (503 only when +unhealthy). `GET /health` lists `services.payment_processor` for information +only. The body holds counts and ratios, never payment ids, amounts, addresses +or asset codes. + +### 6.3 Retry with exponential backoff (#1479) + +The first Supabase read of the status poll and of verification +(`status_lookup`, `verify_lookup`, in both the route and the service) is +retried with full jitter, `random(0, min(max, base * 2^n))`: + +| Variable | Default | Clamp | +|---|---|---| +| `PAYMENT_PROCESSOR_RETRY_MAX_ATTEMPTS` | 3 | 1-5 | +| `PAYMENT_PROCESSOR_RETRY_BASE_DELAY_MS` | 50 | 0-3000 | +| `PAYMENT_PROCESSOR_RETRY_MAX_DELAY_MS` | 500 | base-3000 | + +- Retried: network errors, 408/429/5xx, connection-class Postgres errors + (shared `isRetryableSessionError`). +- Not retried: not-found, invalid ids, missing columns (the status route still + falls through its schema-tolerant query shapes), PostgREST client errors. +- Writes (status transitions, tx claims, refunds) are never wrapped. +- Each query is rebuilt per attempt (PostgREST builders are single-use). +- After the last attempt the final `{ data, error }` is returned unchanged, so + response codes are as before. + +`payment_processor_retries_total{operation, result}` records `scheduled`, +`recovered` and `exhausted`; exhausted retries also degrade health. + +### 6.4 Tests + +```bash +npx vitest run src/lib/payment-processor-telemetry.test.js src/lib/payment-processor-retry.test.js src/lib/backoff-retry.test.js +npx vitest run tests/integration/payment-processor-telemetry.test.js tests/e2e/payment-processor.e2e.test.js +``` 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/app.js b/backend/src/app.js index b330d30..035f39f 100644 --- a/backend/src/app.js +++ b/backend/src/app.js @@ -49,6 +49,7 @@ import { versionDeprecationMiddleware } from "./lib/version-deprecation.js"; import oracleRouter from "./routes/oracle.js"; import { getPaymentSessionValidatorHealth } from "./lib/payment-session-validator.js"; import { getExchangeRateOracleHealth } from "./lib/exchange-rate-oracle-telemetry.js"; +import { getPaymentProcessorHealth } from "./lib/payment-processor-telemetry.js"; import { configureExchangeRateCoordination } from "./services/exchangeRateService.js"; export async function createApp({ redisClient }) { @@ -262,6 +263,7 @@ export async function createApp({ redisClient }) { // Informational only — does not affect `ok` / the status code. payment_session_validator: getPaymentSessionValidatorHealth().status, exchange_rate_oracle_cache: getExchangeRateOracleHealth().status, + payment_processor: getPaymentProcessorHealth().status, }, }); }); @@ -314,6 +316,31 @@ export async function createApp({ redisClient }) { res.status(health.status === "unhealthy" ? 503 : 200).json(health); }); + /** + * @swagger + * /health/payment-processor: + * get: + * summary: Payment Processor telemetry health + * description: > + * Rolling-window session, verification and refund outcome counts, + * error ratios, telemetry rejections and exhausted read retries for + * the payment processor (issue #1478). Returns 503 only when + * unhealthy (an error ratio above the unhealthy threshold); degraded + * still returns 200. The body contains no payment ids, amounts, + * addresses or asset codes. + * tags: [Health] + * security: [] + * responses: + * 200: + * description: Processor healthy or degraded + * 503: + * description: Processor unhealthy + */ + app.get("/health/payment-processor", (_req, res) => { + const health = getPaymentProcessorHealth(); + res.status(health.status === "unhealthy" ? 503 : 200).json(health); + }); + const verifyPaymentRateLimit = createVerifyPaymentRateLimit({ store: redisAvailable ? createRedisRateLimitStore({ client: redisClient }) : undefined, }); 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 847120e..3b0f371 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, @@ -322,11 +321,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 : {}), @@ -378,22 +373,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"); @@ -453,10 +440,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"; @@ -496,7 +483,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; @@ -611,29 +600,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 c1275b6..c765983 100644 --- a/backend/src/routes/sep0001.js +++ b/backend/src/routes/sep0001.js @@ -1,6 +1,14 @@ import express from "express"; import { supabase } from "../lib/supabase.js"; import { generateStellarToml, validateStellarToml } from "../lib/sep0001-generator.js"; +import { + fetchMerchantForStellarToml, + isTransientSep0001Error, +} from "../lib/sep0001-retry.js"; +import { logger } from "../lib/logger.js"; + +/** Seconds a client should wait before retrying after a transient failure. */ +const RETRY_AFTER_SECONDS = 5; const router = express.Router(); @@ -28,6 +36,11 @@ const router = express.Router(); * 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) => { try { @@ -42,16 +55,22 @@ router.get("/.well-known/stellar.toml", async (req, res, next) => { return res.status(400).json({ error: "merchant_id required" }); } - // 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) { - console.error("Failed to fetch merchant:", error); + const transient = isTransientSep0001Error(error); + logger.error( + { code: error.code, attempts: error.retryAttempts ?? 1, transient, err: error.message }, + "SEP-0001: failed to fetch merchant", + ); + // Error responses must never be cached by wallets or CDNs. + res.set("Cache-Control", "no-store"); + if (transient) { + res.set("Retry-After", String(RETRY_AFTER_SECONDS)); + return res.status(503).json({ error: "Merchant lookup temporarily unavailable" }); + } return res.status(500).json({ error: "Failed to fetch merchant" }); } diff --git a/backend/src/routes/sep0001.test.js b/backend/src/routes/sep0001.test.js new file mode 100644 index 0000000..5a5be74 --- /dev/null +++ b/backend/src/routes/sep0001.test.js @@ -0,0 +1,103 @@ +import express from "express"; +import request from "supertest"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { mockMaybeSingle } = vi.hoisted(() => { + // Keep route-level retries instant. + process.env.SEP0001_RETRY_BASE_DELAY_MS = "0"; + process.env.SEP0001_RETRY_MAX_DELAY_MS = "0"; + process.env.SEP0001_RETRY_MAX_ATTEMPTS = "3"; + return { mockMaybeSingle: vi.fn() }; +}); + +vi.mock("../lib/supabase.js", () => { + const builder = { + select: () => builder, + eq: () => builder, + is: () => builder, + maybeSingle: mockMaybeSingle, + }; + return { supabase: { from: () => builder } }; +}); + +vi.mock("../lib/logger.js", () => ({ + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, +})); + +import sep0001Router from "./sep0001.js"; + +const MERCHANT = { + id: "550e8400-e29b-41d4-a716-446655440000", + business_name: "Acme", + email: "a@acme.test", + notification_email: null, + recipient: "GBUQWP3BOUZX34ULNQG23RQ6F4YUSXHTQSXUSMIQSTBE2BRUY4DQAT2B", + branding_config: null, +}; + +function app() { + const a = express(); + a.use("/", sep0001Router); + return a; +} + +const get = () => request(app()).get(`/.well-known/stellar.toml?merchant_id=${MERCHANT.id}`); + +describe("GET /.well-known/stellar.toml retry (#1459)", () => { + beforeEach(() => mockMaybeSingle.mockReset()); + + it("serves stellar.toml after a transient lookup failure", async () => { + mockMaybeSingle + .mockResolvedValueOnce({ data: null, error: { message: "TypeError: fetch failed", code: "" } }) + .mockResolvedValueOnce({ data: MERCHANT, error: null }); + + const res = await get(); + expect(res.status).toBe(200); + expect(res.text).toContain("NETWORK_PASSPHRASE"); + expect(res.headers["cache-control"]).toBe("public, max-age=3600"); + expect(mockMaybeSingle).toHaveBeenCalledTimes(2); + }); + + it("answers 503 with Retry-After and no-store when the outage outlasts retries", async () => { + mockMaybeSingle.mockResolvedValue({ data: null, error: { message: "connection failure", code: "08006" } }); + + const res = await get(); + expect(res.status).toBe(503); + expect(res.headers["retry-after"]).toBe("5"); + expect(res.headers["cache-control"]).toBe("no-store"); + expect(res.body).toEqual({ error: "Merchant lookup temporarily unavailable" }); + expect(mockMaybeSingle).toHaveBeenCalledTimes(3); + }); + + it("keeps 500 for deterministic lookup errors, without retrying", async () => { + mockMaybeSingle.mockResolvedValue({ data: null, error: { message: "bad uuid", code: "22P02" } }); + + const res = await get(); + expect(res.status).toBe(500); + expect(res.headers["cache-control"]).toBe("no-store"); + expect(res.body).toEqual({ error: "Failed to fetch merchant" }); + expect(mockMaybeSingle).toHaveBeenCalledTimes(1); + }); + + it("keeps 404 for an unknown merchant", async () => { + mockMaybeSingle.mockResolvedValue({ data: null, error: null }); + const res = await get(); + expect(res.status).toBe(404); + expect(mockMaybeSingle).toHaveBeenCalledTimes(1); + }); + + it("keeps 400 when no merchant id is given", async () => { + const res = await request(app()).get("/.well-known/stellar.toml"); + expect(res.status).toBe(400); + expect(mockMaybeSingle).not.toHaveBeenCalled(); + }); + + it("never leaks the database error text to the client", async () => { + mockMaybeSingle.mockResolvedValue({ + data: null, + error: { message: "password authentication failed for user admin", code: "08006" }, + }); + const res = await get(); + expect(JSON.stringify(res.body)).not.toMatch(/password|admin/); + }); +}); diff --git a/backend/src/services/paymentService.js b/backend/src/services/paymentService.js index 611631d..c72d771 100644 --- a/backend/src/services/paymentService.js +++ b/backend/src/services/paymentService.js @@ -31,16 +31,16 @@ 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, - paymentProcessorVerificationsTotal, - paymentProcessorVerificationDuration, - paymentProcessorStatusCacheHits, - paymentProcessorStatusCacheMisses, - paymentProcessorRefundsTotal, - paymentProcessorListRequestsTotal, -} from "../lib/payment-processor-metrics.js"; + recordSessionTelemetry, + recordVerificationTelemetry, + recordStatusCacheTelemetry, + recordRefundTelemetry, + recordListRequestTelemetry, +} from "../lib/payment-processor-telemetry.js"; +import { retryProcessorRead } from "../lib/payment-processor-retry.js"; const SAFE_METADATA_KEY_RE = /^[a-zA-Z0-9_-]{1,64}$/; @@ -458,11 +458,7 @@ export const paymentService = { async createPaymentSession(merchant, body) { const sessionStart = Date.now(); const recordSessionOutcome = (outcome) => { - paymentProcessorSessionsTotal.inc({ asset: body?.asset, outcome }); - paymentProcessorSessionDuration.observe( - { asset: body?.asset, outcome }, - (Date.now() - sessionStart) / 1000, - ); + recordSessionTelemetry({ asset: body?.asset, outcome, durationSeconds: (Date.now() - sessionStart) / 1000 }); }; const supabase = await getSupabaseClient(); @@ -580,25 +576,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; @@ -630,27 +626,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; @@ -756,11 +748,7 @@ export const paymentService = { // Record metrics paymentConfirmedCounter.inc({ asset: data.asset }); paymentConfirmationLatency.observe({ asset: data.asset }, latencySeconds); - paymentProcessorVerificationsTotal.inc({ asset: data.asset, outcome: "confirmed" }); - paymentProcessorVerificationDuration.observe( - { asset: data.asset, outcome: "confirmed" }, - (Date.now() - verifyStart) / 1000, - ); + recordVerificationTelemetry({ asset: data.asset, outcome: "confirmed", durationSeconds: (Date.now() - verifyStart) / 1000 }); if (io && data.merchant_id) { io.to(`merchant:${data.merchant_id}`).emit("payment:confirmed", { @@ -828,10 +816,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); } }, @@ -865,14 +853,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; @@ -918,7 +906,7 @@ export const paymentService = { }) .eq("id", payment.id); - paymentProcessorRefundsTotal.inc({ stage: "generate", outcome: "success" }); + recordRefundTelemetry({ stage: "generate", outcome: "success" }); return { xdr: refundTx.xdr, @@ -966,7 +954,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.", ); @@ -975,7 +963,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", ); @@ -998,7 +986,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.", ); @@ -1008,7 +996,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; @@ -1026,7 +1014,7 @@ export const paymentService = { }) .eq("id", payment.id); - paymentProcessorRefundsTotal.inc({ stage: "confirm", outcome: "success" }); + recordRefundTelemetry({ stage: "confirm", outcome: "success" }); return { message: "Refund confirmed successfully" }; }, diff --git a/backend/tests/integration/payment-processor-telemetry.test.js b/backend/tests/integration/payment-processor-telemetry.test.js new file mode 100644 index 0000000..8d42373 --- /dev/null +++ b/backend/tests/integration/payment-processor-telemetry.test.js @@ -0,0 +1,160 @@ +/** + * Payment Processor Telemetry + SEP-0001 retry — HTTP integration + * (issues #1459, #1477, #1478, #1479). + * + * Runs the real Express app. Only the edges are mocked (Supabase, Horizon, + * Redis), as in tests/integration/health.test.js. + */ +import request from "supertest"; +import { describe, it, expect, beforeAll, beforeEach, afterAll, vi } from "vitest"; + +const { mockMaybeSingle } = vi.hoisted(() => { + process.env.SUPABASE_URL ||= "https://example.supabase.co"; + process.env.SUPABASE_SERVICE_ROLE_KEY ||= "test-service-role-key"; + process.env.DATABASE_URL ||= "postgresql://postgres:postgres@127.0.0.1:5432/postgres"; + process.env.SEP0001_RETRY_BASE_DELAY_MS = "0"; + process.env.SEP0001_RETRY_MAX_DELAY_MS = "0"; + return { mockMaybeSingle: vi.fn() }; +}); + +vi.mock("../../src/lib/stellar.js", () => ({ + findMatchingPayment: vi.fn(), + findAnyRecentPayment: vi.fn(), + findStrictReceivePaths: vi.fn(), + getNetworkFeeStats: vi.fn(), + isHorizonReachable: vi.fn(async () => true), + isValidAssetCode: vi.fn(() => true), + isValidStellarAccountId: vi.fn(() => true), + isValidStellarPublicKey: vi.fn(() => true), + validateMemo: vi.fn(() => ({ valid: true })), + verifyTransactionSignature: vi.fn(), + withHorizonRetry: vi.fn(), +})); + +vi.mock("../../src/lib/supabase.js", () => { + const builder = { + select: () => builder, + eq: () => builder, + is: () => builder, + limit: () => Promise.resolve({ data: [], error: null }), + maybeSingle: mockMaybeSingle, + }; + return { supabase: { from: () => builder } }; +}); + +import { createApp } from "../../src/app.js"; +import { closePool } from "../../src/lib/db.js"; +import { + recordProcessorRetryTelemetry, + recordSessionTelemetry, + recordVerificationTelemetry, + resetPaymentProcessorTelemetry, +} from "../../src/lib/payment-processor-telemetry.js"; + +const mockRedisClient = { + ping: vi.fn().mockResolvedValue("PONG"), + on: vi.fn(), + sendCommand: vi.fn().mockResolvedValue("mocked_hash"), +}; + +const MERCHANT_ID = "550e8400-e29b-41d4-a716-446655440000"; + +describe("Payment Processor Telemetry over HTTP (#1477, #1478, #1479)", () => { + let app; + + beforeAll(async () => { + ({ app } = await createApp({ redisClient: mockRedisClient })); + }); + + afterAll(async () => { + await closePool(); + }); + + beforeEach(() => { + resetPaymentProcessorTelemetry(); + mockMaybeSingle.mockReset(); + }); + + it("GET /health/payment-processor is healthy with no traffic", async () => { + const res = await request(app).get("/health/payment-processor"); + expect(res.status).toBe(200); + expect(res.body).toMatchObject({ status: "healthy", sessions: 0, verifications: 0 }); + }); + + it("GET /health/payment-processor returns 503 when persistence keeps failing", async () => { + for (let i = 0; i < 25; i += 1) { + recordSessionTelemetry({ asset: "XLM", outcome: "persistence_failed", durationSeconds: 0.01 }); + } + const res = await request(app).get("/health/payment-processor"); + expect(res.status).toBe(503); + expect(res.body).toMatchObject({ + status: "unhealthy", + reasons: ["session_failure_ratio_critical"], + }); + expect(JSON.stringify(res.body)).not.toContain("XLM"); + }); + + it("GET /health lists payment_processor without changing the overall status", async () => { + for (let i = 0; i < 25; i += 1) recordVerificationTelemetry({ asset: "XLM", outcome: "error" }); + const res = await request(app).get("/health"); + expect(res.status).toBe(200); + expect(res.body.services.payment_processor).toBe("unhealthy"); + }); + + it("GET /metrics exposes sanitised labels and the new alert series", async () => { + recordSessionTelemetry({ asset: 'x"} injected{y="', outcome: "created", durationSeconds: 0.1 }); + recordSessionTelemetry({ asset: "usdc", outcome: "created", durationSeconds: 0.1 }); + recordSessionTelemetry({ asset: "XLM", outcome: "not_an_outcome" }); + recordProcessorRetryTelemetry({ operation: "status_lookup", result: "recovered" }); + + const res = await request(app).get("/metrics"); + expect(res.status).toBe(200); + expect(res.text).toMatch(/payment_processor_sessions_total\{[^}]*asset="invalid"[^}]*\} \d+/); + expect(res.text).toMatch(/payment_processor_sessions_total\{[^}]*asset="USDC"[^}]*\} \d+/); + expect(res.text).not.toContain("injected"); + expect(res.text).not.toContain('outcome="not_an_outcome"'); + expect(res.text).toMatch( + /payment_processor_telemetry_rejected_total\{[^}]*metric="sessions"[^}]*reason="invalid_outcome"[^}]*\} \d+/, + ); + expect(res.text).toMatch(/payment_processor_retries_total\{[^}]*operation="status_lookup"[^}]*\} \d+/); + expect(res.text).toContain("payment_processor_health_state"); + expect(res.text).toContain("sep0001_merchant_lookup_retries_total"); + }); +}); + +describe("SEP-0001 stellar.toml through the app (#1459)", () => { + let app; + + beforeAll(async () => { + ({ app } = await createApp({ redisClient: mockRedisClient })); + }); + + beforeEach(() => mockMaybeSingle.mockReset()); + + it("recovers from a transient merchant lookup failure", async () => { + mockMaybeSingle + .mockResolvedValueOnce({ data: null, error: { message: "TypeError: fetch failed", code: "" } }) + .mockResolvedValueOnce({ + data: { id: MERCHANT_ID, business_name: "Acme", recipient: null, branding_config: null }, + error: null, + }); + + const res = await request(app).get(`/.well-known/stellar.toml?merchant_id=${MERCHANT_ID}`); + expect(res.status).toBe(200); + expect(res.text).toContain('name = "Acme"'); + }); + + it("returns 503 + Retry-After when the lookup stays down", async () => { + mockMaybeSingle.mockResolvedValue({ data: null, error: { message: "down", code: "57P01" } }); + + const res = await request(app).get(`/.well-known/stellar.toml?merchant_id=${MERCHANT_ID}`); + expect(res.status).toBe(503); + expect(res.headers["retry-after"]).toBe("5"); + expect(res.headers["cache-control"]).toBe("no-store"); + + const metrics = await request(app).get("/metrics"); + expect(metrics.text).toMatch( + /sep0001_merchant_lookup_retries_total\{[^}]*result="exhausted"[^}]*\} [1-9]/, + ); + }); +});