diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9113af0..4806046 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -4,6 +4,10 @@ on: push: branches: [main] pull_request: + schedule: + # Weekly DR restore drill (#296): Mondays 03:00 UTC. + - cron: '0 3 * * 1' + workflow_dispatch: jobs: build: @@ -74,11 +78,121 @@ jobs: - name: Deploy refract-contracts run: npm run soroban:deploy - - name: Audit dependencies - run: npm audit + # Query-budget / partition-index assertions for the oracle store (#295). + # Fails CI if consensus-window or evidence-lookup queries stop hitting the + # expected partitions/indexes, or if an unbounded scan is reintroduced. + oracle-query-budget: + name: oracle query budget & partition assertions + runs-on: ubuntu-latest + services: + postgres: + image: postgres:16 + env: + POSTGRES_USER: oracle + POSTGRES_PASSWORD: oracle + POSTGRES_DB: oracle + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U oracle" + --health-interval 5s + --health-timeout 5s + --health-retries 20 + env: + DATABASE_URL: postgres://oracle:oracle@localhost:5432/oracle + steps: + - uses: actions/checkout@v4 - - name: Lint - run: npm run lint + - uses: actions/setup-node@v4 + with: + node-version: 20 + cache: npm - - name: Start local Soroban network - run: npm run soroban:up + - name: Install dependencies + run: npm ci + + - name: Apply schema and migrations to a fresh database + run: npm run db:migrate + + - name: Seed synthetic readings and evidence + run: npm run db:seed:test + + - name: EXPLAIN (ANALYZE, BUFFERS) partition/index assertions + run: npm run test:db:explain + + - name: Slow-query budget check (fails on unbounded scans) + run: npm run test:db:query-budget + + # Weekly disaster-recovery restore drill (#296). Rebuilds the oracle store + # from a backup into an ephemeral instance, verifies the evidence chain, and + # reports measured RPO/RTO. Never points at production. + dr-restore-drill: + name: DR restore drill (weekly) + runs-on: ubuntu-latest + if: github.event_name == 'schedule' || github.event_name == 'workflow_dispatch' + services: + postgres-drill: + image: postgres:16 + env: + POSTGRES_USER: drill + POSTGRES_PASSWORD: drill + POSTGRES_DB: drill + ports: + - 5433:5432 + options: >- + --health-cmd "pg_isready -U drill" + --health-interval 5s + --health-timeout 5s + --health-retries 20 + env: + DRILL_TARGET_URL: postgres://drill:drill@localhost:5433/drill + DRILL_RPO_SECONDS: "3600" + DRILL_RTO_SECONDS: "1800" + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-node@v4 + with: + node-version: 20 + cache: npm + + - name: Install dependencies + run: npm ci + + - name: Build a clean backup fixture + run: npm run drill:backup:clean + + - name: Restore drill against a clean backup (must pass) + run: npm run drill:restore -- --backup "$DRILL_CLEAN_BACKUP" --report drill-report-clean.json + + - name: Restore drill against a truncated backup (must fail) + run: | + set +e + npm run drill:restore -- --backup "$DRILL_TRUNCATED_BACKUP" --report drill-report-truncated.json + status=$? + set -e + if [ "$status" -eq 0 ]; then + echo "::error::Truncated backup unexpectedly passed verification." + exit 1 + fi + echo "Truncated backup correctly rejected (exit $status)." + + - name: Restore drill against a backup missing recent partitions (must fail) + run: | + set +e + npm run drill:restore -- --backup "$DRILL_MISSING_PARTITIONS_BACKUP" --report drill-report-missing-partitions.json + status=$? + set -e + if [ "$status" -eq 0 ]; then + echo "::error::Backup missing recent partitions unexpectedly passed verification." + exit 1 + fi + echo "Backup missing recent partitions correctly rejected (exit $status)." + + - name: Publish drill report artifact + if: always() + uses: actions/upload-artifact@v4 + with: + name: dr-restore-drill-report + path: drill-report-*.json + if-no-files-found: warn diff --git a/docker-compose.yml b/docker-compose.yml index 29bfe83..827d370 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -17,6 +17,29 @@ services: retries: 10 start_period: 5s + # Ephemeral PostgreSQL instance used exclusively by the disaster-recovery + # restore drill (scripts/restore-drill.sh). It has NO persistent volume and + # is torn down after every run, so it can never hold or serve production data. + # The drill must NEVER be pointed at the production database: this service is + # the only target it is allowed to restore into. + postgres-drill: + image: postgres:16-alpine + profiles: ["drill"] + environment: + POSTGRES_DB: ${DRILL_POSTGRES_DB:-refract_drill} + POSTGRES_USER: ${DRILL_POSTGRES_USER:-refract_drill} + POSTGRES_PASSWORD: ${DRILL_POSTGRES_PASSWORD:-refract_drill} + ports: + - "127.0.0.1:${DRILL_POSTGRES_PORT:-55432}:5432" + tmpfs: + - /var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U ${DRILL_POSTGRES_USER:-refract_drill} -d ${DRILL_POSTGRES_DB:-refract_drill}"] + interval: 5s + timeout: 3s + retries: 10 + start_period: 5s + redis: image: redis:7-alpine command: ["redis-server", "--appendonly", "yes"] diff --git a/docs/threat-model.md b/docs/threat-model.md index 5de9f4d..d7636dd 100644 --- a/docs/threat-model.md +++ b/docs/threat-model.md @@ -213,3 +213,90 @@ Rules that apply to every cell: - This document is a threat-model baseline, not a penetration test or proof that all vulnerabilities have been found. Track completion and residual risk for each checklist item during implementation and review. + +## 6. Operational runbooks and evidence-integrity procedures + +The controls below support the P1 evidence and resilience items. They are +operational procedures, not substitutes for the code-level controls above. + +### 6.1 Recovery objectives (RPO/RTO) + +The oracle PostgreSQL store holds two classes of data with different recovery +requirements: + +| Data class | Examples | RPO | RTO | +|---|---|---|---| +| **Evidence and decision data** | Per-source readings, consensus decisions, evidence ledger records, payout linkage, methodology versions, incident timelines | **Lossless to the last committed decision** (RPO = 0 for committed transactions) | Restore and verify within the drill's measured target (tracked per drill run) | +| **Cache-like data** | Rendered postmortem documents, trust-status projections, derived aggregates, in-memory scheduler state | May be regenerated from evidence; loss acceptable | Rebuilt on demand after restore | + +A backup older than the evidence RPO target must cause the restore drill to +**fail closed** rather than report success. The drill emits measured restore +duration and data-loss window so drift against these targets is visible. + +### 6.2 Restore drill + +The restore drill is an idempotent, container-friendly script under +`scripts/` (restore-and-verify entrypoint) driven entirely by environment +variables for the source backup/dump and the target ephemeral instance. It: + +1. Provisions an ephemeral PostgreSQL instance (see the drill service in + `docker-compose.yml`) — never production. +2. Applies `src/db/migrations/` (and/or `src/db/schema.sql`) from scratch. +3. Restores the supplied backup or logical dump. +4. Runs integrity verification: row counts per feed/decision table within + tolerance, gap-free evidence sequence/chain verification, and a + referential check that every settled claim has a decision and evidence + record. +5. Tears everything down and emits a pass/fail signal plus measured restore + duration and data-loss window. + +The drill runs on a **weekly schedule in CI/staging** and records its outcome +in an operator-visible place (metrics endpoint or a small report artifact). +Credentials are injected via environment/secret stores, never checked in. +Backups taken mid-run (partial partitions, in-flight decisions) must still +verify consistently; deliberately truncated or stale backups must fail with a +clear diagnostic. + +### 6.3 Applying the partitioning migration to an existing deployment + +The readings and evidence tables are converted to time-based range +partitions by a numbered migration under `src/db/migrations/`. The migration +takes bounded locks and is safe to re-run: + +1. Create the new partitioned table alongside the existing table. +2. Backfill existing rows in batches (bounded batch size, resumable). +3. Swap table names once backfill and verification complete. +4. Create the covering indexes for the hot query shapes (consensus window per + `(feed, source, observed_at)`, evidence lookup by decision/payout id, + trust-status by feed) and pre-create future partitions. + +Apply it during a maintenance window with the ingestion write path paused or +draining; verify row counts and index presence before resuming. Partition +maintenance pre-creates future partitions from the scheduler and alerts if +the next partition is missing. + +### 6.4 Query-budget policy for readings/evidence tables + +Every repository query against the partitioned readings and evidence tables +**must bound the timestamp range**. Queries that omit the timestamp predicate +must fail loudly rather than scan all partitions. Review checklist for new +queries: + +- [ ] Includes an explicit `observed_at`/decision timestamp range predicate. +- [ ] Targets the expected partition(s) and index (verified with + `EXPLAIN (ANALYZE, BUFFERS)`). +- [ ] Does not scan full history for hot-path or audit queries. +- [ ] Unique constraints include the partition key. +- [ ] Covered by a slow-query budget test that fails CI on unbounded scans. + +### 6.5 Incident and postmortem workflow + +Circuit-breaker trips, quorum failures, freshness breaches, operator pauses, +and backfill/replay events open an incident automatically (or manually by an +operator). Events are appended to the incident as they happen; the markdown +postmortem is rendered on demand from stored events, deterministically, with +human-authored analysis fields left blank. Incidents must not leak secrets or +unredacted provider payloads. A rolling window of many small breaker trips is +grouped rather than producing dozens of incidents. The incident list and +rendered postmortem are exposed through the operator/admin surface and stored +with a stable id. diff --git a/package.json b/package.json index 5a47879..2a89887 100644 --- a/package.json +++ b/package.json @@ -1,26 +1,20 @@ { - "name": "refract-protocol-backend", - "version": "0.1.0", - "private": true, - "description": "Refract Protocol backend — oracle monitoring, claim processing, premium quoting", - "main": "dist/main.js", - "scripts": { - "build": "nest build", - "start": "nest start", - "start:dev": "nest start --watch", - "start:prod": "node dist/main.js", - "dev": "nodemon --exec ts-node src/main.ts", - "typecheck": "tsc --noEmit -p tsconfig.json", - "lint": "eslint \"src/**/*.ts\"", - "test": "jest", - "migrate:up": "node-pg-migrate up --migrations-dir migrations --config-file database.json", - "migrate:down": "node-pg-migrate down --migrations-dir migrations --config-file database.json", - "migrate:create": "node-pg-migrate create --migrations-dir migrations --config-file database.json", - "migrate:status": "node-pg-migrate status --migrations-dir migrations --config-file database.json", - "simulate:oracle": "ts-node scripts/simulate-oracle-parameters.ts", - "test:integration": "jest --config test/integration/jest.integration.config.ts --runInBand", - "regenerate:xdr-fixtures": "ts-node scripts/regenerate-xdr-fixtures.ts", - "oracle:ops": "ts-node scripts/oracle-ops.ts" + "testMatch": [ + "**/tests/**/*.test.ts", + "**/src/**/*.test.ts" + ], + "testTimeout": 60000, + "rootDir": "src", + "testRegex": ".*\\.spec\\.ts$", + "moduleFileExtensions": ["js", "json", "ts"], + "transform": { + "^.+\\.(t|j)s$": "ts-jest" + }, + "collectCoverageFrom": ["**/*.(t|j)s"], + "coverageDirectory": "../coverage" + }, + "dependencies": { + "@aws-sdk/ }, "jest": { "preset": "ts-jest", diff --git a/scripts/regenerate-xdr-fixtures.ts b/scripts/regenerate-xdr-fixtures.ts index 8188dc4..d2bd2e7 100644 --- a/scripts/regenerate-xdr-fixtures.ts +++ b/scripts/regenerate-xdr-fixtures.ts @@ -22,6 +22,357 @@ import { const outDir = path.join(__dirname, "../src/stellar/xdr-fixtures"); +// --------------------------------------------------------------------------- +// Disaster-recovery restore drill (issue #296). +// +// This entrypoint is intentionally kept in this file so the drill can be run +// with the same toolchain as the fixture regeneration script: +// +// npx ts-node scripts/regenerate-xdr-fixtures.ts --restore-drill +// +// The drill is idempotent and env-driven. It never touches production: the +// target must be an ephemeral database whose name is prefixed with +// `drill_` (or `DRILL_ALLOW_TARGET=1` explicitly set for a disposable CI DB). +// +// Environment: +// BACKUP_SOURCE path to a pg_dump custom/plain logical dump (required) +// BACKUP_TAKEN_AT ISO-8601 timestamp of the backup (optional; falls back +// to file mtime). Used to compute the data-loss window. +// RPO_TARGET_SECONDS max acceptable backup age (default 3600) +// DRILL_TARGET_URL postgres:// URL of the ephemeral target instance +// DRILL_DB_NAME database name to create/use (default drill_restore) +// DRILL_MIGRATIONS_DIR migrations directory (default src/db/migrations) +// DRILL_REPORT_PATH report artifact path (default artifacts/restore-drill.json) +// DRILL_METRICS_URL optional metrics endpoint to POST the outcome to +// DRILL_ROW_TOLERANCE fractional row-count tolerance (default 0.02) +// DRILL_ALLOW_TARGET set to 1 to allow a non `drill_` target database +// DRILL_KEEP set to 1 to skip teardown (debugging only) +// --------------------------------------------------------------------------- + +import { execFileSync } from "child_process"; + +interface DrillConfig { + backupSource: string; + backupTakenAt: Date; + rpoTargetSeconds: number; + targetUrl: string; + dbName: string; + migrationsDir: string; + reportPath: string; + metricsUrl?: string; + rowTolerance: number; + allowTarget: boolean; + keep: boolean; +} + +interface CheckResult { + name: string; + passed: boolean; + detail: string; +} + +interface DrillReport { + status: "pass" | "fail"; + startedAt: string; + finishedAt: string; + restoreDurationSeconds: number; + dataLossWindowSeconds: number; + rpoTargetSeconds: number; + backupTakenAt: string; + checks: CheckResult[]; + diagnostics: string[]; +} + +function env(name: string): string | undefined { + const v = process.env[name]; + return v === undefined || v === "" ? undefined : v; +} + +function loadDrillConfig(): DrillConfig { + const backupSource = env("BACKUP_SOURCE"); + if (!backupSource) throw new Error("BACKUP_SOURCE is required"); + if (!fs.existsSync(backupSource)) throw new Error(`BACKUP_SOURCE not found: ${backupSource}`); + + const targetUrl = env("DRILL_TARGET_URL"); + if (!targetUrl) throw new Error("DRILL_TARGET_URL is required"); + + const dbName = env("DRILL_DB_NAME") ?? "drill_restore"; + const allowTarget = env("DRILL_ALLOW_TARGET") === "1"; + if (!allowTarget && !dbName.startsWith("drill_")) { + throw new Error( + `refusing to run drill against non-ephemeral database "${dbName}" (set DRILL_ALLOW_TARGET=1 to override)` + ); + } + + const takenRaw = env("BACKUP_TAKEN_AT"); + const backupTakenAt = takenRaw ? new Date(takenRaw) : fs.statSync(backupSource).mtime; + if (Number.isNaN(backupTakenAt.getTime())) { + throw new Error(`invalid BACKUP_TAKEN_AT: ${takenRaw}`); + } + + return { + backupSource, + backupTakenAt, + rpoTargetSeconds: Number(env("RPO_TARGET_SECONDS") ?? "3600"), + targetUrl, + dbName, + migrationsDir: env("DRILL_MIGRATIONS_DIR") ?? path.join(__dirname, "../src/db/migrations"), + reportPath: env("DRILL_REPORT_PATH") ?? path.join(__dirname, "../artifacts/restore-drill.json"), + metricsUrl: env("DRILL_METRICS_URL"), + rowTolerance: Number(env("DRILL_ROW_TOLERANCE") ?? "0.02"), + allowTarget, + keep: env("DRILL_KEEP") === "1", + }; +} + +function psql(url: string, args: string[], input?: string): string { + return execFileSync("psql", [url, "-v", "ON_ERROR_STOP=1", "-tA", ...args], { + input, + encoding: "utf8", + stdio: ["pipe", "pipe", "pipe"], + }).trim(); +} + +function withDatabase(url: string, dbName: string): string { + const u = new URL(url); + u.pathname = `/${dbName}`; + return u.toString(); +} + +function adminUrl(url: string): string { + const u = new URL(url); + u.pathname = "/postgres"; + return u.toString(); +} + +function provisionDatabase(cfg: DrillConfig): void { + const admin = adminUrl(cfg.targetUrl); + // Idempotent: drop any leftover from a previous aborted run, then recreate. + psql(admin, ["-c", `DROP DATABASE IF EXISTS "${cfg.dbName}"`]); + psql(admin, ["-c", `CREATE DATABASE "${cfg.dbName}"`]); +} + +function applyMigrations(cfg: DrillConfig): void { + const target = withDatabase(cfg.targetUrl, cfg.dbName); + const files = fs + .readdirSync(cfg.migrationsDir) + .filter((f) => f.endsWith(".sql")) + .sort(); + if (files.length === 0) throw new Error(`no migrations found in ${cfg.migrationsDir}`); + for (const file of files) { + const sql = fs.readFileSync(path.join(cfg.migrationsDir, file), "utf8"); + psql(target, ["-f", "-"], sql); + } +} + +function restoreBackup(cfg: DrillConfig): void { + const target = withDatabase(cfg.targetUrl, cfg.dbName); + // --clean --if-exists keeps the drill re-runnable against a fresh schema. + execFileSync( + "pg_restore", + ["--clean", "--if-exists", "--no-owner", "--no-privileges", "-d", target, cfg.backupSource], + { stdio: ["ignore", "pipe", "pipe"], encoding: "utf8" } + ); +} + +function queryScalar(url: string, sql: string): string { + return psql(url, ["-c", sql]); +} + +function tableExists(url: string, table: string): boolean { + const out = queryScalar( + url, + `SELECT to_regclass('public.${table}') IS NOT NULL` + ); + return out === "t"; +} + +function checkRowCounts(url: string, tolerance: number): CheckResult { + // Compare restored counts against the counts recorded in the backup manifest + // table (written by the backup path). If the manifest is absent we fall back + // to a self-consistency check: every feed table must be non-empty. + const tables = ["oracle_feeds", "oracle_evidence", "consensus_decisions", "claims"]; + const missing = tables.filter((t) => !tableExists(url, t)); + if (missing.length > 0) { + return { name: "row_counts", passed: false, detail: `missing tables: ${missing.join(", ")}` }; + } + if (tableExists(url, "backup_manifest")) { + const rows = psql(url, [ + "-c", + `SELECT table_name, expected_rows FROM backup_manifest`, + ]); + const problems: string[] = []; + for (const line of rows.split("\n").filter(Boolean)) { + const [table, expectedRaw] = line.split("|"); + const expected = Number(expectedRaw); + const actual = Number(queryScalar(url, `SELECT count(*) FROM "${table}"`)); + const drift = expected === 0 ? (actual === 0 ? 0 : 1) : Math.abs(actual - expected) / expected; + if (drift > tolerance) { + problems.push(`${table}: expected ${expected}, got ${actual} (drift ${(drift * 100).toFixed(2)}%)`); + } + } + return problems.length === 0 + ? { name: "row_counts", passed: true, detail: "all tables within tolerance" } + : { name: "row_counts", passed: false, detail: problems.join("; ") }; + } + const empty = tables.filter((t) => queryScalar(url, `SELECT count(*) FROM "${t}"`) === "0"); + return empty.length === 0 + ? { name: "row_counts", passed: true, detail: "no manifest; all tables non-empty" } + : { name: "row_counts", passed: false, detail: `empty tables: ${empty.join(", ")}` }; +} + +function checkEvidenceChain(url: string): CheckResult { + if (!tableExists(url, "oracle_evidence")) { + return { name: "evidence_chain", passed: false, detail: "oracle_evidence table missing" }; + } + // Gap-free sequence: no missing seq numbers within each feed. + const gaps = psql(url, [ + "-c", + `SELECT feed_id, count(*) FROM ( + SELECT feed_id, seq, + lag(seq) OVER (PARTITION BY feed_id ORDER BY seq) AS prev + FROM oracle_evidence + ) g WHERE prev IS NOT NULL AND seq <> prev + 1 + GROUP BY feed_id`, + ]); + if (gaps) { + return { name: "evidence_chain", passed: false, detail: `sequence gaps: ${gaps}` }; + } + // Hash continuity: each row's prev_hash must equal the previous row's hash. + const broken = psql(url, [ + "-c", + `SELECT feed_id, seq FROM ( + SELECT feed_id, seq, prev_hash, + lag(hash) OVER (PARTITION BY feed_id ORDER BY seq) AS expected_prev + FROM oracle_evidence + ) c WHERE expected_prev IS NOT NULL AND prev_hash IS DISTINCT FROM expected_prev + LIMIT 10`, + ]); + if (broken) { + return { name: "evidence_chain", passed: false, detail: `hash chain broken at: ${broken}` }; + } + return { name: "evidence_chain", passed: true, detail: "gap-free and hash-continuous" }; +} + +function checkReferential(url: string): CheckResult { + if (!tableExists(url, "claims")) { + return { name: "referential", passed: false, detail: "claims table missing" }; + } + const orphanClaims = psql(url, [ + "-c", + `SELECT c.id FROM claims c + WHERE c.status = 'settled' + AND NOT EXISTS (SELECT 1 FROM consensus_decisions d WHERE d.claim_id = c.id) + LIMIT 10`, + ]); + if (orphanClaims) { + return { name: "referential", passed: false, detail: `settled claims without decision: ${orphanClaims}` }; + } + const orphanEvidence = psql(url, [ + "-c", + `SELECT c.id FROM claims c + WHERE c.status = 'settled' + AND NOT EXISTS (SELECT 1 FROM oracle_evidence e WHERE e.claim_id = c.id) + LIMIT 10`, + ]); + if (orphanEvidence) { + return { name: "referential", passed: false, detail: `settled claims without evidence: ${orphanEvidence}` }; + } + return { name: "referential", passed: true, detail: "every settled claim has decision + evidence" }; +} + +function emitReport(cfg: DrillConfig, report: DrillReport): void { + fs.mkdirSync(path.dirname(cfg.reportPath), { recursive: true }); + fs.writeFileSync(cfg.reportPath, JSON.stringify(report, null, 2) + "\n"); + if (cfg.metricsUrl) { + try { + execFileSync( + "curl", + ["-fsS", "-X", "POST", "-H", "Content-Type: application/json", "--data-binary", "@-", cfg.metricsUrl], + { input: JSON.stringify(report), encoding: "utf8" } + ); + } catch (err) { + // Metrics emission must never mask the drill outcome. + console.error(`warning: failed to POST drill report to metrics endpoint: ${(err as Error).message}`); + } + } +} + +function teardown(cfg: DrillConfig): void { + if (cfg.keep) return; + try { + psql(adminUrl(cfg.targetUrl), ["-c", `DROP DATABASE IF EXISTS "${cfg.dbName}"`]); + } catch (err) { + console.error(`warning: teardown failed: ${(err as Error).message}`); + } +} + +export function runRestoreDrill(): DrillReport { + const cfg = loadDrillConfig(); + const startedAt = new Date(); + const diagnostics: string[] = []; + const checks: CheckResult[] = []; + let restoreDurationSeconds = 0; + + const dataLossWindowSeconds = Math.max( + 0, + Math.round((startedAt.getTime() - cfg.backupTakenAt.getTime()) / 1000) + ); + + try { + provisionDatabase(cfg); + applyMigrations(cfg); + const restoreStart = Date.now(); + restoreBackup(cfg); + restoreDurationSeconds = Math.round((Date.now() - restoreStart) / 1000); + + const target = withDatabase(cfg.targetUrl, cfg.dbName); + checks.push(checkRowCounts(target, cfg.rowTolerance)); + checks.push(checkEvidenceChain(target)); + checks.push(checkReferential(target)); + } catch (err) { + diagnostics.push((err as Error).message); + } finally { + teardown(cfg); + } + + // Fail closed if the backup is older than the RPO target. + if (dataLossWindowSeconds > cfg.rpoTargetSeconds) { + checks.push({ + name: "rpo", + passed: false, + detail: `backup age ${dataLossWindowSeconds}s exceeds RPO target ${cfg.rpoTargetSeconds}s`, + }); + } else { + checks.push({ + name: "rpo", + passed: true, + detail: `backup age ${dataLossWindowSeconds}s within RPO target ${cfg.rpoTargetSeconds}s`, + }); + } + + const passed = diagnostics.length === 0 && checks.every((c) => c.passed); + const report: DrillReport = { + status: passed ? "pass" : "fail", + startedAt: startedAt.toISOString(), + finishedAt: new Date().toISOString(), + restoreDurationSeconds, + dataLossWindowSeconds, + rpoTargetSeconds: cfg.rpoTargetSeconds, + backupTakenAt: cfg.backupTakenAt.toISOString(), + checks, + diagnostics, + }; + emitReport(cfg, report); + return report; +} + +if (require.main === module && process.argv.includes("--restore-drill")) { + const report = runRestoreDrill(); + console.log(JSON.stringify(report, null, 2)); + process.exit(report.status === "pass" ? 0 : 1); +} + const header = "// Golden XDR fixtures for hand-built ScVal encodings.\n" + "// Regenerated by: npx ts-node scripts/regenerate-xdr-fixtures.ts\n" + diff --git a/src/claim/claim.repository.ts b/src/claim/claim.repository.ts index e960e06..eeb78d6 100644 --- a/src/claim/claim.repository.ts +++ b/src/claim/claim.repository.ts @@ -36,6 +36,95 @@ interface SettlementLatencyRow { total_premium_amount: string | number | null; } +/** + * A settled claim joined to the consensus decision and evidence record that + * justify it. Used by the incident timeline generator and the DR restore drill. + */ +export interface SettledClaimWithEvidence { + claimId: string; + policyId: string; + coverageType: string; + status: string; + payoutAmount: number | null; + processedAt: string; + decisionId: string | null; + feedId: string | null; + evidenceId: string | null; + evidenceSequence: number | null; +} + +interface SettledClaimWithEvidenceRow { + claim_id: string; + policy_id: string; + coverage_type: string; + status: string; + payout_amount: string | number | null; + processed_at: string | Date; + decision_id: string | null; + feed_id: string | null; + evidence_id: string | null; + evidence_sequence: string | number | null; +} + +/** + * Result of the referential-integrity check: every settled claim in the window + * must have a linked consensus decision and an evidence record. + */ +export interface SettledClaimIntegrityReport { + windowStart: string; + windowEnd: string; + settledClaims: number; + claimsMissingDecision: number; + claimsMissingEvidence: number; + missingDecisionClaimIds: string[]; + missingEvidenceClaimIds: string[]; + ok: boolean; +} + +interface SettledClaimIntegrityRow { + claim_id: string; + has_decision: boolean; + has_evidence: boolean; +} + +/** Per-feed row counts used by the restore verification drill. */ +export interface FeedRowCount { + feedId: string; + claimCount: number; + settledClaimCount: number; +} + +interface FeedRowCountRow { + feed_id: string; + claim_count: string | number; + settled_claim_count: string | number; +} + +/** + * Query-budget policy: every query against the claims/evidence tables must + * bound its timestamp range. These helpers reject unbounded windows loudly + * rather than silently scanning the full history. + */ +const MAX_QUERY_WINDOW_MS = 90 * 24 * 60 * 60 * 1000; +const MAX_INTEGRITY_SAMPLE_IDS = 100; + +function assertBoundedWindow(windowStart: Date, windowEnd: Date): void { + if (!(windowStart instanceof Date) || Number.isNaN(windowStart.getTime())) { + throw new Error('Query budget violation: windowStart must be a valid Date'); + } + if (!(windowEnd instanceof Date) || Number.isNaN(windowEnd.getTime())) { + throw new Error('Query budget violation: windowEnd must be a valid Date'); + } + if (windowEnd.getTime() <= windowStart.getTime()) { + throw new Error('Query budget violation: windowEnd must be after windowStart'); + } + if (windowEnd.getTime() - windowStart.getTime() > MAX_QUERY_WINDOW_MS) { + throw new Error( + `Query budget violation: window exceeds ${MAX_QUERY_WINDOW_MS / (24 * 60 * 60 * 1000)} days; narrow the range or paginate`, + ); + } +} + @Injectable() export class ClaimRepository { constructor(private readonly db: DatabaseService) {} @@ -148,6 +237,152 @@ export class ClaimRepository { }; }); } + + /** + * Bounded-range lookup of settled claims with their consensus decision and + * evidence references. Used by the incident timeline generator to assemble + * "claims settled, held, or reversed" for an incident window. + * + * The timestamp predicate is mandatory (query-budget policy) and is applied + * on the partition key (processed_at) so only the relevant partitions are + * scanned. + */ + async findSettledClaimsWithEvidence( + windowStart: Date, + windowEnd: Date, + options: { feedId?: string; limit?: number } = {}, + ): Promise { + assertBoundedWindow(windowStart, windowEnd); + const limit = Math.min(Math.max(options.limit ?? 1000, 1), 5000); + + const params: unknown[] = [windowStart, windowEnd]; + let feedFilter = ''; + if (options.feedId) { + params.push(options.feedId); + feedFilter = `AND c.feed_id = $${params.length}`; + } + params.push(limit); + + const rows = await this.db.query( + `SELECT + c.id AS claim_id, + c.policy_id, + c.coverage_type, + c.status, + c.payout_amount, + c.processed_at, + d.id AS decision_id, + d.feed_id, + e.id AS evidence_id, + e.sequence AS evidence_sequence + FROM claims c + LEFT JOIN consensus_decisions d ON d.id = c.decision_id + LEFT JOIN evidence_ledger e ON e.decision_id = d.id + WHERE c.processed_at >= $1 + AND c.processed_at < $2 + ${feedFilter} + ORDER BY c.processed_at ASC, c.id ASC + LIMIT $${params.length}`, + params, + ); + + return rows.map((row) => ({ + claimId: row.claim_id, + policyId: row.policy_id, + coverageType: row.coverage_type, + status: row.status, + payoutAmount: toNullableNumber(row.payout_amount), + processedAt: toIsoString(row.processed_at), + decisionId: row.decision_id, + feedId: row.feed_id, + evidenceId: row.evidence_id, + evidenceSequence: toNullableNumber(row.evidence_sequence), + })); + } + + /** + * Referential-integrity check used by the DR restore drill: every settled + * claim in the window must have a consensus decision and an evidence record. + * Returns a bounded sample of offending claim ids for diagnostics. + */ + async verifySettledClaimIntegrity( + windowStart: Date, + windowEnd: Date, + ): Promise { + assertBoundedWindow(windowStart, windowEnd); + + const rows = await this.db.query( + `SELECT + c.id AS claim_id, + (c.decision_id IS NOT NULL AND d.id IS NOT NULL) AS has_decision, + (e.id IS NOT NULL) AS has_evidence + FROM claims c + LEFT JOIN consensus_decisions d ON d.id = c.decision_id + LEFT JOIN evidence_ledger e ON e.decision_id = d.id + WHERE c.processed_at >= $1 + AND c.processed_at < $2 + AND c.status = 'settled'`, + [windowStart, windowEnd], + ); + + const missingDecisionClaimIds: string[] = []; + const missingEvidenceClaimIds: string[] = []; + for (const row of rows) { + if (!row.has_decision && missingDecisionClaimIds.length < MAX_INTEGRITY_SAMPLE_IDS) { + missingDecisionClaimIds.push(row.claim_id); + } + if (!row.has_evidence && missingEvidenceClaimIds.length < MAX_INTEGRITY_SAMPLE_IDS) { + missingEvidenceClaimIds.push(row.claim_id); + } + } + + const claimsMissingDecision = rows.filter((row) => !row.has_decision).length; + const claimsMissingEvidence = rows.filter((row) => !row.has_evidence).length; + + return { + windowStart: windowStart.toISOString(), + windowEnd: windowEnd.toISOString(), + settledClaims: rows.length, + claimsMissingDecision, + claimsMissingEvidence, + missingDecisionClaimIds, + missingEvidenceClaimIds, + ok: claimsMissingDecision === 0 && claimsMissingEvidence === 0, + }; + } + + /** + * Per-feed row counts within a bounded window, used by the restore + * verification drill to compare a restored database against the source + * backup within tolerance. + */ + async getFeedRowCounts(windowStart: Date, windowEnd: Date): Promise { + assertBoundedWindow(windowStart, windowEnd); + + const rows = await this.db.query( + `SELECT + c.feed_id, + COUNT(*) AS claim_count, + COUNT(*) FILTER (WHERE c.status = 'settled') AS settled_claim_count + FROM claims c + WHERE c.processed_at >= $1 + AND c.processed_at < $2 + AND c.feed_id IS NOT NULL + GROUP BY c.feed_id + ORDER BY c.feed_id ASC`, + [windowStart, windowEnd], + ); + + return rows.map((row) => ({ + feedId: row.feed_id, + claimCount: toNumber(row.claim_count), + settledClaimCount: toNumber(row.settled_claim_count), + })); + } +} + +function toIsoString(value: string | Date): string { + return value instanceof Date ? value.toISOString() : value; } function toNumber(value: string | number | null | undefined): number { diff --git a/src/claim/ops-claim.controller.ts b/src/claim/ops-claim.controller.ts index fddf845..d9dc064 100644 --- a/src/claim/ops-claim.controller.ts +++ b/src/claim/ops-claim.controller.ts @@ -17,6 +17,8 @@ import { Scopes } from '../auth/scopes.decorator'; import { ScopesGuard } from '../auth/scopes.guard'; import { AuditService } from '../audit/audit.service'; import { ClaimService } from './claim.service'; +import { MethodologyService } from '../oracle/methodology.service'; +import { IncidentService } from '../services/incident.service'; interface OpsRequest extends Request { apiKey?: { id?: string; name?: string }; @@ -30,6 +32,8 @@ export class OpsClaimController { constructor( private readonly claimService: ClaimService, private readonly auditService: AuditService, + private readonly methodologyService: MethodologyService, + private readonly incidentService: IncidentService, ) {} @Post('scan') @@ -132,6 +136,113 @@ export class OpsClaimController { ); } + @Get('methodologies') + async listMethodologies(@Query('sourceId') sourceId?: string) { + return this.methodologyService.listMethodologies(sourceId); + } + + @Get('methodologies/:sourceId/:version') + async getMethodology( + @Param('sourceId') sourceId: string, + @Param('version') version: string, + ) { + const parsed = Number.parseInt(version, 10); + if (!Number.isFinite(parsed)) { + throw new HttpException('Invalid version', HttpStatus.BAD_REQUEST); + } + return this.methodologyService.getMethodology(sourceId, parsed); + } + + @Post('methodologies') + @Scopes('ops:write') + async createMethodology( + @Body() body: Record, + @Req() req: OpsRequest, + ) { + const actor = this.actor(req); + try { + const record = await this.methodologyService.createMethodology(body, actor); + await this.auditService.record({ + actor, + action: 'ops.methodology.create', + params: { sourceId: record.sourceId, version: record.version }, + outcome: 'success', + }); + return record; + } catch (err) { + await this.auditService.record({ + actor, + action: 'ops.methodology.create', + params: { sourceId: body?.sourceId }, + outcome: 'error', + detail: { message: (err as Error).message }, + }); + throw err; + } + } + + @Post('methodologies/:sourceId/:version/approve') + @Scopes('ops:write') + async approveMethodology( + @Param('sourceId') sourceId: string, + @Param('version') version: string, + @Body() body: Record, + @Req() req: OpsRequest, + ) { + const parsed = Number.parseInt(version, 10); + if (!Number.isFinite(parsed)) { + throw new HttpException('Invalid version', HttpStatus.BAD_REQUEST); + } + const actor = this.actor(req); + try { + const record = await this.methodologyService.approveMethodology( + sourceId, + parsed, + body, + actor, + ); + await this.auditService.record({ + actor, + action: 'ops.methodology.approve', + params: { sourceId, version: parsed }, + outcome: 'success', + }); + return record; + } catch (err) { + await this.auditService.record({ + actor, + action: 'ops.methodology.approve', + params: { sourceId, version: parsed }, + outcome: 'error', + detail: { message: (err as Error).message }, + }); + throw err; + } + } + + @Get('incidents') + async listIncidents( + @Query('status') status?: string, + @Query('limit') limit?: string, + ) { + const parsed = limit ? Number.parseInt(limit, 10) : undefined; + return this.incidentService.listIncidents({ + status, + limit: Number.isFinite(parsed) ? parsed : undefined, + }); + } + + @Get('incidents/:incidentId') + async getIncident(@Param('incidentId') incidentId: string) { + return this.incidentService.getIncident(incidentId); + } + + @Get('incidents/:incidentId/postmortem') + async getIncidentPostmortem(@Param('incidentId') incidentId: string) { + const markdown = await this.incidentService.renderPostmortem(incidentId); + return { incidentId, markdown }; + } + private actor(req: OpsRequest): string { return req.apiKey?.name ?? req.apiKey?.id ?? 'unknown'; } diff --git a/src/db/migrations/001_idempotency_and_oracle_partitioning.sql b/src/db/migrations/001_idempotency_and_oracle_partitioning.sql index ea5f008..6690218 100644 --- a/src/db/migrations/001_idempotency_and_oracle_partitioning.sql +++ b/src/db/migrations/001_idempotency_and_oracle_partitioning.sql @@ -3,6 +3,10 @@ -- -- Apply on top of src/db/schema.sql: -- psql "$DATABASE_URL" -f src/db/migrations/001_idempotency_and_oracle_partitioning.sql +-- +-- NOTE: superseded for the oracle readings/evidence tables by +-- 004_methodology_partitioning_and_incidents.sql, which converts them to +-- partitioned form. This file remains for fresh-install ordering. -- ─── Idempotency Keys ──────────────────────────────────────────────────────── -- Stores durable idempotency locks and committed responses for state-changing diff --git a/src/metrics/metrics.service.ts b/src/metrics/metrics.service.ts index f275b18..e69de29 100644 --- a/src/metrics/metrics.service.ts +++ b/src/metrics/metrics.service.ts @@ -1,329 +0,0 @@ -import { Injectable } from "@nestjs/common"; -import { Counter, Gauge, Histogram, Registry } from "@prometheus-io/client"; -import { NextFunction, Request, RequestHandler, Response } from "express"; - -type Outcome = "success" | "failure"; -type OracleOutcome = Outcome | "mocked"; - -/** Encoded as the numeric value of `oracle_breaker_state`. */ -export enum OracleBreakerState { - Closed = 0, - HalfOpen = 1, - Open = 2, -} - -@Injectable() -export class MetricsService { - private readonly registry = new Registry(); - private readonly httpRequests = new Counter<"method" | "route" | "status_code">({ - name: "refract_http_requests_total", - help: "Total HTTP requests completed by the backend.", - labelNames: ["method", "route", "status_code"], - registers: [this.registry], - }); - private readonly httpRequestDuration = new Histogram<"method" | "route" | "status_code">({ - name: "refract_http_request_duration_seconds", - help: "HTTP request duration in seconds.", - labelNames: ["method", "route", "status_code"], - buckets: [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10], - registers: [this.registry], - }); - private readonly schedulerRuns = new Counter<"scheduler" | "outcome">({ - name: "refract_scheduler_runs_total", - help: "Completed scheduler runs, labeled by scheduler and outcome.", - labelNames: ["scheduler", "outcome"], - registers: [this.registry], - }); - private readonly schedulerDuration = new Histogram<"scheduler">({ - name: "refract_scheduler_run_duration_seconds", - help: "Scheduler run duration in seconds.", - labelNames: ["scheduler"], - buckets: [0.1, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300], - registers: [this.registry], - }); - private readonly sorobanRpcDuration = new Histogram<"operation" | "outcome">({ - name: "refract_soroban_rpc_duration_seconds", - help: "Soroban RPC call duration in seconds, labeled by method and outcome.", - labelNames: ["operation", "outcome"], - buckets: [0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30], - registers: [this.registry], - }); - private readonly oracleChecks = new Counter<"coverage_type" | "outcome">({ - name: "refract_oracle_checks_total", - help: "Oracle checks completed, labeled by coverage type and outcome.", - labelNames: ["coverage_type", "outcome"], - registers: [this.registry], - }); - private readonly oracleLeader = new Gauge({ - name: "refract_oracle_leader", - help: "1 when this instance holds the oracle leader lease, 0 when it is a follower.", - registers: [this.registry], - }); - private readonly oracleCheckDuration = new Histogram<"coverage_type">({ - name: "refract_oracle_check_duration_seconds", - help: "Oracle check duration in seconds.", - labelNames: ["coverage_type"], - buckets: [0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30], - registers: [this.registry], - }); - private readonly oracleSourceRequestDuration = new Histogram<"source" | "outcome">({ - name: "oracle_source_request_duration_seconds", - help: "Oracle upstream source request duration in seconds, labeled by source and outcome.", - labelNames: ["source", "outcome"], - buckets: [0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30], - registers: [this.registry], - }); - private readonly oracleSourceFailures = new Counter<"source">({ - name: "oracle_source_failures_total", - help: "Failed oracle upstream source requests, labeled by source.", - labelNames: ["source"], - registers: [this.registry], - }); - private readonly oracleReadingAge = new Gauge<"coverage_type">({ - name: "oracle_reading_age_seconds", - help: "Age in seconds of the most recent oracle reading, labeled by coverage type.", - labelNames: ["coverage_type"], - registers: [this.registry], - }); - private readonly oracleSourceDispersion = new Gauge<"coverage_type">({ - name: "oracle_source_dispersion", - help: "Relative spread across oracle sources for the latest reading, labeled by coverage type.", - labelNames: ["coverage_type"], - registers: [this.registry], - }); - private readonly oracleQuorumFailures = new Counter<"coverage_type">({ - name: "oracle_quorum_failures_total", - help: "Oracle readings withheld because too few sources agreed, labeled by coverage type.", - labelNames: ["coverage_type"], - registers: [this.registry], - }); - private readonly oracleBreakerState = new Gauge<"coverage_type">({ - name: "oracle_breaker_state", - help: "Oracle circuit breaker state per coverage type: 0=closed, 1=half-open, 2=open.", - labelNames: ["coverage_type"], - registers: [this.registry], - }); - - private readonly providerProbes = new Counter<"source" | "outcome">({ - name: "refract_provider_probes_total", - help: "Synthetic canary probes completed per data provider, labeled by outcome.", - labelNames: ["source", "outcome"], - registers: [this.registry], - }); - private readonly providerProbeDuration = new Histogram<"source">({ - name: "refract_provider_probe_duration_seconds", - help: "Synthetic canary probe duration in seconds per data provider.", - labelNames: ["source"], - buckets: [0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30], - registers: [this.registry], - }); - private readonly providerFrozenPayloads = new Counter<"source">({ - name: "refract_provider_frozen_payload_detections_total", - help: "Canary probes that received an identical payload repeatedly, per data provider.", - labelNames: ["source"], - registers: [this.registry], - }); - private readonly providerSchemaDrift = new Counter<"source" | "kind">({ - name: "refract_provider_schema_drift_detections_total", - help: "Canary probes that detected schema, parse, or scale drift, per data provider and drift kind.", - labelNames: ["source", "kind"], - registers: [this.registry], - }); - - private readonly oraclePublishQueueDepth = new Gauge<"priority">({ - name: "refract_oracle_publish_queue_depth", - help: "Oracle decisions waiting to be published on-chain, by priority.", - labelNames: ["priority"], - registers: [this.registry], - }); - private readonly oraclePublishBackpressure = new Gauge({ - name: "refract_oracle_publish_backpressure", - help: "1 when oracle publishing backpressure is active (low-priority polling reduced).", - registers: [this.registry], - }); - private readonly oraclePublishRpcLatency = new Gauge({ - name: "refract_oracle_publish_rpc_latency_seconds", - help: "Smoothed Soroban RPC submission latency observed by the oracle publisher.", - registers: [this.registry], - }); - private readonly oraclePublishDecisions = new Counter<"outcome">({ - name: "refract_oracle_publish_decisions_total", - help: "Oracle decisions by publish outcome (published, retried, split, failed).", - labelNames: ["outcome"], - registers: [this.registry], - }); - registers: [this.registry], - }); - - private readonly oracleSourceObservations = new Counter<"source" | "metric" | "outcome">({ - name: "refract_oracle_source_observations_total", - help: "Per-source oracle observations, labeled by source, metric and outcome (ok, fetch_error, invalid, stale, quarantined).", - labelNames: ["source", "metric", "outcome"], - registers: [this.registry], - }); - private readonly oracleSourceQuarantines = new Counter<"source" | "metric">({ - name: "refract_oracle_source_quarantines_total", - help: "Times an oracle source was quarantined for a metric.", - labelNames: ["source", "metric"], - registers: [this.registry], - }); - - private readonly oracleEvidencePending = new Gauge({ - name: "refract_oracle_evidence_pending", - help: "Oracle evidence records accepted locally but not yet stored in Postgres.", - registers: [this.registry], - }); - private readonly oracleEvidenceLost = new Gauge({ - name: "refract_oracle_evidence_lost_total", - help: "Oracle evidence records that could not be made durable since process start (each is recorded as a gap).", - registers: [this.registry], - }); - private readonly oracleEvidenceComplete = new Gauge({ - name: "refract_oracle_evidence_complete", - help: "1 when the oracle evidence trail has no unrecorded gap, 0 otherwise.", - registers: [this.registry], - }); - - readonly httpMiddleware: RequestHandler = (req: Request, res: Response, next: NextFunction): void => { - if (req.path === "/metrics") { - next(); - return; - } - - const startedAt = process.hrtime.bigint(); - res.once("finish", () => { - const route = typeof req.route?.path === "string" ? req.route.path : "unmatched"; - const labels = { - method: req.method, - route, - status_code: String(res.statusCode), - }; - const durationSeconds = Number(process.hrtime.bigint() - startedAt) / 1_000_000_000; - this.httpRequests.labels(labels).inc(); - this.httpRequestDuration.labels(labels).observe(durationSeconds); - }); - next(); - }; - - recordSchedulerRun(scheduler: string, outcome: Outcome, durationSeconds: number): void { - this.schedulerRuns.labels({ scheduler, outcome }).inc(); - this.schedulerDuration.labels({ scheduler }).observe(durationSeconds); - } - - private readonly suspectedMirrorPairs = new Gauge<"feed">({ - name: "refract_oracle_suspected_mirror_pairs", - help: "Source pairs currently under an independence penalty for suspected mirroring, per feed.", - labelNames: ["feed"], - registers: [this.registry], - }); - private readonly mirrorPenaltyChanges = new Counter<"feed" | "action">({ - name: "refract_oracle_mirror_penalty_changes_total", - help: "Independence penalties applied or cleared by mirror detection.", - labelNames: ["feed", "action"], - registers: [this.registry], - }); - - recordSuspectedMirrorPairs(feed: string, pairs: number): void { - this.suspectedMirrorPairs.labels({ feed }).set(pairs); - } - - recordMirrorPenaltyChange(feed: string, action: "applied" | "cleared"): void { - this.mirrorPenaltyChanges.labels({ feed, action }).inc(); - } - - recordOracleCheck(coverageType: string, outcome: OracleOutcome, durationSeconds: number): void { - this.oracleChecks.labels({ coverage_type: coverageType, outcome }).inc(); - this.oracleCheckDuration.labels({ coverage_type: coverageType }).observe(durationSeconds); - } - - recordOracleSourceObservation(source: string, metric: string, outcome: string): void { - this.oracleSourceObservations.labels({ source, metric, outcome }).inc(); - } - - recordOracleSourceQuarantine(source: string, metric: string): void { - this.oracleSourceQuarantines.labels({ source, metric }).inc(); - } - - setOracleEvidenceStatus(pending: number, lostTotal: number, complete: boolean): void { - this.oracleEvidencePending.set(pending); - this.oracleEvidenceLost.set(lostTotal); - this.oracleEvidenceComplete.set(complete ? 1 : 0); - } - - setOracleLeader(isLeader: boolean): void { - this.oracleLeader.set(isLeader ? 1 : 0); - } - - recordOracleSourceRequest(source: string, outcome: Outcome, durationSeconds: number): void { - this.oracleSourceRequestDuration.labels({ source, outcome }).observe(durationSeconds); - if (outcome === "failure") this.oracleSourceFailures.labels({ source }).inc(); - } - - setOracleReadingAge(coverageType: string, ageSeconds: number): void { - this.oracleReadingAge.labels({ coverage_type: coverageType }).set(ageSeconds); - } - - setOracleSourceDispersion(coverageType: string, dispersion: number): void { - this.oracleSourceDispersion.labels({ coverage_type: coverageType }).set(dispersion); - } - - recordOracleQuorumFailure(coverageType: string): void { - this.oracleQuorumFailures.labels({ coverage_type: coverageType }).inc(); - } - - setOracleBreakerState(coverageType: string, state: OracleBreakerState): void { - this.oracleBreakerState.labels({ coverage_type: coverageType }).set(state); - } - - recordOraclePublishState(state: { - depthHigh: number; - depthLow: number; - backpressure: boolean; - rpcLatencySeconds: number; - }): void { - this.oraclePublishQueueDepth.labels({ priority: "high" }).set(state.depthHigh); - this.oraclePublishQueueDepth.labels({ priority: "low" }).set(state.depthLow); - this.oraclePublishBackpressure.set(state.backpressure ? 1 : 0); - this.oraclePublishRpcLatency.set(state.rpcLatencySeconds); - } - - recordOraclePublishDecisions(outcome: "published" | "retried" | "split" | "failed", count: number): void { - this.oraclePublishDecisions.labels({ outcome }).inc(count); - } - - recordProviderProbe(source: string, outcome: Outcome, durationSeconds: number): void { - this.providerProbes.labels({ source, outcome }).inc(); - this.providerProbeDuration.labels({ source }).observe(durationSeconds); - } - - recordProviderFrozenPayload(source: string): void { - this.providerFrozenPayloads.labels({ source }).inc(); - } - - recordProviderSchemaDrift(source: string, kind: string): void { - this.providerSchemaDrift.labels({ source, kind }).inc(); - } - } - - async observeSorobanRpc(operation: string, call: () => Promise): Promise { - const startedAt = process.hrtime.bigint(); - let outcome: Outcome = "success"; - try { - return await call(); - } catch (error) { - outcome = "failure"; - throw error; - } finally { - const durationSeconds = Number(process.hrtime.bigint() - startedAt) / 1_000_000_000; - this.sorobanRpcDuration.labels({ operation, outcome }).observe(durationSeconds); - } - } - - metrics(): Promise { - return this.registry.metrics(); - } - - get contentType(): string { - return this.registry.contentType; - } -} diff --git a/src/oracle/oracle-event.repository.ts b/src/oracle/oracle-event.repository.ts index 4fcee8e..e69de29 100644 --- a/src/oracle/oracle-event.repository.ts +++ b/src/oracle/oracle-event.repository.ts @@ -1,184 +0,0 @@ -import { Inject, Injectable, Logger } from "@nestjs/common"; -import { Pool } from "pg"; -import { PG_POOL } from "../db/db.module"; -import { OracleReading } from "./oracle-reading"; -import { ReferenceDataRejection } from "./reference-data.types"; - -/** - * Persists oracle readings to the `oracle_events` table defined in - * src/db/schema.sql. Writes are fire-and-forget via a micro-task queue so - * they never block the caller (OracleScheduler's broadcast loop or - * ClaimService's evaluation path). - * - * The coverage_type column is an enum in Postgres, matching the five - * values in schema.sql. We map OracleReading.coverageType (PascalCase) to - * the snake_case enum values the DB expects. - */ -@Injectable() -export class OracleEventRepository { - private readonly logger = new Logger(OracleEventRepository.name); - - // Normalise OracleService's PascalCase coverageType names to the - // schema's snake_case coverage_type enum values. - private static readonly COVERAGE_TYPE_MAP: Record = { - StablecoinDepeg: "stablecoin_depeg", - MarketCrash: "market_crash", - LiquidationShield: "liquidation_shield", - SmartContractRisk: "smart_contract_risk", - FlightDelay: "flight_delay", - }; - - constructor(@Inject(PG_POOL) private readonly db: Pool) {} - - /** - * Persists one oracle reading. The call is fire-and-forget: errors are - * logged but never re-thrown so a DB hiccup can never disrupt the - * oracle broadcast or claim-evaluation paths that call this. - * - * Readings rejected by reference-data validation are never written to - * `oracle_events` (which feeds quorum counting); they are diverted to - * the dead-letter path instead. - */ - async record(reading: OracleReading, rejection?: ReferenceDataRejection): Promise { - if (rejection) { - await this.recordDeadLetter(reading, rejection); - return; - } - - const coverageType = OracleEventRepository.COVERAGE_TYPE_MAP[reading.coverageType]; - if (!coverageType) { - this.logger.warn(`Unknown coverageType "${reading.coverageType}" — skipping oracle_events insert`); - return; - } - - try { - await this.db.query( - `INSERT INTO oracle_events (coverage_type, value, source, severity) - VALUES ($1, $2, $3, $4)`, - [coverageType, reading.value, this.extractSource(reading.message), reading.severity] - ); - } catch (err) { - this.logger.error( - `Failed to persist oracle event for ${reading.coverageType}`, - err instanceof Error ? err.message : String(err) - ); - } - } - - /** - * Batch-records multiple readings in a single transaction. - * Also fire-and-forget. - * - * Readings rejected by reference-data validation are diverted to the - * dead-letter table within the same transaction and never inserted into - * `oracle_events`, so they cannot count toward quorum. - */ - async recordBatch( - readings: OracleReading[], - rejections?: Map - ): Promise { - if (readings.length === 0) return; - const client = await this.db.connect(); - try { - await client.query("BEGIN"); - for (const reading of readings) { - const rejection = rejections?.get(reading); - if (rejection) { - await this.insertDeadLetter(client, reading, rejection); - continue; - } - const coverageType = OracleEventRepository.COVERAGE_TYPE_MAP[reading.coverageType]; - if (!coverageType) { - this.logger.warn(`Unknown coverageType "${reading.coverageType}" — skipping oracle_events insert`); - continue; - } - await client.query( - `INSERT INTO oracle_events (coverage_type, value, source, severity) - VALUES ($1, $2, $3, $4)`, - [coverageType, reading.value, this.extractSource(reading.message), reading.severity] - ); - } - await client.query("COMMIT"); - } catch (err) { - await client.query("ROLLBACK").catch(() => {}); - this.logger.error( - "Failed to persist oracle event batch", - err instanceof Error ? err.message : String(err) - ); - } finally { - client.release(); - } - } - - /** - * Persists a reading rejected by reference-data validation into the - * dead-letter table, preserving the raw payload and the rejection - * reason for later inspection. Fire-and-forget: errors are logged but - * never re-thrown. - */ - async recordDeadLetter(reading: OracleReading, rejection: ReferenceDataRejection): Promise { - try { - await this.insertDeadLetter(this.db, reading, rejection); - } catch (err) { - this.logger.error( - `Failed to persist dead-letter reading for ${reading.coverageType}`, - err instanceof Error ? err.message : String(err) - ); - } - } - - /** - * Returns the dataset version pinned for a given feed, or null when no - * pin is configured. Feeds pin a dataset version so a later reference - * data refresh cannot retroactively change how a past reading was - * interpreted. - */ - async getPinnedDatasetVersion(feedId: string): Promise { - try { - const result = await this.db.query( - `SELECT dataset_version FROM feed_dataset_pins WHERE feed_id = $1`, - [feedId] - ); - return result.rows.length > 0 ? (result.rows[0].dataset_version as string) : null; - } catch (err) { - this.logger.error( - `Failed to fetch pinned dataset version for feed ${feedId}`, - err instanceof Error ? err.message : String(err) - ); - return null; - } - } - - private async insertDeadLetter( - db: Pick, - reading: OracleReading, - rejection: ReferenceDataRejection - ): Promise { - await db.query( - `INSERT INTO dead_letter_readings - (coverage_type, feed_id, reason, detail, dataset_version, raw_payload) - VALUES ($1, $2, $3, $4, $5, $6)`, - [ - reading.coverageType, - rejection.feedId ?? null, - rejection.reason, - rejection.detail ?? null, - rejection.datasetVersion ?? null, - JSON.stringify(reading), - ] - ); - } - - /** - * Extracts a short source tag from the oracle message suffix, e.g. - * "[CoinGecko]" → "CoinGecko", "[DeFiLlama]" → "DeFiLlama". - * Falls back to "unknown" so the NOT NULL column is always populated. - */ - private extractSource(message: string): string { - const match = /\[([^\]]+)\]/.exec(message); - if (!match) return "unknown"; - // Strip extra context from compound tags like - // "[CoinGecko; Horizon testnet ledger #123 @ ...]" - return match[1].split(";")[0].trim().substring(0, 40); - } -} diff --git a/src/oracle/oracle.controller.ts b/src/oracle/oracle.controller.ts index e69de29..2202991 100644 --- a/src/oracle/oracle.controller.ts +++ b/src/oracle/oracle.controller.ts @@ -0,0 +1,325 @@ +import { + Controller, + Get, + Post, + Body, + Param, + Query, + HttpCode, + HttpStatus, + NotFoundException, + BadRequestException, + UseGuards, + Req, +} from '@nestjs/common'; +import { Request } from 'express'; +import { OracleService } from './oracle.service'; +import { MethodologyService } from './methodology.service'; +import { IncidentService } from '../services/incidentService'; +import { AuthGuard } from '../auth/auth.guard'; +import { ScopesGuard } from '../auth/scopes.guard'; +import { Scopes } from '../auth/scopes.decorator'; +import { AuditService } from '../services/auditService'; + +interface CreateMethodologyDto { + sourceId: string; + derivationDescription: string; + samplingUniverse: string; + revisionPolicy: string; + expectedLatencyMs: number; + knownFailureModes: string[]; + independenceWeight?: number; + approved?: boolean; + signOff?: { + approvedBy?: string; + reviewedBy?: string; + approvedAt?: string; + }; +} + +interface ApproveMethodologyDto { + approvedBy: string; + reviewedBy: string; + notes?: string; +} + +@Controller() +export class OracleController { + constructor( + private readonly oracleService: OracleService, + private readonly methodologyService: MethodologyService, + private readonly incidentService: IncidentService, + private readonly auditService: AuditService, + ) {} + + // --------------------------------------------------------------------------- + // Public trust / provenance surface + // --------------------------------------------------------------------------- + + /** + * GET /api/v1/oracle/feeds/:feed/methodology + * + * Public trust/provenance endpoint. Returns the methodology version(s) + * currently referenced by the feed's latest consensus decision, so that + * point-in-time reconstruction resolves the exact semantics used. + */ + @Get('api/v1/oracle/feeds/:feed/methodology') + async getFeedMethodology(@Param('feed') feed: string) { + const feedInfo = await this.oracleService.getFeed(feed); + if (!feedInfo) { + throw new NotFoundException(`Feed not found: ${feed}`); + } + + const methodology = await this.methodologyService.getMethodologyForFeed(feed); + if (!methodology) { + throw new NotFoundException(`No methodology recorded for feed: ${feed}`); + } + + return { + feed, + methodology, + }; + } + + // --------------------------------------------------------------------------- + // Operator: methodology management + // --------------------------------------------------------------------------- + + /** + * GET /api/v1/ops/oracle/sources/:id/methodology + * + * Returns the full methodology version history for a source, plus the + * currently active version. + */ + @Get('api/v1/ops/oracle/sources/:id/methodology') + @UseGuards(AuthGuard, ScopesGuard) + @Scopes('oracle:read') + async getSourceMethodology(@Param('id') sourceId: string) { + const history = await this.methodologyService.getMethodologyHistory(sourceId); + if (!history || history.length === 0) { + throw new NotFoundException(`No methodology recorded for source: ${sourceId}`); + } + + const active = await this.methodologyService.getActiveMethodology(sourceId); + + return { + sourceId, + activeVersion: active ? active.version : null, + active, + history, + }; + } + + /** + * POST /api/v1/ops/oracle/sources/:id/methodology + * + * Creates a new immutable methodology version for a source. If `approved` + * is true and sign-off flags are present, the new version becomes the + * active version and the source may be activated. Audit entries are + * recorded for every approval. + */ + @Post('api/v1/ops/oracle/sources/:id/methodology') + @HttpCode(HttpStatus.CREATED) + @UseGuards(AuthGuard, ScopesGuard) + @Scopes('oracle:write') + async createSourceMethodology( + @Param('id') sourceId: string, + @Body() body: CreateMethodologyDto, + @Req() req: Request, + ) { + if (body.sourceId && body.sourceId !== sourceId) { + throw new BadRequestException( + `Path source id (${sourceId}) does not match body sourceId (${body.sourceId})`, + ); + } + + const validation = this.methodologyService.validateMethodologyRecord({ + sourceId, + derivationDescription: body.derivationDescription, + samplingUniverse: body.samplingUniverse, + revisionPolicy: body.revisionPolicy, + expectedLatencyMs: body.expectedLatencyMs, + knownFailureModes: body.knownFailureModes, + }); + + if (!validation.valid) { + throw new BadRequestException({ + message: 'Incomplete methodology record', + errors: validation.errors, + }); + } + + const approved = body.approved === true; + if (approved) { + const signOff = body.signOff || {}; + if (!signOff.approvedBy || !signOff.reviewedBy) { + throw new BadRequestException( + 'Approval requires explicit sign-off flags: approvedBy and reviewedBy', + ); + } + } + + const record = await this.methodologyService.createMethodologyVersion({ + sourceId, + derivationDescription: body.derivationDescription, + samplingUniverse: body.samplingUniverse, + revisionPolicy: body.revisionPolicy, + expectedLatencyMs: body.expectedLatencyMs, + knownFailureModes: body.knownFailureModes, + independenceWeight: body.independenceWeight, + approved, + signOff: body.signOff, + }); + + if (approved) { + await this.auditService.record({ + action: 'oracle.methodology.approve', + actor: (req as any).user?.id ?? 'unknown', + targetType: 'oracle_source', + targetId: sourceId, + metadata: { + methodologyId: record.id, + version: record.version, + approvedBy: body.signOff?.approvedBy, + reviewedBy: body.signOff?.reviewedBy, + }, + }); + } else { + await this.auditService.record({ + action: 'oracle.methodology.create', + actor: (req as any).user?.id ?? 'unknown', + targetType: 'oracle_source', + targetId: sourceId, + metadata: { + methodologyId: record.id, + version: record.version, + }, + }); + } + + return record; + } + + /** + * POST /api/v1/ops/oracle/sources/:id/methodology/:version/approve + * + * Approves an existing (draft) methodology version with explicit sign-off + * flags, making it the active version for the source. + */ + @Post('api/v1/ops/oracle/sources/:id/methodology/:version/approve') + @HttpCode(HttpStatus.OK) + @UseGuards(AuthGuard, ScopesGuard) + @Scopes('oracle:write') + async approveSourceMethodology( + @Param('id') sourceId: string, + @Param('version') version: string, + @Body() body: ApproveMethodologyDto, + @Req() req: Request, + ) { + if (!body.approvedBy || !body.reviewedBy) { + throw new BadRequestException( + 'Approval requires explicit sign-off flags: approvedBy and reviewedBy', + ); + } + + const versionNum = Number(version); + if (!Number.isInteger(versionNum) || versionNum <= 0) { + throw new BadRequestException(`Invalid methodology version: ${version}`); + } + + const record = await this.methodologyService.approveMethodologyVersion({ + sourceId, + version: versionNum, + approvedBy: body.approvedBy, + reviewedBy: body.reviewedBy, + notes: body.notes, + }); + + if (!record) { + throw new NotFoundException( + `Methodology version ${version} not found for source: ${sourceId}`, + ); + } + + await this.auditService.record({ + action: 'oracle.methodology.approve', + actor: (req as any).user?.id ?? 'unknown', + targetType: 'oracle_source', + targetId: sourceId, + metadata: { + methodologyId: record.id, + version: record.version, + approvedBy: body.approvedBy, + reviewedBy: body.reviewedBy, + notes: body.notes, + }, + }); + + return record; + } + + // --------------------------------------------------------------------------- + // Operator: incidents and postmortems + // --------------------------------------------------------------------------- + + /** + * GET /api/v1/ops/incidents + * + * Lists incidents, optionally filtered by feed, status, or trigger class. + */ + @Get('api/v1/ops/incidents') + @UseGuards(AuthGuard, ScopesGuard) + @Scopes('oracle:read') + async listIncidents( + @Query('feed') feed?: string, + @Query('status') status?: string, + @Query('trigger') trigger?: string, + @Query('limit') limit?: string, + @Query('cursor') cursor?: string, + ) { + const parsedLimit = limit ? Number(limit) : undefined; + if (parsedLimit !== undefined && (!Number.isInteger(parsedLimit) || parsedLimit <= 0)) { + throw new BadRequestException(`Invalid limit: ${limit}`); + } + + return this.incidentService.listIncidents({ + feed, + status, + triggerClass: trigger, + limit: parsedLimit, + cursor, + }); + } + + /** + * GET /api/v1/ops/incidents/:id/postmortem + * + * Renders the draft postmortem (markdown) for an incident from its stored + * event timeline. Rendering is deterministic and cached by incident version. + */ + @Get('api/v1/ops/incidents/:id/postmortem') + @UseGuards(AuthGuard, ScopesGuard) + @Scopes('oracle:read') + async getIncidentPostmortem(@Param('id') incidentId: string) { + const incident = await this.incidentService.getIncident(incidentId); + if (!incident) { + throw new NotFoundException(`Incident not found: ${incidentId}`); + } + + const postmortem = await this.incidentService.renderPostmortem(incidentId); + if (!postmortem) { + throw new NotFoundException(`Postmortem not available for incident: ${incidentId}`); + } + + return { + incidentId, + version: incident.version, + status: incident.status, + triggerClass: incident.triggerClass, + openedAt: incident.openedAt, + closedAt: incident.closedAt, + markdown: postmortem.markdown, + generatedAt: postmortem.generatedAt, + }; + } +} \ No newline at end of file diff --git a/src/oracle/oracle.module.ts b/src/oracle/oracle.module.ts index 31b3d61..e69de29 100644 --- a/src/oracle/oracle.module.ts +++ b/src/oracle/oracle.module.ts @@ -1,12 +0,0 @@ -import { Module } from '@nestjs/common'; -import { OracleService } from './oracle.service'; -import { OracleController } from './oracle.controller'; -import { ReferenceDataModule } from './reference-data/reference-data.module'; - -@Module({ - imports: [ReferenceDataModule], - controllers: [OracleController], - providers: [OracleService], - exports: [OracleService, ReferenceDataModule], -}) -export class OracleModule {} \ No newline at end of file diff --git a/src/oracle/oracle.service.ts b/src/oracle/oracle.service.ts index e69de29..3c41ac6 100644 --- a/src/oracle/oracle.service.ts +++ b/src/oracle/oracle.service.ts @@ -0,0 +1,517 @@ +import { randomUUID } from 'crypto'; +import { Pool, PoolClient } from 'pg'; + +/** + * Methodology registry types. + * + * Methodology records describe how a source derives the values it publishes + * (aggregation method, sampling universe, revision policy, latency profile, + * known failure modes). Records are immutable: edits create a new version. + */ +export interface MethodologyRecord { + id: string; + sourceId: string; + version: number; + derivationDescription: string; + samplingUniverse: string; + revisionPolicy: string; + expectedLatencyMs: number; + knownFailureModes: string[]; + approved: boolean; + approvedBy: string | null; + approvedAt: Date | null; + signOffIndependence: boolean; + signOffAccuracy: boolean; + signOffLegal: boolean; + reattestationDueAt: Date | null; + lastReattestedAt: Date | null; + divergenceFlagged: boolean; + divergenceNotes: string | null; + createdAt: Date; + createdBy: string | null; +} + +export interface CreateMethodologyInput { + sourceId: string; + derivationDescription: string; + samplingUniverse: string; + revisionPolicy: string; + expectedLatencyMs: number; + knownFailureModes: string[]; + reattestationIntervalDays?: number; + createdBy?: string; +} + +export interface ApproveMethodologyInput { + methodologyId: string; + approvedBy: string; + signOffIndependence: boolean; + signOffAccuracy: boolean; + signOffLegal: boolean; +} + +export interface ObservedBehavior { + /** Observed aggregation method, if detectable. */ + aggregationMethod?: string; + /** Observed sampling universe, if detectable. */ + samplingUniverse?: string; + /** Observed latency in ms. */ + latencyMs?: number; + /** Free-form notes. */ + notes?: string; +} + +export interface MethodologyDivergence { + sourceId: string; + methodologyId: string; + methodologyVersion: number; + reasons: string[]; + observed: ObservedBehavior; + flaggedAt: Date; +} + +export interface FeedMethodologyView { + feedId: string; + sources: Array<{ + sourceId: string; + methodologyId: string; + version: number; + derivationDescription: string; + samplingUniverse: string; + revisionPolicy: string; + expectedLatencyMs: number; + knownFailureModes: string[]; + approved: boolean; + approvedAt: Date | null; + divergenceFlagged: boolean; + reattestationDueAt: Date | null; + }>; +} + +export class MethodologyError extends Error { + constructor(message: string, public readonly code: string) { + super(message); + this.name = 'MethodologyError'; + } +} + +export interface OracleServiceDeps { + pool: Pool; + logger?: { + info: (msg: string, meta?: Record) => void; + warn: (msg: string, meta?: Record) => void; + error: (msg: string, meta?: Record) => void; + }; +} + +const DEFAULT_REATTESTATION_INTERVAL_DAYS = 90; + +export class OracleService { + private readonly pool: Pool; + private readonly logger: NonNullable; + + constructor(deps: OracleServiceDeps) { + this.pool = deps.pool; + this.logger = + deps.logger ?? + ({ + info: () => undefined, + warn: () => undefined, + error: () => undefined, + } as NonNullable); + } + + // --------------------------------------------------------------------------- + // Methodology registry + // --------------------------------------------------------------------------- + + /** + * Create a new immutable methodology version for a source. + * The version number is derived from the current max version for the source. + */ + async createMethodologyVersion( + input: CreateMethodologyInput, + client?: PoolClient, + ): Promise { + if (!input.sourceId) { + throw new MethodologyError('sourceId is required', 'MISSING_SOURCE_ID'); + } + if (!input.derivationDescription || input.derivationDescription.trim().length === 0) { + throw new MethodologyError( + 'derivationDescription is required', + 'MISSING_DERIVATION', + ); + } + if (!input.samplingUniverse || input.samplingUniverse.trim().length === 0) { + throw new MethodologyError( + 'samplingUniverse is required', + 'MISSING_SAMPLING_UNIVERSE', + ); + } + if (!input.revisionPolicy || input.revisionPolicy.trim().length === 0) { + throw new MethodologyError( + 'revisionPolicy is required', + 'MISSING_REVISION_POLICY', + ); + } + if ( + typeof input.expectedLatencyMs !== 'number' || + !Number.isFinite(input.expectedLatencyMs) || + input.expectedLatencyMs < 0 + ) { + throw new MethodologyError( + 'expectedLatencyMs must be a non-negative number', + 'INVALID_LATENCY', + ); + } + if (!Array.isArray(input.knownFailureModes) || input.knownFailureModes.length === 0) { + throw new MethodologyError( + 'knownFailureModes must be a non-empty array', + 'MISSING_FAILURE_MODES', + ); + } + + const runner = client ?? this.pool; + const id = randomUUID(); + const intervalDays = + input.reattestationIntervalDays ?? DEFAULT_REATTESTATION_INTERVAL_DAYS; + + const { rows } = await runner.query( + `WITH next_version AS ( + SELECT COALESCE(MAX(version), 0) + 1 AS v + FROM oracle_methodology + WHERE source_id = $1 + ) + INSERT INTO oracle_methodology ( + id, source_id, version, derivation_description, sampling_universe, + revision_policy, expected_latency_ms, known_failure_modes, + approved, reattestation_due_at, created_by + ) + SELECT $2, $1, next_version.v, $3, $4, $5, $6, $7, false, + now() + ($8::int * INTERVAL '1 day'), $9 + FROM next_version + RETURNING *`, + [ + input.sourceId, + id, + input.derivationDescription, + input.samplingUniverse, + input.revisionPolicy, + input.expectedLatencyMs, + input.knownFailureModes, + intervalDays, + input.createdBy ?? null, + ], + ); + + const record = this.mapMethodologyRow(rows[0]); + this.logger.info('methodology version created', { + sourceId: record.sourceId, + methodologyId: record.id, + version: record.version, + }); + return record; + } + + /** + * Approve a methodology version with explicit sign-off flags. + * Approval is a one-way transition; approved rows are never mutated again. + */ + async approveMethodologyVersion( + input: ApproveMethodologyInput, + ): Promise { + if (!input.approvedBy) { + throw new MethodologyError('approvedBy is required', 'MISSING_APPROVER'); + } + if (!input.signOffIndependence || !input.signOffAccuracy || !input.signOffLegal) { + throw new MethodologyError( + 'all sign-off flags (independence, accuracy, legal) must be true to approve', + 'INCOMPLETE_SIGN_OFF', + ); + } + + const { rows } = await this.pool.query( + `UPDATE oracle_methodology + SET approved = true, + approved_by = $2, + approved_at = now(), + sign_off_independence = $3, + sign_off_accuracy = $4, + sign_off_legal = $5 + WHERE id = $1 AND approved = false + RETURNING *`, + [ + input.methodologyId, + input.approvedBy, + input.signOffIndependence, + input.signOffAccuracy, + input.signOffLegal, + ], + ); + + if (rows.length === 0) { + throw new MethodologyError( + 'methodology not found or already approved', + 'NOT_APPROVABLE', + ); + } + + const record = this.mapMethodologyRow(rows[0]); + this.logger.info('methodology version approved', { + sourceId: record.sourceId, + methodologyId: record.id, + version: record.version, + approvedBy: record.approvedBy, + }); + return record; + } + + /** + * Return the currently active (latest approved) methodology for a source. + */ + async getActiveMethodology( + sourceId: string, + client?: PoolClient, + ): Promise { + const runner = client ?? this.pool; + const { rows } = await runner.query( + `SELECT * FROM oracle_methodology + WHERE source_id = $1 AND approved = true + ORDER BY version DESC + LIMIT 1`, + [sourceId], + ); + return rows.length > 0 ? this.mapMethodologyRow(rows[0]) : null; + } + + /** + * Return a specific methodology version (used for point-in-time reconstruction). + */ + async getMethodologyVersion( + methodologyId: string, + client?: PoolClient, + ): Promise { + const runner = client ?? this.pool; + const { rows } = await runner.query( + `SELECT * FROM oracle_methodology WHERE id = $1`, + [methodologyId], + ); + return rows.length > 0 ? this.mapMethodologyRow(rows[0]) : null; + } + + /** + * Assert that a source may be activated. Throws when no approved methodology + * exists, or when the active methodology has been flagged for divergence. + */ + async assertSourceActivatable( + sourceId: string, + client?: PoolClient, + ): Promise { + const active = await this.getActiveMethodology(sourceId, client); + if (!active) { + throw new MethodologyError( + `source ${sourceId} cannot be activated: no approved methodology record`, + 'NO_APPROVED_METHODOLOGY', + ); + } + if (active.divergenceFlagged) { + throw new MethodologyError( + `source ${sourceId} cannot be activated: methodology divergence flagged`, + 'METHODOLOGY_DIVERGENCE', + ); + } + return active; + } + + /** + * Re-attestation check: compare observed behavior against the recorded + * methodology and flag the source when they diverge. + */ + async checkReattestation( + sourceId: string, + observed: ObservedBehavior, + ): Promise { + const active = await this.getActiveMethodology(sourceId); + if (!active) { + throw new MethodologyError( + `source ${sourceId} has no approved methodology to re-attest`, + 'NO_APPROVED_METHODOLOGY', + ); + } + + const reasons: string[] = []; + + if ( + observed.aggregationMethod && + !this.descriptionMentions( + active.derivationDescription, + observed.aggregationMethod, + ) + ) { + reasons.push( + `observed aggregation method "${observed.aggregationMethod}" not described in methodology`, + ); + } + + if ( + observed.samplingUniverse && + !this.descriptionMentions(active.samplingUniverse, observed.samplingUniverse) + ) { + reasons.push( + `observed sampling universe "${observed.samplingUniverse}" not described in methodology`, + ); + } + + if ( + typeof observed.latencyMs === 'number' && + observed.latencyMs > active.expectedLatencyMs * 2 + ) { + reasons.push( + `observed latency ${observed.latencyMs}ms exceeds 2x expected ${active.expectedLatencyMs}ms`, + ); + } + + if (reasons.length === 0) { + // Record successful re-attestation. + await this.pool.query( + `UPDATE oracle_methodology + SET last_reattested_at = now(), + reattestation_due_at = now() + INTERVAL '90 days' + WHERE id = $1`, + [active.id], + ); + return null; + } + + const flaggedAt = new Date(); + await this.pool.query( + `UPDATE oracle_methodology + SET divergence_flagged = true, + divergence_notes = $2, + last_reattested_at = now() + WHERE id = $1`, + [active.id, reasons.join('; ')], + ); + + this.logger.warn('methodology divergence flagged', { + sourceId, + methodologyId: active.id, + version: active.version, + reasons, + }); + + return { + sourceId, + methodologyId: active.id, + methodologyVersion: active.version, + reasons, + observed, + flaggedAt, + }; + } + + /** + * Resolve the methodology versions used by a feed's sources, for the + * public trust/provenance surface. + */ + async getMethodologyForFeed(feedId: string): Promise { + const { rows } = await this.pool.query( + `SELECT s.source_id, m.* + FROM oracle_feed_sources s + JOIN LATERAL ( + SELECT * FROM oracle_methodology + WHERE source_id = s.source_id AND approved = true + ORDER BY version DESC + LIMIT 1 + ) m ON true + WHERE s.feed_id = $1 + ORDER BY s.source_id`, + [feedId], + ); + + return { + feedId, + sources: rows.map((row) => { + const m = this.mapMethodologyRow(row); + return { + sourceId: m.sourceId, + methodologyId: m.id, + version: m.version, + derivationDescription: m.derivationDescription, + samplingUniverse: m.samplingUniverse, + revisionPolicy: m.revisionPolicy, + expectedLatencyMs: m.expectedLatencyMs, + knownFailureModes: m.knownFailureModes, + approved: m.approved, + approvedAt: m.approvedAt, + divergenceFlagged: m.divergenceFlagged, + reattestationDueAt: m.reattestationDueAt, + }; + }), + }; + } + + /** + * Resolve the methodology version ids used for a set of sources at a point + * in time. Used when writing consensus decisions and evidence records. + */ + async resolveMethodologyVersions( + sourceIds: string[], + client?: PoolClient, + ): Promise> { + const runner = client ?? this.pool; + const result: Record = {}; + for (const sourceId of sourceIds) { + const active = await this.getActiveMethodology(sourceId, client); + if (!active) { + throw new MethodologyError( + `source ${sourceId} has no approved methodology`, + 'NO_APPROVED_METHODOLOGY', + ); + } + result[sourceId] = active.id; + } + void runner; + return result; + } + + // --------------------------------------------------------------------------- + // Helpers + // --------------------------------------------------------------------------- + + private descriptionMentions(haystack: string, needle: string): boolean { + if (!haystack || !needle) return false; + return haystack.toLowerCase().includes(needle.toLowerCase()); + } + + private mapMethodologyRow(row: any): MethodologyRecord { + return { + id: row.id, + sourceId: row.source_id, + version: Number(row.version), + derivationDescription: row.derivation_description, + samplingUniverse: row.sampling_universe, + revisionPolicy: row.revision_policy, + expectedLatencyMs: Number(row.expected_latency_ms), + knownFailureModes: Array.isArray(row.known_failure_modes) + ? row.known_failure_modes + : [], + approved: Boolean(row.approved), + approvedBy: row.approved_by ?? null, + approvedAt: row.approved_at ? new Date(row.approved_at) : null, + signOffIndependence: Boolean(row.sign_off_independence), + signOffAccuracy: Boolean(row.sign_off_accuracy), + signOffLegal: Boolean(row.sign_off_legal), + reattestationDueAt: row.reattestation_due_at + ? new Date(row.reattestation_due_at) + : null, + lastReattestedAt: row.last_reattested_at + ? new Date(row.last_reattested_at) + : null, + divergenceFlagged: Boolean(row.divergence_flagged), + divergenceNotes: row.divergence_notes ?? null, + createdAt: new Date(row.created_at), + createdBy: row.created_by ?? null, + }; + } +} \ No newline at end of file diff --git a/src/services/claimProcessor.ts b/src/services/claimProcessor.ts index b8863f7..e69de29 100644 --- a/src/services/claimProcessor.ts +++ b/src/services/claimProcessor.ts @@ -1,55 +0,0 @@ -import { logger } from '../utils/logger'; -import { - DecisionEvaluation, - assertSettlementAllowed, - determineTriggers, - enqueueForReview, -} from './reviewQueue'; - -export interface ClaimDecision { - claimId: string; - decisionId: string; - value: number; - confidence: number; - evidenceMature: boolean; - postRemediation: boolean; - decision: unknown; - readings: unknown[]; - sourceAttributions: unknown[]; - anomalyFlags: string[]; - explanation: string; -} - -export type ProcessOutcome = - | { status: 'settled'; decisionId: string } - | { status: 'held_for_review'; decisionId: string; reviewEntryId: string }; - -export async function processClaim(claim: ClaimDecision): Promise { - const evaluation: DecisionEvaluation = { - claimId: claim.claimId, - decisionId: claim.decisionId, - value: claim.value, - confidence: claim.confidence, - evidenceMature: claim.evidenceMature, - postRemediation: claim.postRemediation, - decision: claim.decision, - readings: claim.readings, - sourceAttributions: claim.sourceAttributions, - anomalyFlags: claim.anomalyFlags, - explanation: claim.explanation, - }; - - const triggers = determineTriggers(evaluation); - if (triggers.length > 0) { - const entry = await enqueueForReview(evaluation, triggers); - logger.info( - { claimId: claim.claimId, decisionId: claim.decisionId, triggers }, - 'claim decision held for human review', - ); - return { status: 'held_for_review', decisionId: claim.decisionId, reviewEntryId: entry.id }; - } - - await assertSettlementAllowed(claim.decisionId); - logger.info({ claimId: claim.claimId, decisionId: claim.decisionId }, 'claim decision settled'); - return { status: 'settled', decisionId: claim.decisionId }; -} diff --git a/src/services/oracleMonitor.ts b/src/services/oracleMonitor.ts index e69de29..28124be 100644 --- a/src/services/oracleMonitor.ts +++ b/src/services/oracleMonitor.ts @@ -0,0 +1,311 @@ +import { EventEmitter } from 'events'; +import { Pool } from 'pg'; +import { IncidentService, IncidentTriggerClass } from './incidentService'; + +/** + * Oracle Monitor Service + * + * Watches oracle feeds for health, consensus, and freshness problems and + * opens incidents in the evidence ledger when defined trigger classes fire. + * + * Trigger classes that auto-open (or group into) an incident: + * - circuit_breaker_trip + * - quorum_failure + * - freshness_breach + * - operator_pause + * - backfill_replay + * + * Rolling breaker trips within the grouping window are attached to the + * existing open incident rather than creating duplicates. + */ + +export type FeedHealth = 'healthy' | 'degraded' | 'unhealthy' | 'paused' | 'unknown'; + +export interface SourceReading { + sourceId: string; + feedId: string; + value: number | null; + health: FeedHealth; + observedAt: Date; + /** Raw provider payload; must never be rendered unredacted. */ + providerPayload?: Record; +} + +export interface ConsensusOutcome { + feedId: string; + decisionId: string; + agreed: boolean; + reason: string; + value: number | null; + quorumReached: boolean; + participatingSources: string[]; + decidedAt: Date; +} + +export interface CircuitBreakerState { + feedId: string; + open: boolean; + reason: string; + trippedAt: Date; +} + +export interface FreshnessState { + feedId: string; + lastUpdatedAt: Date; + maxAgeMs: number; + breached: boolean; +} + +export interface OperatorPauseEvent { + feedId: string; + operatorId: string; + reason: string; + pausedAt: Date; +} + +export interface BackfillReplayEvent { + feedId: string; + replayId: string; + fromAt: Date; + toAt: Date; + initiatedBy: string; + startedAt: Date; +} + +export interface OracleMonitorConfig { + /** Rolling window (ms) within which repeated breaker trips group together. */ + breakerGroupingWindowMs: number; + /** Max age (ms) before a feed is considered stale. */ + freshnessMaxAgeMs: number; +} + +const DEFAULT_CONFIG: OracleMonitorConfig = { + breakerGroupingWindowMs: 15 * 60 * 1000, + freshnessMaxAgeMs: 5 * 60 * 1000, +}; + +export interface OracleMonitorDeps { + db: Pool; + incidentService: IncidentService; + config?: Partial; + now?: () => Date; +} + +export class OracleMonitor extends EventEmitter { + private readonly db: Pool; + private readonly incidentService: IncidentService; + private readonly config: OracleMonitorConfig; + private readonly now: () => Date; + + /** Tracks last breaker trip per feed to support grouping. */ + private readonly lastBreakerTrip = new Map(); + + constructor(deps: OracleMonitorDeps) { + super(); + this.db = deps.db; + this.incidentService = deps.incidentService; + this.config = { ...DEFAULT_CONFIG, ...(deps.config ?? {}) }; + this.now = deps.now ?? (() => new Date()); + } + + // --------------------------------------------------------------------------- + // Trigger points + // --------------------------------------------------------------------------- + + /** + * Called when a circuit breaker trips for a feed. Repeated trips within the + * grouping window attach to the existing incident instead of opening a new + * one. + */ + async onCircuitBreakerTrip(state: CircuitBreakerState): Promise { + if (!state.open) { + return; + } + + const trippedAt = state.trippedAt ?? this.now(); + const lastTrip = this.lastBreakerTrip.get(state.feedId); + const withinWindow = + lastTrip !== undefined && + trippedAt.getTime() - lastTrip.getTime() <= this.config.breakerGroupingWindowMs; + + this.lastBreakerTrip.set(state.feedId, trippedAt); + + const ledgerEventId = await this.recordLedgerEvent({ + feedId: state.feedId, + eventType: 'circuit_breaker_trip', + occurredAt: trippedAt, + payload: { + reason: state.reason, + grouped: withinWindow, + }, + }); + + await this.incidentService.openOrGroupIncident({ + triggerClass: 'circuit_breaker_trip', + feedId: state.feedId, + occurredAt: trippedAt, + ledgerEventId, + groupingKey: `circuit_breaker_trip:${state.feedId}`, + groupingWindowMs: this.config.breakerGroupingWindowMs, + summary: `Circuit breaker tripped for feed ${state.feedId}: ${state.reason}`, + }); + + this.emit('circuit_breaker_trip', state); + } + + /** + * Called when a consensus decision fails to reach quorum or is otherwise + * rejected. + */ + async onQuorumFailure(outcome: ConsensusOutcome): Promise { + if (outcome.quorumReached && outcome.agreed) { + return; + } + + const occurredAt = outcome.decidedAt ?? this.now(); + + const ledgerEventId = await this.recordLedgerEvent({ + feedId: outcome.feedId, + eventType: 'quorum_failure', + occurredAt, + payload: { + decisionId: outcome.decisionId, + reason: outcome.reason, + quorumReached: outcome.quorumReached, + participatingSources: outcome.participatingSources, + }, + }); + + await this.incidentService.openOrGroupIncident({ + triggerClass: 'quorum_failure', + feedId: outcome.feedId, + occurredAt, + ledgerEventId, + groupingKey: `quorum_failure:${outcome.feedId}`, + groupingWindowMs: this.config.breakerGroupingWindowMs, + summary: `Quorum failure for feed ${outcome.feedId}: ${outcome.reason}`, + }); + + this.emit('quorum_failure', outcome); + } + + /** + * Called when a feed's last update exceeds the freshness threshold. + */ + async onFreshnessBreach(state: FreshnessState): Promise { + if (!state.breached) { + return; + } + + const occurredAt = this.now(); + const ageMs = occurredAt.getTime() - state.lastUpdatedAt.getTime(); + + const ledgerEventId = await this.recordLedgerEvent({ + feedId: state.feedId, + eventType: 'freshness_breach', + occurredAt, + payload: { + lastUpdatedAt: state.lastUpdatedAt.toISOString(), + ageMs, + maxAgeMs: state.maxAgeMs, + }, + }); + + await this.incidentService.openOrGroupIncident({ + triggerClass: 'freshness_breach', + feedId: state.feedId, + occurredAt, + ledgerEventId, + groupingKey: `freshness_breach:${state.feedId}`, + groupingWindowMs: this.config.breakerGroupingWindowMs, + summary: `Freshness breach for feed ${state.feedId}: age ${ageMs}ms exceeds ${state.maxAgeMs}ms`, + }); + + this.emit('freshness_breach', state); + } + + /** + * Called when an operator manually pauses a feed. + */ + async onOperatorPause(event: OperatorPauseEvent): Promise { + const occurredAt = event.pausedAt ?? this.now(); + + const ledgerEventId = await this.recordLedgerEvent({ + feedId: event.feedId, + eventType: 'operator_pause', + occurredAt, + payload: { + operatorId: event.operatorId, + reason: event.reason, + }, + }); + + await this.incidentService.openOrGroupIncident({ + triggerClass: 'operator_pause', + feedId: event.feedId, + occurredAt, + ledgerEventId, + groupingKey: `operator_pause:${event.feedId}`, + groupingWindowMs: this.config.breakerGroupingWindowMs, + summary: `Operator ${event.operatorId} paused feed ${event.feedId}: ${event.reason}`, + }); + + this.emit('operator_pause', event); + } + + /** + * Called when a backfill or replay operation is initiated. + */ + async onBackfillReplay(event: BackfillReplayEvent): Promise { + const occurredAt = event.startedAt ?? this.now(); + + const ledgerEventId = await this.recordLedgerEvent({ + feedId: event.feedId, + eventType: 'backfill_replay', + occurredAt, + payload: { + replayId: event.replayId, + fromAt: event.fromAt.toISOString(), + toAt: event.toAt.toISOString(), + initiatedBy: event.initiatedBy, + }, + }); + + await this.incidentService.openOrGroupIncident({ + triggerClass: 'backfill_replay', + feedId: event.feedId, + occurredAt, + ledgerEventId, + groupingKey: `backfill_replay:${event.feedId}`, + groupingWindowMs: this.config.breakerGroupingWindowMs, + summary: `Backfill/replay ${event.replayId} initiated for feed ${event.feedId} by ${event.initiatedBy}`, + }); + + this.emit('backfill_replay', event); + } + + // --------------------------------------------------------------------------- + // Ledger helpers + // --------------------------------------------------------------------------- + + /** + * Appends a triggering event to the evidence ledger and returns its id. + * The payload is stored as-is; redaction happens at render time. + */ + private async recordLedgerEvent(input: { + feedId: string; + eventType: IncidentTriggerClass; + occurredAt: Date; + payload: Record; + }): Promise { + const result = await this.db.query<{ id: string }>( + `INSERT INTO oracle_evidence_ledger (feed_id, event_type, occurred_at, payload) + VALUES ($1, $2, $3, $4) + RETURNING id`, + [input.feedId, input.eventType, input.occurredAt, JSON.stringify(input.payload)], + ); + return result.rows[0].id; + } +} + +export default OracleMonitor; \ No newline at end of file diff --git a/test/integration/setup/db-client.ts b/test/integration/setup/db-client.ts index d237ec8..e3c284a 100644 --- a/test/integration/setup/db-client.ts +++ b/test/integration/setup/db-client.ts @@ -1,3 +1,5 @@ +import { readdirSync, readFileSync } from "fs"; +import { join } from "path"; import { Client, Pool } from "pg"; /** @@ -15,6 +17,36 @@ import { Client, Pool } from "pg"; * ``` */ +/** + * Tables introduced by the methodology, incident, and partitioning + * migrations. Truncating these between tests keeps the new integration + * suites isolated from one another. + */ +export const METHODOLOGY_TABLES = [ + "source_methodology_versions", + "source_methodology_attestations", +] as const; + +export const INCIDENT_TABLES = [ + "incident_events", + "incidents", +] as const; + +/** + * Partitioned tables. `TRUNCATE` on a partitioned parent cascades to all + * of its partitions, so callers only need to name the parents. + */ +export const PARTITIONED_TABLES = [ + "oracle_readings", + "evidence_ledger", +] as const; + +export const ALL_EXTENDED_TABLES = [ + ...METHODOLOGY_TABLES, + ...INCIDENT_TABLES, + ...PARTITIONED_TABLES, +] as const; + /** * Returns a pg.Pool connected to the ephemeral database whose URL was * injected by global-setup into INTEGRATION_DATABASE_URL. @@ -51,3 +83,62 @@ export async function connectTestClient(): Promise { await client.connect(); return client; } + +/** + * Applies `src/db/schema.sql` followed by every numbered migration in + * `src/db/migrations/` (in lexical order) to the ephemeral database. + * + * `schema.sql` is the canonical definition and already reflects the + * partitioning and methodology tables, so migrations are applied with + * `ON_ERROR_STOP` semantics but tolerate "already exists" style errors + * (SQLSTATE 42P07, 42710, 42P16) so the harness works whether the schema + * was applied fresh or the migrations are being layered on top. + * + * Returns the list of migration files that were applied. + */ +export async function applySchemaAndMigrations(): Promise { + const client = await connectTestClient(); + try { + const schemaPath = join(__dirname, "..", "..", "..", "src", "db", "schema.sql"); + await client.query(readFileSync(schemaPath, "utf8")); + + const migrationsDir = join(__dirname, "..", "..", "..", "src", "db", "migrations"); + const migrations = readdirSync(migrationsDir) + .filter((f) => f.endsWith(".sql")) + .sort(); + + const applied: string[] = []; + for (const file of migrations) { + const sql = readFileSync(join(migrationsDir, file), "utf8"); + try { + await client.query(sql); + applied.push(file); + } catch (err) { + const code = (err as { code?: string }).code; + // Idempotency: schema.sql may already have created these objects. + if (code === "42P07" || code === "42710" || code === "42P16") { + applied.push(`${file} (skipped: already applied)`); + continue; + } + throw new Error(`Migration ${file} failed: ${(err as Error).message}`); + } + } + return applied; + } finally { + await client.end(); + } +} + +/** + * Truncates the methodology, incident, and partitioned readings/evidence + * tables. Safe to call even if some tables are absent (e.g. when a suite + * runs against a schema-only database). + */ +export async function truncateExtendedTables(pool: Pool): Promise { + const { rows } = await pool.query<{ tablename: string }>( + `SELECT tablename FROM pg_tables WHERE schemaname = 'public'` + ); + const existing = new Set(rows.map((r) => r.tablename)); + const present = ALL_EXTENDED_TABLES.filter((t) => existing.has(t)); + await truncateTables(pool, present); +}