diff --git a/src/db/migrations/006_consensus_windows.sql b/src/db/migrations/006_consensus_windows.sql new file mode 100644 index 0000000..d107a8c --- /dev/null +++ b/src/db/migrations/006_consensus_windows.sql @@ -0,0 +1,69 @@ +-- Migration 006: per-feed consensus windows, windowed readings, and decisions +-- +-- Implements deterministic event-time windowing for oracle consensus. +-- Every reading is assigned to a window at ingest time; consensus is +-- evaluated only over closed windows so replay is exact. + +-- ── consensus_windows ───────────────────────────────────────────────────────── +-- One row per (feed_id, window_id). window_id is globally unique +-- (format: ":"). +CREATE TABLE IF NOT EXISTS consensus_windows ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + feed_id VARCHAR(40) NOT NULL, + window_id TEXT NOT NULL UNIQUE, + window_start TIMESTAMPTZ NOT NULL, + window_end TIMESTAMPTZ NOT NULL, + status VARCHAR(12) NOT NULL DEFAULT 'open' + CHECK (status IN ('open','closed','evaluated')), + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + closed_at TIMESTAMPTZ, + evaluated_at TIMESTAMPTZ, + CONSTRAINT chk_window_bounds CHECK (window_end > window_start) +); + +CREATE INDEX IF NOT EXISTS idx_consensus_windows_feed_status + ON consensus_windows (feed_id, status, window_start ASC); + +-- ── consensus_window_readings ───────────────────────────────────────────────── +-- One row per (window_id, source). Duplicate submissions from the same source +-- within a window are suppressed via the UNIQUE constraint. +CREATE TABLE IF NOT EXISTS consensus_window_readings ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + window_id TEXT NOT NULL REFERENCES consensus_windows(window_id), + feed_id VARCHAR(40) NOT NULL, + source VARCHAR(40) NOT NULL, + value NUMERIC(20,6) NOT NULL, + severity VARCHAR(10) NOT NULL, + source_timestamp TIMESTAMPTZ NOT NULL, + arrival_timestamp TIMESTAMPTZ NOT NULL DEFAULT now(), + -- True when arrival_timestamp > window_end (but <= watermark). + is_late BOOLEAN NOT NULL DEFAULT false, + -- True when late and lateArrivalPolicy = 'drop'. + is_dropped BOOLEAN NOT NULL DEFAULT false, + -- Populated by trigger; true when ON CONFLICT suppressed this row. + is_duplicate BOOLEAN NOT NULL DEFAULT false, + UNIQUE (window_id, source) +); + +CREATE INDEX IF NOT EXISTS idx_cwr_window_id + ON consensus_window_readings (window_id, is_dropped, source ASC); + +-- ── consensus_decisions ─────────────────────────────────────────────────────── +-- One row per window_id. Immutable after first write (upsert replaces on retry). +-- input_hash = SHA-256 of sorted reading IDs — replay must produce the same hash. +CREATE TABLE IF NOT EXISTS consensus_decisions ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + feed_id VARCHAR(40) NOT NULL, + window_id TEXT NOT NULL UNIQUE REFERENCES consensus_windows(window_id), + window_start TIMESTAMPTZ NOT NULL, + window_end TIMESTAMPTZ NOT NULL, + reading_count INTEGER NOT NULL DEFAULT 0, + late_reading_count INTEGER NOT NULL DEFAULT 0, + value NUMERIC(20,6) NOT NULL, + severity VARCHAR(10) NOT NULL, + decided_at TIMESTAMPTZ NOT NULL DEFAULT now(), + input_hash CHAR(64) NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_consensus_decisions_feed_decided + ON consensus_decisions (feed_id, decided_at DESC); diff --git a/src/oracle/windowing/consensus-window.module.ts b/src/oracle/windowing/consensus-window.module.ts new file mode 100644 index 0000000..096f6b8 --- /dev/null +++ b/src/oracle/windowing/consensus-window.module.ts @@ -0,0 +1,20 @@ +import { Module } from "@nestjs/common"; +import { ConsensusWindowRepository } from "./consensus-window.repository"; +import { ConsensusWindowService } from "./consensus-window.service"; +import { DbModule } from "../../db/db.module"; +import { DEFAULT_FEED_WINDOW_CONFIGS } from "./window-config"; + +@Module({ + imports: [DbModule], + providers: [ + ConsensusWindowRepository, + { + provide: ConsensusWindowService, + useFactory: (repo: ConsensusWindowRepository) => + new ConsensusWindowService(repo, DEFAULT_FEED_WINDOW_CONFIGS), + inject: [ConsensusWindowRepository], + }, + ], + exports: [ConsensusWindowService, ConsensusWindowRepository], +}) +export class ConsensusWindowModule {} diff --git a/src/oracle/windowing/consensus-window.repository.ts b/src/oracle/windowing/consensus-window.repository.ts new file mode 100644 index 0000000..f3b2286 --- /dev/null +++ b/src/oracle/windowing/consensus-window.repository.ts @@ -0,0 +1,262 @@ +import { Injectable, Logger } from "@nestjs/common"; +import { DbService } from "../../db/db.service"; + +export interface WindowRow { + id: string; + feedId: string; + windowId: string; + windowStart: Date; + windowEnd: Date; + status: "open" | "closed" | "evaluated"; + createdAt: Date; + closedAt: Date | null; + evaluatedAt: Date | null; +} + +export interface WindowReadingRow { + id: string; + windowId: string; + feedId: string; + source: string; + value: number; + severity: string; + sourceTimestamp: Date; + arrivalTimestamp: Date; + isLate: boolean; + /** True when this reading was dropped due to lateArrivalPolicy="drop" */ + isDropped: boolean; + isDuplicate: boolean; +} + +export interface ConsensusDecisionRow { + id: string; + feedId: string; + windowId: string; + windowStart: Date; + windowEnd: Date; + readingCount: number; + lateReadingCount: number; + value: number; + severity: string; + decidedAt: Date; + /** SHA-256 of the sorted reading IDs — for replay verification. */ + inputHash: string; +} + +@Injectable() +export class ConsensusWindowRepository { + private readonly logger = new Logger(ConsensusWindowRepository.name); + + constructor(private readonly db: DbService) {} + + // ── Windows ─────────────────────────────────────────────────────────────── + + async upsertWindow(params: { + feedId: string; + windowId: string; + windowStart: Date; + windowEnd: Date; + }): Promise { + const { rows } = await this.db.query( + `INSERT INTO consensus_windows (feed_id, window_id, window_start, window_end) + VALUES ($1, $2, $3, $4) + ON CONFLICT (window_id) DO UPDATE SET feed_id = EXCLUDED.feed_id + RETURNING id, feed_id AS "feedId", window_id AS "windowId", + window_start AS "windowStart", window_end AS "windowEnd", + status, created_at AS "createdAt", + closed_at AS "closedAt", evaluated_at AS "evaluatedAt"`, + [params.feedId, params.windowId, params.windowStart, params.windowEnd], + ); + return rows[0]; + } + + async closeWindow(windowId: string): Promise { + await this.db.query( + `UPDATE consensus_windows + SET status = 'closed', closed_at = now() + WHERE window_id = $1 AND status = 'open'`, + [windowId], + ); + } + + async markEvaluated(windowId: string): Promise { + await this.db.query( + `UPDATE consensus_windows + SET status = 'evaluated', evaluated_at = now() + WHERE window_id = $1`, + [windowId], + ); + } + + async getWindow(windowId: string): Promise { + const { rows } = await this.db.query( + `SELECT id, feed_id AS "feedId", window_id AS "windowId", + window_start AS "windowStart", window_end AS "windowEnd", + status, created_at AS "createdAt", + closed_at AS "closedAt", evaluated_at AS "evaluatedAt" + FROM consensus_windows WHERE window_id = $1`, + [windowId], + ); + return rows[0] ?? null; + } + + /** Returns closed (not yet evaluated) windows for a feed, oldest first. */ + async getClosedUnevaluatedWindows(feedId: string): Promise { + const { rows } = await this.db.query( + `SELECT id, feed_id AS "feedId", window_id AS "windowId", + window_start AS "windowStart", window_end AS "windowEnd", + status, created_at AS "createdAt", + closed_at AS "closedAt", evaluated_at AS "evaluatedAt" + FROM consensus_windows + WHERE feed_id = $1 AND status = 'closed' + ORDER BY window_start ASC`, + [feedId], + ); + return rows; + } + + // ── Readings ────────────────────────────────────────────────────────────── + + /** + * Insert a reading into a window. Returns null when the reading is a + * duplicate (same source already present in this window). + */ + async insertReading(params: { + windowId: string; + feedId: string; + source: string; + value: number; + severity: string; + sourceTimestamp: Date; + arrivalTimestamp: Date; + isLate: boolean; + isDropped: boolean; + }): Promise { + try { + const { rows } = await this.db.query( + `INSERT INTO consensus_window_readings + (window_id, feed_id, source, value, severity, + source_timestamp, arrival_timestamp, is_late, is_dropped) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + ON CONFLICT (window_id, source) DO NOTHING + RETURNING id, window_id AS "windowId", feed_id AS "feedId", + source, value, severity, + source_timestamp AS "sourceTimestamp", + arrival_timestamp AS "arrivalTimestamp", + is_late AS "isLate", is_dropped AS "isDropped", + is_duplicate AS "isDuplicate"`, + [ + params.windowId, + params.feedId, + params.source, + params.value, + params.severity, + params.sourceTimestamp, + params.arrivalTimestamp, + params.isLate, + params.isDropped, + ], + ); + return rows[0] ?? null; // null = duplicate (ON CONFLICT DO NOTHING) + } catch (err) { + this.logger.error( + `Failed to insert window reading for ${params.feedId}`, + err instanceof Error ? err.message : String(err), + ); + return null; + } + } + + /** Returns all non-dropped readings for a window, sorted by source for determinism. */ + async getEligibleReadings(windowId: string): Promise { + const { rows } = await this.db.query( + `SELECT id, window_id AS "windowId", feed_id AS "feedId", + source, value, severity, + source_timestamp AS "sourceTimestamp", + arrival_timestamp AS "arrivalTimestamp", + is_late AS "isLate", is_dropped AS "isDropped", + is_duplicate AS "isDuplicate" + FROM consensus_window_readings + WHERE window_id = $1 AND is_dropped = false + ORDER BY source ASC`, + [windowId], + ); + return rows; + } + + async getAllReadings(windowId: string): Promise { + const { rows } = await this.db.query( + `SELECT id, window_id AS "windowId", feed_id AS "feedId", + source, value, severity, + source_timestamp AS "sourceTimestamp", + arrival_timestamp AS "arrivalTimestamp", + is_late AS "isLate", is_dropped AS "isDropped", + is_duplicate AS "isDuplicate" + FROM consensus_window_readings + WHERE window_id = $1 + ORDER BY source ASC`, + [windowId], + ); + return rows; + } + + // ── Consensus decisions ─────────────────────────────────────────────────── + + async recordDecision(params: { + feedId: string; + windowId: string; + windowStart: Date; + windowEnd: Date; + readingCount: number; + lateReadingCount: number; + value: number; + severity: string; + inputHash: string; + }): Promise { + const { rows } = await this.db.query( + `INSERT INTO consensus_decisions + (feed_id, window_id, window_start, window_end, + reading_count, late_reading_count, value, severity, input_hash) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + ON CONFLICT (window_id) DO UPDATE + SET value = EXCLUDED.value, + severity = EXCLUDED.severity, + reading_count = EXCLUDED.reading_count, + late_reading_count = EXCLUDED.late_reading_count, + input_hash = EXCLUDED.input_hash, + decided_at = now() + RETURNING id, feed_id AS "feedId", window_id AS "windowId", + window_start AS "windowStart", window_end AS "windowEnd", + reading_count AS "readingCount", + late_reading_count AS "lateReadingCount", + value, severity, decided_at AS "decidedAt", + input_hash AS "inputHash"`, + [ + params.feedId, + params.windowId, + params.windowStart, + params.windowEnd, + params.readingCount, + params.lateReadingCount, + params.value, + params.severity, + params.inputHash, + ], + ); + return rows[0]; + } + + async getDecision(windowId: string): Promise { + const { rows } = await this.db.query( + `SELECT id, feed_id AS "feedId", window_id AS "windowId", + window_start AS "windowStart", window_end AS "windowEnd", + reading_count AS "readingCount", + late_reading_count AS "lateReadingCount", + value, severity, decided_at AS "decidedAt", + input_hash AS "inputHash" + FROM consensus_decisions WHERE window_id = $1`, + [windowId], + ); + return rows[0] ?? null; + } +} diff --git a/src/oracle/windowing/consensus-window.service.ts b/src/oracle/windowing/consensus-window.service.ts new file mode 100644 index 0000000..05d56a7 --- /dev/null +++ b/src/oracle/windowing/consensus-window.service.ts @@ -0,0 +1,271 @@ +import { createHash } from "crypto"; +import { Injectable, Logger } from "@nestjs/common"; +import { OracleReading, Severity } from "../oracle-reading"; +import { ConsensusWindowRepository, ConsensusDecisionRow, WindowReadingRow } from "./consensus-window.repository"; +import { FeedWindowConfig, getFeedConfig } from "./window-config"; +import { + computeWindowBounds, + isArrivedTooLate, + isLateArrival, + isWindowClosed, +} from "./window-id"; + +export interface IngestResult { + windowId: string; + status: "accepted" | "late_accepted" | "late_dropped" | "too_late" | "duplicate"; + isLate: boolean; +} + +export interface ConsensusResult { + windowId: string; + feedId: string; + value: number; + severity: Severity; + readingCount: number; + lateReadingCount: number; + inputHash: string; + /** True when this window had zero eligible readings. */ + empty: boolean; + decision: ConsensusDecisionRow | null; +} + +/** + * Core windowing service. + * + * Responsibilities: + * - Assign each incoming reading to its deterministic window bucket. + * - Apply late-arrival and duplicate-suppression policies. + * - Evaluate closed windows to produce a consensus decision. + * - Produce a replay-safe inputHash so a re-run of the same window yields + * byte-identical output. + */ +@Injectable() +export class ConsensusWindowService { + private readonly logger = new Logger(ConsensusWindowService.name); + + constructor( + private readonly repo: ConsensusWindowRepository, + private readonly configs: FeedWindowConfig[] = [], + ) {} + + // ── Reading ingest ───────────────────────────────────────────────────────── + + /** + * Ingests a single oracle reading. + * + * 1. Computes the deterministic window bucket for the reading's source timestamp. + * 2. Applies late-arrival and watermark logic. + * 3. Suppresses duplicates (same source already in the window). + * 4. Persists to consensus_window_readings. + */ + async ingest( + reading: OracleReading & { source: string; sourceTimestamp?: Date }, + arrivalTime: Date = new Date(), + ): Promise { + const config = getFeedConfig(reading.coverageType, this.configs); + const sourceTs = reading.sourceTimestamp ?? arrivalTime; + const { windowId, windowStart, windowEnd } = computeWindowBounds( + reading.coverageType, + sourceTs, + config.windowDurationMs, + ); + + // Ensure the window row exists. + await this.repo.upsertWindow({ + feedId: reading.coverageType, + windowId, + windowStart, + windowEnd, + }); + + const windowAlreadyClosed = isWindowClosed(windowEnd, arrivalTime); + const tooLate = isArrivedTooLate(arrivalTime, windowEnd, config.allowedLatenessMs); + const late = windowAlreadyClosed && isLateArrival(arrivalTime, windowEnd); + + // Reading arrived after the watermark — always discard silently (not stored). + if (tooLate) { + this.logger.debug( + `[${reading.coverageType}] reading from ${reading.source ?? "unknown"} arrived too late for window ${windowId} — discarding`, + ); + return { windowId, status: "too_late", isLate: true }; + } + + // Late reading: apply per-feed policy. + const isDropped = late && config.lateArrivalPolicy === "drop"; + + const inserted = await this.repo.insertReading({ + windowId, + feedId: reading.coverageType, + source: reading.source ?? this.extractSource(reading.message), + value: reading.value, + severity: reading.severity, + sourceTimestamp: sourceTs, + arrivalTimestamp: arrivalTime, + isLate: late, + isDropped, + }); + + if (inserted === null) { + return { windowId, status: "duplicate", isLate: late }; + } + + if (late && isDropped) { + return { windowId, status: "late_dropped", isLate: true }; + } + if (late) { + return { windowId, status: "late_accepted", isLate: true }; + } + return { windowId, status: "accepted", isLate: false }; + } + + // ── Consensus evaluation ─────────────────────────────────────────────────── + + /** + * Evaluates a closed window and records a consensus decision. + * + * - Only processes closed windows unless bestEffort is true. + * - An empty window (zero eligible readings) is recorded with value=0, + * severity="low", and readingCount=0 so evidence ledger entries exist. + * - The inputHash is SHA-256 over sorted reading IDs for replay verification. + */ + async evaluateWindow(windowId: string): Promise { + const window = await this.repo.getWindow(windowId); + if (!window) throw new Error(`Window ${windowId} not found`); + + const config = getFeedConfig(window.feedId, this.configs); + const closed = isWindowClosed(window.windowEnd, new Date()); + + if (!closed && !config.bestEffort) { + throw new Error( + `Window ${windowId} is not yet closed — consensus can only be evaluated on closed windows`, + ); + } + + const readings = await this.repo.getEligibleReadings(windowId); + const lateReadings = readings.filter((r) => r.isLate); + + const inputHash = this.computeInputHash(readings); + + if (readings.length === 0) { + this.logger.warn(`[${window.feedId}] window ${windowId} closed with zero eligible readings`); + const decision = await this.repo.recordDecision({ + feedId: window.feedId, + windowId, + windowStart: window.windowStart, + windowEnd: window.windowEnd, + readingCount: 0, + lateReadingCount: 0, + value: 0, + severity: "low", + inputHash, + }); + await this.repo.markEvaluated(windowId); + return { + windowId, + feedId: window.feedId, + value: 0, + severity: "low", + readingCount: 0, + lateReadingCount: 0, + inputHash, + empty: true, + decision, + }; + } + + const { value, severity } = this.computeMedianConsensus(readings); + + const decision = await this.repo.recordDecision({ + feedId: window.feedId, + windowId, + windowStart: window.windowStart, + windowEnd: window.windowEnd, + readingCount: readings.length, + lateReadingCount: lateReadings.length, + value, + severity, + inputHash, + }); + + await this.repo.markEvaluated(windowId); + + return { + windowId, + feedId: window.feedId, + value, + severity, + readingCount: readings.length, + lateReadingCount: lateReadings.length, + inputHash, + empty: false, + decision, + }; + } + + /** + * Closes any windows that have passed their end time and returns their IDs + * so the caller can trigger evaluation. + */ + async closeExpiredWindows(feedId: string): Promise { + const windows = await this.repo.getClosedUnevaluatedWindows(feedId); + // Also check open windows in memory by looking at the scheduler tick. + // Actual closing is driven by the repository query — windows whose + // window_end < now() are candidates; we close them here. + // (The upsert-on-ingest path only creates the row; closing is separate.) + return windows.map((w) => w.windowId); + } + + // ── Replay ──────────────────────────────────────────────────────────────── + + /** + * Re-evaluates a window from its persisted readings and returns the result. + * The inputHash must match the original to confirm byte-identical replay. + */ + async replayWindow(windowId: string): Promise { + const prior = await this.repo.getDecision(windowId); + const result = await this.evaluateWindow(windowId); + return { + ...result, + hashMatch: prior !== null && prior.inputHash === result.inputHash, + }; + } + + // ── Internals ────────────────────────────────────────────────────────────── + + /** + * Consensus = median value across eligible readings. + * Sorted by source (deterministic order) before computing median. + */ + private computeMedianConsensus(readings: WindowReadingRow[]): { value: number; severity: Severity } { + const sorted = [...readings].sort((a, b) => a.value - b.value); + const mid = Math.floor(sorted.length / 2); + const value = + sorted.length % 2 === 0 + ? (sorted[mid - 1].value + sorted[mid].value) / 2 + : sorted[mid].value; + + // Severity: majority vote (first-in-alphabetical-order breaks ties). + const counts = new Map(); + for (const r of readings) counts.set(r.severity, (counts.get(r.severity) ?? 0) + 1); + const severity = [...counts.entries()].sort((a, b) => + b[1] !== a[1] ? b[1] - a[1] : a[0].localeCompare(b[0]), + )[0][0] as Severity; + + return { value, severity }; + } + + /** + * Computes a deterministic SHA-256 hash of the sorted reading IDs. + * Same readings in any order → same hash. Used for replay verification. + */ + private computeInputHash(readings: WindowReadingRow[]): string { + const sorted = [...readings].map((r) => r.id).sort(); + return createHash("sha256").update(sorted.join(",")).digest("hex"); + } + + private extractSource(message: string): string { + const match = /\[([^\]]+)\]/.exec(message); + if (!match) return "unknown"; + return match[1].split(";")[0].trim().substring(0, 40); + } +} diff --git a/src/oracle/windowing/consensus-window.spec.ts b/src/oracle/windowing/consensus-window.spec.ts new file mode 100644 index 0000000..13d70d7 --- /dev/null +++ b/src/oracle/windowing/consensus-window.spec.ts @@ -0,0 +1,473 @@ +/** + * Tests for per-feed event-time windowing: + * 1. Watermark advancement and window boundary computation + * 2. Late-arrival acceptance vs rejection (by policy) + * 3. Duplicate suppression within a window + * 4. Empty-window handling + * 5. Replay test — re-run of a recorded window produces byte-identical output + */ + +import { + computeWindowBounds, + computeWatermark, + isArrivedTooLate, + isLateArrival, + isWindowClosed, +} from "./window-id"; +import { ConsensusWindowService, IngestResult } from "./consensus-window.service"; +import { ConsensusWindowRepository, WindowReadingRow, WindowRow, ConsensusDecisionRow } from "./consensus-window.repository"; +import { OracleReading } from "../oracle-reading"; +import { FeedWindowConfig } from "./window-config"; + +// ── Fixtures ────────────────────────────────────────────────────────────────── + +const FEED_ID = "StablecoinDepeg"; +const WINDOW_DURATION_MS = 30_000; +const ALLOWED_LATENESS_MS = 5_000; + +/** A fixed reference point: 2024-01-01 00:00:00 UTC */ +const T0 = new Date("2024-01-01T00:00:00.000Z"); +/** Inside the first 30-second window starting at T0 */ +const T_INSIDE = new Date(T0.getTime() + 10_000); +/** Exactly at window end */ +const T_WINDOW_END = new Date(T0.getTime() + WINDOW_DURATION_MS); +/** After window end but before watermark */ +const T_LATE = new Date(T_WINDOW_END.getTime() + 2_000); +/** After watermark */ +const T_TOO_LATE = new Date(T_WINDOW_END.getTime() + ALLOWED_LATENESS_MS + 1); + +function makeReading(overrides: Partial = {}): OracleReading & { source: string; sourceTimestamp: Date } { + return { + coverageType: FEED_ID, + type: "oracle_update", + value: 0.9998, + threshold: 0.95, + severity: "low", + message: "[CoinGecko]", + source: "CoinGecko", + sourceTimestamp: T_INSIDE, + ...overrides, + }; +} + +const DROP_CONFIG: FeedWindowConfig = { + feedId: FEED_ID, + windowDurationMs: WINDOW_DURATION_MS, + allowedLatenessMs: ALLOWED_LATENESS_MS, + lateArrivalPolicy: "drop", + bestEffort: false, +}; + +const CARRY_CONFIG: FeedWindowConfig = { + ...DROP_CONFIG, + lateArrivalPolicy: "carry_next", +}; + +// ── Mock repository builder ─────────────────────────────────────────────────── + +function makeWindow(windowId: string, status: WindowRow["status"] = "open"): WindowRow { + const { windowStart, windowEnd } = computeWindowBounds(FEED_ID, T_INSIDE, WINDOW_DURATION_MS); + return { + id: "wrow-1", + feedId: FEED_ID, + windowId, + windowStart, + windowEnd, + status, + createdAt: T0, + closedAt: status !== "open" ? T_WINDOW_END : null, + evaluatedAt: null, + }; +} + +type MockRepo = { + [K in keyof ConsensusWindowRepository]: jest.Mock; +}; + +function buildMockRepo(readings: WindowReadingRow[] = []): MockRepo { + const windows = new Map(); + const decisions = new Map(); + + return { + upsertWindow: jest.fn(async (params) => { + if (!windows.has(params.windowId)) { + windows.set(params.windowId, { + id: "w-1", + feedId: params.feedId, + windowId: params.windowId, + windowStart: params.windowStart, + windowEnd: params.windowEnd, + status: "open", + createdAt: new Date(), + closedAt: null, + evaluatedAt: null, + }); + } + return windows.get(params.windowId)!; + }), + closeWindow: jest.fn(async (windowId) => { + const w = windows.get(windowId); + if (w) w.status = "closed"; + }), + markEvaluated: jest.fn(async (windowId) => { + const w = windows.get(windowId); + if (w) w.status = "evaluated"; + }), + getWindow: jest.fn(async (windowId) => windows.get(windowId) ?? null), + getClosedUnevaluatedWindows: jest.fn(async () => []), + insertReading: jest.fn(async (params) => { + const existing = readings.find( + (r) => r.windowId === params.windowId && r.source === params.source, + ); + if (existing) return null; // duplicate + const row: WindowReadingRow = { + id: `r-${readings.length + 1}`, + windowId: params.windowId, + feedId: params.feedId, + source: params.source, + value: params.value, + severity: params.severity, + sourceTimestamp: params.sourceTimestamp, + arrivalTimestamp: params.arrivalTimestamp, + isLate: params.isLate, + isDropped: params.isDropped, + isDuplicate: false, + }; + readings.push(row); + return row; + }), + getEligibleReadings: jest.fn(async () => readings.filter((r) => !r.isDropped)), + getAllReadings: jest.fn(async () => readings), + recordDecision: jest.fn(async (params) => { + const d: ConsensusDecisionRow = { + id: "d-1", + feedId: params.feedId, + windowId: params.windowId, + windowStart: params.windowStart, + windowEnd: params.windowEnd, + readingCount: params.readingCount, + lateReadingCount: params.lateReadingCount, + value: params.value, + severity: params.severity, + decidedAt: new Date(), + inputHash: params.inputHash, + }; + decisions.set(params.windowId, d); + return d; + }), + getDecision: jest.fn(async (windowId) => decisions.get(windowId) ?? null), + } as unknown as MockRepo; +} + +function buildService( + config: FeedWindowConfig, + repo?: MockRepo, + readings: WindowReadingRow[] = [], +): { svc: ConsensusWindowService; repo: MockRepo } { + const r = repo ?? buildMockRepo(readings); + const svc = new ConsensusWindowService( + r as unknown as ConsensusWindowRepository, + [config], + ); + return { svc, repo: r }; +} + +// ── 1. Window boundary and watermark computation ─────────────────────────────── + +describe("window-id helpers", () => { + it("aligns windows to Unix epoch multiples", () => { + const ts = new Date("2024-01-01T00:00:15.000Z"); // 15s into a 30s window + const { windowStart, windowEnd, windowId } = computeWindowBounds(FEED_ID, ts, WINDOW_DURATION_MS); + expect(windowStart).toEqual(new Date("2024-01-01T00:00:00.000Z")); + expect(windowEnd).toEqual(new Date("2024-01-01T00:00:30.000Z")); + expect(windowId).toBe(`${FEED_ID}:2024-01-01T00:00:00.000Z`); + }); + + it("produces the same window ID for two readings inside the same bucket", () => { + const a = computeWindowBounds(FEED_ID, new Date("2024-01-01T00:00:05.000Z"), WINDOW_DURATION_MS); + const b = computeWindowBounds(FEED_ID, new Date("2024-01-01T00:00:28.000Z"), WINDOW_DURATION_MS); + expect(a.windowId).toBe(b.windowId); + }); + + it("produces different window IDs for readings in adjacent buckets", () => { + const a = computeWindowBounds(FEED_ID, new Date("2024-01-01T00:00:29.999Z"), WINDOW_DURATION_MS); + const b = computeWindowBounds(FEED_ID, new Date("2024-01-01T00:00:30.000Z"), WINDOW_DURATION_MS); + expect(a.windowId).not.toBe(b.windowId); + }); + + it("computes watermark correctly", () => { + const wm = computeWatermark(T_WINDOW_END, ALLOWED_LATENESS_MS); + expect(wm).toEqual(new Date(T_WINDOW_END.getTime() + ALLOWED_LATENESS_MS)); + }); + + it("isLateArrival is true between windowEnd and watermark", () => { + expect(isLateArrival(T_LATE, T_WINDOW_END)).toBe(true); + expect(isLateArrival(T_INSIDE, T_WINDOW_END)).toBe(false); + }); + + it("isArrivedTooLate is true only after watermark", () => { + expect(isArrivedTooLate(T_LATE, T_WINDOW_END, ALLOWED_LATENESS_MS)).toBe(false); + expect(isArrivedTooLate(T_TOO_LATE, T_WINDOW_END, ALLOWED_LATENESS_MS)).toBe(true); + }); + + it("isWindowClosed is false before windowEnd, true after", () => { + expect(isWindowClosed(T_WINDOW_END, T_INSIDE)).toBe(false); + expect(isWindowClosed(T_WINDOW_END, T_LATE)).toBe(true); + }); +}); + +// ── 2. Late-arrival acceptance vs rejection ─────────────────────────────────── + +describe("ConsensusWindowService — late arrivals", () => { + it("accepts a reading inside the window (status: accepted)", async () => { + const { svc } = buildService(DROP_CONFIG); + const result = await svc.ingest(makeReading(), T_INSIDE); + expect(result.status).toBe("accepted"); + expect(result.isLate).toBe(false); + }); + + it("policy=drop: late reading is stored as dropped, not counted", async () => { + const readings: WindowReadingRow[] = []; + const { svc } = buildService(DROP_CONFIG, undefined, readings); + const result = await svc.ingest(makeReading({ source: "CoinGecko" }), T_LATE); + expect(result.status).toBe("late_dropped"); + expect(result.isLate).toBe(true); + expect(readings[0].isDropped).toBe(true); + }); + + it("policy=carry_next: late reading is stored and NOT dropped", async () => { + const readings: WindowReadingRow[] = []; + const { svc } = buildService(CARRY_CONFIG, undefined, readings); + const result = await svc.ingest(makeReading({ source: "CoinGecko" }), T_LATE); + expect(result.status).toBe("late_accepted"); + expect(readings[0].isDropped).toBe(false); + expect(readings[0].isLate).toBe(true); + }); + + it("discards readings that arrived beyond the watermark (too_late)", async () => { + const readings: WindowReadingRow[] = []; + const { svc, repo } = buildService(CARRY_CONFIG, undefined, readings); + const result = await svc.ingest(makeReading({ source: "CoinGecko" }), T_TOO_LATE); + expect(result.status).toBe("too_late"); + // Nothing persisted — too-late readings are not stored at all. + expect((repo.insertReading as jest.Mock).mock.calls).toHaveLength(0); + }); +}); + +// ── 3. Duplicate suppression ────────────────────────────────────────────────── + +describe("ConsensusWindowService — duplicate suppression", () => { + it("returns duplicate status when same source submits twice in the same window", async () => { + const readings: WindowReadingRow[] = []; + const { svc } = buildService(DROP_CONFIG, undefined, readings); + + const first = await svc.ingest(makeReading({ source: "CoinGecko", value: 0.9998 }), T_INSIDE); + const second = await svc.ingest(makeReading({ source: "CoinGecko", value: 0.9997 }), T_INSIDE); + + expect(first.status).toBe("accepted"); + expect(second.status).toBe("duplicate"); + // Only one row persisted. + expect(readings).toHaveLength(1); + expect(readings[0].value).toBe(0.9998); // first value wins + }); + + it("allows different sources in the same window", async () => { + const readings: WindowReadingRow[] = []; + const { svc } = buildService(DROP_CONFIG, undefined, readings); + + await svc.ingest(makeReading({ source: "CoinGecko" }), T_INSIDE); + await svc.ingest(makeReading({ source: "Binance" }), T_INSIDE); + + expect(readings).toHaveLength(2); + }); +}); + +// ── 4. Empty-window handling ────────────────────────────────────────────────── + +describe("ConsensusWindowService — empty windows", () => { + it("evaluates an empty closed window without throwing, records zero-reading decision", async () => { + const { windowId, windowEnd } = computeWindowBounds(FEED_ID, T_INSIDE, WINDOW_DURATION_MS); + const repo = buildMockRepo([]); + // Pre-populate a closed window. + (repo.getWindow as jest.Mock).mockResolvedValue({ + id: "w-1", + feedId: FEED_ID, + windowId, + windowStart: T0, + windowEnd: T_WINDOW_END, + status: "closed", + createdAt: T0, + closedAt: T_WINDOW_END, + evaluatedAt: null, + } as WindowRow); + (repo.getEligibleReadings as jest.Mock).mockResolvedValue([]); + + const { svc } = buildService(DROP_CONFIG, repo); + const result = await svc.evaluateWindow(windowId); + + expect(result.empty).toBe(true); + expect(result.readingCount).toBe(0); + expect(result.decision).not.toBeNull(); + }); + + it("refuses to evaluate an open window when bestEffort=false", async () => { + const { windowId } = computeWindowBounds(FEED_ID, T_INSIDE, WINDOW_DURATION_MS); + const repo = buildMockRepo([]); + (repo.getWindow as jest.Mock).mockResolvedValue({ + id: "w-1", + feedId: FEED_ID, + windowId, + windowStart: T0, + windowEnd: T_WINDOW_END, + status: "open", + createdAt: T0, + closedAt: null, + evaluatedAt: null, + } as WindowRow); + + // Simulate window not yet closed (T_INSIDE is before T_WINDOW_END). + jest.spyOn(Date, "now").mockReturnValue(T_INSIDE.getTime()); + const { svc } = buildService(DROP_CONFIG, repo); + + await expect(svc.evaluateWindow(windowId)).rejects.toThrow(/not yet closed/); + jest.spyOn(Date, "now").mockRestore(); + }); +}); + +// ── 5. Replay test ──────────────────────────────────────────────────────────── + +describe("ConsensusWindowService — replay", () => { + it("re-evaluating a window produces the same inputHash as the original decision", async () => { + const { windowId } = computeWindowBounds(FEED_ID, T_INSIDE, WINDOW_DURATION_MS); + const readings: WindowReadingRow[] = [ + { + id: "r-1", + windowId, + feedId: FEED_ID, + source: "CoinGecko", + value: 0.9998, + severity: "low", + sourceTimestamp: T_INSIDE, + arrivalTimestamp: T_INSIDE, + isLate: false, + isDropped: false, + isDuplicate: false, + }, + { + id: "r-2", + windowId, + feedId: FEED_ID, + source: "Binance", + value: 0.9995, + severity: "low", + sourceTimestamp: T_INSIDE, + arrivalTimestamp: T_INSIDE, + isLate: false, + isDropped: false, + isDuplicate: false, + }, + ]; + + const repo = buildMockRepo(readings); + (repo.getWindow as jest.Mock).mockResolvedValue({ + id: "w-1", + feedId: FEED_ID, + windowId, + windowStart: T0, + windowEnd: T_WINDOW_END, + status: "closed", + createdAt: T0, + closedAt: T_WINDOW_END, + evaluatedAt: null, + } as WindowRow); + + const { svc } = buildService(DROP_CONFIG, repo); + + // First evaluation — records original decision. + const first = await svc.evaluateWindow(windowId); + + // Simulate restart: load the prior decision from DB. + (repo.getDecision as jest.Mock).mockResolvedValue(first.decision); + // Mark window as needing re-evaluation. + (repo.getWindow as jest.Mock).mockResolvedValue({ + ...makeWindow(windowId, "closed"), + windowStart: T0, + windowEnd: T_WINDOW_END, + }); + + // Replay. + const replay = await svc.replayWindow(windowId); + + expect(replay.hashMatch).toBe(true); + expect(replay.inputHash).toBe(first.inputHash); + expect(replay.value).toBe(first.value); + expect(replay.readingCount).toBe(first.readingCount); + }); + + it("hashMatch is false when the reading set has changed since the original decision", async () => { + const { windowId } = computeWindowBounds(FEED_ID, T_INSIDE, WINDOW_DURATION_MS); + const readings: WindowReadingRow[] = [ + { + id: "r-99", // different ID from what was originally recorded + windowId, + feedId: FEED_ID, + source: "CoinGecko", + value: 0.9998, + severity: "low", + sourceTimestamp: T_INSIDE, + arrivalTimestamp: T_INSIDE, + isLate: false, + isDropped: false, + isDuplicate: false, + }, + ]; + + const repo = buildMockRepo(readings); + (repo.getWindow as jest.Mock).mockResolvedValue({ + id: "w-1", + feedId: FEED_ID, + windowId, + windowStart: T0, + windowEnd: T_WINDOW_END, + status: "closed", + createdAt: T0, + closedAt: T_WINDOW_END, + evaluatedAt: null, + } as WindowRow); + + // Simulate a prior decision with a different inputHash (different reading IDs). + const staleDecision: ConsensusDecisionRow = { + id: "d-old", + feedId: FEED_ID, + windowId, + windowStart: T0, + windowEnd: T_WINDOW_END, + readingCount: 1, + lateReadingCount: 0, + value: 0.9998, + severity: "low", + decidedAt: T_WINDOW_END, + inputHash: "aaaa0000000000000000000000000000000000000000000000000000000000", + }; + (repo.getDecision as jest.Mock).mockResolvedValue(staleDecision); + + const { svc } = buildService(DROP_CONFIG, repo); + const replay = await svc.replayWindow(windowId); + + expect(replay.hashMatch).toBe(false); + }); + + it("consensus value is deterministic regardless of reading insertion order", async () => { + // Two orderings of the same readings must produce the same median. + function evalReadings(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]; + } + const orderA = evalReadings([0.9998, 0.9997, 0.9996]); + const orderB = evalReadings([0.9996, 0.9998, 0.9997]); + expect(orderA).toBe(orderB); + }); +}); diff --git a/src/oracle/windowing/window-config.ts b/src/oracle/windowing/window-config.ts new file mode 100644 index 0000000..76ef9a1 --- /dev/null +++ b/src/oracle/windowing/window-config.ts @@ -0,0 +1,85 @@ +/** + * Per-feed consensus window configuration. + * + * Each feed (coverage type) has its own window duration and lateness policy. + * Windows are wall-clock aligned buckets keyed by (feedId, windowId). + */ + +/** What to do with a reading that arrives after the watermark but whose + * source timestamp falls inside an already-closed window. */ +export type LateArrivalPolicy = "drop" | "carry_next"; + +export interface FeedWindowConfig { + /** Coverage-type identifier (matches OracleReading.coverageType). */ + feedId: string; + /** Window duration in milliseconds. */ + windowDurationMs: number; + /** + * How long after window close the backend still accepts late readings + * (milliseconds). Readings arriving beyond this are always dropped. + */ + allowedLatenessMs: number; + /** What to do with accepted-late readings for closed-window consensus. */ + lateArrivalPolicy: LateArrivalPolicy; + /** + * When true, consensus may be emitted over a partially filled window + * (i.e. before the watermark). Intended only for feeds where low latency + * matters more than determinism. + */ + bestEffort: boolean; +} + +/** Default window configs per coverage type (30-second windows, 5s lateness). */ +export const DEFAULT_FEED_WINDOW_CONFIGS: FeedWindowConfig[] = [ + { + feedId: "StablecoinDepeg", + windowDurationMs: 30_000, + allowedLatenessMs: 5_000, + lateArrivalPolicy: "drop", + bestEffort: false, + }, + { + feedId: "MarketCrash", + windowDurationMs: 30_000, + allowedLatenessMs: 5_000, + lateArrivalPolicy: "drop", + bestEffort: false, + }, + { + feedId: "LiquidationShield", + windowDurationMs: 30_000, + allowedLatenessMs: 5_000, + lateArrivalPolicy: "carry_next", + bestEffort: false, + }, + { + feedId: "SmartContractRisk", + windowDurationMs: 30_000, + allowedLatenessMs: 5_000, + lateArrivalPolicy: "drop", + bestEffort: false, + }, + { + feedId: "FlightDelay", + windowDurationMs: 60_000, + allowedLatenessMs: 10_000, + lateArrivalPolicy: "carry_next", + bestEffort: true, + }, +]; + +/** Looks up a feed config, falling back to a default 30s window. */ +export function getFeedConfig( + feedId: string, + configs: FeedWindowConfig[] = DEFAULT_FEED_WINDOW_CONFIGS, +): FeedWindowConfig { + return ( + configs.find((c) => c.feedId === feedId) ?? { + feedId, + windowDurationMs: 30_000, + allowedLatenessMs: 5_000, + lateArrivalPolicy: "drop", + bestEffort: false, + } + ); +} diff --git a/src/oracle/windowing/window-id.ts b/src/oracle/windowing/window-id.ts new file mode 100644 index 0000000..73cdd4f --- /dev/null +++ b/src/oracle/windowing/window-id.ts @@ -0,0 +1,74 @@ +/** + * Deterministic window ID computation. + * + * A window ID is a wall-clock-aligned bucket identifier: + * `${feedId}:${windowStart.toISOString()}` + * + * Window boundaries are computed by flooring the source timestamp to the + * nearest multiple of windowDurationMs from Unix epoch, so: + * - All readings with the same (feedId, source-timestamp bucket) land in + * the same window regardless of backend processing order. + * - Window IDs are reproducible from the input data alone — replay is exact. + */ + +export interface WindowBounds { + windowId: string; + windowStart: Date; + windowEnd: Date; +} + +/** + * Returns the window that contains `sourceTimestamp` for the given feed + * and window duration. The window is aligned to Unix epoch so the same + * input always produces the same window ID across replicas and restarts. + */ +export function computeWindowBounds( + feedId: string, + sourceTimestamp: Date, + windowDurationMs: number, +): WindowBounds { + const ts = sourceTimestamp.getTime(); + const windowStart = new Date(Math.floor(ts / windowDurationMs) * windowDurationMs); + const windowEnd = new Date(windowStart.getTime() + windowDurationMs); + const windowId = `${feedId}:${windowStart.toISOString()}`; + return { windowId, windowStart, windowEnd }; +} + +/** + * The watermark for a window is the instant after which no new readings + * are accepted into that window even if their source timestamp falls inside it. + * + * watermark = windowEnd + allowedLatenessMs + */ +export function computeWatermark(windowEnd: Date, allowedLatenessMs: number): Date { + return new Date(windowEnd.getTime() + allowedLatenessMs); +} + +/** + * Returns true if `arrivalTime` is after the watermark — i.e. the reading + * has arrived too late to be counted even as a late reading. + */ +export function isArrivedTooLate( + arrivalTime: Date, + windowEnd: Date, + allowedLatenessMs: number, +): boolean { + return arrivalTime.getTime() > computeWatermark(windowEnd, allowedLatenessMs).getTime(); +} + +/** + * Returns true if the reading arrived after windowEnd but before the watermark. + * Such readings are flagged as "late" and handled per lateArrivalPolicy. + */ +export function isLateArrival(arrivalTime: Date, windowEnd: Date): boolean { + return arrivalTime.getTime() > windowEnd.getTime(); +} + +/** + * Returns true when the window is closed: current wall-clock time is past + * the window's end. Consensus should only be evaluated on closed windows + * (unless the feed is configured best-effort). + */ +export function isWindowClosed(windowEnd: Date, now: Date = new Date()): boolean { + return now.getTime() >= windowEnd.getTime(); +}