Fix: Harden Oracle Reading Timestamps Against Clock Skew, Future-Dating, and Replay

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);

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<string, string> = {
+    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<void> {
+    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<void> {
+    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);
+  }
+}

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;
+}

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<string, number[]>();
+
+  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];
+}