diff --git a/src/db/migrations/004_oracle_reading_admission.sql b/src/db/migrations/004_oracle_reading_admission.sql new file mode 100644 index 0000000..ecf484d --- /dev/null +++ b/src/db/migrations/004_oracle_reading_admission.sql @@ -0,0 +1,24 @@ +-- Apply to an existing database to store upstream observation time and payload +-- digest on oracle readings, and to dedupe replayed readings. +-- psql "$DATABASE_URL" -f src/db/migrations/004_oracle_reading_admission.sql + +-- Nullable: rows written before this migration have no observation time. +-- The application rejects new readings that lack one (see oracle-timestamp.ts). +ALTER TABLE oracle_events + ADD COLUMN IF NOT EXISTS observed_at TIMESTAMPTZ, + ADD COLUMN IF NOT EXISTS payload_digest CHAR(64); + +-- Replay guard: one receipt per (source, observation time, payload digest). +-- Kept separate from the monthly-partitioned oracle_events, whose unique keys +-- must include recorded_at and therefore cannot enforce this globally. +CREATE TABLE IF NOT EXISTS oracle_reading_receipts ( + source VARCHAR(40) NOT NULL, + observed_at TIMESTAMPTZ NOT NULL, + payload_digest CHAR(64) NOT NULL, + received_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (source, observed_at, payload_digest) +); + +-- Supports retention cleanup of old receipts. +CREATE INDEX IF NOT EXISTS idx_oracle_reading_receipts_received_at + ON oracle_reading_receipts (received_at); diff --git a/src/oracle/oracle-event.repository.ts b/src/oracle/oracle-event.repository.ts index e69de29..059d2c0 100644 --- a/src/oracle/oracle-event.repository.ts +++ b/src/oracle/oracle-event.repository.ts @@ -0,0 +1,164 @@ +import { Inject, Injectable, Logger } from "@nestjs/common"; +import { Pool } from "pg"; +import { PG_POOL } from "../db/db.module"; +import { OracleDeadLetterRepository } from "./oracle-dead-letter.repository"; +import { CacheService } from "../cache/cache.service"; +import { OracleReading } from "./oracle-reading"; +import { ReadingAdmissionGate, SourceClockOffset, readingDigest } from "./oracle-timestamp"; +import { OracleStatusService } from "./oracle-status.service"; + +type InsertParams = [string, number, string, string, Date, string]; + +/** + * 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). + * + * Every reading either lands in oracle_events or in oracle_dead_letters + * (src/db/migrations/004_oracle_dead_letter_queue.sql). Nothing is dropped. + * + * 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. + * + * Every reading passes through ReadingAdmissionGate before it is written. + * Rejected readings are logged with a reason code and never persisted. + * Replays (same source + observation time + payload digest) are dropped by + * the oracle_reading_receipts primary key in the same statement as the insert. + */ +@Injectable() +export class OracleEventRepository { + private readonly logger = new Logger(OracleEventRepository.name); + private readonly admission = new ReadingAdmissionGate(); + + // Normalise OracleService's PascalCase coverageType names to the + // schema's snake_case coverage_type enum values. + static readonly COVERAGE_TYPE_MAP: Record = { + StablecoinDepeg: "stablecoin_depeg", + MarketCrash: "market_crash", + LiquidationShield: "liquidation_shield", + SmartContractRisk: "smart_contract_risk", + FlightDelay: "flight_delay", + }; + + // Receipt insert and event insert are one statement, so a replay can never + // leave an event row without its receipt (or vice versa). The event row is + // written only when the receipt insert actually inserted. + private static readonly INSERT_SQL = ` + WITH receipt AS ( + INSERT INTO oracle_reading_receipts (source, observed_at, payload_digest) + VALUES ($3::varchar, $5::timestamptz, $6::char(64)) + ON CONFLICT DO NOTHING + RETURNING 1 + ) + INSERT INTO oracle_events (coverage_type, value, source, severity, observed_at, payload_digest) + SELECT $1::coverage_type, $2::numeric, $3::varchar, $4::varchar, $5::timestamptz, $6::char(64) FROM receipt`; + + constructor( + @Inject(PG_POOL) private readonly db: Pool, + private readonly deadLetters: OracleDeadLetterRepository, + private readonly cache: CacheService + ) {} + + /** + * 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. + * Failed readings are written to the dead-letter queue. + */ + async record(reading: OracleReading): Promise { + const params = this.admitForInsert(reading); + if (!params) return; + + try { + const result = await this.db.query(OracleEventRepository.INSERT_SQL, params); + if (result.rowCount === 0) { + this.logger.debug(`Discarded replayed oracle reading from ${params[2]} for ${reading.coverageType}`); + } + await OracleStatusService.invalidate(this.cache); + } catch (err) { + this.logger.error( + `Failed to persist oracle event for ${reading.coverageType}`, + err instanceof Error ? err.message : String(err) + ); + await this.deadLetters.enqueue([reading], err instanceof Error ? err.constructor.name : "UnknownError", err); + } + } + + /** + * Batch-records multiple readings in a single transaction. + * Also fire-and-forget. If the transaction rolls back, every reading in + * the batch that passed validation is dead-lettered. + */ + async recordBatch(readings: OracleReading[]): Promise { + if (readings.length === 0) return; + const client = await this.db.connect(); + const valid: OracleReading[] = []; + try { + await client.query("BEGIN"); + for (const reading of readings) { + const coverageType = OracleEventRepository.COVERAGE_TYPE_MAP[reading.coverageType]; + if (!coverageType) { + this.logger.warn(`Unknown coverageType "${reading.coverageType}" — dead-lettering oracle reading`); + await this.deadLetters.enqueue([reading], "UnknownCoverageType", new Error(`Unknown coverageType "${reading.coverageType}"`)); + continue; + } + const params = this.admitForInsert(reading); + if (!params) continue; + valid.push(reading); + await client.query(OracleEventRepository.INSERT_SQL, params); + } + await client.query("COMMIT"); + await OracleStatusService.invalidate(this.cache); + } catch (err) { + await client.query("ROLLBACK").catch(() => {}); + this.logger.error( + "Failed to persist oracle event batch", + err instanceof Error ? err.message : String(err) + ); + await this.deadLetters.enqueue(valid, err instanceof Error ? err.constructor.name : "UnknownError", err); + } finally { + client.release(); + } + } + + /** Per-source clock offset (receipt − observation), for operators to spot skewed providers. */ + getClockOffsets(): SourceClockOffset[] { + return this.admission.clockOffsets(); + } + + /** + * Maps and admits a reading. Returns the insert parameters, or undefined when + * the reading must not be persisted (unknown coverage type or rejected timestamp). + */ + private admitForInsert(reading: OracleReading): InsertParams | undefined { + const coverageType = OracleEventRepository.COVERAGE_TYPE_MAP[reading.coverageType]; + if (!coverageType) { + this.logger.warn(`Unknown coverageType "${reading.coverageType}" — skipping oracle_events insert`); + return undefined; + } + + const source = this.extractSource(reading.message); + const admission = this.admission.admit(source, reading.observedAt); + if (!admission.accepted) { + this.logger.warn(`Rejected oracle reading ${reading.coverageType} from ${source}: ${admission.reason}`); + return undefined; + } + + return [coverageType, reading.value, source, reading.severity, admission.observedAt, readingDigest(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. + */ + static 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-reading.ts b/src/oracle/oracle-reading.ts index e69de29..b61e9ba 100644 --- a/src/oracle/oracle-reading.ts +++ b/src/oracle/oracle-reading.ts @@ -0,0 +1,28 @@ +import { FreshnessRejectionReason } from "./freshnessPolicy"; + +export type Severity = "low" | "medium" | "high" | "triggered"; + +export interface OracleReading { + coverageType: string; + type: "oracle_update"; + value: number; + threshold: number; + severity: Severity; + message: string; + /** + * True when this reading came from the fail-safe fallback (upstream outage). + * OraclePublisherService must never publish degraded readings on-chain. + */ + degraded?: boolean; + /** + * Upstream observation time (ISO-8601 with explicit offset, or a Date). + * Consensus uses this; readings without it are rejected at ingestion. + */ + observedAt?: string | Date; + /** + * False when the freshness guard rejected this reading (issue #186). Such + * readings are persisted for audit but excluded from corroboration. + */ + usable?: boolean; + rejectionReason?: FreshnessRejectionReason; +} diff --git a/src/oracle/oracle-timestamp.ts b/src/oracle/oracle-timestamp.ts new file mode 100644 index 0000000..57480b9 --- /dev/null +++ b/src/oracle/oracle-timestamp.ts @@ -0,0 +1,134 @@ +import { createHash } from "crypto"; +import { performance } from "perf_hooks"; +import { OracleReading } from "./oracle-reading"; + +export type ObservationRejectionReason = + | "MISSING_OBSERVATION_TIME" + | "DATE_ONLY_OBSERVATION_TIME" + | "INVALID_OBSERVATION_TIME" + | "FUTURE_OBSERVATION_TIME" + | "STALE_OBSERVATION_TIME"; + +export type ObservationAdmission = + | { accepted: true; observedAt: Date } + | { accepted: false; reason: ObservationRejectionReason; observedAt?: Date }; + +export interface SourceClockOffset { + source: string; + samples: number; + /** receipt − observation, in ms. Positive: provider clock is behind. Negative: ahead. */ + medianOffsetMs: number; + skewed: boolean; +} + +export interface AdmissionOptions { + /** How far an observation time may lead receipt time before it is rejected. */ + futureToleranceMs: number; + /** How old an observation may be at receipt before it is rejected. */ + freshnessWindowMs: number; + /** |median offset| above which a source is reported as skewed. */ + skewThresholdMs: number; +} + +export const DEFAULT_ADMISSION_OPTIONS: AdmissionOptions = { + futureToleranceMs: 5_000, + freshnessWindowMs: 30 * 60_000, // matches STALENESS_LIMIT_SECONDS in claim.service.ts + skewThresholdMs: 10_000, +}; + +const MAX_OFFSET_SAMPLES = 50; +const DATE_ONLY = /^\d{4}-\d{2}-\d{2}$/; +// Requires an explicit UTC offset so a provider's local time is never silently +// interpreted as the backend's local time. +const ISO_WITH_OFFSET = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}(:\d{2}(\.\d{1,9})?)?(Z|[+-]\d{2}:?\d{2})$/; + +// Wall-clock anchor taken once at boot. performance.now() does not jump when the +// host clock is stepped (NTP correction, VM resume), so receipt times derived +// from it cannot mass-reject readings after a local clock change. +const WALL_CLOCK_ANCHOR_MS = Date.now() - performance.now(); + +export function monotonicNow(): Date { + return new Date(WALL_CLOCK_ANCHOR_MS + performance.now()); +} + +/** + * Parses an upstream observation time. Date-only strings are rejected explicitly + * rather than being defaulted to midnight or to receipt time. + */ +export function parseObservedAt(value: unknown): Date | ObservationRejectionReason { + if (value === undefined || value === null || value === "") return "MISSING_OBSERVATION_TIME"; + if (value instanceof Date) { + return Number.isNaN(value.getTime()) ? "INVALID_OBSERVATION_TIME" : value; + } + if (typeof value !== "string") return "INVALID_OBSERVATION_TIME"; + if (DATE_ONLY.test(value)) return "DATE_ONLY_OBSERVATION_TIME"; + if (!ISO_WITH_OFFSET.test(value)) return "INVALID_OBSERVATION_TIME"; + const parsed = new Date(value); + return Number.isNaN(parsed.getTime()) ? "INVALID_OBSERVATION_TIME" : parsed; +} + +/** SHA-256 over the reading payload. Observation time is excluded: it is part of the dedupe key. */ +export function readingDigest(reading: OracleReading): string { + const canonical = JSON.stringify([ + reading.coverageType, + reading.value, + reading.threshold, + reading.severity, + reading.message, + ]); + return createHash("sha256").update(canonical).digest("hex"); +} + +/** + * Shared ingestion gate: validates observation times against the backend's + * monotonic receipt clock and tracks per-source clock offset. + */ +export class ReadingAdmissionGate { + private readonly offsets = new Map(); + + constructor( + private readonly options: AdmissionOptions = DEFAULT_ADMISSION_OPTIONS, + private readonly receivedNow: () => Date = monotonicNow + ) {} + + admit(source: string, observedAt: unknown): ObservationAdmission { + const parsed = parseObservedAt(observedAt); + if (typeof parsed === "string") return { accepted: false, reason: parsed }; + + const receivedAt = this.receivedNow().getTime(); + this.recordOffset(source, receivedAt - parsed.getTime()); + + if (parsed.getTime() - receivedAt > this.options.futureToleranceMs) { + return { accepted: false, reason: "FUTURE_OBSERVATION_TIME", observedAt: parsed }; + } + if (receivedAt - parsed.getTime() > this.options.freshnessWindowMs) { + return { accepted: false, reason: "STALE_OBSERVATION_TIME", observedAt: parsed }; + } + return { accepted: true, observedAt: parsed }; + } + + clockOffsets(): SourceClockOffset[] { + return [...this.offsets.entries()].map(([source, samples]) => { + const medianOffsetMs = median(samples); + return { + source, + samples: samples.length, + medianOffsetMs, + skewed: Math.abs(medianOffsetMs) > this.options.skewThresholdMs, + }; + }); + } + + private recordOffset(source: string, offsetMs: number): void { + const samples = this.offsets.get(source) ?? []; + samples.push(offsetMs); + if (samples.length > MAX_OFFSET_SAMPLES) samples.shift(); + this.offsets.set(source, samples); + } +} + +function median(values: number[]): number { + const sorted = [...values].sort((a, b) => a - b); + const mid = Math.floor(sorted.length / 2); + return sorted.length % 2 === 0 ? (sorted[mid - 1] + sorted[mid]) / 2 : sorted[mid]; +}