diff --git a/backend/.env.example b/backend/.env.example index 6295ebb9..51b93792 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -97,3 +97,15 @@ 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 + +# Log Retention & Archival Service (docs/LOG_RETENTION.md) +# LOG_RETENTION_DAYS=30 +# LOG_PURGE_BATCH_SIZE=1000 +# LOG_PURGE_MAX_DURATION_MS=300000 +# LOG_PURGE_RETRY_MAX_ATTEMPTS=4 +# LOG_PURGE_RETRY_BASE_DELAY_MS=250 +# LOG_PURGE_RETRY_MAX_DELAY_MS=5000 +# Run the purge inside the API on a cron schedule (safe on every instance): +# LOG_RETENTION_PURGE_CRON=30 3 * * * +# Push the cron script's metrics here (http/https only): +# LOG_RETENTION_PUSHGATEWAY_URL=http://pushgateway:9091 diff --git a/backend/docs/LOG_RETENTION.md b/backend/docs/LOG_RETENTION.md new file mode 100644 index 00000000..61e178cf --- /dev/null +++ b/backend/docs/LOG_RETENTION.md @@ -0,0 +1,143 @@ +# Log Retention & Archival Service + +Purges old rows from `webhook_delivery_logs` in small batches +(`npm run purge:webhook-logs`, `src/lib/log-retention.js`). + +Covers issues **#1473** (Prometheus alert metrics and health telemetry), +**#1474** (automated retry with exponential backoff), +**#1475** (distributed concurrency control and locking) and +**#1476** (integration and stress test suite). + +--- + +## 1. Layout + +| File | Role | +|---|---| +| `src/lib/log-retention.js` | `purgeWebhookLogs()` (safety checks + batched delete) and `runLogRetentionPurge()` (the same, under the purge lock) | +| `src/lib/log-retention-telemetry.js` | Alert metrics, run-history health, separate registry, Pushgateway push (#1473) | +| `src/lib/log-retention-retry.js` | Full-jitter exponential backoff for transient database errors (#1474) | +| `src/lib/log-retention-lock.js` | In-process single flight + Postgres advisory lock (#1475) | +| `scripts/purge-webhook-logs.js` | Cron entry point; uses `runLogRetentionPurge()` | +| `docs/alerts/log-retention.rules.yml` | Prometheus alert rules | + +``` +runLogRetentionPurge({ pool }) + ├─ in-process: already running here? ──────────────▶ skipped + ├─ pg_try_advisory_lock(hashtext(key)) on its own + │ pooled connection ── false ─────────────────────▶ skipped (another instance) + └─ purgeWebhookLogs() + ├─ index check, FK check (retried if transient) + ├─ loop: DELETE ... LIMIT n FOR UPDATE SKIP LOCKED (each batch retried) + └─ telemetry: batches, rows, duration, outcome + pg_advisory_unlock, connection returned to the pool +``` + +## 2. Configuration + +| Variable | Default | Meaning | +|---|---|---| +| `LOG_RETENTION_DAYS` | `30` | Rows older than this are deleted | +| `LOG_PURGE_BATCH_SIZE` | `1000` | Rows per delete batch | +| `LOG_PURGE_MAX_DURATION_MS` | `300000` | Time budget; overrunning raises the duration alert, and no retry starts after it | +| `LOG_PURGE_RETRY_MAX_ATTEMPTS` | `4` | Attempts per query, clamped to 1-8 | +| `LOG_PURGE_RETRY_BASE_DELAY_MS` | `250` | First backoff ceiling, clamped to 0-60000 | +| `LOG_PURGE_RETRY_MAX_DELAY_MS` | `5000` | Backoff cap, clamped to base-60000 | +| `LOG_RETENTION_PURGE_CRON` | unset | Also run the purge inside the API on this cron schedule (for example `30 3 * * *`). Safe on every instance | +| `LOG_RETENTION_PUSHGATEWAY_URL` | unset | Pushgateway (http/https) the cron script pushes its metrics to | +| `LOG_RETENTION_HEALTH_MAX_CONSECUTIVE_FAILURES` | `3` | Failure streak that makes health `unhealthy` | +| `LOG_RETENTION_HEALTH_MAX_STALENESS_MS` | `180000000` (50 h) | No success for this long makes health `unhealthy` | +| `LOG_RETENTION_HEALTH_STUCK_AFTER_MS` | `900000` (15 min) | A run in progress this long makes health `degraded` | + +## 3. Metrics (#1473) + +All in `logRetentionRegister`, merged into `GET /metrics`. Labels come from +fixed sets; no row ids, log contents or error messages ever become labels. + +| Metric | Type | Labels | +|---|---|---| +| `log_retention_purge_runs_total` | counter | `outcome`: success, failure, skipped | +| `log_retention_purge_duration_seconds` | histogram | `outcome` | +| `log_retention_purge_deleted_rows_total` | counter | `table` | +| `log_retention_purge_batches_total` | counter | `table` | +| `log_retention_purge_duration_exceeded_total` | counter | | +| `log_retention_purge_retries_total` | counter | `result`: scheduled, recovered, exhausted | +| `log_retention_purge_lock_total` | counter | `result`: acquired, contended, error | +| `log_retention_health_state` | gauge | 0 healthy/idle, 1 degraded, 2 unhealthy | +| `log_retention_purge_in_progress` | gauge | | +| `log_retention_purge_consecutive_failures` | gauge | | +| `log_retention_purge_last_success_timestamp_seconds` | gauge | | +| `log_retention_purge_last_run_timestamp_seconds` | gauge | | + +## 4. Health (#1473) + +`GET /health/log-retention` returns the run history of this process and a +status. It answers 503 only when `unhealthy`. `GET /health` reports the +status under `services.log_retention` for information only. + +| Status | When | +|---|---| +| `idle` | No purge has run in this process (normal for API instances when the purge runs as a cron script) | +| `healthy` | Last run succeeded within budget | +| `degraded` | Last run failed (below the streak limit), overran `LOG_PURGE_MAX_DURATION_MS`, or a run is overdue | +| `unhealthy` | `consecutive_failures` reached the limit, or the last success is older than the staleness limit | + +A run skipped because another instance holds the lock is counted in +`skipped_runs`; it neither resets nor extends the failure streak. + +## 5. Retry (#1474) + +Each query (index check, FK check, every delete batch, lock connect and +acquire) runs through `withLogRetentionRetry()`: + +- Retried: SQLSTATE `40001`, `40P01`, `55P03`, `57014`, `57P01`-`57P03`, + `53300`, any class `08` connection error, and socket errors + (`ECONNRESET`, `ETIMEDOUT`, ...). +- Never retried: safety-check failures (`LogRetentionSafetyError`: missing + timestamp index, referencing foreign keys), and any other SQLSTATE + (syntax, permission, constraint). +- Delay: `random(0, min(maxDelay, base * 2^n))` (full jitter), so instances + that failed together do not retry together. +- Retrying stops once the run is past `LOG_PURGE_MAX_DURATION_MS`. + +Retrying a delete batch is safe: the statement deletes "rows older than the +cutoff", so a batch whose acknowledgement was lost deletes nothing new or the +next oldest rows, never a row inside the retention window. + +## 6. Concurrency (#1475) + +- A second call in the same process while a purge runs returns + `{ skipped: true, reason: "locked" }` without touching the database. +- Across instances, `pg_try_advisory_lock(hashtext('stellar-payment-api:log-retention:webhook_delivery_logs'))` + is taken on a dedicated connection and never waits. If the holder crashes, + Postgres drops the lock with its connection, so it cannot get stuck. +- If releasing the lock fails, the connection is destroyed instead of + returned to the pool, which also drops the lock. +- Delete batches use `FOR UPDATE SKIP LOCKED`, so even callers that bypass + the lock (direct `purgeWebhookLogs()`) never block on or double-delete rows. + +`purgeWebhookLogs()` keeps its original signature and behaviour (no lock) for +existing callers; schedulers should use `runLogRetentionPurge()`. + +## 7. Tests (#1476) + +| Suite | Command | +|---|---| +| Unit: purge, telemetry, retry, lock | `npx vitest run src/lib/log-retention` | +| Integration: lock + retry + telemetry over an in-memory Postgres double, plus `/metrics` and `/health/log-retention` | `npx vitest run tests/integration/log-retention.test.js` | +| Stress: many concurrent instances, injected transient failures and latency | `npm run test:load -- log-retention-stress` | + +The stress suite checks that every expired row is deleted exactly once, no +row inside the retention window is touched, at most one instance holds the +lock at any time, no lock or in-process guard leaks, and the run counters add +up to the number of attempts. + +## 8. Security notes + +- Table, column and lock key are server-side constants; only the retention + days and batch size reach SQL, and only as bind parameters. +- The Pushgateway URL must be http(s) and is never logged (it may carry + credentials). +- Health and metrics expose counts and timestamps only. +- The safety checks (timestamp index present, no referencing foreign keys) + still run first and fail closed. diff --git a/backend/docs/alerts/log-retention.rules.yml b/backend/docs/alerts/log-retention.rules.yml new file mode 100644 index 00000000..ead3faed --- /dev/null +++ b/backend/docs/alerts/log-retention.rules.yml @@ -0,0 +1,83 @@ +# Prometheus alerting rules for the Log Retention & Archival Service (issue #1473). +# +# Load with: rule_files: ["docs/alerts/log-retention.rules.yml"] +# Validate: promtool check rules docs/alerts/log-retention.rules.yml +# +# Metric reference: docs/LOG_RETENTION.md +# +# The purge usually runs as a cron script that pushes to a Pushgateway +# (LOG_RETENTION_PUSHGATEWAY_URL), or inside the API when +# LOG_RETENTION_PURGE_CRON is set. Either way the series below are emitted. + +groups: + - name: log-retention + rules: + - alert: LogRetentionPurgeUnhealthy + expr: max(log_retention_health_state) >= 2 + for: 10m + labels: + severity: critical + annotations: + summary: Webhook log purge reports unhealthy + description: > + The purge failed several times in a row or has not succeeded within + the staleness limit (LOG_RETENTION_HEALTH_MAX_STALENESS_MS). Old + webhook delivery logs are accumulating. See GET /health/log-retention + for the reasons and the last error in the logs. + + - alert: LogRetentionPurgeFailing + expr: max(log_retention_purge_consecutive_failures) >= 1 + for: 30m + labels: + severity: warning + annotations: + summary: Last webhook log purge run failed + description: > + At least one purge run failed and no run has succeeded since. + Check the "Log retention purge failed" log line for the error code. + + - alert: LogRetentionPurgeStale + expr: | + max(log_retention_purge_last_success_timestamp_seconds) > 0 + and + time() - max(log_retention_purge_last_success_timestamp_seconds) > 2 * 86400 + for: 15m + labels: + severity: warning + annotations: + summary: No successful webhook log purge in 48 hours + description: > + The daily purge has not succeeded for two days. Check that the cron + (npm run purge:webhook-logs) or LOG_RETENTION_PURGE_CRON still runs. + + - alert: LogRetentionPurgeDurationExceeded + expr: increase(log_retention_purge_duration_exceeded_total[1d]) > 0 + labels: + severity: warning + annotations: + summary: Webhook log purge overran its time budget + description: > + A purge run took longer than LOG_PURGE_MAX_DURATION_MS. The backlog + may be large or the database slow; consider a larger + LOG_PURGE_BATCH_SIZE or running the purge more often. + + - alert: LogRetentionRetriesExhausted + expr: increase(log_retention_purge_retries_total{result="exhausted"}[1h]) > 0 + labels: + severity: warning + annotations: + summary: Webhook log purge gave up after retries + description: > + A purge query kept failing with transient database errors until the + retry budget (LOG_PURGE_RETRY_MAX_ATTEMPTS) ran out. + + - alert: LogRetentionLockErrors + expr: increase(log_retention_purge_lock_total{result="error"}[1h]) > 0 + labels: + severity: warning + annotations: + summary: Webhook log purge could not take its advisory lock + description: > + Acquiring the Postgres advisory lock failed (not merely contended). + The purge did not run. Check database connectivity from the host + that runs the purge. diff --git a/backend/load-tests/log-retention-stress.test.js b/backend/load-tests/log-retention-stress.test.js new file mode 100644 index 00000000..8c3cd556 --- /dev/null +++ b/backend/load-tests/log-retention-stress.test.js @@ -0,0 +1,190 @@ +/** + * Log Retention & Archival Service — stress suite (issue #1476). + * + * Many simulated instances (one pool each) purge one shared in-memory + * Postgres double concurrently, with injected latency and transient + * failures. No network and no real database. + * + * Invariants checked under load: + * - every expired row is deleted exactly once; rows inside the retention + * window are never touched + * - at most one instance holds the purge lock at any moment + * - no advisory lock, connection or in-process guard leaks, including + * after failures + * - run counters add up: success + failure + skipped == attempts + * - retries absorb a high rate of transient errors + * + * Run with: npm run test:load -- log-retention-stress + */ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; + +vi.mock("../src/lib/logger.js", () => ({ + logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, +})); + +import { purgeWebhookLogs, runLogRetentionPurge } from "../src/lib/log-retention.js"; +import { + logRetentionRunsTotal, + resetLogRetentionHealth, +} from "../src/lib/log-retention-telemetry.js"; +import { resetLogRetentionLocks } from "../src/lib/log-retention-lock.js"; +import { createFakeLogRetentionDb, makeRows, pgError } from "../tests/helpers/fake-log-retention-db.js"; + +const fastRetry = { sleep: async () => {}, baseDelayMs: 0, maxAttempts: 8 }; + +/** Deterministic PRNG so a failing seed can be replayed. */ +function prng(seed) { + let s = seed >>> 0; + return () => { + s = (s * 1664525 + 1013904223) >>> 0; + return s / 2 ** 32; + }; +} + +async function runs(outcome) { + const { values } = await logRetentionRunsTotal.get(); + return values.find((v) => v.labels.outcome === outcome)?.value ?? 0; +} + +describe("log retention stress", () => { + beforeEach(() => { + resetLogRetentionHealth(); + resetLogRetentionLocks(); + vi.spyOn(console, "log").mockImplementation(() => {}); + vi.spyOn(console, "error").mockImplementation(() => {}); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it("50 instances racing: one purges, the rest skip, every expired row deleted once", async () => { + const expired = makeRows(5_000, 90, "old"); + const kept = makeRows(500, 3, "new"); + const db = createFakeLogRetentionDb({ rows: [...expired, ...kept], latencyMs: 1 }); + + const before = { success: await runs("success"), skipped: await runs("skipped") }; + const results = await Promise.all( + Array.from({ length: 50 }, () => + runLogRetentionPurge({ + pool: db.createPool(), + retentionDays: 30, + batchSize: 250, + retry: fastRetry, + }), + ), + ); + + const ran = results.filter((r) => !r.skipped); + expect(ran).toHaveLength(1); + expect(ran[0].totalDeleted).toBe(5_000); + expect(results.filter((r) => r.skipped)).toHaveLength(49); + expect(db.stats.maxLockHolders).toBe(1); + expect(db.doubleDeletes).toEqual([]); + expect(new Set(db.deleted).size).toBe(db.deleted.length); + expect(db.remainingIds().sort()).toEqual(kept.map((r) => r.id).sort()); + expect(db.heldLocks()).toEqual([]); + expect(db.stats.openConnections).toBe(0); + expect(await runs("success")).toBe(before.success + 1); + expect(await runs("skipped")).toBe(before.skipped + 49); + }); + + it("repeated waves with 20% transient failures converge with no leaks", async () => { + const random = prng(1476); + const transient = ["40001", "40P01", "55P03", "57014", "08006", "ECONNRESET"]; + const db = createFakeLogRetentionDb({ + rows: [...makeRows(3_000, 45, "old"), ...makeRows(300, 10, "new")], + latencyMs: () => Math.floor(random() * 3), + fail: (kind) => + kind !== "unlock" && random() < 0.2 + ? pgError(transient[Math.floor(random() * transient.length)]) + : null, + }); + + let attempts = 0; + let outcomes = { ok: 0, skipped: 0, failed: 0 }; + for (let wave = 0; wave < 10 && db.remainingIds().length > 300; wave += 1) { + const results = await Promise.allSettled( + Array.from({ length: 8 }, () => { + attempts += 1; + return runLogRetentionPurge({ + pool: db.createPool(), + retentionDays: 30, + batchSize: 100, + retry: fastRetry, + }); + }), + ); + for (const r of results) { + if (r.status === "rejected") outcomes.failed += 1; + else if (r.value.skipped) outcomes.skipped += 1; + else outcomes.ok += 1; + } + expect(db.heldLocks()).toEqual([]); + expect(db.stats.openConnections).toBe(0); + } + + expect(outcomes.ok + outcomes.skipped + outcomes.failed).toBe(attempts); + expect(outcomes.ok).toBeGreaterThanOrEqual(1); + expect(db.remainingIds()).toHaveLength(300); + expect(db.remainingIds().every((id) => id.startsWith("new-"))).toBe(true); + expect(db.doubleDeletes).toEqual([]); + expect(db.stats.maxLockHolders).toBe(1); + }); + + it("unlocked direct callers under contention still delete each row once (SKIP LOCKED)", async () => { + const db = createFakeLogRetentionDb({ + rows: makeRows(4_000, 120), + latencyMs: () => 1, + }); + + const results = await Promise.all( + Array.from({ length: 16 }, () => + purgeWebhookLogs({ + pool: db.createPool(), + retentionDays: 30, + batchSize: 64, + retry: fastRetry, + }), + ), + ); + + expect(results.reduce((sum, r) => sum + r.totalDeleted, 0)).toBe(4_000); + expect(db.doubleDeletes).toEqual([]); + expect(db.remainingIds()).toEqual([]); + }); + + it("a burst of triggers on one instance runs at most one purge at a time", async () => { + const db = createFakeLogRetentionDb({ rows: makeRows(1_000, 40), latencyMs: 1 }); + const pool = db.createPool(); + + const results = await Promise.all( + Array.from({ length: 100 }, () => + runLogRetentionPurge({ pool, retentionDays: 30, batchSize: 100, retry: fastRetry }), + ), + ); + + expect(results.filter((r) => !r.skipped)).toHaveLength(1); + // Skipped in-process: only the winner ever reached the database lock. + expect(db.callCount("lock")).toBe(1); + expect(db.remainingIds()).toEqual([]); + }); + + it("keeps purge throughput reasonable for a large backlog", async () => { + const db = createFakeLogRetentionDb({ rows: makeRows(50_000, 365) }); + const started = Date.now(); + + const result = await runLogRetentionPurge({ + pool: db.createPool(), + retentionDays: 30, + batchSize: 1_000, + retry: fastRetry, + }); + + expect(result.totalDeleted).toBe(50_000); + expect(db.callCount("delete")).toBe(51); + // Generous ceiling: this guards against accidental O(n^2) per-row work + // in the service, not against a slow CI machine. + expect(Date.now() - started).toBeLessThan(10_000); + }); +}); diff --git a/backend/scripts/purge-webhook-logs.js b/backend/scripts/purge-webhook-logs.js index e3afad7d..cae57430 100644 --- a/backend/scripts/purge-webhook-logs.js +++ b/backend/scripts/purge-webhook-logs.js @@ -1,15 +1,24 @@ import "dotenv/config"; import { pool, closePool } from "../src/lib/db.js"; -import { purgeWebhookLogs } from "../src/lib/log-retention.js"; +import { runLogRetentionPurge } from "../src/lib/log-retention.js"; +import { pushLogRetentionMetrics } from "../src/lib/log-retention-telemetry.js"; async function run() { const startedAt = Date.now(); try { - const result = await purgeWebhookLogs({ pool }); + // Takes the cross-instance purge lock (issue #1475): when the cron runs + // on several hosts, exactly one purges and the others exit cleanly. + const result = await runLogRetentionPurge({ pool }); const elapsedMs = Date.now() - startedAt; - console.log( - `[log-retention] Run succeeded: table=${result.tableName} deleted_rows=${result.totalDeleted} elapsed_ms=${elapsedMs}`, - ); + if (result.skipped) { + console.log( + `[log-retention] Run skipped: table=${result.tableName} another purge holds the lock elapsed_ms=${elapsedMs}`, + ); + } else { + console.log( + `[log-retention] Run succeeded: table=${result.tableName} deleted_rows=${result.totalDeleted} elapsed_ms=${elapsedMs}`, + ); + } } catch (error) { const elapsedMs = Date.now() - startedAt; console.error( @@ -18,6 +27,11 @@ async function run() { ); process.exitCode = 1; } finally { + // This process exits before any scrape could reach it, so push the run's + // metrics (issue #1473) when a Pushgateway is configured. + if (process.env.LOG_RETENTION_PUSHGATEWAY_URL) { + await pushLogRetentionMetrics(process.env.LOG_RETENTION_PUSHGATEWAY_URL); + } await closePool(); } } diff --git a/backend/src/app.js b/backend/src/app.js index b330d308..797aad56 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 { getLogRetentionHealth } from "./lib/log-retention-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, + log_retention: getLogRetentionHealth().status, }, }); }); @@ -314,6 +316,31 @@ export async function createApp({ redisClient }) { res.status(health.status === "unhealthy" ? 503 : 200).json(health); }); + /** + * @swagger + * /health/log-retention: + * get: + * summary: Log Retention & Archival Service health telemetry + * description: > + * Run history and derived status for the webhook delivery log purge + * in this process (issue #1473): last run / last success times, + * consecutive failures, duration budget overruns and whether a run is + * in progress. Returns 503 only when unhealthy (repeated failures or + * no success within the staleness limit). "idle" means no purge has + * run in this process. The body contains no log contents or ids. + * tags: [Health] + * security: [] + * responses: + * 200: + * description: Purge healthy, degraded or idle + * 503: + * description: Purge unhealthy + */ + app.get("/health/log-retention", (_req, res) => { + const health = getLogRetentionHealth(); + 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-lock.js b/backend/src/lib/log-retention-lock.js new file mode 100644 index 00000000..38ed587a --- /dev/null +++ b/backend/src/lib/log-retention-lock.js @@ -0,0 +1,170 @@ +/** + * Distributed concurrency control for the Log Retention & Archival Service + * (issue #1475). + * + * Problem + * ------- + * The webhook log purge can be started from several places at once: the + * `npm run purge:webhook-logs` cron on more than one host, the optional + * in-server schedule on every API instance (LOG_RETENTION_PURGE_CRON), or an + * operator running it by hand. Concurrent purges race on the same rows, hold + * row locks against each other, multiply load on the primary and make the + * duration alert fire for the wrong reason. + * + * Solution + * -------- + * 1. In-process single flight: a second call on the same pool while one is + * running is skipped without touching the database. + * 2. Cross-instance: a PostgreSQL session advisory lock, + * `pg_try_advisory_lock(hashtext())`, taken on a dedicated pooled + * connection. It never waits: a contended lock means another instance is + * already purging, so this run is skipped (not failed). + * Postgres releases a session lock automatically when the connection + * closes, so a crashed or killed purge can never leave the lock stuck. + * 3. The delete batches also use `FOR UPDATE SKIP LOCKED` (log-retention.js), + * so even an unlocked caller never blocks on, or double-deletes, rows a + * running purge already holds. + * + * The lock key is a fixed server-side constant per table, never input. + */ + +import { logger } from './logger.js'; +import { recordLogRetentionLock } from './log-retention-telemetry.js'; +import { withLogRetentionRetry } from './log-retention-retry.js'; + +export const LOG_RETENTION_LOCK_PREFIX = 'stellar-payment-api:log-retention'; + +const ACQUIRE_SQL = 'SELECT pg_try_advisory_lock(hashtext($1)) AS acquired'; +const RELEASE_SQL = 'SELECT pg_advisory_unlock(hashtext($1)) AS released'; + +/** + * pool -> keys held (or being acquired) by a run in this process. Keyed by + * pool so two pools in one process (as in the multi-instance stress tests) + * behave like two instances and contend on the database lock instead. + */ +let inProcess = new WeakMap(); + +function heldKeys(pool) { + let keys = inProcess.get(pool); + if (!keys) { + keys = new Set(); + inProcess.set(pool, keys); + } + return keys; +} + +/** @param {string} tableName */ +export function buildLogRetentionLockKey(tableName) { + if (!tableName || typeof tableName !== 'string' || !/^[a-z_][a-z0-9_]*$/.test(tableName)) { + throw new TypeError('tableName must be a plain lower-case SQL identifier'); + } + return `${LOG_RETENTION_LOCK_PREFIX}:${tableName}`; +} + +/** Test-only: forget in-process holders. */ +export function resetLogRetentionLocks() { + inProcess = new WeakMap(); +} + +/** + * Try to take the purge lock without waiting. + * + * @param {object} pool pg Pool (uses pool.connect()) or a single client + * @param {string} key From buildLogRetentionLockKey() + * @param {object} [retryOptions] Passed to withLogRetentionRetry + * @returns {Promise Promise}>} + * `null` when another run (this process or another instance) holds it. + */ +export async function acquireLogRetentionLock(pool, key, retryOptions = {}) { + if (!pool || typeof pool.query !== 'function') { + throw new Error('A pg pool/client is required to acquire the log retention lock.'); + } + + const held = heldKeys(pool); + if (held.has(key)) { + recordLogRetentionLock('contended'); + return null; + } + held.add(key); + + let client = null; + let ownsClient = false; + try { + if (typeof pool.connect === 'function') { + client = await withLogRetentionRetry(() => pool.connect(), { + ...retryOptions, + label: 'log-retention-lock-connect', + }); + ownsClient = true; + } else { + client = pool; + } + + const result = await withLogRetentionRetry(() => client.query(ACQUIRE_SQL, [key]), { + ...retryOptions, + label: 'log-retention-lock-acquire', + }); + const acquired = result?.rows?.[0]?.acquired === true; + + if (!acquired) { + recordLogRetentionLock('contended'); + if (ownsClient) client.release(); + held.delete(key); + return null; + } + + recordLogRetentionLock('acquired'); + let released = false; + return { + key, + release: async () => { + if (released) return false; + released = true; + try { + const res = await client.query(RELEASE_SQL, [key]); + const ok = res?.rows?.[0]?.released === true; + if (!ok) { + logger.warn({ key }, 'Log retention lock was not held at release'); + } + if (ownsClient) client.release(); + return ok; + } catch (err) { + // Destroy the connection instead of returning it to the pool: the + // server then drops the session lock with it. + logger.error({ err: err?.message, key }, 'Log retention lock release failed'); + if (ownsClient) client.release(err instanceof Error ? err : true); + return false; + } finally { + held.delete(key); + } + }, + }; + } catch (err) { + recordLogRetentionLock('error'); + if (ownsClient && client) client.release(err instanceof Error ? err : true); + held.delete(key); + throw err; + } +} + +/** + * Run `fn` while holding the purge lock. Returns `{ skipped: true }` without + * calling `fn` when the lock is held elsewhere. + * + * @template T + * @param {object} pool + * @param {string} key + * @param {() => Promise} fn + * @param {object} [retryOptions] + * @returns {Promise<{skipped:true} | {skipped:false, result:T}>} + */ +export async function withLogRetentionLock(pool, key, fn, retryOptions = {}) { + const lock = await acquireLogRetentionLock(pool, key, retryOptions); + if (!lock) return { skipped: true }; + try { + const result = await fn(); + return { skipped: false, result }; + } finally { + await lock.release(); + } +} diff --git a/backend/src/lib/log-retention-lock.test.js b/backend/src/lib/log-retention-lock.test.js new file mode 100644 index 00000000..b85075b6 --- /dev/null +++ b/backend/src/lib/log-retention-lock.test.js @@ -0,0 +1,239 @@ +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 { + LOG_RETENTION_LOCK_PREFIX, + acquireLogRetentionLock, + buildLogRetentionLockKey, + resetLogRetentionLocks, + withLogRetentionLock, +} from "./log-retention-lock.js"; +import { logRetentionLockTotal } from "./log-retention-telemetry.js"; + +const KEY = buildLogRetentionLockKey("webhook_delivery_logs"); +const noRetry = { maxAttempts: 1 }; + +async function lockCount(result) { + const { values } = await logRetentionLockTotal.get(); + return values.find((v) => v.labels.result === result)?.value ?? 0; +} + +/** pg Pool double: connect() hands out one client whose query is scripted. */ +function makePool(queryImpl) { + const client = { + query: vi.fn(queryImpl), + release: vi.fn(), + }; + return { + client, + pool: { + query: vi.fn(), + connect: vi.fn(async () => client), + }, + }; +} + +const granted = async (sql) => + /advisory_unlock/.test(sql) + ? { rows: [{ released: true }] } + : { rows: [{ acquired: true }] }; + +describe("buildLogRetentionLockKey (#1475)", () => { + it("namespaces the table name", () => { + expect(KEY).toBe(`${LOG_RETENTION_LOCK_PREFIX}:webhook_delivery_logs`); + }); + + it.each(["", null, "Webhook", "logs; DROP TABLE x", "a-b", "1abc", 7])( + "rejects %p", + (name) => { + expect(() => buildLogRetentionLockKey(name)).toThrow(TypeError); + }, + ); +}); + +describe("acquireLogRetentionLock", () => { + beforeEach(() => { + resetLogRetentionLocks(); + }); + + it("takes a session advisory lock on a dedicated connection and releases it", async () => { + const acquiredBefore = await lockCount("acquired"); + const { pool, client } = makePool(granted); + + const lock = await acquireLogRetentionLock(pool, KEY, noRetry); + + expect(lock).not.toBeNull(); + expect(pool.connect).toHaveBeenCalledTimes(1); + expect(client.query).toHaveBeenCalledWith( + expect.stringContaining("pg_try_advisory_lock(hashtext($1))"), + [KEY], + ); + expect(await lockCount("acquired")).toBe(acquiredBefore + 1); + + await expect(lock.release()).resolves.toBe(true); + expect(client.query).toHaveBeenLastCalledWith( + expect.stringContaining("pg_advisory_unlock(hashtext($1))"), + [KEY], + ); + expect(client.release).toHaveBeenCalledWith(); + }); + + it("returns null and frees the connection when another session holds the lock", async () => { + const before = await lockCount("contended"); + const { pool, client } = makePool(async () => ({ rows: [{ acquired: false }] })); + + await expect(acquireLogRetentionLock(pool, KEY, noRetry)).resolves.toBeNull(); + expect(client.release).toHaveBeenCalledTimes(1); + expect(await lockCount("contended")).toBe(before + 1); + + // The in-process guard was cleared: a later attempt reaches the database. + await acquireLogRetentionLock(pool, KEY, noRetry); + expect(pool.connect).toHaveBeenCalledTimes(2); + }); + + it("skips a second acquire on the same pool without touching the database", async () => { + const { pool } = makePool(granted); + const first = await acquireLogRetentionLock(pool, KEY, noRetry); + + await expect(acquireLogRetentionLock(pool, KEY, noRetry)).resolves.toBeNull(); + expect(pool.connect).toHaveBeenCalledTimes(1); + + await first.release(); + const again = await acquireLogRetentionLock(pool, KEY, noRetry); + expect(again).not.toBeNull(); + await again.release(); + }); + + it("treats separate pools as separate instances", async () => { + const a = makePool(granted); + const b = makePool(granted); + const lockA = await acquireLogRetentionLock(a.pool, KEY, noRetry); + const lockB = await acquireLogRetentionLock(b.pool, KEY, noRetry); + expect(b.pool.connect).toHaveBeenCalledTimes(1); + await lockA.release(); + await lockB.release(); + }); + + it("destroys the connection and rethrows when acquiring errors", async () => { + const before = await lockCount("error"); + const err = Object.assign(new Error("permission denied"), { code: "42501" }); + const { pool, client } = makePool(async () => { + throw err; + }); + + await expect(acquireLogRetentionLock(pool, KEY, noRetry)).rejects.toBe(err); + expect(client.release).toHaveBeenCalledWith(err); + expect(await lockCount("error")).toBe(before + 1); + + // Guard cleared after the failure. + await expect(acquireLogRetentionLock(pool, KEY, noRetry)).rejects.toBe(err); + expect(pool.connect).toHaveBeenCalledTimes(2); + }); + + it("retries a transient acquire failure", async () => { + let calls = 0; + const { pool } = makePool(async (sql) => { + if (/try_advisory_lock/.test(sql) && calls++ === 0) { + throw Object.assign(new Error("deadlock"), { code: "40P01" }); + } + return granted(sql); + }); + + const lock = await acquireLogRetentionLock(pool, KEY, { + maxAttempts: 3, + baseDelayMs: 0, + sleep: async () => {}, + }); + expect(lock).not.toBeNull(); + await lock.release(); + }); + + it("rethrows when the pool cannot hand out a connection", async () => { + const err = new Error("too many clients"); + const pool = { query: vi.fn(), connect: vi.fn().mockRejectedValue(err) }; + await expect(acquireLogRetentionLock(pool, KEY, noRetry)).rejects.toBe(err); + // Not left marked as held. + pool.connect.mockResolvedValueOnce({ query: granted, release: vi.fn() }); + await expect(acquireLogRetentionLock(pool, KEY, noRetry)).resolves.not.toBeNull(); + }); + + it("destroys the connection when release fails, and only releases once", async () => { + const { pool, client } = makePool(async (sql) => { + if (/advisory_unlock/.test(sql)) throw new Error("connection lost"); + return { rows: [{ acquired: true }] }; + }); + const lock = await acquireLogRetentionLock(pool, KEY, noRetry); + + await expect(lock.release()).resolves.toBe(false); + expect(client.release).toHaveBeenCalledWith(expect.any(Error)); + await expect(lock.release()).resolves.toBe(false); + expect(client.release).toHaveBeenCalledTimes(1); + + // Guard cleared even though the release failed. + const next = await acquireLogRetentionLock(pool, KEY, noRetry); + expect(next).not.toBeNull(); + }); + + it("reports false when the lock was already gone at release", async () => { + const { pool, client } = makePool(async (sql) => + /advisory_unlock/.test(sql) ? { rows: [{ released: false }] } : { rows: [{ acquired: true }] }, + ); + const lock = await acquireLogRetentionLock(pool, KEY, noRetry); + await expect(lock.release()).resolves.toBe(false); + expect(client.release).toHaveBeenCalledWith(); + }); + + it("works on a single client without connect()", async () => { + const client = { query: vi.fn(granted) }; + const lock = await acquireLogRetentionLock(client, KEY, noRetry); + expect(lock).not.toBeNull(); + await expect(lock.release()).resolves.toBe(true); + }); + + it("requires a pool", async () => { + await expect(acquireLogRetentionLock(null, KEY)).rejects.toThrow(/pg pool/); + }); +}); + +describe("withLogRetentionLock", () => { + beforeEach(() => resetLogRetentionLocks()); + + it("runs fn under the lock and releases it afterwards", async () => { + const { pool, client } = makePool(granted); + const fn = vi.fn(async () => 42); + + await expect(withLogRetentionLock(pool, KEY, fn, noRetry)).resolves.toEqual({ + skipped: false, + result: 42, + }); + expect(fn).toHaveBeenCalledTimes(1); + expect(client.release).toHaveBeenCalledTimes(1); + }); + + it("skips fn when the lock is held elsewhere", async () => { + const { pool } = makePool(async () => ({ rows: [{ acquired: false }] })); + const fn = vi.fn(); + + await expect(withLogRetentionLock(pool, KEY, fn, noRetry)).resolves.toEqual({ skipped: true }); + expect(fn).not.toHaveBeenCalled(); + }); + + it("releases the lock when fn throws", async () => { + const { pool, client } = makePool(granted); + const err = new Error("purge failed"); + + await expect( + withLogRetentionLock(pool, KEY, async () => { + throw err; + }, noRetry), + ).rejects.toBe(err); + expect(client.query).toHaveBeenLastCalledWith( + expect.stringContaining("pg_advisory_unlock"), + [KEY], + ); + expect(client.release).toHaveBeenCalledTimes(1); + }); +}); diff --git a/backend/src/lib/log-retention-retry.js b/backend/src/lib/log-retention-retry.js new file mode 100644 index 00000000..925d30fe --- /dev/null +++ b/backend/src/lib/log-retention-retry.js @@ -0,0 +1,197 @@ +/** + * Automated retry with exponential backoff for the Log Retention & Archival + * Service (issue #1474). + * + * Every query the purge issues is safe to repeat: the safety checks are + * read-only and each delete batch removes "rows older than the cutoff", so a + * batch that is retried after a lost acknowledgement simply deletes nothing + * or the next rows. Only transient database/network failures are retried: + * + * 40001 serialization_failure 40P01 deadlock_detected + * 55P03 lock_not_available 57014 query_canceled (statement timeout) + * 57P01/57P02/57P03 server restart 53300 too_many_connections + * 08xxx connection exceptions ECONNRESET / ETIMEDOUT / ... socket errors + * + * Safety-check rejections (missing index, foreign keys), syntax and + * permission errors are never retried, so a purge that must not run cannot + * be forced through by repetition. + * + * Delay schedule: full jitter + * delay(n) = random(0, min(maxDelayMs, baseDelayMs * 2^n)) + * Attempts and delays are clamped to hard ceilings so a misconfigured + * environment cannot turn a failing purge into a load generator against + * the primary database. + */ + +import { logger } from './logger.js'; +import { recordLogRetentionRetry } from './log-retention-telemetry.js'; + +export const DEFAULT_LOG_RETENTION_RETRY_OPTIONS = Object.freeze({ + maxAttempts: 4, + baseDelayMs: 250, + maxDelayMs: 5_000, +}); + +const MAX_ATTEMPTS_CEILING = 8; +const MAX_DELAY_CEILING_MS = 60_000; + +const RETRYABLE_SQLSTATES = new Set([ + '40001', + '40P01', + '55P03', + '57014', + '57P01', + '57P02', + '57P03', + '53300', +]); + +const RETRYABLE_NETWORK_CODES = new Set([ + 'ECONNRESET', + 'ECONNREFUSED', + 'ECONNABORTED', + 'ETIMEDOUT', + 'EPIPE', + 'EAI_AGAIN', + 'EHOSTUNREACH', + 'ENETUNREACH', +]); + +/** + * @param {unknown} err + * @returns {boolean} + */ +export function isRetryableLogRetentionError(err) { + if (!err || typeof err !== 'object') return false; + if (err.retryable === false) return false; + if (err.retryable === true) return true; + + const code = typeof err.code === 'string' ? err.code : ''; + if (RETRYABLE_SQLSTATES.has(code)) return true; + // Class 08: connection exceptions (08000, 08003, 08006, ...). + if (/^08[0-9A-Z]{3}$/.test(code)) return true; + if (RETRYABLE_NETWORK_CODES.has(code)) return true; + // Any other SQLSTATE (syntax, permission, constraint, ...) is deterministic. + if (/^[0-9A-Z]{5}$/.test(code)) return false; + + const message = typeof err.message === 'string' ? err.message : ''; + return /connection terminated|connection timeout|timeout exceeded when trying to connect|socket hang up/i.test( + message, + ); +} + +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); +} + +export function resolveLogRetentionRetryOptions(options = {}, env = process.env) { + const maxAttempts = clampInt( + options.maxAttempts ?? env.LOG_PURGE_RETRY_MAX_ATTEMPTS, + DEFAULT_LOG_RETENTION_RETRY_OPTIONS.maxAttempts, + 1, + MAX_ATTEMPTS_CEILING, + ); + const baseDelayMs = clampInt( + options.baseDelayMs ?? env.LOG_PURGE_RETRY_BASE_DELAY_MS, + DEFAULT_LOG_RETENTION_RETRY_OPTIONS.baseDelayMs, + 0, + MAX_DELAY_CEILING_MS, + ); + const maxDelayMs = clampInt( + options.maxDelayMs ?? env.LOG_PURGE_RETRY_MAX_DELAY_MS, + DEFAULT_LOG_RETENTION_RETRY_OPTIONS.maxDelayMs, + baseDelayMs, + MAX_DELAY_CEILING_MS, + ); + return { maxAttempts, baseDelayMs, maxDelayMs }; +} + +/** + * @param {number} retryIndex 0 for the first retry + * @param {{baseDelayMs:number, maxDelayMs:number}} opts + * @param {() => number} [random] + */ +export function computeLogRetentionBackoffDelay( + retryIndex, + { baseDelayMs, maxDelayMs }, + random = Math.random, +) { + const ceiling = Math.min(maxDelayMs, baseDelayMs * 2 ** retryIndex); + if (ceiling <= 0) return 0; + return Math.floor(random() * ceiling); +} + +const defaultSleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); + +/** + * Run `operation`, retrying transient failures with full-jitter exponential + * backoff. The final error is rethrown with `retryAttempts` set. + * + * @template T + * @param {(ctx:{attempt:number}) => Promise} operation + * @param {object} [options] + * @param {string} [options.label] Fixed, server-side step name for logs + * @param {number} [options.maxAttempts] + * @param {number} [options.baseDelayMs] + * @param {number} [options.maxDelayMs] + * @param {(err:unknown) => boolean} [options.shouldRetry] + * @param {(ms:number) => Promise} [options.sleep] + * @param {() => number} [options.random] + * @param {() => boolean} [options.shouldAbort] Stop retrying (e.g. deadline passed) + * @returns {Promise} + */ +export async function withLogRetentionRetry(operation, options = {}) { + const { + label = 'log-retention-query', + shouldRetry = isRetryableLogRetentionError, + sleep = defaultSleep, + random = Math.random, + shouldAbort = () => false, + } = options; + const resolved = resolveLogRetentionRetryOptions(options); + + let lastError; + for (let attempt = 1; attempt <= resolved.maxAttempts; attempt += 1) { + try { + const result = await operation({ attempt }); + if (attempt > 1) recordLogRetentionRetry('recovered'); + return result; + } catch (err) { + lastError = err; + const retryable = shouldRetry(err); + const exhausted = attempt >= resolved.maxAttempts || (retryable && shouldAbort()); + + if (!retryable || exhausted) { + if (err && typeof err === 'object') err.retryAttempts = attempt; + if (retryable && exhausted) { + recordLogRetentionRetry('exhausted'); + logger.error( + { label, attempts: attempt, err: err?.message, code: err?.code }, + 'Log retention query failed after exhausting retries', + ); + } + throw err; + } + + const delayMs = computeLogRetentionBackoffDelay(attempt - 1, resolved, random); + logger.warn( + { + label, + attempt, + maxAttempts: resolved.maxAttempts, + delayMs, + err: err?.message, + code: err?.code, + }, + 'Log retention query failed, retrying with backoff', + ); + recordLogRetentionRetry('scheduled'); + await sleep(delayMs); + } + } + + throw lastError; +} diff --git a/backend/src/lib/log-retention-retry.test.js b/backend/src/lib/log-retention-retry.test.js new file mode 100644 index 00000000..4536c3c5 --- /dev/null +++ b/backend/src/lib/log-retention-retry.test.js @@ -0,0 +1,229 @@ +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_LOG_RETENTION_RETRY_OPTIONS, + computeLogRetentionBackoffDelay, + isRetryableLogRetentionError, + resolveLogRetentionRetryOptions, + withLogRetentionRetry, +} from "./log-retention-retry.js"; +import { logRetentionRetriesTotal } from "./log-retention-telemetry.js"; +import { logger } from "./logger.js"; + +const pgError = (code, message = `pg ${code}`) => Object.assign(new Error(message), { code }); + +async function retryCount(result) { + const { values } = await logRetentionRetriesTotal.get(); + return values.find((v) => v.labels.result === result)?.value ?? 0; +} + +describe("isRetryableLogRetentionError (#1474)", () => { + it.each(["40001", "40P01", "55P03", "57014", "57P01", "57P02", "57P03", "53300"])( + "retries transient SQLSTATE %s", + (code) => { + expect(isRetryableLogRetentionError(pgError(code))).toBe(true); + }, + ); + + it.each(["08000", "08003", "08006", "08P01"])("retries connection exception %s", (code) => { + expect(isRetryableLogRetentionError(pgError(code))).toBe(true); + }); + + it.each(["ECONNRESET", "ETIMEDOUT", "ECONNREFUSED", "EPIPE"])("retries socket error %s", (code) => { + expect(isRetryableLogRetentionError(pgError(code))).toBe(true); + }); + + it.each(["42601", "42501", "23503", "42P01", "22P02"])( + "never retries deterministic SQLSTATE %s", + (code) => { + expect(isRetryableLogRetentionError(pgError(code))).toBe(false); + }, + ); + + it("retries pg-pool connection messages without a code", () => { + expect(isRetryableLogRetentionError(new Error("Connection terminated unexpectedly"))).toBe(true); + expect( + isRetryableLogRetentionError(new Error("timeout exceeded when trying to connect")), + ).toBe(true); + }); + + it("honours an explicit retryable flag either way", () => { + expect(isRetryableLogRetentionError(Object.assign(pgError("40001"), { retryable: false }))).toBe( + false, + ); + expect(isRetryableLogRetentionError(Object.assign(new Error("x"), { retryable: true }))).toBe(true); + }); + + it.each([null, undefined, "boom", 42, new Error("Missing index for table")])( + "does not retry %p", + (err) => { + expect(isRetryableLogRetentionError(err)).toBe(false); + }, + ); +}); + +describe("resolveLogRetentionRetryOptions", () => { + it("uses defaults", () => { + expect(resolveLogRetentionRetryOptions({}, {})).toEqual(DEFAULT_LOG_RETENTION_RETRY_OPTIONS); + }); + + it("reads env vars", () => { + expect( + resolveLogRetentionRetryOptions( + {}, + { + LOG_PURGE_RETRY_MAX_ATTEMPTS: "6", + LOG_PURGE_RETRY_BASE_DELAY_MS: "10", + LOG_PURGE_RETRY_MAX_DELAY_MS: "100", + }, + ), + ).toEqual({ maxAttempts: 6, baseDelayMs: 10, maxDelayMs: 100 }); + }); + + it("lets explicit options win over env", () => { + expect( + resolveLogRetentionRetryOptions({ maxAttempts: 2 }, { LOG_PURGE_RETRY_MAX_ATTEMPTS: "6" }) + .maxAttempts, + ).toBe(2); + }); + + it("clamps to safe ceilings and floors", () => { + expect( + resolveLogRetentionRetryOptions( + { maxAttempts: 1000, baseDelayMs: -5, maxDelayMs: 10_000_000 }, + {}, + ), + ).toEqual({ maxAttempts: 8, baseDelayMs: 0, maxDelayMs: 60_000 }); + expect(resolveLogRetentionRetryOptions({ maxAttempts: 0 }, {}).maxAttempts).toBe(1); + }); + + it("never lets maxDelay drop below baseDelay", () => { + expect(resolveLogRetentionRetryOptions({ baseDelayMs: 500, maxDelayMs: 100 }, {})).toMatchObject({ + baseDelayMs: 500, + maxDelayMs: 500, + }); + }); + + it("ignores non-numeric env values", () => { + expect( + resolveLogRetentionRetryOptions({}, { LOG_PURGE_RETRY_MAX_ATTEMPTS: "lots" }).maxAttempts, + ).toBe(DEFAULT_LOG_RETENTION_RETRY_OPTIONS.maxAttempts); + }); +}); + +describe("computeLogRetentionBackoffDelay", () => { + const opts = { baseDelayMs: 100, maxDelayMs: 1000 }; + + it("grows the ceiling exponentially and caps it", () => { + const top = () => 0.999999; + expect(computeLogRetentionBackoffDelay(0, opts, top)).toBe(99); + expect(computeLogRetentionBackoffDelay(1, opts, top)).toBe(199); + expect(computeLogRetentionBackoffDelay(2, opts, top)).toBe(399); + expect(computeLogRetentionBackoffDelay(10, opts, top)).toBe(999); + }); + + it("applies full jitter", () => { + expect(computeLogRetentionBackoffDelay(3, opts, () => 0)).toBe(0); + expect(computeLogRetentionBackoffDelay(3, opts, () => 0.5)).toBe(400); + }); + + it("returns 0 with a zero base delay", () => { + expect(computeLogRetentionBackoffDelay(4, { baseDelayMs: 0, maxDelayMs: 0 })).toBe(0); + }); +}); + +describe("withLogRetentionRetry", () => { + const sleep = vi.fn(async () => {}); + + beforeEach(() => { + sleep.mockClear(); + vi.mocked(logger.warn).mockClear(); + vi.mocked(logger.error).mockClear(); + }); + + it("returns the first successful result without sleeping", async () => { + const op = vi.fn(async () => "ok"); + await expect(withLogRetentionRetry(op, { sleep })).resolves.toBe("ok"); + expect(op).toHaveBeenCalledTimes(1); + expect(sleep).not.toHaveBeenCalled(); + }); + + it("retries transient failures with backoff and records recovery", async () => { + const before = await retryCount("recovered"); + const scheduledBefore = await retryCount("scheduled"); + const op = vi + .fn() + .mockRejectedValueOnce(pgError("40P01")) + .mockRejectedValueOnce(pgError("ECONNRESET")) + .mockResolvedValueOnce("done"); + + const result = await withLogRetentionRetry(op, { + sleep, + random: () => 0.5, + baseDelayMs: 100, + maxDelayMs: 1000, + maxAttempts: 4, + }); + + expect(result).toBe("done"); + expect(op).toHaveBeenCalledTimes(3); + expect(op.mock.calls.map(([ctx]) => ctx.attempt)).toEqual([1, 2, 3]); + expect(sleep.mock.calls.map(([ms]) => ms)).toEqual([50, 100]); + expect(await retryCount("recovered")).toBe(before + 1); + expect(await retryCount("scheduled")).toBe(scheduledBefore + 2); + expect(logger.warn).toHaveBeenCalledTimes(2); + }); + + it("throws a non-retryable error immediately", async () => { + const err = pgError("42501", "permission denied"); + const op = vi.fn().mockRejectedValue(err); + + await expect(withLogRetentionRetry(op, { sleep })).rejects.toBe(err); + expect(op).toHaveBeenCalledTimes(1); + expect(sleep).not.toHaveBeenCalled(); + expect(err.retryAttempts).toBe(1); + }); + + it("gives up after maxAttempts and records exhaustion", async () => { + const before = await retryCount("exhausted"); + const op = vi.fn().mockRejectedValue(pgError("40001")); + + await expect( + withLogRetentionRetry(op, { sleep, maxAttempts: 3, baseDelayMs: 1 }), + ).rejects.toMatchObject({ code: "40001", retryAttempts: 3 }); + expect(op).toHaveBeenCalledTimes(3); + expect(sleep).toHaveBeenCalledTimes(2); + expect(await retryCount("exhausted")).toBe(before + 1); + expect(logger.error).toHaveBeenCalledTimes(1); + }); + + it("stops retrying when shouldAbort says the budget is spent", async () => { + const op = vi.fn().mockRejectedValue(pgError("57014")); + + await expect( + withLogRetentionRetry(op, { sleep, maxAttempts: 5, shouldAbort: () => true }), + ).rejects.toMatchObject({ retryAttempts: 1 }); + expect(op).toHaveBeenCalledTimes(1); + }); + + it("supports a custom shouldRetry predicate", async () => { + const op = vi.fn().mockRejectedValueOnce(new Error("custom")).mockResolvedValueOnce(1); + await expect( + withLogRetentionRetry(op, { sleep, shouldRetry: () => true, baseDelayMs: 0 }), + ).resolves.toBe(1); + }); + + it("logs only the fixed label, code and message, never query parameters", async () => { + const op = vi.fn().mockRejectedValueOnce(pgError("40001")).mockResolvedValueOnce(1); + await withLogRetentionRetry(op, { sleep, label: "log-retention-delete-batch" }); + const [fields] = vi.mocked(logger.warn).mock.calls[0]; + expect(Object.keys(fields).sort()).toEqual( + ["attempt", "code", "delayMs", "err", "label", "maxAttempts"].sort(), + ); + expect(fields.label).toBe("log-retention-delete-batch"); + }); +}); diff --git a/backend/src/lib/log-retention-telemetry.js b/backend/src/lib/log-retention-telemetry.js new file mode 100644 index 00000000..2b7c4b73 --- /dev/null +++ b/backend/src/lib/log-retention-telemetry.js @@ -0,0 +1,385 @@ +/** + * Log Retention & Archival Service telemetry (issue #1473). + * + * Prometheus alert metrics and a health snapshot for the webhook delivery + * log purge (log-retention.js). Every label value comes from a fixed + * server-side set, never from table contents, row ids or error messages, so + * the purge cannot inflate series cardinality or leak log data into metrics. + * + * The metrics live in their own registry so they can be tested in isolation + * and pushed from the standalone purge script. /metrics merges this registry + * with the others. + * + * Health is derived from the run history kept in this process: + * unhealthy - N consecutive failed runs, or the last success is older than + * the staleness limit + * degraded - the last run failed, overran its time budget, or a run has + * been in progress for longer than the stuck limit + * healthy - otherwise + * idle - no run recorded in this process yet (not an error) + */ + +import client from 'prom-client'; +import { logger } from './logger.js'; + +const register = new client.Registry(); + +register.setDefaultLabels({ + app: 'stellar-payment-api', +}); + +/** outcome: success | failure | skipped */ +export const logRetentionRunsTotal = new client.Counter({ + name: 'log_retention_purge_runs_total', + help: 'Log retention purge runs, by outcome (skipped = another instance held the lock)', + labelNames: ['outcome'], +}); + +export const logRetentionRunDuration = new client.Histogram({ + name: 'log_retention_purge_duration_seconds', + help: 'Wall time of a log retention purge run, by outcome', + labelNames: ['outcome'], + buckets: [0.1, 0.5, 1, 5, 15, 30, 60, 120, 300, 600, 1200], +}); + +export const logRetentionDeletedRowsTotal = new client.Counter({ + name: 'log_retention_purge_deleted_rows_total', + help: 'Rows deleted by the log retention purge', + labelNames: ['table'], +}); + +export const logRetentionBatchesTotal = new client.Counter({ + name: 'log_retention_purge_batches_total', + help: 'Delete batches executed by the log retention purge', + labelNames: ['table'], +}); + +/** Alert counter: a run took longer than LOG_PURGE_MAX_DURATION_MS. */ +export const logRetentionDurationExceededTotal = new client.Counter({ + name: 'log_retention_purge_duration_exceeded_total', + help: 'Log retention purge runs that exceeded their max duration budget', +}); + +/** result: scheduled | recovered | exhausted (issue #1474) */ +export const logRetentionRetriesTotal = new client.Counter({ + name: 'log_retention_purge_retries_total', + help: 'Log retention purge query retries with exponential backoff, by result', + labelNames: ['result'], +}); + +/** result: acquired | contended | error (issue #1475) */ +export const logRetentionLockTotal = new client.Counter({ + name: 'log_retention_purge_lock_total', + help: 'Log retention purge distributed lock attempts, by result', + labelNames: ['result'], +}); + +/** 0 = healthy or idle, 1 = degraded, 2 = unhealthy. Refreshed at scrape time. */ +export const logRetentionHealthState = new client.Gauge({ + name: 'log_retention_health_state', + help: 'Log retention purge health (0 = healthy/idle, 1 = degraded, 2 = unhealthy)', +}); + +export const logRetentionInProgress = new client.Gauge({ + name: 'log_retention_purge_in_progress', + help: 'Whether a log retention purge is running in this process (1) or not (0)', +}); + +export const logRetentionConsecutiveFailures = new client.Gauge({ + name: 'log_retention_purge_consecutive_failures', + help: 'Consecutive failed log retention purge runs in this process', +}); + +export const logRetentionLastSuccessTimestamp = new client.Gauge({ + name: 'log_retention_purge_last_success_timestamp_seconds', + help: 'Unix time of the last successful log retention purge', +}); + +export const logRetentionLastRunTimestamp = new client.Gauge({ + name: 'log_retention_purge_last_run_timestamp_seconds', + help: 'Unix time of the last finished log retention purge run (any outcome)', +}); + +const RUN_OUTCOMES = new Set(['success', 'failure', 'skipped']); +const RETRY_RESULTS = new Set(['scheduled', 'recovered', 'exhausted']); +const LOCK_RESULTS = new Set(['acquired', 'contended', 'error']); +/** Only tables this service purges may appear as a label value. */ +const KNOWN_TABLES = new Set(['webhook_delivery_logs']); + +export const HEALTH_STATE_VALUES = Object.freeze({ + idle: 0, + healthy: 0, + degraded: 1, + unhealthy: 2, +}); + +function readNumberEnv(name, fallback, env = process.env) { + const raw = Number.parseFloat(env[name] ?? ''); + return Number.isFinite(raw) && raw >= 0 ? raw : fallback; +} + +export function resolveHealthOptions(options = {}, env = process.env) { + return { + maxConsecutiveFailures: Math.max( + 1, + Math.trunc( + options.maxConsecutiveFailures ?? + readNumberEnv('LOG_RETENTION_HEALTH_MAX_CONSECUTIVE_FAILURES', 3, env), + ), + ), + // A daily job: allow one missed run plus slack before alerting. + maxStalenessMs: + options.maxStalenessMs ?? + readNumberEnv('LOG_RETENTION_HEALTH_MAX_STALENESS_MS', 50 * 60 * 60 * 1000, env), + stuckAfterMs: + options.stuckAfterMs ?? + readNumberEnv('LOG_RETENTION_HEALTH_STUCK_AFTER_MS', 15 * 60 * 1000, env), + }; +} + +/** + * Run-history health tracker. Constant memory: only the latest run state and + * a failure streak are kept. + */ +export class LogRetentionHealthMonitor { + constructor(options = {}, now = () => Date.now()) { + this.options = resolveHealthOptions(options); + this.now = now; + this.reset(); + } + + reset() { + this.runs = 0; + this.inProgress = 0; + this.runningSince = null; + this.lastRunAt = null; + this.lastSuccessAt = null; + this.lastOutcome = null; + this.lastDurationMs = null; + this.lastDeletedRows = null; + this.lastDurationExceeded = false; + this.consecutiveFailures = 0; + this.skipped = 0; + } + + recordStart() { + this.inProgress += 1; + if (this.runningSince === null) this.runningSince = this.now(); + } + + /** + * @param {{outcome:'success'|'failure'|'skipped', durationMs?:number, deletedRows?:number, durationExceeded?:boolean}} run + */ + recordFinish({ outcome, durationMs = 0, deletedRows = 0, durationExceeded = false }) { + // A skipped run (the lock is held elsewhere) never started here: it must + // not touch the in-progress count, nor reset or extend the failure streak. + if (outcome === 'skipped') { + this.skipped += 1; + return; + } + + if (this.inProgress > 0) this.inProgress -= 1; + if (this.inProgress === 0) this.runningSince = null; + + const time = this.now(); + this.runs += 1; + this.lastRunAt = time; + this.lastOutcome = outcome; + this.lastDurationMs = durationMs; + this.lastDurationExceeded = Boolean(durationExceeded); + if (outcome === 'success') { + this.lastSuccessAt = time; + this.lastDeletedRows = deletedRows; + this.consecutiveFailures = 0; + } else { + this.consecutiveFailures += 1; + } + } + + snapshot() { + const time = this.now(); + const { maxConsecutiveFailures, maxStalenessMs, stuckAfterMs } = this.options; + const reasons = []; + let status = this.runs === 0 && this.inProgress === 0 ? 'idle' : 'healthy'; + + const escalate = (next, reason) => { + reasons.push(reason); + if (next === 'unhealthy' || status === 'idle' || status === 'healthy') status = next; + }; + + if (this.consecutiveFailures >= maxConsecutiveFailures) { + escalate('unhealthy', 'consecutive_failures'); + } + if (this.lastSuccessAt !== null && time - this.lastSuccessAt > maxStalenessMs) { + escalate('unhealthy', 'last_success_stale'); + } + if (this.lastOutcome === 'failure' && this.consecutiveFailures < maxConsecutiveFailures) { + escalate('degraded', 'last_run_failed'); + } + if (this.lastDurationExceeded) { + escalate('degraded', 'duration_exceeded'); + } + const runningForMs = this.runningSince === null ? null : time - this.runningSince; + if (runningForMs !== null && runningForMs > stuckAfterMs) { + escalate('degraded', 'run_overdue'); + } + + const iso = (t) => (t === null ? null : new Date(t).toISOString()); + return { + status, + reasons, + runs: this.runs, + skipped_runs: this.skipped, + in_progress: this.inProgress > 0, + running_for_ms: runningForMs, + consecutive_failures: this.consecutiveFailures, + last_outcome: this.lastOutcome, + last_run_at: iso(this.lastRunAt), + last_success_at: iso(this.lastSuccessAt), + last_duration_ms: this.lastDurationMs, + last_deleted_rows: this.lastDeletedRows, + thresholds: { + max_consecutive_failures: maxConsecutiveFailures, + max_staleness_ms: maxStalenessMs, + stuck_after_ms: stuckAfterMs, + }, + }; + } +} + +const healthMonitor = new LogRetentionHealthMonitor(); + +logRetentionHealthState.collect = function collectLogRetentionHealth() { + const snap = healthMonitor.snapshot(); + this.set(HEALTH_STATE_VALUES[snap.status] ?? 0); + logRetentionInProgress.set(snap.in_progress ? 1 : 0); + logRetentionConsecutiveFailures.set(snap.consecutive_failures); + if (healthMonitor.lastSuccessAt !== null) { + logRetentionLastSuccessTimestamp.set(healthMonitor.lastSuccessAt / 1000); + } + if (healthMonitor.lastRunAt !== null) { + logRetentionLastRunTimestamp.set(healthMonitor.lastRunAt / 1000); + } +}; + +/** Current purge health, for /health endpoints. No log contents or ids. */ +export function getLogRetentionHealth() { + return healthMonitor.snapshot(); +} + +/** Test-only: clear run history. Counters are left intact. */ +export function resetLogRetentionHealth() { + healthMonitor.reset(); +} + +function safeRecord(fn) { + try { + fn(); + } catch (err) { + logger.error({ err: err?.message }, 'Failed to record log retention telemetry'); + } +} + +const safeTable = (table) => (KNOWN_TABLES.has(table) ? table : 'other'); + +export function recordLogRetentionRunStart() { + safeRecord(() => healthMonitor.recordStart()); +} + +/** + * @param {object} run + * @param {'success'|'failure'|'skipped'} run.outcome + * @param {number} [run.durationMs] + * @param {number} [run.deletedRows] + * @param {boolean} [run.durationExceeded] + */ +export function recordLogRetentionRunFinish(run) { + const outcome = RUN_OUTCOMES.has(run?.outcome) ? run.outcome : 'failure'; + const durationMs = + Number.isFinite(run?.durationMs) && run.durationMs > 0 ? run.durationMs : 0; + const deletedRows = + Number.isFinite(run?.deletedRows) && run.deletedRows > 0 ? Math.trunc(run.deletedRows) : 0; + const durationExceeded = Boolean(run?.durationExceeded); + safeRecord(() => { + logRetentionRunsTotal.inc({ outcome }); + if (outcome !== 'skipped') { + logRetentionRunDuration.observe({ outcome }, durationMs / 1000); + } + if (durationExceeded) logRetentionDurationExceededTotal.inc(); + healthMonitor.recordFinish({ outcome, durationMs, deletedRows, durationExceeded }); + }); +} + +/** @param {string} table @param {number} deletedRows */ +export function recordLogRetentionBatch(table, deletedRows) { + const label = { table: safeTable(table) }; + const rows = Number.isFinite(deletedRows) && deletedRows > 0 ? Math.trunc(deletedRows) : 0; + safeRecord(() => { + logRetentionBatchesTotal.inc(label); + if (rows > 0) logRetentionDeletedRowsTotal.inc(label, rows); + }); +} + +/** @param {'scheduled'|'recovered'|'exhausted'} result */ +export function recordLogRetentionRetry(result) { + const safe = RETRY_RESULTS.has(result) ? result : 'exhausted'; + safeRecord(() => logRetentionRetriesTotal.inc({ result: safe })); +} + +/** @param {'acquired'|'contended'|'error'} result */ +export function recordLogRetentionLock(result) { + const safe = LOCK_RESULTS.has(result) ? result : 'error'; + safeRecord(() => logRetentionLockTotal.inc({ result: safe })); +} + +/** + * Push this registry to a Prometheus Pushgateway. Used by the standalone + * purge script, which exits before any scrape could reach it. + * + * Only http(s) URLs are accepted. Failures are logged and swallowed: losing + * a metrics push must never fail the purge. + * + * @param {string} url + * @param {{jobName?:string, Pushgateway?:typeof client.Pushgateway}} [options] + * @returns {Promise} whether the push succeeded + */ +export async function pushLogRetentionMetrics(url, options = {}) { + const { jobName = 'log_retention_purge', Pushgateway = client.Pushgateway } = options; + let parsed; + try { + parsed = new URL(url); + } catch { + logger.warn('Log retention: invalid Pushgateway URL, metrics not pushed'); + return false; + } + if (parsed.protocol !== 'http:' && parsed.protocol !== 'https:') { + logger.warn({ protocol: parsed.protocol }, 'Log retention: Pushgateway URL must be http(s)'); + return false; + } + try { + // Make sure the scrape-time gauges carry the final state. + logRetentionHealthState.collect(); + const gateway = new Pushgateway(parsed.toString(), {}, register); + await gateway.pushAdd({ jobName }); + return true; + } catch (err) { + // Never log the URL itself: it may embed basic-auth credentials. + logger.warn({ err: err?.message }, 'Log retention: Pushgateway push failed'); + return false; + } +} + +register.registerMetric(logRetentionRunsTotal); +register.registerMetric(logRetentionRunDuration); +register.registerMetric(logRetentionDeletedRowsTotal); +register.registerMetric(logRetentionBatchesTotal); +register.registerMetric(logRetentionDurationExceededTotal); +register.registerMetric(logRetentionRetriesTotal); +register.registerMetric(logRetentionLockTotal); +register.registerMetric(logRetentionHealthState); +register.registerMetric(logRetentionInProgress); +register.registerMetric(logRetentionConsecutiveFailures); +register.registerMetric(logRetentionLastSuccessTimestamp); +register.registerMetric(logRetentionLastRunTimestamp); + +export { register as logRetentionRegister }; diff --git a/backend/src/lib/log-retention-telemetry.test.js b/backend/src/lib/log-retention-telemetry.test.js new file mode 100644 index 00000000..8eebeaea --- /dev/null +++ b/backend/src/lib/log-retention-telemetry.test.js @@ -0,0 +1,326 @@ +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, + LogRetentionHealthMonitor, + getLogRetentionHealth, + logRetentionBatchesTotal, + logRetentionDeletedRowsTotal, + logRetentionDurationExceededTotal, + logRetentionHealthState, + logRetentionLockTotal, + logRetentionRegister, + logRetentionRetriesTotal, + logRetentionRunsTotal, + pushLogRetentionMetrics, + recordLogRetentionBatch, + recordLogRetentionLock, + recordLogRetentionRetry, + recordLogRetentionRunFinish, + recordLogRetentionRunStart, + resetLogRetentionHealth, + resolveHealthOptions, +} from "./log-retention-telemetry.js"; + +async function value(metric, labels = {}) { + const { values } = await metric.get(); + const match = values.find((v) => + Object.entries(labels).every(([k, val]) => v.labels[k] === val), + ); + return match?.value ?? 0; +} + +const HOUR = 60 * 60 * 1000; + +function monitorAt(start = 1_700_000_000_000, options = {}) { + let t = start; + const monitor = new LogRetentionHealthMonitor( + { maxConsecutiveFailures: 3, maxStalenessMs: 50 * HOUR, stuckAfterMs: 15 * 60 * 1000, ...options }, + () => t, + ); + return { monitor, advance: (ms) => (t += ms) }; +} + +function finish(monitor, outcome, extra = {}) { + monitor.recordStart(); + monitor.recordFinish({ outcome, durationMs: 100, deletedRows: 5, ...extra }); +} + +describe("LogRetentionHealthMonitor (#1473)", () => { + it("is idle before any run", () => { + const { monitor } = monitorAt(); + expect(monitor.snapshot()).toMatchObject({ + status: "idle", + reasons: [], + runs: 0, + in_progress: false, + last_run_at: null, + last_success_at: null, + }); + }); + + it("is healthy after a successful run and reports its numbers", () => { + const { monitor } = monitorAt(); + finish(monitor, "success", { deletedRows: 12, durationMs: 250 }); + const snap = monitor.snapshot(); + expect(snap).toMatchObject({ + status: "healthy", + runs: 1, + last_outcome: "success", + last_deleted_rows: 12, + last_duration_ms: 250, + consecutive_failures: 0, + }); + expect(snap.last_success_at).toBe(snap.last_run_at); + }); + + it("is degraded after a single failure and unhealthy at the streak limit", () => { + const { monitor } = monitorAt(); + finish(monitor, "failure"); + expect(monitor.snapshot()).toMatchObject({ status: "degraded", reasons: ["last_run_failed"] }); + finish(monitor, "failure"); + finish(monitor, "failure"); + expect(monitor.snapshot()).toMatchObject({ + status: "unhealthy", + reasons: ["consecutive_failures"], + consecutive_failures: 3, + }); + }); + + it("resets the failure streak on success", () => { + const { monitor } = monitorAt(); + finish(monitor, "failure"); + finish(monitor, "failure"); + finish(monitor, "success"); + expect(monitor.snapshot()).toMatchObject({ status: "healthy", consecutive_failures: 0 }); + }); + + it("is unhealthy when the last success is older than the staleness limit", () => { + const { monitor, advance } = monitorAt(); + finish(monitor, "success"); + advance(51 * HOUR); + expect(monitor.snapshot()).toMatchObject({ + status: "unhealthy", + reasons: ["last_success_stale"], + }); + }); + + it("is degraded when a run overran its time budget", () => { + const { monitor } = monitorAt(); + finish(monitor, "success", { durationExceeded: true }); + expect(monitor.snapshot()).toMatchObject({ + status: "degraded", + reasons: ["duration_exceeded"], + }); + }); + + it("reports an in-progress run and degrades when it is overdue", () => { + const { monitor, advance } = monitorAt(); + monitor.recordStart(); + expect(monitor.snapshot()).toMatchObject({ status: "healthy", in_progress: true, running_for_ms: 0 }); + advance(16 * 60 * 1000); + expect(monitor.snapshot()).toMatchObject({ status: "degraded", reasons: ["run_overdue"] }); + monitor.recordFinish({ outcome: "success" }); + expect(monitor.snapshot()).toMatchObject({ in_progress: false, running_for_ms: null }); + }); + + it("does not let a skipped run touch the streak or the in-progress count", () => { + const { monitor } = monitorAt(); + finish(monitor, "failure"); + monitor.recordStart(); + monitor.recordFinish({ outcome: "skipped" }); + const snap = monitor.snapshot(); + expect(snap).toMatchObject({ + skipped_runs: 1, + runs: 1, + consecutive_failures: 1, + in_progress: true, + }); + }); + + it("keeps unhealthy above degraded when several reasons apply", () => { + const { monitor } = monitorAt(); + finish(monitor, "failure", { durationExceeded: true }); + finish(monitor, "failure", { durationExceeded: true }); + finish(monitor, "failure", { durationExceeded: true }); + const snap = monitor.snapshot(); + expect(snap.status).toBe("unhealthy"); + expect(snap.reasons).toEqual(["consecutive_failures", "duration_exceeded"]); + }); + + it("exposes no table contents, ids or error messages", () => { + const { monitor } = monitorAt(); + finish(monitor, "failure"); + expect(Object.keys(monitor.snapshot()).sort()).toEqual( + [ + "consecutive_failures", + "in_progress", + "last_deleted_rows", + "last_duration_ms", + "last_outcome", + "last_run_at", + "last_success_at", + "reasons", + "running_for_ms", + "runs", + "skipped_runs", + "status", + "thresholds", + ].sort(), + ); + }); +}); + +describe("resolveHealthOptions", () => { + it("reads env and falls back on bad values", () => { + expect( + resolveHealthOptions( + {}, + { + LOG_RETENTION_HEALTH_MAX_CONSECUTIVE_FAILURES: "5", + LOG_RETENTION_HEALTH_MAX_STALENESS_MS: "1000", + LOG_RETENTION_HEALTH_STUCK_AFTER_MS: "nope", + }, + ), + ).toEqual({ maxConsecutiveFailures: 5, maxStalenessMs: 1000, stuckAfterMs: 15 * 60 * 1000 }); + }); + + it("never allows a streak limit below 1", () => { + expect(resolveHealthOptions({ maxConsecutiveFailures: 0 }, {}).maxConsecutiveFailures).toBe(1); + }); +}); + +describe("record* helpers", () => { + beforeEach(() => resetLogRetentionHealth()); + + it("counts runs by outcome and feeds the shared health monitor", async () => { + const success = await value(logRetentionRunsTotal, { outcome: "success" }); + const exceeded = await value(logRetentionDurationExceededTotal); + + recordLogRetentionRunStart(); + recordLogRetentionRunFinish({ outcome: "success", durationMs: 1200, deletedRows: 7, durationExceeded: true }); + + expect(await value(logRetentionRunsTotal, { outcome: "success" })).toBe(success + 1); + expect(await value(logRetentionDurationExceededTotal)).toBe(exceeded + 1); + expect(getLogRetentionHealth()).toMatchObject({ status: "degraded", last_deleted_rows: 7 }); + }); + + it("maps an unknown outcome to failure and ignores bad numbers", async () => { + const failure = await value(logRetentionRunsTotal, { outcome: "failure" }); + recordLogRetentionRunStart(); + recordLogRetentionRunFinish({ outcome: "bogus", durationMs: -5, deletedRows: Number.NaN }); + expect(await value(logRetentionRunsTotal, { outcome: "failure" })).toBe(failure + 1); + expect(getLogRetentionHealth()).toMatchObject({ last_outcome: "failure", last_duration_ms: 0 }); + }); + + it("counts skipped runs without a duration sample", async () => { + const skipped = await value(logRetentionRunsTotal, { outcome: "skipped" }); + recordLogRetentionRunFinish({ outcome: "skipped" }); + expect(await value(logRetentionRunsTotal, { outcome: "skipped" })).toBe(skipped + 1); + expect(getLogRetentionHealth()).toMatchObject({ status: "idle", skipped_runs: 1 }); + }); + + it("records batches and deleted rows per known table only", async () => { + const batches = await value(logRetentionBatchesTotal, { table: "webhook_delivery_logs" }); + const rows = await value(logRetentionDeletedRowsTotal, { table: "webhook_delivery_logs" }); + recordLogRetentionBatch("webhook_delivery_logs", 25); + recordLogRetentionBatch("webhook_delivery_logs", 0); + recordLogRetentionBatch("users; drop table x", 3); + + expect(await value(logRetentionBatchesTotal, { table: "webhook_delivery_logs" })).toBe(batches + 2); + expect(await value(logRetentionDeletedRowsTotal, { table: "webhook_delivery_logs" })).toBe(rows + 25); + expect(await value(logRetentionBatchesTotal, { table: "other" })).toBeGreaterThanOrEqual(1); + const { values } = await logRetentionBatchesTotal.get(); + expect(values.every((v) => ["webhook_delivery_logs", "other"].includes(v.labels.table))).toBe(true); + }); + + it("sanitises retry and lock results to fixed label sets", async () => { + recordLogRetentionRetry("scheduled"); + recordLogRetentionRetry("