From 84708a5d7fbf84dac443bd0cecbfda54b5a54bff Mon Sep 17 00:00:00 2001 From: only1dreamgene Date: Mon, 19 Oct 2026 15:46:17 -0700 Subject: [PATCH] Validate oracle observation timestamps and dedupe replayed readings - Add ReadingAdmissionGate: rejects missing, date-only, invalid, future-dated, and stale observation times with structured reason codes; receipt time comes from a monotonic clock so local clock jumps do not mass-reject readings. - Track per-source clock offset (receipt - observation) and expose it via OracleEventRepository.getClockOffsets() with a skew flag. - Persist observed_at and payload_digest; drop replays atomically via the oracle_reading_receipts primary key (source, observed_at, payload_digest). - Migration 004 adds the columns and receipts table. Co-Authored-By: Claude Haiku 5.5 --- .../004_oracle_reading_admission.sql | 24 ++++ src/oracle/oracle-event.repository.ts | 77 +++++++--- src/oracle/oracle-reading.ts | 5 + src/oracle/oracle-timestamp.ts | 134 ++++++++++++++++++ 4 files changed, 220 insertions(+), 20 deletions(-) create mode 100644 src/db/migrations/004_oracle_reading_admission.sql create mode 100644 src/oracle/oracle-timestamp.ts 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 51bafc9..accda90 100644 --- a/src/oracle/oracle-event.repository.ts +++ b/src/oracle/oracle-event.repository.ts @@ -2,6 +2,9 @@ import { Inject, Injectable, Logger } from "@nestjs/common"; import { Pool } from "pg"; import { PG_POOL } from "../db/db.module"; import { OracleReading } from "./oracle-reading"; +import { ReadingAdmissionGate, SourceClockOffset, readingDigest } from "./oracle-timestamp"; + +type InsertParams = [string, number, string, string, Date, string]; /** * Persists oracle readings to the `oracle_events` table defined in @@ -12,10 +15,16 @@ import { OracleReading } from "./oracle-reading"; * 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. @@ -27,6 +36,19 @@ export class OracleEventRepository { 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) {} /** @@ -35,18 +57,14 @@ export class OracleEventRepository { * oracle broadcast or claim-evaluation paths that call this. */ async record(reading: OracleReading): Promise { - const coverageType = OracleEventRepository.COVERAGE_TYPE_MAP[reading.coverageType]; - if (!coverageType) { - this.logger.warn(`Unknown coverageType "${reading.coverageType}" — skipping oracle_events insert`); - return; - } + const params = this.admitForInsert(reading); + if (!params) 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] - ); + 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}`); + } } catch (err) { this.logger.error( `Failed to persist oracle event for ${reading.coverageType}`, @@ -65,16 +83,9 @@ export class OracleEventRepository { 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}" — 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] - ); + const params = this.admitForInsert(reading); + if (!params) continue; + await client.query(OracleEventRepository.INSERT_SQL, params); } await client.query("COMMIT"); } catch (err) { @@ -88,6 +99,32 @@ export class OracleEventRepository { } } + /** 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". diff --git a/src/oracle/oracle-reading.ts b/src/oracle/oracle-reading.ts index 6dd9583..6a3de59 100644 --- a/src/oracle/oracle-reading.ts +++ b/src/oracle/oracle-reading.ts @@ -12,4 +12,9 @@ export interface OracleReading { * 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; } 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]; +}