diff --git a/backend/docs/LOG_RETENTION.md b/backend/docs/LOG_RETENTION.md new file mode 100644 index 00000000..f542b0b7 --- /dev/null +++ b/backend/docs/LOG_RETENTION.md @@ -0,0 +1,45 @@ +# Log Retention & Archival Service + +`src/lib/log-retention.js` purges operational webhook delivery logs +(`webhook_delivery_logs`) older than the retention window, in batches. Run it +with `npm run purge:webhook-logs` (see `scripts/purge-webhook-logs.js` and +`.github/workflows/webhook-log-retention.yml`). + +> Compliance: confirm the table holds operational delivery telemetry only. +> Payment audit records may need longer retention by policy or regulation. + +## Configuration + +| Option | Environment variable | Default | Allowed | +|---|---|---|---| +| `retentionDays` | `LOG_RETENTION_DAYS` | 30 | 1-3650 | +| `batchSize` | `LOG_PURGE_BATCH_SIZE` | 1000 | 1-50000 | +| `maxDurationMs` | `LOG_PURGE_MAX_DURATION_MS` | 300000 | 1000-86400000 | +| `maxBatches` | `LOG_PURGE_MAX_BATCHES` | 100000 | 1-10000000 | + +Explicit options win over environment variables. `logger` (an object with +`log` and `error`) may be passed instead of `console`. + +## Strict validation and sanitization (#1472) + +- Numbers must be plain base-10 integers in range. Values such as `30days`, + `1e3`, `-5`, `0x10` or `1.5` are rejected with a + `LogRetentionValidationError` naming the option or environment variable. + Before this change `parseInt` silently turned `30days` into `30`. +- Unknown option keys (for example a typo like `retentiondays`) are rejected + rather than silently falling back to the default retention. +- `pool` must expose `query()`. Everything is validated before the first + query runs. +- Only allow-listed table/column pairs can be purged, and identifiers are + checked against `^[a-z_][a-z0-9_]{0,62}$` before they reach SQL. The + retention period and batch size are always bind parameters. +- Each batch's `rowCount` must be an integer between 0 and `batchSize`; a + bad value stops the purge. `maxBatches` caps the loop so a misbehaving + driver cannot spin forever. +- The returned summary is a frozen object with `tableName`, `retentionDays`, + `batchSize`, `totalDeleted`, `durationMs` and `batches` only. +- `sanitizeLogMessage()` masks credentials (`scheme://user:pass@`, + `password=`, `token=`, `Bearer ...`), strips control characters (log + injection) and caps length. The purge script logs only the scrubbed message + and the error's name/code, never the raw error object, which can echo the + database connection string. diff --git a/backend/docs/PAYMENT_PROCESSOR.md b/backend/docs/PAYMENT_PROCESSOR.md index d39b2142..876d360e 100644 --- a/backend/docs/PAYMENT_PROCESSOR.md +++ b/backend/docs/PAYMENT_PROCESSOR.md @@ -110,7 +110,35 @@ Include the printed percentile tables in PRs touching the processor. --- -## 5. Security notes +## 5. Telemetry integration and stress suite (#1481) + +| Suite | File | Run | +|---|---|---| +| Integration | `tests/integration/payment-processor-telemetry.test.js` | `npx vitest run tests/integration/payment-processor-telemetry.test.js` | +| Stress | `load-tests/payment-processor-telemetry-stress.test.js` | `npm run test:load -- payment-processor-telemetry` | + +The integration suite scrapes the real `/metrics` route and parses the +exposition text a Prometheus server would receive. It checks that every +`payment_processor_*` series is merged into the scrape, that counts are +exact per asset and outcome, that histograms are consistent (cumulative +buckets, `+Inf` equals `_count`, `_sum` matches), that concurrent scrapes +during writes never fail, and that the output always parses. + +The stress suite checks there are no lost increments (100 concurrent workers +x 1,000 events), that cardinality stays bounded under 50,000 adversarial +asset values, that histogram count and sum stay exact over 100,000 +observations, and that scrape p95 latency stays under 100 ms while writes +continue. + +**Fix found by the suite:** session creation used `body.asset` straight from +the request as the `asset` label, so a client could create one time series +per arbitrary string (including on `validation_failed`). Labels now go +through `toAssetLabel()`: well-formed asset codes are upper-cased, anything +else becomes `invalid`, and a missing value becomes `unknown`. + +--- + +## 6. Security notes - API keys are hashed for rate-limit bucketing (`getCreatePaymentRateLimitKey`) - Metadata keys must match `^[a-zA-Z0-9_-]{1,64}$`; unsafe keys never reach SQL @@ -118,3 +146,4 @@ 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) +- Metric labels never carry raw client input; `asset` is bounded by `toAssetLabel()` diff --git a/backend/docs/WEBSOCKET_RELAY.md b/backend/docs/WEBSOCKET_RELAY.md new file mode 100644 index 00000000..a552d482 --- /dev/null +++ b/backend/docs/WEBSOCKET_RELAY.md @@ -0,0 +1,101 @@ +# WebSocket Relay Service + +Covers issues **#1454** (automated retry with exponential backoff) and +**#1453** (Prometheus alert metrics and health telemetry). + +## 1. Layout + +| File | Role | +|---|---| +| `src/lib/websocket-relay-security.js` | Origin, size, JWT and message-shape checks | +| `src/lib/websocket-relay-recovery.js` | Circuit breaker, dead letter queue, `reconnectWithBackoff` | +| `src/lib/websocket-relay-retry.js` | Retry policy, `deliverRelayMessage`, `redriveDeadLetters` (#1454) | +| `src/lib/websocket-relay-metrics.js` | Granular relay counters (#1083) | +| `src/lib/websocket-relay-telemetry.js` | Alert metrics and rolling-window health (#1453) | +| `docs/alerts/websocket-relay.rules.yml` | Prometheus alert rules | + +## 2. Retry with exponential backoff (#1454) + +```js +import { deliverRelayMessage, redriveDeadLetters, withRelayRetry } from "./lib/websocket-relay-retry.js"; +import { CircuitBreaker, DeadLetterQueue } from "./lib/websocket-relay-recovery.js"; +import { attachRelayHealthSource } from "./lib/websocket-relay-telemetry.js"; + +const circuitBreaker = new CircuitBreaker(); +const dlq = new DeadLetterQueue(); +attachRelayHealthSource({ circuitBreaker, dlq }); + +// Deliver, retrying transient failures, dead-lettering what still fails. +await deliverRelayMessage(message, (msg) => socket.sendAsync(msg), { circuitBreaker, dlq }); + +// Reconnect with the same policy. +await withRelayRetry(() => connect(), { operation: "reconnect", signal }); + +// Later, once the downstream recovers: +await redriveDeadLetters(dlq, (msg) => socket.sendAsync(msg), { limit: 100 }); +``` + +- **Delay:** full jitter, `random(0, min(maxDelayMs, baseDelayMs * 2^n))`. +- **Retried:** network error codes (`ECONNRESET`, `ETIMEDOUT`, ...), timeouts, + HTTP 408/429/5xx (not 501), WebSocket close codes 1006/1011/1012/1013, and + errors flagged `retryable: true`. +- **Not retried:** an OPEN circuit breaker (`circuitOpen`), other 4xx + (validation, signature, auth), policy close codes, `retryable: false`, + and programming errors. These fail or dead-letter immediately. +- **Bounds:** `maxAttempts`, a total `maxElapsedMs` budget including waits, + and an optional `AbortSignal`. +- **Failures** carry `retryAttempts`. With a `dlq`, `deliverRelayMessage` + dead-letters the message (with the total retry count) instead of throwing. +- **Logs** contain the operation label, attempt counts, delay and the + error's code/status/close code, with the message truncated to 200 + characters. Payloads are never logged. + +| Variable | Default | Range | +|---|---|---| +| `RELAY_RETRY_MAX_ATTEMPTS` | `5` | 1-10 | +| `RELAY_RETRY_BASE_DELAY_MS` | `250` | 0-60000 | +| `RELAY_RETRY_MAX_DELAY_MS` | `10000` | base-60000 | +| `RELAY_RETRY_MAX_ELAPSED_MS` | `30000` | 0-300000 | + +`reconnectWithBackoff` in `websocket-relay-recovery.js` is unchanged. + +## 3. Alert metrics and health (#1453) + +Exposed on `GET /metrics` (merged from their own registry, together with the +existing `relay_*` series, which were not previously scraped). + +| Metric | Type | Labels | +|---|---|---| +| `websocket_relay_deliveries_total` | counter | `outcome`: delivered, recovered, dead_lettered, failed | +| `websocket_relay_retries_total` | counter | `operation`: deliver, reconnect, redrive, other; `result`: scheduled, recovered, exhausted, aborted | +| `websocket_relay_retry_delay_seconds` | histogram | `operation` | +| `websocket_relay_health_state` | gauge | 0 healthy, 1 degraded, 2 unhealthy | +| `websocket_relay_delivery_failure_ratio` | gauge | rolling window | +| `websocket_relay_dlq_depth` | gauge | live DLQ size | +| `websocket_relay_circuit_open` | gauge | 1 OPEN, 0.5 HALF_OPEN, 0 CLOSED | +| `websocket_relay_last_delivery_timestamp_seconds` | gauge | | + +Gauges are computed at scrape time from a fixed ring of 5-second buckets, so +they decay when traffic stops and memory stays constant. + +`GET /health/websocket-relay` returns the snapshot (no payloads or payment +ids). Status rules: + +| Status | When | +|---|---| +| unhealthy (HTTP 503) | circuit OPEN, failure ratio >= `RELAY_FAILURE_RATIO_THRESHOLD` (default 0.1) with at least `RELAY_HEALTH_MIN_SAMPLES` (20) deliveries, or DLQ >= `RELAY_DLQ_UNHEALTHY_THRESHOLD` (100) | +| degraded (HTTP 200) | circuit HALF_OPEN, DLQ >= `RELAY_DLQ_DEGRADED_THRESHOLD` (1), or retries exhausted in the window | +| healthy | otherwise | + +The window is `RELAY_HEALTH_WINDOW_MS` (default 5 minutes). `GET /health` +also reports `services.websocket_relay` (informational; it does not change +the overall status code). + +## 4. Security notes + +- Label values come from fixed server-side sets; message fields, origins, + payment ids and error text never become labels. +- The health snapshot and logs contain counts, states and error codes only. +- An OPEN breaker is never retried, so retries cannot defeat it. +- Retry configuration is clamped, so a misconfigured environment cannot + disable backoff or make a delivery wait for minutes. diff --git a/backend/docs/alerts/websocket-relay.rules.yml b/backend/docs/alerts/websocket-relay.rules.yml new file mode 100644 index 00000000..bd0d9c51 --- /dev/null +++ b/backend/docs/alerts/websocket-relay.rules.yml @@ -0,0 +1,83 @@ +# Prometheus alerting rules for the WebSocket Relay Service (issue #1453). +# +# Load with: rule_files: ["docs/alerts/websocket-relay.rules.yml"] +# Validate: promtool check rules docs/alerts/websocket-relay.rules.yml +# +# Metric reference: docs/WEBSOCKET_RELAY.md + +groups: + - name: websocket-relay + rules: + - alert: WebSocketRelayUnhealthy + expr: max(websocket_relay_health_state) >= 2 + for: 5m + labels: + severity: critical + annotations: + summary: WebSocket relay reports unhealthy + description: > + The circuit breaker is open, the delivery failure ratio is above + threshold, or the dead letter queue has a large backlog. See + GET /health/websocket-relay for the reasons. + + - alert: WebSocketRelayCircuitOpen + expr: max(websocket_relay_circuit_open) >= 1 + for: 2m + labels: + severity: critical + annotations: + summary: WebSocket relay circuit breaker is open + description: > + Deliveries are being rejected without being attempted and go + straight to the dead letter queue. + + - alert: WebSocketRelayHighFailureRatio + expr: | + sum(rate(websocket_relay_deliveries_total{outcome=~"failed|dead_lettered"}[10m])) + / + clamp_min(sum(rate(websocket_relay_deliveries_total[10m])), 1e-9) + > 0.1 + and sum(increase(websocket_relay_deliveries_total[10m])) >= 20 + for: 10m + labels: + severity: critical + annotations: + summary: More than 10% of WebSocket relay deliveries are failing + description: > + Check logs for "WebSocket relay operation failed after exhausting + retries". + + - alert: WebSocketRelayRetriesExhausted + expr: sum(increase(websocket_relay_retries_total{result="exhausted"}[10m])) > 5 + for: 5m + labels: + severity: warning + annotations: + summary: WebSocket relay retries are being exhausted + description: > + Transient failures are lasting longer than the retry budget + (RELAY_RETRY_MAX_ATTEMPTS / RELAY_RETRY_MAX_ELAPSED_MS). + + - alert: WebSocketRelayDeadLetterBacklog + expr: max(websocket_relay_dlq_depth) > 0 + for: 15m + labels: + severity: warning + annotations: + summary: WebSocket relay dead letter queue is not draining + description: > + Messages have been waiting in the DLQ for 15 minutes. Run + redriveDeadLetters() once the downstream recovers. + + - alert: WebSocketRelayNoDeliveries + expr: | + time() - max(websocket_relay_last_delivery_timestamp_seconds) > 1800 + and sum(increase(websocket_relay_deliveries_total[30m])) > 0 + for: 5m + labels: + severity: warning + annotations: + summary: WebSocket relay has attempted deliveries but none succeeded for 30 minutes + description: > + Deliveries are being attempted but none succeed. Check the + relay_errors_total breakdown and the circuit breaker state. diff --git a/backend/load-tests/payment-processor-telemetry-stress.test.js b/backend/load-tests/payment-processor-telemetry-stress.test.js new file mode 100644 index 00000000..27db5221 --- /dev/null +++ b/backend/load-tests/payment-processor-telemetry-stress.test.js @@ -0,0 +1,160 @@ +/** + * Payment Processor Telemetry — stress suite (issue #1481). + * + * Pushes the payment-processor metric series far past normal volume and + * checks the invariants monitoring relies on: + * - no lost increments under heavy concurrent writes + * - series cardinality stays bounded under adversarial asset values + * - histogram sums and counts stay exact at volume + * - scrape latency stays bounded while writes continue + * + * Run with: npm run test:load -- payment-processor-telemetry + */ +import { describe, it, expect, beforeEach } from "vitest"; +import { + paymentProcessorRefundsTotal, + paymentProcessorRegister, + paymentProcessorSessionDuration, + paymentProcessorSessionsTotal, + paymentProcessorVerificationDuration, + paymentProcessorVerificationsTotal, + toAssetLabel, +} from "../src/lib/payment-processor-metrics.js"; + +const SESSION_OUTCOMES = ["created", "validation_failed", "persistence_failed", "sandbox_skipped"]; +const VERIFY_OUTCOMES = [ + "confirmed", + "already_confirmed", + "pending_no_match", + "signature_invalid", + "underpayment", + "overpayment", + "tx_claim_conflict", + "error", +]; + +/** Seeded PRNG so runs are reproducible. */ +function mulberry32(seed) { + let a = seed; + return () => { + a |= 0; + a = (a + 0x6d2b79f5) | 0; + let t = Math.imul(a ^ (a >>> 15), 1 | a); + t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t; + return ((t ^ (t >>> 14)) >>> 0) / 4294967296; + }; +} + +const yieldToLoop = () => new Promise((resolve) => setImmediate(resolve)); + +async function seriesCount(metric) { + return (await metric.get()).values.length; +} + +async function total(metric, filter = () => true) { + return (await metric.get()).values.filter((v) => filter(v.labels)).reduce((s, v) => s + v.value, 0); +} + +describe("Payment Processor Telemetry — stress (#1481)", () => { + beforeEach(() => { + paymentProcessorSessionsTotal.reset(); + paymentProcessorSessionDuration.reset(); + paymentProcessorVerificationsTotal.reset(); + paymentProcessorVerificationDuration.reset(); + paymentProcessorRefundsTotal.reset(); + }); + + it("loses no increments across 100 concurrent workers x 1,000 events", async () => { + const random = mulberry32(1481); + const expected = new Map(); + + await Promise.all( + Array.from({ length: 100 }, async () => { + for (let i = 0; i < 1_000; i += 1) { + const outcome = SESSION_OUTCOMES[Math.floor(random() * SESSION_OUTCOMES.length)]; + paymentProcessorSessionsTotal.inc({ asset: "XLM", outcome }); + expected.set(outcome, (expected.get(outcome) ?? 0) + 1); + if (i % 250 === 0) await yieldToLoop(); + } + }), + ); + + expect(await total(paymentProcessorSessionsTotal)).toBe(100_000); + for (const [outcome, count] of expected) { + expect(await total(paymentProcessorSessionsTotal, (l) => l.outcome === outcome)).toBe(count); + } + }); + + it("keeps cardinality bounded under 50,000 adversarial asset values", async () => { + const random = mulberry32(7); + for (let i = 0; i < 50_000; i += 1) { + const raw = `${Math.floor(random() * 1e12).toString(36)}-${i}`; // always invalid (contains "-") + const asset = toAssetLabel(i % 10 === 0 ? "XLM" : raw); + paymentProcessorSessionsTotal.inc({ asset, outcome: "validation_failed" }); + } + + // XLM + invalid, one outcome each: two series, not 50,000. + expect(await seriesCount(paymentProcessorSessionsTotal)).toBe(2); + expect(await total(paymentProcessorSessionsTotal)).toBe(50_000); + }); + + it("caps the label space at assets x outcomes even with every valid code", async () => { + const codes = ["XLM", "USDC", "EURC", "BTC", "ETH", "AQUA", "YXLM", "SHX"]; + for (let i = 0; i < 20_000; i += 1) { + const outcome = VERIFY_OUTCOMES[i % VERIFY_OUTCOMES.length]; + paymentProcessorVerificationsTotal.inc({ asset: toAssetLabel(codes[i % codes.length]), outcome }); + } + expect(await seriesCount(paymentProcessorVerificationsTotal)).toBeLessThanOrEqual(codes.length * VERIFY_OUTCOMES.length); + }); + + it("keeps histogram count and sum exact at volume", async () => { + const random = mulberry32(99); + let expectedSum = 0; + const n = 100_000; + for (let i = 0; i < n; i += 1) { + const seconds = Math.round(random() * 10_000) / 1_000; // millisecond resolution, 0-10s + expectedSum += seconds; + paymentProcessorVerificationDuration.observe({ asset: "XLM", outcome: "confirmed" }, seconds); + } + + const { values } = await paymentProcessorVerificationDuration.get(); + const count = values.find((v) => v.metricName === "payment_processor_verification_duration_seconds_count").value; + const sum = values.find((v) => v.metricName === "payment_processor_verification_duration_seconds_sum").value; + const inf = values.find((v) => v.metricName.endsWith("_bucket") && v.labels.le === "+Inf").value; + expect(count).toBe(n); + expect(inf).toBe(n); + expect(sum).toBeCloseTo(expectedSum, 3); + }); + + it("serves scrapes quickly while writes continue", async () => { + for (const outcome of SESSION_OUTCOMES) { + paymentProcessorSessionDuration.observe({ asset: "XLM", outcome }, 0.01); + } + + let writing = true; + const writer = (async () => { + let i = 0; + while (writing) { + paymentProcessorRefundsTotal.inc({ stage: i % 2 ? "generate" : "confirm", outcome: "success" }); + i += 1; + if (i % 500 === 0) await yieldToLoop(); + } + })(); + + const latencies = []; + for (let i = 0; i < 50; i += 1) { + const started = performance.now(); + const text = await paymentProcessorRegister.metrics(); + latencies.push(performance.now() - started); + expect(text).toContain("payment_processor_refunds_total"); + await yieldToLoop(); + } + writing = false; + await writer; + + latencies.sort((a, b) => a - b); + const p95 = latencies[Math.floor(latencies.length * 0.95)]; + console.log(`[telemetry-stress] scrape p95=${p95.toFixed(2)}ms max=${latencies.at(-1).toFixed(2)}ms`); + expect(p95).toBeLessThan(100); + }); +}); diff --git a/backend/scripts/purge-webhook-logs.js b/backend/scripts/purge-webhook-logs.js index e3afad7d..e6428a2c 100644 --- a/backend/scripts/purge-webhook-logs.js +++ b/backend/scripts/purge-webhook-logs.js @@ -1,6 +1,6 @@ import "dotenv/config"; import { pool, closePool } from "../src/lib/db.js"; -import { purgeWebhookLogs } from "../src/lib/log-retention.js"; +import { purgeWebhookLogs, sanitizeLogMessage } from "../src/lib/log-retention.js"; async function run() { const startedAt = Date.now(); @@ -12,9 +12,11 @@ async function run() { ); } catch (error) { const elapsedMs = Date.now() - startedAt; + // Driver errors can echo the connection string; log a scrubbed message + // and the error name/code only, never the raw error object (#1472). console.error( - `[ALERT][log-retention] Run failed after ${elapsedMs}ms: ${error.message}`, - error, + `[ALERT][log-retention] Run failed after ${elapsedMs}ms: ${sanitizeLogMessage(error)}`, + { name: error?.name, code: error?.code, field: error?.field }, ); process.exitCode = 1; } finally { diff --git a/backend/src/app.js b/backend/src/app.js index b330d308..c244b7bc 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 { getWebSocketRelayHealth } from "./lib/websocket-relay-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, + websocket_relay: getWebSocketRelayHealth().status, }, }); }); @@ -314,6 +316,31 @@ export async function createApp({ redisClient }) { res.status(health.status === "unhealthy" ? 503 : 200).json(health); }); + /** + * @swagger + * /health/websocket-relay: + * get: + * summary: WebSocket Relay Service health telemetry + * description: > + * Rolling-window delivery counts, failure ratio, retry exhaustion, + * circuit breaker state and dead letter queue depth for the WebSocket + * relay (issue #1453). Returns 503 only when the relay is unhealthy + * (circuit open, failure ratio above threshold or a large DLQ + * backlog). A degraded status still returns 200. The body contains + * no message payloads or payment ids. + * tags: [Health] + * security: [] + * responses: + * 200: + * description: Relay healthy or degraded + * 503: + * description: Relay unhealthy + */ + app.get("/health/websocket-relay", (_req, res) => { + const health = getWebSocketRelayHealth(); + res.status(health.status === "unhealthy" ? 503 : 200).json(health); + }); + const verifyPaymentRateLimit = createVerifyPaymentRateLimit({ store: redisAvailable ? createRedisRateLimitStore({ client: redisClient }) : undefined, }); diff --git a/backend/src/lib/log-retention-validation.test.js b/backend/src/lib/log-retention-validation.test.js new file mode 100644 index 00000000..f974e1a3 --- /dev/null +++ b/backend/src/lib/log-retention-validation.test.js @@ -0,0 +1,220 @@ +import { describe, expect, it, vi } from "vitest"; +import { + DEFAULT_BATCH_SIZE, + DEFAULT_RETENTION_DAYS, + LogRetentionValidationError, + assertPurgeTarget, + parsePositiveInteger, + purgeWebhookLogs, + sanitizeLogMessage, + validatePurgeOptions, +} from "./log-retention.js"; + +/** A pool whose safety checks pass and whose deletes return the given counts. */ +function poolWithDeletes(...counts) { + const query = vi + .fn() + .mockResolvedValueOnce({ rowCount: 1, rows: [{}] }) + .mockResolvedValueOnce({ rowCount: 0, rows: [] }); + for (const count of counts) { + query.mockResolvedValueOnce({ rowCount: count, rows: [] }); + } + return { query }; +} + +const quietLogger = () => ({ log: vi.fn(), error: vi.fn() }); + +describe("parsePositiveInteger (issue #1472)", () => { + it.each([ + ["30", 30], + [" 7 ", 7], + [45, 45], + ])("accepts %j", (value, expected) => { + expect(parsePositiveInteger(value, 1)).toBe(expected); + }); + + it.each([undefined, null, ""])("returns the fallback for %j", (value) => { + expect(parsePositiveInteger(value, 12)).toBe(12); + }); + + it.each(["30days", "1e3", "-5", "0", "0x10", "1.5", " ", "9".repeat(20), 1.5, -1, 0, NaN, Infinity, true, {}, []])( + "rejects %j instead of coercing it", + (value) => { + expect(() => parsePositiveInteger(value, 1, { name: "LOG_RETENTION_DAYS" })).toThrow(LogRetentionValidationError); + }, + ); + + it("enforces bounds and names the setting", () => { + expect(() => parsePositiveInteger("4000", 1, { name: "retentionDays", min: 1, max: 3650 })).toThrow( + "retentionDays must be between 1 and 3650, got 4000", + ); + }); + + it("does not echo control characters or long input back", () => { + let message = ""; + try { + parsePositiveInteger(`12\n[ALERT] forged log line ${"x".repeat(100)}`, 1, { name: "x" }); + } catch (err) { + message = err.message; + } + expect(message).not.toContain("\n"); + expect(message.length).toBeLessThan(120); + }); +}); + +describe("validatePurgeOptions", () => { + it("applies defaults and environment variables", () => { + const pool = { query: vi.fn() }; + expect(validatePurgeOptions({ pool }, {})).toMatchObject({ + retentionDays: DEFAULT_RETENTION_DAYS, + batchSize: DEFAULT_BATCH_SIZE, + }); + expect(validatePurgeOptions({ pool }, { LOG_RETENTION_DAYS: "90", LOG_PURGE_BATCH_SIZE: "500" })).toMatchObject({ + retentionDays: 90, + batchSize: 500, + }); + }); + + it("lets explicit options override the environment", () => { + const opts = validatePurgeOptions({ pool: { query: vi.fn() }, retentionDays: 14 }, { LOG_RETENTION_DAYS: "90" }); + expect(opts.retentionDays).toBe(14); + }); + + it("names the environment variable that is invalid", () => { + expect(() => validatePurgeOptions({ pool: { query: vi.fn() } }, { LOG_PURGE_BATCH_SIZE: "lots" })).toThrow( + /LOG_PURGE_BATCH_SIZE/, + ); + }); + + it.each([ + [{ retentionDays: 0 }, "retentionDays"], + [{ retentionDays: 5000 }, "retentionDays"], + [{ batchSize: 100_000 }, "batchSize"], + [{ maxDurationMs: 10 }, "maxDurationMs"], + [{ maxBatches: -1 }, "maxBatches"], + ])("rejects out-of-range option %j", (extra, field) => { + let caught; + try { + validatePurgeOptions({ pool: { query: vi.fn() }, ...extra }, {}); + } catch (err) { + caught = err; + } + expect(caught).toBeInstanceOf(LogRetentionValidationError); + expect(caught.field).toBe(field); + }); + + it("rejects unknown option keys", () => { + expect(() => validatePurgeOptions({ pool: { query: vi.fn() }, retentiondays: 7 }, {})).toThrow(/Unknown purgeWebhookLogs option/); + }); + + it("rejects non-object options and invalid pools or loggers", () => { + expect(() => validatePurgeOptions(null, {})).toThrow(LogRetentionValidationError); + expect(() => validatePurgeOptions([], {})).toThrow(LogRetentionValidationError); + expect(() => validatePurgeOptions({}, {})).toThrow("A pg pool/client is required for purgeWebhookLogs."); + expect(() => validatePurgeOptions({ pool: {} }, {})).toThrow(/query\(\) function/); + expect(() => validatePurgeOptions({ pool: { query: vi.fn() }, logger: {} }, {})).toThrow(/logger/); + }); +}); + +describe("assertPurgeTarget", () => { + it("allows the webhook log table", () => { + expect(() => assertPurgeTarget("webhook_delivery_logs", "timestamp")).not.toThrow(); + }); + + it.each([ + ["payments", "created_at"], + ["webhook_delivery_logs", "id"], + ["webhook_delivery_logs; DROP TABLE payments", "timestamp"], + ['"webhook_delivery_logs"', "timestamp"], + ])("rejects %s.%s", (table, column) => { + expect(() => assertPurgeTarget(table, column)).toThrow(LogRetentionValidationError); + }); +}); + +describe("sanitizeLogMessage", () => { + it("masks credentials in connection strings and key/value pairs", () => { + const text = sanitizeLogMessage( + 'connect failed for postgresql://admin:hunter2@db.internal:5432/app password=hunter2 token: "abc" Bearer eyJhbGci.x.y', + ); + expect(text).toContain("postgresql://***:***@db.internal:5432/app"); + expect(text).not.toContain("hunter2"); + expect(text).not.toContain("abc"); + expect(text).not.toContain("eyJhbGci"); + }); + + it("strips control characters and caps length", () => { + expect(sanitizeLogMessage("line1\nline2\r\u0000")).toBe("line1 line2"); + expect(sanitizeLogMessage("x".repeat(2000)).length).toBe(503); + }); + + it("accepts errors and other values", () => { + expect(sanitizeLogMessage(new Error("boom"))).toBe("boom"); + expect(sanitizeLogMessage(42)).toBe("42"); + }); +}); + +describe("purgeWebhookLogs strict behaviour", () => { + it("passes validated values only as bind parameters", async () => { + const pool = poolWithDeletes(0); + await purgeWebhookLogs({ pool, retentionDays: 7, batchSize: 10, logger: quietLogger() }); + const [sql, params] = pool.query.mock.calls[2]; + expect(params).toEqual([7, 10]); + expect(sql).not.toContain("7"); + }); + + it("returns a frozen summary with only known fields", async () => { + const result = await purgeWebhookLogs({ pool: poolWithDeletes(3, 0), batchSize: 5, logger: quietLogger() }); + expect(Object.isFrozen(result)).toBe(true); + expect(Object.keys(result).sort()).toEqual( + ["batchSize", "batches", "durationMs", "retentionDays", "tableName", "totalDeleted"].sort(), + ); + expect(result).toMatchObject({ totalDeleted: 3, batches: 2 }); + }); + + it.each([ + ["a row count above the batch size", 11], + ["a negative row count", -1], + ["a non-numeric row count", "5"], + ])("aborts on %s from the driver", async (_label, count) => { + await expect( + purgeWebhookLogs({ pool: poolWithDeletes(count), batchSize: 10, logger: quietLogger() }), + ).rejects.toThrow(/Unexpected rowCount/); + }); + + it("stops at maxBatches instead of looping forever", async () => { + const query = vi + .fn() + .mockResolvedValueOnce({ rowCount: 1, rows: [{}] }) + .mockResolvedValueOnce({ rowCount: 0, rows: [] }) + .mockResolvedValue({ rowCount: 1, rows: [] }); + + await expect( + purgeWebhookLogs({ pool: { query }, batchSize: 1, maxBatches: 3, logger: quietLogger() }), + ).rejects.toThrow(/stopped after 3 batches \(3 rows deleted\)/); + expect(query).toHaveBeenCalledTimes(5); + }); + + it("validates options before touching the database", async () => { + const query = vi.fn(); + await expect(purgeWebhookLogs({ pool: { query }, retentionDays: "30; DROP TABLE x" })).rejects.toThrow( + LogRetentionValidationError, + ); + expect(query).not.toHaveBeenCalled(); + }); + + it("scrubs the foreign-key constraint name from the error", async () => { + const query = vi + .fn() + .mockResolvedValueOnce({ rowCount: 1, rows: [{}] }) + .mockResolvedValueOnce({ rowCount: 1, rows: [{ conname: "fk\nforged" }] }); + await expect(purgeWebhookLogs({ pool: { query }, logger: quietLogger() })).rejects.toThrow( + "referenced by foreign keys (example: fk forged)", + ); + }); + + it("writes summary and alert lines through the injected logger", async () => { + const logger = quietLogger(); + await purgeWebhookLogs({ pool: poolWithDeletes(0), logger }); + expect(logger.log).toHaveBeenCalledWith(expect.stringContaining("[log-retention] Purge complete for webhook_delivery_logs")); + }); +}); diff --git a/backend/src/lib/log-retention.js b/backend/src/lib/log-retention.js index 9f3fc24a..53917c63 100644 --- a/backend/src/lib/log-retention.js +++ b/backend/src/lib/log-retention.js @@ -1,20 +1,216 @@ +/** + * Log Retention & Archival Service. + * + * Purges operational webhook delivery logs older than the retention window, + * in bounded batches. + * + * Strict validation and sanitization (issue #1472): + * - Every numeric setting (argument or environment variable) must be a + * plain positive integer within a safe range. "30days", "1e3", "-5", + * "0x10" and fractions are rejected instead of being silently coerced. + * - Unknown option keys are rejected, so a typo cannot silently fall back + * to the default retention. + * - SQL identifiers are checked against an allow-list before they are + * interpolated; values always go through bind parameters. + * - The driver's row counts are validated, and a hard batch cap stops a + * misbehaving driver from looping forever. + * - Log lines and returned errors are scrubbed of connection strings and + * credentials before they are written. + */ + const DEFAULT_RETENTION_DAYS = 30; const DEFAULT_BATCH_SIZE = 1000; const DEFAULT_MAX_DURATION_MS = 5 * 60 * 1000; +const DEFAULT_MAX_BATCHES = 100_000; + +/** Inclusive bounds for each setting. */ +const LIMITS = Object.freeze({ + retentionDays: { min: 1, max: 3650, env: "LOG_RETENTION_DAYS" }, + batchSize: { min: 1, max: 50_000, env: "LOG_PURGE_BATCH_SIZE" }, + maxDurationMs: { min: 1000, max: 24 * 60 * 60 * 1000, env: "LOG_PURGE_MAX_DURATION_MS" }, + maxBatches: { min: 1, max: 10_000_000, env: "LOG_PURGE_MAX_BATCHES" }, +}); + +const ALLOWED_OPTIONS = new Set(["pool", "retentionDays", "batchSize", "maxDurationMs", "maxBatches", "logger"]); const WEBHOOK_LOGS_TABLE = "webhook_delivery_logs"; const WEBHOOK_LOGS_TIMESTAMP_COLUMN = "timestamp"; -function parsePositiveInteger(value, fallback) { - if (!value) return fallback; - const parsed = Number.parseInt(value, 10); - if (!Number.isInteger(parsed) || parsed <= 0) { - throw new Error(`Expected positive integer value but got: ${value}`); +/** Tables and columns this service is allowed to purge. */ +const PURGEABLE_TARGETS = Object.freeze({ + [WEBHOOK_LOGS_TABLE]: new Set([WEBHOOK_LOGS_TIMESTAMP_COLUMN]), +}); + +const IDENTIFIER_PATTERN = /^[a-z_][a-z0-9_]{0,62}$/; +const STRICT_INTEGER_PATTERN = /^\d{1,15}$/; + +export class LogRetentionValidationError extends Error { + constructor(message, field) { + super(message); + this.name = "LogRetentionValidationError"; + this.field = field; + } +} + +/** + * Parse a strictly positive integer setting. + * + * Backwards compatible with the old `(value, fallback)` form: an empty value + * returns the fallback. Anything else must be a plain base-10 integer (or a + * safe integer number) within `[min, max]`. + * + * @param {unknown} value + * @param {number} fallback + * @param {{ name?: string, min?: number, max?: number }} [opts] + * @returns {number} + * @throws {LogRetentionValidationError} + */ +function parsePositiveInteger(value, fallback, { name = "value", min = 1, max = Number.MAX_SAFE_INTEGER } = {}) { + if (value === undefined || value === null || value === "") return fallback; + + let parsed; + if (typeof value === "number") { + parsed = Number.isSafeInteger(value) ? value : NaN; + } else if (typeof value === "string" && STRICT_INTEGER_PATTERN.test(value.trim())) { + parsed = Number(value.trim()); + } else { + parsed = NaN; + } + + if (!Number.isSafeInteger(parsed) || parsed <= 0) { + throw new LogRetentionValidationError( + `Expected positive integer value for ${name} but got: ${describeValue(value)}`, + name, + ); + } + if (parsed < min || parsed > max) { + throw new LogRetentionValidationError( + `${name} must be between ${min} and ${max}, got ${parsed}`, + name, + ); } return parsed; } +/** Short, safe description of a rejected value for error messages. */ +function describeValue(value) { + if (typeof value === "string") { + const printable = value.replace(/[^\x20-\x7e]/g, "?"); + return JSON.stringify(printable.length > 32 ? `${printable.slice(0, 32)}...` : printable); + } + if (typeof value === "number") return String(value); + return `<${value === null ? "null" : typeof value}>`; +} + +/** + * Resolve one setting from an explicit option or its environment variable. + * @param {keyof typeof LIMITS} key + */ +function resolveSetting(key, explicit, fallback, env = process.env) { + const limit = LIMITS[key]; + const source = explicit !== undefined ? explicit : env[limit.env]; + const name = explicit !== undefined ? key : limit.env; + return parsePositiveInteger(source, fallback, { name, min: limit.min, max: limit.max }); +} + +/** + * Validate and normalise purge options. + * + * @param {object} options + * @param {NodeJS.ProcessEnv} [env] + * @returns {{ pool: { query: Function }, retentionDays: number, batchSize: number, maxDurationMs: number, maxBatches: number, logger: Console }} + * @throws {LogRetentionValidationError} + */ +function validatePurgeOptions(options, env = process.env) { + if (options === null || typeof options !== "object" || Array.isArray(options)) { + throw new LogRetentionValidationError("purgeWebhookLogs options must be an object", "options"); + } + const unknown = Object.keys(options).filter((key) => !ALLOWED_OPTIONS.has(key)); + if (unknown.length > 0) { + throw new LogRetentionValidationError( + `Unknown purgeWebhookLogs option(s): ${unknown.map((k) => describeValue(k)).join(", ")}`, + unknown[0], + ); + } + + const { pool } = options; + if (!pool) { + throw new Error("A pg pool/client is required for purgeWebhookLogs."); + } + if (typeof pool.query !== "function") { + throw new LogRetentionValidationError("pool must expose a query() function", "pool"); + } + + const logger = options.logger ?? console; + if (typeof logger.log !== "function" || typeof logger.error !== "function") { + throw new LogRetentionValidationError("logger must expose log() and error()", "logger"); + } + + return { + pool, + logger, + retentionDays: resolveSetting("retentionDays", options.retentionDays, DEFAULT_RETENTION_DAYS, env), + batchSize: resolveSetting("batchSize", options.batchSize, DEFAULT_BATCH_SIZE, env), + maxDurationMs: resolveSetting("maxDurationMs", options.maxDurationMs, DEFAULT_MAX_DURATION_MS, env), + maxBatches: resolveSetting("maxBatches", options.maxBatches, DEFAULT_MAX_BATCHES, env), + }; +} + +/** + * Allow-list check for an identifier that will be interpolated into SQL. + * @throws {LogRetentionValidationError} + */ +function assertPurgeTarget(tableName, timestampColumn) { + if (!IDENTIFIER_PATTERN.test(tableName) || !IDENTIFIER_PATTERN.test(timestampColumn)) { + throw new LogRetentionValidationError("Purge target contains an invalid SQL identifier", "tableName"); + } + if (!PURGEABLE_TARGETS[tableName]?.has(timestampColumn)) { + throw new LogRetentionValidationError( + `${tableName}.${timestampColumn} is not an allowed purge target`, + "tableName", + ); + } +} + +const SECRET_PATTERNS = [ + // postgres://user:password@host/db and similar URLs with credentials + [/\b([a-z][a-z0-9+.-]*:\/\/)[^\s:@/]+:[^\s@/]+@/gi, "$1***:***@"], + [/\b(password|passwd|pwd|secret|token|api[_-]?key)\s*[=:]\s*("[^"]*"|'[^']*'|[^\s,;]+)/gi, "$1=***"], + [/\bBearer\s+[A-Za-z0-9._~+/-]+=*/g, "Bearer ***"], +]; + +/** + * Scrub credentials from a message before it is logged. Also strips control + * characters (log injection) and caps the length. + * + * @param {unknown} message + * @returns {string} + */ +function sanitizeLogMessage(message) { + let text = typeof message === "string" ? message : message instanceof Error ? message.message : String(message); + for (const [pattern, replacement] of SECRET_PATTERNS) { + text = text.replace(pattern, replacement); + } + // eslint-disable-next-line no-control-regex + text = text.replace(/[\u0000-\u001f\u007f]+/g, " ").trim(); + return text.length > 500 ? `${text.slice(0, 500)}...` : text; +} + +/** Validate a driver result's row count. */ +function readRowCount(result, batchSize) { + const count = result?.rowCount ?? 0; + if (!Number.isSafeInteger(count) || count < 0 || count > batchSize) { + throw new LogRetentionValidationError( + `Unexpected rowCount from database driver: ${describeValue(count)}`, + "rowCount", + ); + } + return count; +} + async function validatePurgeSafety({ pool, tableName, timestampColumn }) { + assertPurgeTarget(tableName, timestampColumn); + const indexQuery = ` SELECT 1 FROM pg_indexes @@ -24,7 +220,7 @@ async function validatePurgeSafety({ pool, tableName, timestampColumn }) { LIMIT 1 `; const indexResult = await pool.query(indexQuery, [tableName, timestampColumn]); - if (indexResult.rowCount === 0) { + if (!indexResult || indexResult.rowCount === 0) { throw new Error( `Missing index for ${tableName}.${timestampColumn}; aborting purge to avoid sequential scans.`, ); @@ -38,31 +234,16 @@ async function validatePurgeSafety({ pool, tableName, timestampColumn }) { LIMIT 1 `; const fkResult = await pool.query(fkQuery, [tableName]); - if (fkResult.rowCount > 0) { + if (fkResult?.rowCount > 0) { + const constraint = String(fkResult.rows?.[0]?.conname ?? "unknown"); throw new Error( - `Table ${tableName} is referenced by foreign keys (example: ${fkResult.rows[0].conname}); aborting purge.`, + `Table ${tableName} is referenced by foreign keys (example: ${sanitizeLogMessage(constraint)}); aborting purge.`, ); } } -async function purgeWebhookLogs({ - pool, - retentionDays = parsePositiveInteger( - process.env.LOG_RETENTION_DAYS, - DEFAULT_RETENTION_DAYS, - ), - batchSize = parsePositiveInteger( - process.env.LOG_PURGE_BATCH_SIZE, - DEFAULT_BATCH_SIZE, - ), - maxDurationMs = parsePositiveInteger( - process.env.LOG_PURGE_MAX_DURATION_MS, - DEFAULT_MAX_DURATION_MS, - ), -} = {}) { - if (!pool) { - throw new Error("A pg pool/client is required for purgeWebhookLogs."); - } +async function purgeWebhookLogs(options = {}) { + const { pool, logger, retentionDays, batchSize, maxDurationMs, maxBatches } = validatePurgeOptions(options); const startedAt = Date.now(); const tableName = WEBHOOK_LOGS_TABLE; @@ -74,8 +255,16 @@ async function purgeWebhookLogs({ // Confirm this table contains operational delivery telemetry only before enforcing 30-day retention. let totalDeleted = 0; let deletedInBatch = 0; + let batches = 0; do { + if (batches >= maxBatches) { + throw new LogRetentionValidationError( + `Purge stopped after ${maxBatches} batches (${totalDeleted} rows deleted); raise LOG_PURGE_MAX_BATCHES if this is expected.`, + "maxBatches", + ); + } + const deleteQuery = ` WITH rows_to_delete AS ( SELECT id @@ -91,35 +280,42 @@ async function purgeWebhookLogs({ `; const deleteResult = await pool.query(deleteQuery, [retentionDays, batchSize]); - deletedInBatch = deleteResult.rowCount ?? 0; + deletedInBatch = readRowCount(deleteResult, batchSize); totalDeleted += deletedInBatch; + batches += 1; } while (deletedInBatch > 0); const durationMs = Date.now() - startedAt; const durationSec = (durationMs / 1000).toFixed(2); - console.log( + logger.log( `[log-retention] Purge complete for ${tableName}: deleted_rows=${totalDeleted} duration_ms=${durationMs} duration_sec=${durationSec}`, ); if (durationMs > maxDurationMs) { - console.error( + logger.error( `[ALERT][log-retention] Purge exceeded max duration (${maxDurationMs}ms): actual=${durationMs}ms`, ); } - return { + // Only known, non-sensitive fields are returned. + return Object.freeze({ tableName, retentionDays, batchSize, totalDeleted, durationMs, - }; + batches, + }); } export { purgeWebhookLogs, parsePositiveInteger, + validatePurgeOptions, + sanitizeLogMessage, + assertPurgeTarget, DEFAULT_RETENTION_DAYS, DEFAULT_BATCH_SIZE, DEFAULT_MAX_DURATION_MS, + DEFAULT_MAX_BATCHES, }; diff --git a/backend/src/lib/payment-processor-metrics.js b/backend/src/lib/payment-processor-metrics.js index f0ad8e70..430ba4cc 100644 --- a/backend/src/lib/payment-processor-metrics.js +++ b/backend/src/lib/payment-processor-metrics.js @@ -92,6 +92,25 @@ export const paymentProcessorListRequestsTotal = new client.Counter({ labelNames: ["outcome"], }); +/** + * Bound the `asset` label (issue #1481). + * + * Session creation records the asset straight from the request body, and a + * validation_failed request can carry any string. Unbounded label values + * grow one time series per distinct value, which a hostile client can use to + * exhaust Prometheus memory. Only well-formed Stellar asset codes are kept; + * everything else is bucketed as "invalid" (or "unknown" when absent). + * + * @param {unknown} asset + * @returns {string} + */ +export function toAssetLabel(asset) { + if (asset === undefined || asset === null || asset === "") return "unknown"; + if (typeof asset !== "string") return "invalid"; + const code = asset.trim().toUpperCase(); + return /^[A-Z0-9]{1,12}$/.test(code) ? code : "invalid"; +} + register.registerMetric(paymentProcessorSessionsTotal); register.registerMetric(paymentProcessorSessionDuration); register.registerMetric(paymentProcessorVerificationsTotal); diff --git a/backend/src/lib/websocket-relay-retry.js b/backend/src/lib/websocket-relay-retry.js new file mode 100644 index 00000000..edfba424 --- /dev/null +++ b/backend/src/lib/websocket-relay-retry.js @@ -0,0 +1,338 @@ +/** + * websocket-relay-retry.js + * + * Automated retry with exponential backoff for the WebSocket Relay Service + * (issue #1454). + * + * reconnectWithBackoff() in websocket-relay-recovery.js retries a connection + * on every error. Relay deliveries need more care: + * + * - Only transient failures are retried (network errors, timeouts, 408, + * 429, 5xx). Validation, signature and auth failures and an OPEN circuit + * breaker fail immediately: retrying them cannot succeed, and retrying + * into an open breaker would defeat it. + * - Full-jitter backoff: delay(n) = random(0, min(maxDelayMs, baseDelayMs * 2^n)), + * so many relay instances never retry in lockstep. + * - A total time budget (maxElapsedMs) and an AbortSignal bound every call. + * - Messages that still fail are moved to the dead letter queue instead of + * being dropped, and can be redriven later. + * - Every retry and outcome is recorded in websocket-relay-telemetry.js. + * + * Logs never include message payloads, only the operation label, attempt + * counts, delays and the error's code/status. + */ + +import { logger } from "./logger.js"; +import { recordRelayDelivery, recordRelayRetry } from "./websocket-relay-telemetry.js"; + +export const DEFAULT_RELAY_RETRY_OPTIONS = Object.freeze({ + maxAttempts: 5, + baseDelayMs: 250, + maxDelayMs: 10_000, + maxElapsedMs: 30_000, +}); + +const LIMITS = Object.freeze({ + maxAttempts: 10, + maxDelayMs: 60_000, + maxElapsedMs: 300_000, +}); + +const RETRYABLE_NETWORK_CODES = new Set([ + "ECONNRESET", + "ECONNREFUSED", + "ECONNABORTED", + "ETIMEDOUT", + "EPIPE", + "EAI_AGAIN", + "ENOTFOUND", + "EHOSTUNREACH", + "ENETUNREACH", + "UND_ERR_SOCKET", + "UND_ERR_CONNECT_TIMEOUT", +]); + +/** WebSocket close codes worth retrying (1006 abnormal, 1011 server error, 1012 restart, 1013 try again later). */ +const RETRYABLE_CLOSE_CODES = new Set([1006, 1011, 1012, 1013]); + +export class RelayRetryAbortedError extends Error { + constructor(message = "Relay retry aborted") { + super(message); + this.name = "RelayRetryAbortedError"; + this.retryable = false; + } +} + +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. Precedence: explicit > environment > defaults. + * Values are clamped so bad configuration can neither disable backoff nor + * stall the relay. + */ +export function resolveRelayRetryOptions(options = {}, env = process.env) { + const maxAttempts = clampInt( + options.maxAttempts ?? env.RELAY_RETRY_MAX_ATTEMPTS, + DEFAULT_RELAY_RETRY_OPTIONS.maxAttempts, + 1, + LIMITS.maxAttempts, + ); + const baseDelayMs = clampInt( + options.baseDelayMs ?? env.RELAY_RETRY_BASE_DELAY_MS, + DEFAULT_RELAY_RETRY_OPTIONS.baseDelayMs, + 0, + LIMITS.maxDelayMs, + ); + const maxDelayMs = clampInt( + options.maxDelayMs ?? env.RELAY_RETRY_MAX_DELAY_MS, + DEFAULT_RELAY_RETRY_OPTIONS.maxDelayMs, + baseDelayMs, + LIMITS.maxDelayMs, + ); + const maxElapsedMs = clampInt( + options.maxElapsedMs ?? env.RELAY_RETRY_MAX_ELAPSED_MS, + DEFAULT_RELAY_RETRY_OPTIONS.maxElapsedMs, + 0, + LIMITS.maxElapsedMs, + ); + return Object.freeze({ maxAttempts, baseDelayMs, maxDelayMs, maxElapsedMs }); +} + +/** + * Full-jitter backoff delay before retry number `retryIndex` (0 = first retry). + * + * @param {number} retryIndex + * @param {{ baseDelayMs: number, maxDelayMs: number }} opts + * @param {() => number} [random] + */ +export function computeRelayBackoffDelay(retryIndex, { baseDelayMs, maxDelayMs }, random = Math.random) { + const index = Math.min(Math.max(Math.trunc(Number(retryIndex) || 0), 0), 30); + const ceiling = Math.min(maxDelayMs, baseDelayMs * 2 ** index); + if (!Number.isFinite(ceiling) || ceiling <= 0) return 0; + const r = Math.min(Math.max(Number(random()) || 0, 0), 1); + return Math.floor(r * ceiling); +} + +function getStatus(err) { + const status = err?.status ?? err?.statusCode ?? err?.response?.status; + return typeof status === "number" ? status : null; +} + +/** + * Whether a relay failure is transient and worth retrying. + * @param {unknown} err + */ +export function isRetryableRelayError(err) { + if (!err || typeof err !== "object") return false; + if (err.circuitOpen === true) return false; + if (err.retryable === false) return false; + if (err.retryable === true) return true; + + const status = getStatus(err); + if (status !== null) { + if (status === 408 || status === 429) return true; + if (status >= 500 && status !== 501) return true; + if (status >= 400) return false; + } + + if (typeof err.closeCode === "number") return RETRYABLE_CLOSE_CODES.has(err.closeCode); + if (RETRYABLE_NETWORK_CODES.has(err.code)) return true; + + const message = typeof err.message === "string" ? err.message : ""; + return /timed? ?out|socket hang up|network error|connection (reset|closed)/i.test(message); +} + +function sleepWithSignal(ms, signal) { + return new Promise((resolve, reject) => { + if (signal?.aborted) { + reject(new RelayRetryAbortedError()); + return; + } + const timer = setTimeout(() => { + signal?.removeEventListener?.("abort", onAbort); + resolve(); + }, ms); + timer.unref?.(); + function onAbort() { + clearTimeout(timer); + reject(new RelayRetryAbortedError()); + } + signal?.addEventListener?.("abort", onAbort, { once: true }); + }); +} + +function errorFields(err) { + return { + err: typeof err?.message === "string" ? err.message.slice(0, 200) : undefined, + code: err?.code, + status: getStatus(err) ?? undefined, + closeCode: err?.closeCode, + }; +} + +/** + * Run `operation` with automated retries and full-jitter exponential backoff. + * + * @template T + * @param {(ctx: { attempt: number, signal?: AbortSignal }) => Promise} operation + * @param {object} [options] + * @param {'deliver'|'reconnect'|'redrive'|'other'} [options.operation='other'] telemetry label + * @param {number} [options.maxAttempts] + * @param {number} [options.baseDelayMs] + * @param {number} [options.maxDelayMs] + * @param {number} [options.maxElapsedMs] total budget including waits + * @param {AbortSignal} [options.signal] + * @param {(err: unknown) => boolean} [options.shouldRetry] + * @param {(info: { attempt: number, delayMs: number, error: unknown }) => void} [options.onRetry] + * @param {(ms: number, signal?: AbortSignal) => Promise} [options.sleep] injectable for tests + * @param {() => number} [options.random] injectable for tests + * @param {() => number} [options.now] injectable for tests + * @returns {Promise<{ value: T, attempts: number }>} + * @throws the last error, with `retryAttempts` set + */ +export async function withRelayRetry(operation, options = {}) { + const { + operation: label = "other", + signal, + shouldRetry = isRetryableRelayError, + onRetry, + sleep = sleepWithSignal, + random = Math.random, + now = Date.now, + } = options; + const resolved = resolveRelayRetryOptions(options); + const deadline = now() + resolved.maxElapsedMs; + + for (let attempt = 1; ; attempt += 1) { + if (signal?.aborted) { + recordRelayRetry(label, "aborted"); + throw Object.assign(new RelayRetryAbortedError(), { retryAttempts: attempt - 1 }); + } + + try { + const value = await operation({ attempt, signal }); + if (attempt > 1) recordRelayRetry(label, "recovered"); + return { value, attempts: attempt }; + } catch (err) { + const retryable = !(err instanceof RelayRetryAbortedError) && shouldRetry(err); + const delayMs = computeRelayBackoffDelay(attempt - 1, resolved, random); + const outOfAttempts = attempt >= resolved.maxAttempts; + const outOfTime = now() + delayMs > deadline; + + if (!retryable || outOfAttempts || outOfTime) { + if (err && typeof err === "object") err.retryAttempts = attempt; + if (retryable) { + recordRelayRetry(label, "exhausted"); + logger.error( + { operation: label, attempts: attempt, reason: outOfAttempts ? "max_attempts" : "time_budget", ...errorFields(err) }, + "WebSocket relay operation failed after exhausting retries", + ); + } + throw err; + } + + recordRelayRetry(label, "scheduled", delayMs); + logger.warn( + { operation: label, attempt, maxAttempts: resolved.maxAttempts, delayMs, ...errorFields(err) }, + "WebSocket relay operation failed, retrying with backoff", + ); + try { + onRetry?.({ attempt, delayMs, error: err }); + } catch { + // An observer must never break the retry loop. + } + + try { + await sleep(delayMs, signal); + } catch (sleepErr) { + recordRelayRetry(label, "aborted"); + throw Object.assign( + sleepErr instanceof Error ? sleepErr : new RelayRetryAbortedError(), + { retryAttempts: attempt }, + ); + } + } + } +} + +/** + * Deliver one relay message with retries, through the circuit breaker when + * one is given, dead-lettering it when every attempt fails. + * + * @param {unknown} message + * @param {(message: unknown, ctx: { attempt: number }) => Promise} send + * @param {object} [options] retry options plus: + * @param {{ call: (fn: () => Promise) => Promise }} [options.circuitBreaker] + * @param {{ push: (message: unknown, error: unknown, retryCount: number) => void }} [options.dlq] + * @param {number} [options.previousAttempts=0] attempts already spent (redrive) + * @returns {Promise<{ delivered: true, attempts: number, value: unknown } | { delivered: false, attempts: number, deadLettered: true, error: unknown }>} + * @throws the delivery error when no `dlq` is given + */ +export async function deliverRelayMessage(message, send, options = {}) { + if (typeof send !== "function") { + throw new TypeError("deliverRelayMessage requires a send function"); + } + const { circuitBreaker, dlq, previousAttempts = 0, operation = "deliver", ...retryOptions } = options; + + try { + const { value, attempts } = await withRelayRetry( + (ctx) => (circuitBreaker ? circuitBreaker.call(() => send(message, ctx)) : send(message, ctx)), + { ...retryOptions, operation }, + ); + recordRelayDelivery(attempts > 1 ? "recovered" : "delivered"); + return { delivered: true, attempts, value }; + } catch (error) { + const attempts = typeof error?.retryAttempts === "number" ? error.retryAttempts : 1; + if (!dlq) { + recordRelayDelivery("failed"); + throw error; + } + dlq.push(message, error, previousAttempts + attempts); + recordRelayDelivery("dead_lettered"); + logger.warn( + { operation, attempts: previousAttempts + attempts, circuitOpen: error?.circuitOpen === true, ...errorFields(error) }, + "WebSocket relay message moved to dead letter queue", + ); + return { delivered: false, attempts, deadLettered: true, error }; + } +} + +/** + * Re-attempt up to `limit` dead-lettered messages, oldest first. Messages + * that fail again go back on the queue with their retry count increased. + * + * @param {{ shift: () => any, push: Function, size: number }} dlq + * @param {(message: unknown, ctx: { attempt: number }) => Promise} send + * @param {object} [options] deliverRelayMessage options plus `limit` (default 50) + * @returns {Promise<{ redelivered: number, failed: number }>} + */ +export async function redriveDeadLetters(dlq, send, options = {}) { + if (!dlq || typeof dlq.shift !== "function") { + throw new TypeError("redriveDeadLetters requires a dead letter queue"); + } + const { limit = 50, ...deliverOptions } = options; + const max = clampInt(limit, 50, 0, 10_000); + let redelivered = 0; + let failed = 0; + + // Snapshot the count first so entries re-queued by this run are not retried again. + const toProcess = Math.min(max, dlq.size); + for (let i = 0; i < toProcess; i += 1) { + const entry = dlq.shift(); + if (!entry) break; + const result = await deliverRelayMessage(entry.message, send, { + ...deliverOptions, + operation: "redrive", + dlq, + previousAttempts: Number(entry.retryCount) || 0, + }); + if (result.delivered) redelivered += 1; + else failed += 1; + } + return { redelivered, failed }; +} diff --git a/backend/src/lib/websocket-relay-retry.test.js b/backend/src/lib/websocket-relay-retry.test.js new file mode 100644 index 00000000..0ce672a6 --- /dev/null +++ b/backend/src/lib/websocket-relay-retry.test.js @@ -0,0 +1,328 @@ +import { describe, it, expect, beforeEach, vi } from "vitest"; + +vi.mock("./logger.js", () => ({ + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }, +})); + +import { logger } from "./logger.js"; +import { + DEFAULT_RELAY_RETRY_OPTIONS, + RelayRetryAbortedError, + computeRelayBackoffDelay, + deliverRelayMessage, + isRetryableRelayError, + redriveDeadLetters, + resolveRelayRetryOptions, + withRelayRetry, +} from "./websocket-relay-retry.js"; +import { CircuitBreaker, DeadLetterQueue } from "./websocket-relay-recovery.js"; +import { + getWebSocketRelayHealth, + resetWebSocketRelayHealth, + websocketRelayTelemetryRegister, +} from "./websocket-relay-telemetry.js"; + +const netError = (code = "ECONNRESET") => Object.assign(new Error("socket hang up"), { code }); +const httpError = (status) => Object.assign(new Error(`HTTP ${status}`), { status }); + +/** Instant, recorded sleeps and deterministic jitter (max of the window). */ +function fastOptions(extra = {}) { + const sleeps = []; + return { + sleeps, + options: { + baseDelayMs: 100, + maxDelayMs: 1_000, + maxAttempts: 4, + random: () => 0.999, + sleep: async (ms) => { + sleeps.push(ms); + }, + ...extra, + }, + }; +} + +async function metricValue(name, labels) { + const metric = await websocketRelayTelemetryRegister.getSingleMetric(name).get(); + return metric.values + .filter((v) => Object.entries(labels).every(([k, val]) => v.labels[k] === val)) + .reduce((sum, v) => sum + v.value, 0); +} + +beforeEach(() => { + vi.clearAllMocks(); + resetWebSocketRelayHealth(); +}); + +describe("resolveRelayRetryOptions", () => { + it("uses defaults", () => { + expect(resolveRelayRetryOptions({}, {})).toEqual(DEFAULT_RELAY_RETRY_OPTIONS); + }); + + it("reads the environment and lets explicit options win", () => { + const env = { RELAY_RETRY_MAX_ATTEMPTS: "3", RELAY_RETRY_BASE_DELAY_MS: "50", RELAY_RETRY_MAX_DELAY_MS: "500" }; + expect(resolveRelayRetryOptions({}, env)).toMatchObject({ maxAttempts: 3, baseDelayMs: 50, maxDelayMs: 500 }); + expect(resolveRelayRetryOptions({ maxAttempts: 2 }, env).maxAttempts).toBe(2); + }); + + it("clamps hostile or broken values", () => { + const resolved = resolveRelayRetryOptions( + { maxAttempts: 1_000, baseDelayMs: -5, maxDelayMs: "abc", maxElapsedMs: 1e12 }, + {}, + ); + expect(resolved.maxAttempts).toBe(10); + expect(resolved.baseDelayMs).toBe(0); + expect(resolved.maxDelayMs).toBe(DEFAULT_RELAY_RETRY_OPTIONS.maxDelayMs); + expect(resolved.maxElapsedMs).toBe(300_000); + }); + + it("never allows fewer than one attempt", () => { + expect(resolveRelayRetryOptions({ maxAttempts: 0 }, {}).maxAttempts).toBe(1); + }); +}); + +describe("computeRelayBackoffDelay", () => { + const opts = { baseDelayMs: 100, maxDelayMs: 1_000 }; + + it("doubles the ceiling per retry and caps it", () => { + expect([0, 1, 2, 3, 4, 10].map((n) => computeRelayBackoffDelay(n, opts, () => 1))).toEqual([ + 100, 200, 400, 800, 1000, 1000, + ]); + }); + + it("applies full jitter", () => { + expect(computeRelayBackoffDelay(2, opts, () => 0)).toBe(0); + expect(computeRelayBackoffDelay(2, opts, () => 0.5)).toBe(200); + }); + + it("tolerates bad random sources and indices", () => { + expect(computeRelayBackoffDelay(1, opts, () => NaN)).toBe(0); + expect(computeRelayBackoffDelay(1, opts, () => 7)).toBe(200); + expect(computeRelayBackoffDelay(-4, opts, () => 1)).toBe(100); + expect(computeRelayBackoffDelay(1e9, opts, () => 1)).toBe(1_000); + }); +}); + +describe("isRetryableRelayError", () => { + it.each([ + ["network code", netError("ECONNREFUSED")], + ["timeout message", new Error("request timed out")], + ["503", httpError(503)], + ["429", httpError(429)], + ["abnormal close 1006", Object.assign(new Error("closed"), { closeCode: 1006 })], + ["explicit retryable", Object.assign(new Error("x"), { retryable: true })], + ])("retries %s", (_label, err) => { + expect(isRetryableRelayError(err)).toBe(true); + }); + + it.each([ + ["open circuit", Object.assign(new Error("open"), { circuitOpen: true })], + ["400 validation", httpError(400)], + ["401 auth", httpError(401)], + ["501", httpError(501)], + ["policy close 1008", Object.assign(new Error("policy"), { closeCode: 1008 })], + ["explicit non-retryable", Object.assign(netError(), { retryable: false })], + ["plain bug", new TypeError("x is undefined")], + ["non-object", "boom"], + ])("does not retry %s", (_label, err) => { + expect(isRetryableRelayError(err)).toBe(false); + }); +}); + +describe("withRelayRetry", () => { + it("returns the first successful result without sleeping", async () => { + const { options, sleeps } = fastOptions(); + const op = vi.fn().mockResolvedValue("ok"); + await expect(withRelayRetry(op, options)).resolves.toEqual({ value: "ok", attempts: 1 }); + expect(sleeps).toEqual([]); + }); + + it("retries transient failures with exponential backoff and records recovery", async () => { + const { options, sleeps } = fastOptions({ operation: "reconnect" }); + const op = vi.fn().mockRejectedValueOnce(netError()).mockRejectedValueOnce(httpError(502)).mockResolvedValue("ok"); + + const result = await withRelayRetry(op, options); + + expect(result).toEqual({ value: "ok", attempts: 3 }); + expect(sleeps).toEqual([99, 199]); + expect(op).toHaveBeenNthCalledWith(2, expect.objectContaining({ attempt: 2 })); + expect(await metricValue("websocket_relay_retries_total", { operation: "reconnect", result: "scheduled" })).toBeGreaterThanOrEqual(2); + expect(await metricValue("websocket_relay_retries_total", { operation: "reconnect", result: "recovered" })).toBeGreaterThanOrEqual(1); + }); + + it("stops after maxAttempts and marks the error", async () => { + const { options } = fastOptions({ maxAttempts: 3 }); + const op = vi.fn().mockRejectedValue(netError()); + + const err = await withRelayRetry(op, options).catch((e) => e); + + expect(op).toHaveBeenCalledTimes(3); + expect(err.retryAttempts).toBe(3); + expect(logger.error).toHaveBeenCalledWith( + expect.objectContaining({ attempts: 3, reason: "max_attempts" }), + "WebSocket relay operation failed after exhausting retries", + ); + expect(getWebSocketRelayHealth().retries_exhausted).toBe(1); + }); + + it("does not retry non-retryable errors", async () => { + const { options, sleeps } = fastOptions(); + const op = vi.fn().mockRejectedValue(httpError(400)); + const err = await withRelayRetry(op, options).catch((e) => e); + expect(op).toHaveBeenCalledTimes(1); + expect(sleeps).toEqual([]); + expect(err.retryAttempts).toBe(1); + expect(logger.error).not.toHaveBeenCalled(); + }); + + it("respects the total time budget", async () => { + let clock = 0; + const { options } = fastOptions({ + maxAttempts: 10, + maxElapsedMs: 250, + now: () => clock, + sleep: async (ms) => { + clock += ms; + }, + }); + const op = vi.fn().mockRejectedValue(netError()); + + const err = await withRelayRetry(op, options).catch((e) => e); + + // 99 + 199 = 298 > 250, so only the first wait fits. + expect(op).toHaveBeenCalledTimes(2); + expect(err.retryAttempts).toBe(2); + }); + + it("aborts before the first attempt when the signal is already aborted", async () => { + const controller = new AbortController(); + controller.abort(); + const op = vi.fn(); + await expect(withRelayRetry(op, { signal: controller.signal })).rejects.toBeInstanceOf(RelayRetryAbortedError); + expect(op).not.toHaveBeenCalled(); + }); + + it("aborts during a backoff wait", async () => { + const controller = new AbortController(); + const op = vi.fn().mockRejectedValue(netError()); + const pending = withRelayRetry(op, { signal: controller.signal, baseDelayMs: 10_000, maxDelayMs: 10_000, random: () => 1 }); + await vi.waitFor(() => expect(op).toHaveBeenCalledTimes(1)); + controller.abort(); + await expect(pending).rejects.toBeInstanceOf(RelayRetryAbortedError); + }); + + it("keeps message content out of the logs", async () => { + const { options } = fastOptions({ maxAttempts: 2 }); + const secret = "payment_id=pay_123 token=abc"; + const op = vi.fn().mockRejectedValue(Object.assign(new Error("x".repeat(500)), { code: "ECONNRESET" })); + await withRelayRetry(op, { ...options, operation: "deliver" }).catch(() => {}); + const logged = JSON.stringify([...logger.warn.mock.calls, ...logger.error.mock.calls]); + expect(logged).not.toContain(secret); + for (const [fields] of logger.warn.mock.calls) { + expect(fields.err.length).toBeLessThanOrEqual(200); + } + }); + + it("ignores a throwing onRetry observer", async () => { + const { options } = fastOptions({ onRetry: () => { throw new Error("observer"); } }); + const op = vi.fn().mockRejectedValueOnce(netError()).mockResolvedValue("ok"); + await expect(withRelayRetry(op, options)).resolves.toMatchObject({ value: "ok" }); + }); +}); + +describe("deliverRelayMessage", () => { + const message = { type: "event", payment_id: "pay_1", event_type: "payment.confirmed" }; + + it("delivers and counts a first-attempt success", async () => { + const { options } = fastOptions(); + const send = vi.fn().mockResolvedValue("ack"); + const result = await deliverRelayMessage(message, send, options); + expect(result).toEqual({ delivered: true, attempts: 1, value: "ack" }); + expect(send).toHaveBeenCalledWith(message, expect.objectContaining({ attempt: 1 })); + expect(getWebSocketRelayHealth().delivered).toBe(1); + }); + + it("counts a delivery after retries as recovered", async () => { + const { options } = fastOptions(); + const send = vi.fn().mockRejectedValueOnce(netError()).mockResolvedValue("ack"); + await deliverRelayMessage(message, send, options); + expect(getWebSocketRelayHealth().recovered).toBe(1); + }); + + it("dead-letters a message after exhausting retries", async () => { + const { options } = fastOptions({ maxAttempts: 3 }); + const dlq = new DeadLetterQueue(); + const send = vi.fn().mockRejectedValue(netError()); + + const result = await deliverRelayMessage(message, send, { ...options, dlq }); + + expect(result).toMatchObject({ delivered: false, deadLettered: true, attempts: 3 }); + expect(dlq.size).toBe(1); + expect(dlq.getAll()[0]).toMatchObject({ message, retryCount: 3, errorMessage: "socket hang up" }); + expect(getWebSocketRelayHealth().dead_lettered).toBe(1); + }); + + it("throws when there is no DLQ", async () => { + const { options } = fastOptions({ maxAttempts: 2 }); + await expect(deliverRelayMessage(message, vi.fn().mockRejectedValue(netError()), options)).rejects.toThrow("socket hang up"); + expect(getWebSocketRelayHealth().failed).toBe(1); + }); + + it("does not retry into an open circuit breaker", async () => { + const { options, sleeps } = fastOptions(); + const breaker = new CircuitBreaker({ failureThreshold: 1, resetTimeoutMs: 60_000 }); + const dlq = new DeadLetterQueue(); + const send = vi.fn().mockRejectedValue(httpError(503)); + + // First message trips the breaker on its first failure, then the retry + // hits the OPEN breaker and stops immediately. + const result = await deliverRelayMessage(message, send, { ...options, circuitBreaker: breaker, dlq }); + + expect(breaker.state).toBe("OPEN"); + expect(send).toHaveBeenCalledTimes(1); + expect(sleeps).toHaveLength(1); + expect(result.error.circuitOpen).toBe(true); + expect(dlq.size).toBe(1); + }); + + it("validates its arguments", async () => { + await expect(deliverRelayMessage(message, null)).rejects.toThrow(TypeError); + }); +}); + +describe("redriveDeadLetters", () => { + it("redelivers queued messages and re-queues the ones that still fail", async () => { + const { options } = fastOptions({ maxAttempts: 2 }); + const dlq = new DeadLetterQueue(); + dlq.push({ id: 1 }, new Error("old"), 3); + dlq.push({ id: 2 }, new Error("old"), 1); + const send = vi.fn(async (msg) => { + if (msg.id === 2) throw netError(); + return "ack"; + }); + + const result = await redriveDeadLetters(dlq, send, options); + + expect(result).toEqual({ redelivered: 1, failed: 1 }); + expect(dlq.size).toBe(1); + expect(dlq.getAll()[0]).toMatchObject({ message: { id: 2 }, retryCount: 3 }); + }); + + it("processes at most `limit` entries and never re-processes re-queued ones", async () => { + const { options } = fastOptions({ maxAttempts: 1 }); + const dlq = new DeadLetterQueue(); + for (let i = 0; i < 5; i += 1) dlq.push({ id: i }, new Error("old")); + const send = vi.fn().mockRejectedValue(netError()); + + const result = await redriveDeadLetters(dlq, send, { ...options, limit: 3 }); + + expect(send).toHaveBeenCalledTimes(3); + expect(result).toEqual({ redelivered: 0, failed: 3 }); + expect(dlq.size).toBe(5); + }); + + it("requires a queue", async () => { + await expect(redriveDeadLetters(null, vi.fn())).rejects.toThrow(TypeError); + }); +}); diff --git a/backend/src/lib/websocket-relay-telemetry.js b/backend/src/lib/websocket-relay-telemetry.js new file mode 100644 index 00000000..ff453c09 --- /dev/null +++ b/backend/src/lib/websocket-relay-telemetry.js @@ -0,0 +1,307 @@ +/** + * websocket-relay-telemetry.js + * + * Prometheus alert metrics and health telemetry for the WebSocket Relay + * Service (issue #1453). + * + * websocket-relay-metrics.js (#1083) exposes raw relay counters. This module + * adds what alerting needs on top of them: + * + * - Delivery and retry outcome counters with a fixed label set + * - A rolling-window health monitor (constant memory) that derives + * healthy / degraded / unhealthy from the delivery failure ratio, retry + * exhaustion, the circuit breaker state and the dead letter queue depth + * - Scrape-time gauges for that health, so alerts decay when traffic stops + * - getWebSocketRelayHealth() for GET /health/websocket-relay + * + * Labels are never taken from message content, payment ids, origins or error + * messages, so relay traffic cannot inflate cardinality or leak data into + * metrics. The metrics live in their own registry; /metrics merges it. + */ + +import client from "prom-client"; +import { logger } from "./logger.js"; + +const register = new client.Registry(); + +register.setDefaultLabels({ + app: "stellar-payment-api", + module: "websocket-relay", +}); + +/** outcome: delivered | recovered | dead_lettered | failed */ +export const websocketRelayDeliveriesTotal = new client.Counter({ + name: "websocket_relay_deliveries_total", + help: "Relay message deliveries by final outcome (recovered = delivered after at least one retry)", + labelNames: ["outcome"], +}); + +/** result: scheduled | recovered | exhausted | aborted */ +export const websocketRelayRetriesTotal = new client.Counter({ + name: "websocket_relay_retries_total", + help: "Relay retries with exponential backoff, by result", + labelNames: ["operation", "result"], +}); + +export const websocketRelayRetryDelay = new client.Histogram({ + name: "websocket_relay_retry_delay_seconds", + help: "Backoff delay scheduled before a relay retry", + labelNames: ["operation"], + buckets: [0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30], +}); + +/** health_state: 0 = healthy, 1 = degraded, 2 = unhealthy */ +export const websocketRelayHealthState = new client.Gauge({ + name: "websocket_relay_health_state", + help: "WebSocket relay health (0 = healthy, 1 = degraded, 2 = unhealthy)", +}); + +export const websocketRelayFailureRatio = new client.Gauge({ + name: "websocket_relay_delivery_failure_ratio", + help: "Share of relay deliveries that failed or were dead-lettered over the rolling window", +}); + +export const websocketRelayDlqDepth = new client.Gauge({ + name: "websocket_relay_dlq_depth", + help: "Messages currently in the relay dead letter queue", +}); + +export const websocketRelayCircuitOpen = new client.Gauge({ + name: "websocket_relay_circuit_open", + help: "1 while the relay circuit breaker is OPEN, 0.5 while HALF_OPEN, otherwise 0", +}); + +export const websocketRelayLastDeliveryTimestamp = new client.Gauge({ + name: "websocket_relay_last_delivery_timestamp_seconds", + help: "Unix time of the most recent successful relay delivery", +}); + +const DELIVERY_OUTCOMES = new Set(["delivered", "recovered", "dead_lettered", "failed"]); +const RETRY_RESULTS = new Set(["scheduled", "recovered", "exhausted", "aborted"]); +const OPERATIONS = new Set(["deliver", "reconnect", "redrive", "other"]); + +export const HEALTH_STATE_VALUES = Object.freeze({ healthy: 0, degraded: 1, unhealthy: 2 }); + +function readNumberEnv(name, fallback) { + const raw = Number.parseFloat(process.env[name] ?? ""); + return Number.isFinite(raw) && raw >= 0 ? raw : fallback; +} + +export const DEFAULT_RELAY_HEALTH_OPTIONS = Object.freeze({ + windowMs: readNumberEnv("RELAY_HEALTH_WINDOW_MS", 300_000), + bucketMs: 5_000, + minSamples: readNumberEnv("RELAY_HEALTH_MIN_SAMPLES", 20), + failureRatioThreshold: readNumberEnv("RELAY_FAILURE_RATIO_THRESHOLD", 0.1), + dlqDegradedThreshold: readNumberEnv("RELAY_DLQ_DEGRADED_THRESHOLD", 1), + dlqUnhealthyThreshold: readNumberEnv("RELAY_DLQ_UNHEALTHY_THRESHOLD", 100), +}); + +function emptyBucket() { + return { slot: -1, delivered: 0, recovered: 0, deadLettered: 0, failed: 0, exhausted: 0 }; +} + +/** Restrict a label value to a known set. */ +export function safeLabel(value, allowed, fallback) { + return allowed.has(value) ? value : fallback; +} + +/** + * Rolling-window relay health. Counts live in a fixed ring of time buckets, + * so memory is constant regardless of traffic. Circuit breaker state and DLQ + * depth are read live from whatever source was attached. + */ +export class WebSocketRelayHealthMonitor { + constructor(options = {}, now = () => Date.now()) { + this.options = { ...DEFAULT_RELAY_HEALTH_OPTIONS, ...options }; + this.now = now; + this.bucketCount = Math.max(1, Math.ceil(this.options.windowMs / this.options.bucketMs)); + this.source = null; + this.reset(); + } + + reset() { + this.buckets = Array.from({ length: this.bucketCount }, () => emptyBucket()); + this.lastDeliveryAt = null; + } + + /** + * Attach the relay's circuit breaker and dead letter queue so health + * reflects their live state. Pass null to detach. + * + * @param {{ circuitBreaker?: { getStatus(): { state: string } }, dlq?: { size: number } } | null} source + */ + attach(source) { + this.source = source ?? null; + } + + _bucketFor(time) { + const slot = Math.floor(time / this.options.bucketMs); + const bucket = this.buckets[slot % this.bucketCount]; + if (bucket.slot !== slot) Object.assign(bucket, emptyBucket(), { slot }); + return bucket; + } + + /** @param {'delivered'|'recovered'|'dead_lettered'|'failed'} outcome */ + recordDelivery(outcome) { + const time = this.now(); + const bucket = this._bucketFor(time); + if (outcome === "delivered") bucket.delivered += 1; + else if (outcome === "recovered") bucket.recovered += 1; + else if (outcome === "dead_lettered") bucket.deadLettered += 1; + else bucket.failed += 1; + if (outcome === "delivered" || outcome === "recovered") this.lastDeliveryAt = time; + } + + recordRetryExhausted() { + this._bucketFor(this.now()).exhausted += 1; + } + + _liveState() { + let circuitState = "CLOSED"; + let dlqSize = 0; + try { + const status = this.source?.circuitBreaker?.getStatus?.(); + if (status && typeof status.state === "string") circuitState = status.state; + const size = Number(this.source?.dlq?.size ?? 0); + dlqSize = Number.isFinite(size) && size > 0 ? size : 0; + } catch (err) { + logger.warn({ err: err?.message }, "WebSocket relay health source could not be read"); + } + return { circuitState, dlqSize }; + } + + snapshot() { + const time = this.now(); + const currentSlot = Math.floor(time / this.options.bucketMs); + const oldestSlot = currentSlot - this.bucketCount + 1; + const counts = emptyBucket(); + for (const bucket of this.buckets) { + if (bucket.slot < oldestSlot || bucket.slot > currentSlot) continue; + counts.delivered += bucket.delivered; + counts.recovered += bucket.recovered; + counts.deadLettered += bucket.deadLettered; + counts.failed += bucket.failed; + counts.exhausted += bucket.exhausted; + } + + const deliveries = counts.delivered + counts.recovered + counts.deadLettered + counts.failed; + const failures = counts.deadLettered + counts.failed; + const failureRatio = deliveries > 0 ? failures / deliveries : 0; + const { circuitState, dlqSize } = this._liveState(); + const { minSamples, failureRatioThreshold, dlqDegradedThreshold, dlqUnhealthyThreshold } = this.options; + + const reasons = []; + let status = "healthy"; + const degrade = (reason) => { + if (status === "healthy") status = "degraded"; + reasons.push(reason); + }; + const fail = (reason) => { + status = "unhealthy"; + reasons.push(reason); + }; + + if (circuitState === "OPEN") fail("circuit_open"); + else if (circuitState === "HALF_OPEN") degrade("circuit_half_open"); + if (deliveries >= minSamples && failureRatio >= failureRatioThreshold) fail("failure_ratio_exceeded"); + if (dlqUnhealthyThreshold > 0 && dlqSize >= dlqUnhealthyThreshold) fail("dlq_backlog"); + else if (dlqDegradedThreshold > 0 && dlqSize >= dlqDegradedThreshold) degrade("dlq_not_empty"); + if (counts.exhausted > 0) degrade("retries_exhausted"); + + return { + status, + reasons, + window_ms: this.bucketCount * this.options.bucketMs, + deliveries, + delivered: counts.delivered, + recovered: counts.recovered, + dead_lettered: counts.deadLettered, + failed: counts.failed, + retries_exhausted: counts.exhausted, + failure_ratio: Number(failureRatio.toFixed(4)), + circuit_state: circuitState, + dlq_size: dlqSize, + last_delivery_at: this.lastDeliveryAt === null ? null : new Date(this.lastDeliveryAt).toISOString(), + thresholds: { + min_samples: minSamples, + failure_ratio: failureRatioThreshold, + dlq_degraded: dlqDegradedThreshold, + dlq_unhealthy: dlqUnhealthyThreshold, + }, + }; + } +} + +const healthMonitor = new WebSocketRelayHealthMonitor(); + +websocketRelayHealthState.collect = function collectRelayHealth() { + const snap = healthMonitor.snapshot(); + this.set(HEALTH_STATE_VALUES[snap.status] ?? 0); + websocketRelayFailureRatio.set(snap.failure_ratio); + websocketRelayDlqDepth.set(snap.dlq_size); + websocketRelayCircuitOpen.set(snap.circuit_state === "OPEN" ? 1 : snap.circuit_state === "HALF_OPEN" ? 0.5 : 0); + if (healthMonitor.lastDeliveryAt !== null) { + websocketRelayLastDeliveryTimestamp.set(healthMonitor.lastDeliveryAt / 1000); + } +}; + +function safeRecord(fn) { + try { + fn(); + } catch (err) { + logger.error({ err: err?.message }, "Failed to record WebSocket relay telemetry"); + } +} + +/** @param {'delivered'|'recovered'|'dead_lettered'|'failed'} outcome */ +export function recordRelayDelivery(outcome) { + const safe = safeLabel(outcome, DELIVERY_OUTCOMES, "failed"); + safeRecord(() => { + websocketRelayDeliveriesTotal.inc({ outcome: safe }); + healthMonitor.recordDelivery(safe); + }); +} + +/** + * @param {'deliver'|'reconnect'|'redrive'|'other'} operation + * @param {'scheduled'|'recovered'|'exhausted'|'aborted'} result + * @param {number} [delayMs] backoff delay, for `scheduled` + */ +export function recordRelayRetry(operation, result, delayMs) { + const op = safeLabel(operation, OPERATIONS, "other"); + const safe = safeLabel(result, RETRY_RESULTS, "exhausted"); + safeRecord(() => { + websocketRelayRetriesTotal.inc({ operation: op, result: safe }); + if (safe === "scheduled" && Number.isFinite(delayMs) && delayMs >= 0) { + websocketRelayRetryDelay.observe({ operation: op }, delayMs / 1000); + } + if (safe === "exhausted") healthMonitor.recordRetryExhausted(); + }); +} + +/** Attach the relay's circuit breaker and DLQ to the health snapshot. */ +export function attachRelayHealthSource(source) { + healthMonitor.attach(source); +} + +/** Current relay health for /health endpoints. Contains no message data. */ +export function getWebSocketRelayHealth() { + return healthMonitor.snapshot(); +} + +/** Test-only: clear the rolling window and detach any source. Counters are left intact. */ +export function resetWebSocketRelayHealth() { + healthMonitor.reset(); + healthMonitor.attach(null); +} + +register.registerMetric(websocketRelayDeliveriesTotal); +register.registerMetric(websocketRelayRetriesTotal); +register.registerMetric(websocketRelayRetryDelay); +register.registerMetric(websocketRelayHealthState); +register.registerMetric(websocketRelayFailureRatio); +register.registerMetric(websocketRelayDlqDepth); +register.registerMetric(websocketRelayCircuitOpen); +register.registerMetric(websocketRelayLastDeliveryTimestamp); + +export { register as websocketRelayTelemetryRegister }; diff --git a/backend/src/lib/websocket-relay-telemetry.test.js b/backend/src/lib/websocket-relay-telemetry.test.js new file mode 100644 index 00000000..b1ba790b --- /dev/null +++ b/backend/src/lib/websocket-relay-telemetry.test.js @@ -0,0 +1,178 @@ +import { readFileSync } from "node:fs"; +import { describe, it, expect, beforeEach, vi } from "vitest"; + +vi.mock("./logger.js", () => ({ + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }, +})); + +import { + WebSocketRelayHealthMonitor, + attachRelayHealthSource, + getWebSocketRelayHealth, + recordRelayDelivery, + recordRelayRetry, + resetWebSocketRelayHealth, + safeLabel, + websocketRelayTelemetryRegister, +} from "./websocket-relay-telemetry.js"; +import { CircuitBreaker, DeadLetterQueue } from "./websocket-relay-recovery.js"; + +function make(overrides = {}) { + let clock = 1_700_000_000_000; + const monitor = new WebSocketRelayHealthMonitor( + { windowMs: 60_000, bucketMs: 5_000, minSamples: 10, failureRatioThreshold: 0.1, dlqDegradedThreshold: 1, dlqUnhealthyThreshold: 50, ...overrides }, + () => clock, + ); + return { + monitor, + advance(ms) { + clock += ms; + }, + }; +} + +const record = (monitor, outcome, times) => { + for (let i = 0; i < times; i += 1) monitor.recordDelivery(outcome); +}; + +beforeEach(() => { + resetWebSocketRelayHealth(); +}); + +describe("WebSocketRelayHealthMonitor", () => { + it("starts healthy with no data", () => { + const { monitor } = make(); + expect(monitor.snapshot()).toMatchObject({ status: "healthy", reasons: [], deliveries: 0, failure_ratio: 0, circuit_state: "CLOSED", dlq_size: 0 }); + }); + + it("goes unhealthy when the failure ratio crosses the threshold with enough samples", () => { + const { monitor } = make(); + record(monitor, "delivered", 8); + record(monitor, "dead_lettered", 2); + const snap = monitor.snapshot(); + expect(snap.failure_ratio).toBe(0.2); + expect(snap).toMatchObject({ status: "unhealthy", reasons: ["failure_ratio_exceeded"] }); + }); + + it("ignores the ratio below minSamples", () => { + const { monitor } = make(); + record(monitor, "failed", 3); + expect(monitor.snapshot().status).toBe("healthy"); + }); + + it("counts recovered deliveries as successes", () => { + const { monitor } = make(); + record(monitor, "recovered", 20); + expect(monitor.snapshot()).toMatchObject({ status: "healthy", recovered: 20, failure_ratio: 0 }); + }); + + it("forgets data outside the rolling window", () => { + const { monitor, advance } = make(); + record(monitor, "failed", 20); + expect(monitor.snapshot().status).toBe("unhealthy"); + advance(61_000); + expect(monitor.snapshot()).toMatchObject({ status: "healthy", deliveries: 0 }); + }); + + it("reflects the live circuit breaker state", async () => { + const { monitor } = make(); + const breaker = new CircuitBreaker({ failureThreshold: 1, resetTimeoutMs: 1_000, now: () => 0 }); + monitor.attach({ circuitBreaker: breaker, dlq: new DeadLetterQueue() }); + await breaker.call(() => Promise.reject(new Error("down"))).catch(() => {}); + expect(monitor.snapshot()).toMatchObject({ status: "unhealthy", circuit_state: "OPEN", reasons: ["circuit_open"] }); + }); + + it("degrades on a non-empty DLQ and fails on a large backlog", () => { + const { monitor } = make(); + const dlq = new DeadLetterQueue(); + monitor.attach({ dlq }); + dlq.push({}, new Error("x")); + expect(monitor.snapshot()).toMatchObject({ status: "degraded", reasons: ["dlq_not_empty"], dlq_size: 1 }); + for (let i = 0; i < 60; i += 1) dlq.push({}, new Error("x")); + expect(monitor.snapshot()).toMatchObject({ status: "unhealthy", reasons: ["dlq_backlog"] }); + }); + + it("degrades after retries were exhausted in the window", () => { + const { monitor } = make(); + monitor.recordRetryExhausted(); + expect(monitor.snapshot()).toMatchObject({ status: "degraded", retries_exhausted: 1 }); + }); + + it("survives a health source that throws", () => { + const { monitor } = make(); + monitor.attach({ circuitBreaker: { getStatus: () => { throw new Error("boom"); } } }); + expect(monitor.snapshot().status).toBe("healthy"); + }); + + it("keeps memory constant regardless of traffic", () => { + const { monitor, advance } = make(); + for (let i = 0; i < 10_000; i += 1) { + monitor.recordDelivery("delivered"); + advance(37); + } + expect(monitor.buckets).toHaveLength(12); + }); +}); + +describe("recording helpers", () => { + it("restricts labels to known values", () => { + expect(safeLabel("evil\nlabel", new Set(["a"]), "fallback")).toBe("fallback"); + }); + + it("records deliveries and retries with bounded labels", async () => { + recordRelayDelivery("delivered"); + recordRelayDelivery("