diff --git a/Dockerfile b/Dockerfile index ad70306..0126147 100644 --- a/Dockerfile +++ b/Dockerfile @@ -32,8 +32,8 @@ RUN mkdir -p /var/lib/refract/oracle-evidence && chown node:node /var/lib/refrac USER node EXPOSE 4001 -HEALTHCHECK --interval=30s --timeout=3s --start-period=10s --retries=3 \ - CMD node -e "fetch('http://127.0.0.1:4001/health').then(r=>process.exit(r.ok?0:1)).catch(()=>process.exit(1))" +HEALTHCHECK --interval=30s --timeout=10s --start-period=10s --retries=3 \ + CMD node -e "fetch('http://127.0.0.1:4001/health/ready').then(r=>process.exit(r.ok?0:1)).catch(()=>process.exit(1))" # Exec form so node is PID 1 and receives SIGTERM directly. Give it more time # than SHUTDOWN_GRACE_PERIOD_MS before SIGKILL (docker-compose sets diff --git a/docker-compose.yml b/docker-compose.yml index 156bf2a..b302ce9 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -98,9 +98,9 @@ services: redis: condition: service_healthy healthcheck: - test: ["CMD", "node", "-e", "fetch('http://127.0.0.1:4001/health').then(r=>process.exit(r.ok?0:1)).catch(()=>process.exit(1))"] + test: ["CMD", "node", "-e", "fetch('http://127.0.0.1:4001/health/ready').then(r=>process.exit(r.ok?0:1)).catch(()=>process.exit(1))"] interval: 30s - timeout: 3s + timeout: 10s retries: 3 start_period: 10s diff --git a/package.json b/package.json index bdf2ca9..a4b67f4 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "refract-protocol-backend", "version": "0.1.0", - "description": "Refract Protocol backend \u2014 oracle monitoring, claim processing, premium quoting", + "description": "Refract Protocol backend — oracle monitoring, claim processing, premium quoting", "main": "dist/main.js", "private": true, "workspaces": [ @@ -127,7 +127,7 @@ ] }, "transformIgnorePatterns": [ - "^(?!.*/node_modules/(@noble|uint8array-extras)/).*/node_modules/" + "/node_modules/(?!(?:@stellar/stellar-sdk|@noble/.*|@scure/.*|uint8array-extras|hexoid|bignumber.js|@stellar|@noble|@scure|stellar-base|crc|tweetnacl|base32.js|@adraffy|@noble/curves))" ], "collectCoverageFrom": [ "**/*.(t|j)s" diff --git a/src/auth/public.decorator.ts b/src/auth/public.decorator.ts new file mode 100644 index 0000000..cdd98b1 --- /dev/null +++ b/src/auth/public.decorator.ts @@ -0,0 +1,5 @@ +import { SetMetadata } from "@nestjs/common"; + +export const IS_PUBLIC_KEY = "isPublic"; +export const Public = (): MethodDecorator & ClassDecorator => + SetMetadata(IS_PUBLIC_KEY, true); diff --git a/src/cache/cache.service.ts b/src/cache/cache.service.ts index 9c41e78..7803746 100644 --- a/src/cache/cache.service.ts +++ b/src/cache/cache.service.ts @@ -3,6 +3,23 @@ import { ConfigService } from "@nestjs/config"; import Redis from "ioredis"; import { AppConfig } from "../config/configuration"; +/** + * CacheService — thin, typed wrapper around ioredis. + * + * Design goals + * ------------ + * - Single Redis connection per process, shared across all consumers. + * - Fail-open: every public method swallows Redis errors and falls back + * to the non-cached path so a Redis outage never brings down the API. + * - `wrap()` is the primary interface: "return cached value or run + * the loader, cache the result, return it." + * + * TTL conventions (seconds) + * ------------------------- + * ORACLE_TTL 55 s — just under the 60 s scheduler interval so the + * HTTP endpoint always returns data that is at most one + * cycle stale without issuing a redundant upstream call. + */ export const ORACLE_TTL = 55; @Injectable() @@ -11,41 +28,54 @@ export class CacheService implements OnModuleInit, OnModuleDestroy, OnApplicatio private readonly client: Redis; private closed = false; - constructor(configService: ConfigService) { - const configured = configService.get("redis", { infer: true }); + constructor(private readonly configService: ConfigService) { + const configured = this.configService.get("redis", { infer: true }); const url = process.env.REDIS_URL ?? configured.url; - this.client = new Redis(url, { lazyConnect: false }); - this.client.on("error", (err) => this.logger.error(`Redis client error: ${err.message}`)); + this.client = new Redis(url, { + // Don't retry indefinitely — give up after 3 reconnection attempts + // so a missing Redis doesn't stall startup. + maxRetriesPerRequest: 1, + enableReadyCheck: false, + lazyConnect: false, + }); + + this.client.on("connect", () => this.logger.log("Redis connected")); + this.client.on("error", (err: Error) => + this.logger.warn(`Redis error (cache degraded): ${err.message}`) + ); } onModuleInit(): void {} + /** Returns the parsed JSON value for `key`, or `null` on miss or error. */ async get(key: string): Promise { try { - const value = await this.client.get(key); - return value === null ? null : (JSON.parse(value) as T); - } catch (error) { - this.logger.warn(`Redis cache read failed for ${key}: ${this.errorMessage(error)}`); + const raw = await this.client.get(key); + return raw ? (JSON.parse(raw) as T) : null; + } catch (err) { + this.logger.warn(`cache.get("${key}") failed: ${this.errorMessage(err)}`); return null; } } - async set(key: string, value: unknown, ttlSeconds?: number): Promise { + /** Serialises `value` to JSON and stores it under `key` for `ttlSeconds`. */ + async set(key: string, value: T, ttlSeconds?: number): Promise { try { const serialized = JSON.stringify(value); if (ttlSeconds) await this.client.set(key, serialized, "EX", ttlSeconds); else await this.client.set(key, serialized); - } catch (error) { - this.logger.warn(`Redis cache write failed for ${key}: ${this.errorMessage(error)}`); + } catch (err) { + this.logger.warn(`cache.set("${key}") failed: ${this.errorMessage(err)}`); } } + /** Deletes one or more keys (best-effort). */ async del(...keys: string[]): Promise { if (keys.length === 0) return; try { await this.client.del(...keys); - } catch (error) { - this.logger.warn(`Redis cache delete failed: ${this.errorMessage(error)}`); + } catch (err) { + this.logger.warn(`cache.del(${keys.join(",")}) failed: ${this.errorMessage(err)}`); } } @@ -62,9 +92,25 @@ export class CacheService implements OnModuleInit, OnModuleDestroy, OnApplicatio return this.client.status === "ready"; } + /** + * Cache-aside helper. + * + * Returns the cached value for `key` when present. Otherwise calls + * `loader`, stores the result for `ttlSeconds`, and returns it. + * If either the cache read or write throws, the loader result is + * returned without caching (fail-open). + * + * @example + * const readings = await this.cacheService.wrap( + * "oracle:checkAll", + * ORACLE_TTL, + * () => this.runAllChecks() + * ); + */ async wrap(key: string, ttlSeconds: number, loader: () => Promise): Promise { const cached = await this.get(key); if (cached !== null) return cached; + const fresh = await loader(); await this.set(key, fresh, ttlSeconds); return fresh; @@ -74,11 +120,6 @@ export class CacheService implements OnModuleInit, OnModuleDestroy, OnApplicatio await this.close(); } - /** - * Gracefully close the Redis connection pool during application shutdown. - * Called by Nest when `app.enableShutdownHooks()` is active and the process - * receives SIGTERM/SIGINT, ensuring in-flight commands drain before exit. - */ async onApplicationShutdown(signal?: string): Promise { this.logger.log( `Closing Redis connection pool${signal ? ` (signal: ${signal})` : ''}...`, diff --git a/src/config/configuration.ts b/src/config/configuration.ts index 4ed723a..cde1c9e 100644 --- a/src/config/configuration.ts +++ b/src/config/configuration.ts @@ -95,6 +95,7 @@ export interface AppConfig { defiLlamaBaseUrl: string; defiLlamaProtocolSlug: string; httpTimeoutMs: number; + feedLatencyBudgetsMs?: Record; }; /** Soroban inclusion-fee estimation strategy. */ fees: { @@ -111,6 +112,7 @@ export interface AppConfig { pollIntervalMs: number; lagAlertLedgers: number; pageLimit: number; + contractId: string; }; /** Ops-facing auth (bearer token + read-only community keys). */ ops: { @@ -134,6 +136,7 @@ export interface AppConfig { default: { ttl: number; limit: number }; catalog: { ttl: number; limit: number }; }; + oracleTrustMode: "strict" | "quorum" | "development"; /** Pending off-chain policies that never get submitted expire after this many ms. */ policyPendingTtlMs: number; } @@ -348,21 +351,17 @@ export default (): AppConfig => { redis: { url: env.REDIS_URL ?? "redis://localhost:6379" }, stellar: { network, - sorobanRpcUrl: env.SOROBAN_RPC_URL ?? selected.sorobanRpcUrl, - sorobanRpcUrls: (env.SOROBAN_RPC_URLS ?? selected.sorobanRpcUrl) + sorobanRpcUrl: env.SOROBAN_RPC_URL || selectedNetwork.sorobanRpcUrl, + sorobanRpcUrls: (env.SOROBAN_RPC_URLS ?? env.SOROBAN_RPC_URL ?? selectedNetwork.sorobanRpcUrl) .split(",") .map((url) => url.trim()) .filter(Boolean), eventStartLedger: numberFromEnv("EVENT_START_LEDGER", 1), - horizonUrl: env.STELLAR_HORIZON_URL ?? selected.horizonUrl, - networkPassphrase: - env.STELLAR_NETWORK_PASSPHRASE ?? selected.networkPassphrase, - poolContractId: - networkSpecificPoolId ?? env.REFRACT_POOL_CONTRACT_ID ?? "", - policyContractId: - networkSpecificPolicyId ?? env.REFRACT_POLICY_CONTRACT_ID ?? "", - oracleContractId: - networkSpecificOracleId ?? env.REFRACT_ORACLE_CONTRACT_ID ?? "", + horizonUrl: env.STELLAR_HORIZON_URL || selectedNetwork.horizonUrl, + networkPassphrase: env.STELLAR_NETWORK_PASSPHRASE || selectedNetwork.networkPassphrase, + poolContractId: networkSpecificPoolId ?? env.REFRACT_POOL_CONTRACT_ID ?? "", + policyContractId: networkSpecificPolicyId ?? env.REFRACT_POLICY_CONTRACT_ID ?? "", + oracleContractId: networkSpecificOracleId ?? env.REFRACT_ORACLE_CONTRACT_ID ?? "", relayerSecret: env.ORACLE_RELAYER_SECRET ?? "", relayerSecretId: env.ORACLE_RELAYER_SECRET_ID ?? "", relayerSecretRegion: env.ORACLE_RELAYER_SECRET_REGION ?? "us-east-1", @@ -397,20 +396,20 @@ export default (): AppConfig => { minIntervalMs: env.ORACLE_PUBLISH_MIN_INTERVAL_MS ?? 0, }, oracles: { - coingeckoBaseUrl: - env.COINGECKO_BASE_URL ?? "https://api.coingecko.com/api/v3", - horizonUrl: env.STELLAR_HORIZON_URL ?? selected.horizonUrl, - defiLlamaBaseUrl: env.DEFILLAMA_BASE_URL ?? "https://api.llama.fi", + coingeckoBaseUrl: env.COINGECKO_BASE_URL || "https://api.coingecko.com/api/v3", + horizonUrl: env.STELLAR_HORIZON_URL || selectedNetwork.horizonUrl, + defiLlamaBaseUrl: env.DEFILLAMA_BASE_URL || "https://api.llama.fi", // Placeholder "covered protocol" for the SmartContractRisk TVL-drop // check until Refract defines a real list of covered Soroban // protocols. Defaults to a large, consistently-tracked protocol so // the drop-detection logic has real data to run against. - defiLlamaProtocolSlug: env.DEFILLAMA_PROTOCOL_SLUG ?? "aave", - httpTimeoutMs: env.ORACLE_HTTP_TIMEOUT_MS ?? 5000, + defiLlamaProtocolSlug: env.DEFILLAMA_PROTOCOL_SLUG || "aave", + httpTimeoutMs: numberFromEnv("ORACLE_HTTP_TIMEOUT_MS", 5_000), + feedLatencyBudgetsMs: parseFeedBudgets(env.ORACLE_FEED_LATENCY_BUDGETS_MS), }, fees: { statsTtlMs: numberFromEnv("FEE_STATS_TTL_MS", 15_000), - ceilingStroops: String(numberFromEnv("FEE_CEILING_STROOPS", 10_000_000)), + ceilingStroops: env.FEE_CEILING_STROOPS || String(numberFromEnv("FEE_CEILING_STROOPS", 10_000_000)), profiles: { moderate: { percentile: numberFromEnv("FEE_MODERATE_PERCENTILE", 90), @@ -427,13 +426,11 @@ export default (): AppConfig => { pollIntervalMs: numberFromEnv("EVENT_INDEXER_POLL_INTERVAL_MS", 5_000), lagAlertLedgers: numberFromEnv("EVENT_INDEXER_LAG_ALERT_LEDGERS", 100), pageLimit: numberFromEnv("EVENT_INDEXER_PAGE_LIMIT", 100), + contractId: env.REFRACT_POOL_CONTRACT_ID || "", }, ops: { - apiToken: env.OPS_API_TOKEN ?? "", - communityApiKeys: (env.COMMUNITY_API_KEYS ?? "") - .split(",") - .map((k) => k.trim()) - .filter(Boolean), + apiToken: env.OPS_API_TOKEN || "", + communityApiKeys: parseCsv(env.COMMUNITY_API_KEYS), }, adminApiKey: env.ADMIN_API_KEY ?? "", communityApiKeys: [], @@ -474,6 +471,47 @@ export default (): AppConfig => { limit: parseInt(process.env.THROTTLE_CATALOG_LIMIT || "300", 10), }, }, + oracleTrustMode: parseTrustMode(env.ORACLE_TRUST_MODE), policyPendingTtlMs: numberFromEnv("POLICY_PENDING_TTL_MS", 3_600_000), }; }; + +export function isSandboxMode(env: NodeJS.ProcessEnv = process.env): boolean { + const raw = (env.SANDBOX_MODE ?? "").trim().toLowerCase(); + return raw === "true" || raw === "1" || raw === "yes" || raw === "on"; +} + +export function assertSandboxSafety(env: NodeJS.ProcessEnv = process.env): void { + if (isSandboxMode(env) && (env.NODE_ENV ?? "").trim().toLowerCase() === "production") { + throw new Error("SANDBOX_MODE is not allowed when NODE_ENV=production"); + } +} + +function parseCsv(value?: string): string[] { + return (value ?? "").split(",").map((entry) => entry.trim()).filter(Boolean); +} + +function parseFeedBudgets(value?: string): Record | undefined { + if (!value) return undefined; + let result: unknown; + try { + result = JSON.parse(value); + } catch { + throw new Error("ORACLE_FEED_LATENCY_BUDGETS_MS must be valid JSON"); + } + if (!result || typeof result !== "object" || Array.isArray(result)) { + throw new Error("ORACLE_FEED_LATENCY_BUDGETS_MS must be a JSON object"); + } + for (const [feed, budget] of Object.entries(result)) { + if (!Number.isInteger(budget) || Number(budget) <= 0) { + throw new Error(`Latency budget for ${feed} must be a positive integer`); + } + } + return result as Record; +} + +function parseTrustMode(value?: string): AppConfig["oracleTrustMode"] { + if (value === undefined || value === "") return "strict"; + if (value === "strict" || value === "quorum" || value === "development") return value; + throw new Error("ORACLE_TRUST_MODE must be strict, quorum, or development"); +} diff --git a/src/config/oracle-feed.config.ts b/src/config/oracle-feed.config.ts new file mode 100644 index 0000000..6d4f97f --- /dev/null +++ b/src/config/oracle-feed.config.ts @@ -0,0 +1,124 @@ +export interface OracleFeedConfiguration { + latencyBudgetMs: number; + requiredQuorum: number; + unit: string; + sourceType: "http" | "mock"; +} + +export const ORACLE_FEEDS = { + "coingecko-usdc": { + latencyBudgetMs: 30_000, + requiredQuorum: 1, + unit: "USD", + sourceType: "http", + }, + "coingecko-stellar": { + latencyBudgetMs: 30_000, + requiredQuorum: 1, + unit: "percent_change_24h", + sourceType: "http", + }, + "defillama-aave": { + latencyBudgetMs: 45_000, + requiredQuorum: 1, + unit: "percent_change_24h", + sourceType: "http", + }, + "nexus-liquidations": { + latencyBudgetMs: 10_000, + requiredQuorum: 1, + unit: "collateral_ratio", + sourceType: "mock", + }, + aviationstack: { + latencyBudgetMs: 10_000, + requiredQuorum: 1, + unit: "minutes", + sourceType: "mock", + }, +} as const satisfies Record; + +export type OracleFeedId = keyof typeof ORACLE_FEEDS; + +export interface FeedPayloadRule { + requiredFields: readonly string[]; + numericRanges: Readonly>; + maxStepChanges?: Readonly>; + allowedCategoricalValues?: Readonly>; + expectedUnits: Readonly>; +} + +export const ORACLE_FEED_RULES: Readonly> = { + "coingecko-usdc": { + requiredFields: ["price", "unit"], + numericRanges: { price: [0, 2] }, + maxStepChanges: { price: 0.2 }, + expectedUnits: { unit: "USD" }, + }, + "coingecko-stellar": { + requiredFields: ["price", "change24h", "unit", "changeUnit"], + numericRanges: { price: [0, 1_000_000], change24h: [-100, 1_000] }, + maxStepChanges: { price: 25, change24h: 50 }, + expectedUnits: { unit: "USD", changeUnit: "percent" }, + }, + "defillama-aave": { + requiredFields: ["unit", "tvl[].date", "tvl[].totalLiquidityUSD"], + numericRanges: { + "tvl[].date": [0, 4_102_444_800], + "tvl[].totalLiquidityUSD": [0, 1_000_000_000_000_000], + }, + maxStepChanges: { "tvl[].totalLiquidityUSD": 20_000_000_000 }, + expectedUnits: { unit: "USD" }, + }, + "nexus-liquidations": { + requiredFields: ["collateralRatio", "status", "unit"], + numericRanges: { collateralRatio: [0, 2] }, + allowedCategoricalValues: { status: ["healthy", "liquidated"] }, + expectedUnits: { unit: "ratio" }, + }, + aviationstack: { + requiredFields: ["flightNumber", "delayMinutes", "status", "unit"], + numericRanges: { delayMinutes: [0, 10_080] }, + allowedCategoricalValues: { status: ["scheduled", "delayed", "cancelled"] }, + expectedUnits: { unit: "minutes" }, + }, +}; + +export const COVERAGE_REQUIRED_FEEDS = { + StablecoinDepeg: ["coingecko-usdc"], + MarketCrash: ["coingecko-stellar"], + LiquidationShield: ["nexus-liquidations"], + SmartContractRisk: ["defillama-aave"], + FlightDelay: ["aviationstack"], +} as const; + +export function feedLatencyBudgetMs( + feedId: string, + configuredBudgets?: Readonly>, +): number { + const configured = ORACLE_FEEDS[feedId as OracleFeedId]; + const configuredOverride = configuredBudgets?.[feedId]; + if (configuredOverride !== undefined) { + if (!Number.isInteger(configuredOverride) || configuredOverride <= 0) { + throw new Error(`Latency budget for ${feedId} must be a positive integer`); + } + return configuredOverride; + } + const override = process.env.ORACLE_FEED_LATENCY_BUDGETS_MS; + if (override) { + let budgets: Record; + try { + budgets = JSON.parse(override) as Record; + } catch { + throw new Error("ORACLE_FEED_LATENCY_BUDGETS_MS must be a JSON object of feed IDs to positive integer milliseconds"); + } + const value = budgets[feedId]; + if (value !== undefined) { + if (!Number.isInteger(value) || value <= 0) { + throw new Error(`Latency budget for ${feedId} must be a positive integer`); + } + return value; + } + } + return configured?.latencyBudgetMs ?? 30_000; +} diff --git a/src/health/health.module.ts b/src/health/health.module.ts index 6bd5081..5be6124 100644 --- a/src/health/health.module.ts +++ b/src/health/health.module.ts @@ -3,9 +3,10 @@ import { CacheModule } from "../cache/cache.module"; import { DbModule } from "../db/db.module"; import { HealthController } from "./health.controller"; import { HealthService } from "./health.service"; +import { OracleTrustModule } from "../oracle/oracle-trust.module"; @Module({ - imports: [CacheModule, DbModule], + imports: [CacheModule, DbModule, OracleTrustModule], controllers: [HealthController], providers: [HealthService], exports: [HealthService], diff --git a/src/health/health.service.spec.ts b/src/health/health.service.spec.ts new file mode 100644 index 0000000..24b65de --- /dev/null +++ b/src/health/health.service.spec.ts @@ -0,0 +1,73 @@ +import { DbService } from "../db/db.service"; +import { ORACLE_FEEDS } from "../config/oracle-feed.config"; +import { OracleTrustService } from "../oracle/oracle-trust.service"; +import { SorobanRpcService } from "../stellar/soroban-rpc.service"; +import { HealthService } from "./health.service"; + +describe("HealthService readiness", () => { + const originalTrustMode = process.env.ORACLE_TRUST_MODE; + + afterEach(() => { + if (originalTrustMode === undefined) delete process.env.ORACLE_TRUST_MODE; + else process.env.ORACLE_TRUST_MODE = originalTrustMode; + }); + + it("requires reachable trusted feeds, a satisfiable quorum, healthy RPC, and writable evidence storage", async () => { + process.env.ORACLE_TRUST_MODE = "development"; + const trust = new OracleTrustService(); + for (const [feedId, config] of Object.entries(ORACLE_FEEDS)) { + if (config.sourceType === "http") trust.recordObservation(feedId, true); + } + trust.setPublisherConfigured(true); + + const rpc = { + server: { getHealth: jest.fn().mockResolvedValue({ status: "healthy" }) }, + } as unknown as SorobanRpcService; + const db = { + isAvailable: jest.fn().mockReturnValue(true), + withClient: jest.fn(async (fn: (client: { query: jest.Mock }) => Promise) => + fn({ query: jest.fn().mockResolvedValue({}) }), + ), + } as unknown as DbService; + const service = new HealthService(trust, rpc, db); + + await expect(service.getReadiness()).resolves.toMatchObject({ + ready: true, + checks: { + sourcesReachable: true, + quorumSatisfiable: true, + rpcHealthy: true, + evidenceStoreWritable: true, + publisherConfigured: true, + }, + }); + + (db.withClient as jest.Mock).mockRejectedValueOnce(new Error("read-only database")); + await expect(service.getReadiness()).resolves.toMatchObject({ + ready: false, + checks: { evidenceStoreWritable: false }, + }); + }); + + it("is not ready when no fresh feed observations can satisfy trust", async () => { + const rpc = { + server: { getHealth: jest.fn().mockRejectedValue(new Error("RPC down")) }, + } as unknown as SorobanRpcService; + const db = { + isAvailable: jest.fn().mockReturnValue(false), + } as unknown as DbService; + const service = new HealthService(new OracleTrustService(), rpc, db); + + await expect(service.getReadiness()).resolves.toMatchObject({ + ready: false, + status: "not_ready", + checks: { + sourcesReachable: false, + quorumSatisfiable: false, + rpcHealthy: false, + evidenceStoreWritable: false, + publisherConfigured: false, + }, + }); + }); +}); diff --git a/src/health/health.service.ts b/src/health/health.service.ts index 75087ba..74eb0f3 100644 --- a/src/health/health.service.ts +++ b/src/health/health.service.ts @@ -1,36 +1,143 @@ -import { Injectable } from "@nestjs/common"; +import { Injectable, Logger, Optional } from "@nestjs/common"; import { CacheService } from "../cache/cache.service"; import { DbService } from "../db/db.service"; +import { OracleTrustService } from "../oracle/oracle-trust.service"; +import { SorobanRpcService } from "../stellar/soroban-rpc.service"; @Injectable() export class HealthService { + private readonly logger = new Logger(HealthService.name); + constructor( - private readonly database: DbService, - private readonly cache: CacheService, + @Optional() private readonly oracleTrust?: OracleTrustService, + @Optional() private readonly rpcService?: SorobanRpcService, + @Optional() private readonly dbService?: DbService, + @Optional() private readonly database?: DbService, + @Optional() private readonly cache?: CacheService, ) {} getLiveness(): { status: "ok" } { return { status: "ok" }; } - async getReadiness() { + async getReadiness(): Promise<{ + ready: boolean; + status: "ready" | "not_ready"; + checks: { + sourcesReachable: boolean; + quorumSatisfiable: boolean; + rpcHealthy: boolean; + evidenceStoreWritable: boolean; + publisherConfigured: boolean; + }; + trustMode: string; + feeds: ReturnType["feeds"]; + dependencies: { postgres: boolean; redis: boolean }; + }> { + const feedReadiness = this.oracleTrust?.readiness(); + const infrastructure = await this.getInfrastructureReadiness(); const [postgres, redis] = await Promise.all([ - Promise.resolve(this.database.isAvailable()), - this.cache.ping(), + Promise.resolve(this.database?.isAvailable() ?? this.dbService?.isAvailable() ?? false), + this.cache ? this.cache.ping() : Promise.resolve(false), ]); + const checks = { + sourcesReachable: feedReadiness?.sourcesReachable ?? false, + quorumSatisfiable: feedReadiness?.quorumSatisfiable ?? false, + ...infrastructure.checks, + }; + const ready = Object.values(checks).every(Boolean) && postgres && redis; return { - ready: postgres && redis, + ready, + status: ready ? "ready" : "not_ready", + checks, + trustMode: feedReadiness?.trustMode ?? "strict", + feeds: feedReadiness?.feeds ?? [], dependencies: { postgres, redis }, }; } - getDetail() { + async getInfrastructureReadiness(): Promise<{ + ready: boolean; + checks: { + rpcHealthy: boolean; + evidenceStoreWritable: boolean; + publisherConfigured: boolean; + }; + }> { + const feedReadiness = this.oracleTrust?.readiness(); + const [rpcHealthy, evidenceStoreWritable] = await Promise.all([ + this.checkRpc(), + this.checkEvidenceStore(), + ]); + const checks = { + rpcHealthy, + evidenceStoreWritable, + publisherConfigured: feedReadiness?.publisherConfigured ?? false, + }; + return { ready: Object.values(checks).every(Boolean), checks }; + } + + async getDetail(): Promise> { + const readiness = await this.getReadiness(); return { - status: "ok", + status: readiness.status, + trustMode: readiness.trustMode, dependencies: { - postgres: this.database.isAvailable(), - redis: this.cache.isConnected(), + ...readiness.checks, + postgres: this.database?.isAvailable() ?? this.dbService?.isAvailable() ?? false, + redis: this.cache ? this.cache.isConnected() : false, }, + feeds: readiness.feeds, }; } + + private async checkRpc(): Promise { + if (!this.rpcService) return false; + try { + await withTimeout(this.rpcService.server.getHealth(), 3_000, "Soroban RPC health probe timed out"); + return true; + } catch (error) { + this.logger.warn(`Soroban RPC readiness probe failed: ${errorMessage(error)}`); + return false; + } + } + + private async checkEvidenceStore(): Promise { + if (!this.dbService?.isAvailable()) return false; + try { + await this.dbService.withClient(async (client) => { + await client.query("BEGIN"); + try { + await client.query( + `INSERT INTO oracle_events (coverage_type, value, source, severity) + VALUES ('stablecoin_depeg', 0, 'readiness_probe', 'low')`, + ); + await client.query("ROLLBACK"); + } catch (error) { + await client.query("ROLLBACK"); + throw error; + } + }); + return true; + } catch (error) { + this.logger.warn(`Oracle evidence store readiness probe failed: ${errorMessage(error)}`); + return false; + } + } +} + +function withTimeout(promise: Promise, timeoutMs: number, message: string): Promise { + let timer: NodeJS.Timeout | undefined; + return Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(message)), timeoutMs); + }), + ]).finally(() => { + if (timer) clearTimeout(timer); + }); +} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); } diff --git a/src/metrics/metrics.service.spec.ts b/src/metrics/metrics.service.spec.ts index ba9c925..257869a 100644 --- a/src/metrics/metrics.service.spec.ts +++ b/src/metrics/metrics.service.spec.ts @@ -9,6 +9,8 @@ describe("MetricsService", () => { const metrics = new MetricsService(); metrics.recordSchedulerRun("claim_settlement", "success", 1.2); metrics.recordOracleCheck("StablecoinDepeg", "failure", 0.3); + metrics.recordOracleStage("coingecko-usdc", "consensus", 0.4); + metrics.recordOracleDeadlineBreach("coingecko-usdc", "confirmation", 0.2); await metrics.observeSorobanRpc("send_transaction", async () => "hash"); const output = await metrics.metrics(); @@ -17,6 +19,9 @@ describe("MetricsService", () => { expect(output).toContain('refract_scheduler_run_duration_seconds_count{scheduler="claim_settlement"} 1'); expect(output).toContain('refract_oracle_checks_total{coverage_type="StablecoinDepeg",outcome="failure"} 1'); expect(output).toContain('refract_oracle_check_duration_seconds_count{coverage_type="StablecoinDepeg"} 1'); + expect(output).toContain('refract_oracle_stage_duration_seconds_count{feed="coingecko-usdc",stage="consensus"} 1'); + expect(output).toContain('refract_oracle_deadline_breaches_total{feed="coingecko-usdc",stage="confirmation"} 1'); + expect(output).toContain('refract_oracle_deadline_overrun_seconds_count{feed="coingecko-usdc",stage="confirmation"} 1'); expect(output).toContain('refract_soroban_rpc_duration_seconds_count{operation="send_transaction",outcome="success"} 1'); }); diff --git a/src/metrics/metrics.service.ts b/src/metrics/metrics.service.ts index 768b0cd..e74fe42 100644 --- a/src/metrics/metrics.service.ts +++ b/src/metrics/metrics.service.ts @@ -56,6 +56,26 @@ export class MetricsService implements RelayerMetricsSink { buckets: [0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30], registers: [this.registry], }); + private readonly oracleStageDuration = new Histogram<"feed" | "stage">({ + name: "refract_oracle_stage_duration_seconds", + help: "Oracle feed processing duration by feed and pipeline stage.", + labelNames: ["feed", "stage"], + buckets: [0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60], + registers: [this.registry], + }); + private readonly oracleDeadlineBreaches = new Counter<"feed" | "stage">({ + name: "refract_oracle_deadline_breaches_total", + help: "Oracle feed publication deadlines breached, labeled by feed and stage.", + labelNames: ["feed", "stage"], + registers: [this.registry], + }); + private readonly oracleDeadlineOverrun = new Histogram<"feed" | "stage">({ + name: "refract_oracle_deadline_overrun_seconds", + help: "Time by which oracle feed processing exceeded its publication deadline.", + labelNames: ["feed", "stage"], + buckets: [0.01, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30], + registers: [this.registry], + }); private readonly relayerBalance = new Gauge({ name: "refract_relayer_balance_stroops", @@ -131,6 +151,16 @@ export class MetricsService implements RelayerMetricsSink { this.oracleCheckDuration.labels({ coverage_type: coverageType }).observe(durationSeconds); } + recordOracleStage(feed: string, stage: string, durationSeconds: number): void { + this.oracleStageDuration.labels({ feed, stage }).observe(durationSeconds); + } + + recordOracleDeadlineBreach(feed: string, stage: string, overrunSeconds: number): void { + const labels = { feed, stage }; + this.oracleDeadlineBreaches.labels(labels).inc(); + this.oracleDeadlineOverrun.labels(labels).observe(Math.max(0, overrunSeconds)); + } + recordRelayerBalance(a: BalanceAssessment, spend: FeeSpendSummary): void { this.relayerBalance.set(Number(a.balanceStroops)); this.relayerSpendable.set(Number(a.spendableStroops)); diff --git a/src/oracle/feed-payload.validator.spec.ts b/src/oracle/feed-payload.validator.spec.ts new file mode 100644 index 0000000..547cda1 --- /dev/null +++ b/src/oracle/feed-payload.validator.spec.ts @@ -0,0 +1,56 @@ +import { FeedPayloadValidator, InvalidFeedPayloadError } from "./feed-payload.validator"; + +describe("FeedPayloadValidator", () => { + it("rejects missing fields, out-of-range values, and unexpected units", () => { + const validator = new FeedPayloadValidator(); + + expect(() => + validator.validateAndRemember("coingecko-usdc", { + price: 4, + unit: "EUR", + }), + ).toThrow(InvalidFeedPayloadError); + expect(() => + validator.validateAndRemember("coingecko-usdc", { + price: 1, + }), + ).toThrow(/required field unit is missing/); + }); + + it("rejects an excessive step change without updating the accepted baseline", () => { + const validator = new FeedPayloadValidator(); + validator.validateAndRemember("coingecko-usdc", { price: 1, unit: "USD" }); + + expect(() => + validator.validateAndRemember("coingecko-usdc", { price: 0.7, unit: "USD" }), + ).toThrow(/changed by more than 0.2/); + + expect(() => + validator.validateAndRemember("coingecko-usdc", { price: 0.9, unit: "USD" }), + ).not.toThrow(); + }); + + it("rejects categorical values outside the configured set", () => { + const validator = new FeedPayloadValidator(); + + expect(() => + validator.validateAndRemember("aviationstack", { + flightNumber: "RF123", + delayMinutes: 0, + status: "unknown", + unit: "minutes", + }), + ).toThrow(/status must be one of scheduled, delayed, cancelled/); + }); + + it("validates every numeric entry in array-valued provider fields", () => { + const validator = new FeedPayloadValidator(); + + expect(() => + validator.validateAndRemember("defillama-aave", { + unit: "USD", + tvl: [{ date: 1, totalLiquidityUSD: -10 }], + }), + ).toThrow(/tvl\[\]\.totalLiquidityUSD/); + }); +}); diff --git a/src/oracle/feed-payload.validator.ts b/src/oracle/feed-payload.validator.ts new file mode 100644 index 0000000..96c7080 --- /dev/null +++ b/src/oracle/feed-payload.validator.ts @@ -0,0 +1,114 @@ +import { ORACLE_FEED_RULES } from "../config/oracle-feed.config"; + +export class InvalidFeedPayloadError extends Error { + constructor(feedId: string, failures: string[]) { + super(`Feed ${feedId} payload rejected: ${failures.join("; ")}`); + this.name = "InvalidFeedPayloadError"; + } +} + +export class FeedPayloadValidator { + private readonly previousPayloads = new Map(); + + validateAndRemember(feedId: string, payload: unknown): void { + const rule = ORACLE_FEED_RULES[feedId]; + if (!rule) { + throw new Error(`No validation rules configured for feed ${feedId}`); + } + + const previous = this.previousPayloads.get(feedId); + const failures: string[] = []; + const record = asRecord(payload); + if (!record) { + throw new InvalidFeedPayloadError(feedId, ["payload must be an object"]); + } + + for (const path of rule.requiredFields) { + if (readPathValues(record, path).length === 0) { + failures.push(`required field ${path} is missing`); + } + } + + for (const [path, [minimum, maximum]] of Object.entries(rule.numericRanges)) { + const values = readPathValues(record, path); + if (values.length === 0) continue; + if ( + values.some( + (value) => + typeof value !== "number" || + !Number.isFinite(value) || + value < minimum || + value > maximum, + ) + ) { + failures.push(`${path} must contain finite numbers in [${minimum}, ${maximum}]`); + } + } + + if (previous && asRecord(previous)) { + for (const [path, maximumStep] of Object.entries(rule.maxStepChanges ?? {})) { + const currentValues = readPathValues(record, path); + const previousValues = readPathValues(asRecord(previous)!, path); + const count = Math.min(currentValues.length, previousValues.length); + for (let index = 0; index < count; index += 1) { + const current = currentValues[index]; + const prior = previousValues[index]; + if ( + typeof current === "number" && + typeof prior === "number" && + Math.abs(current - prior) > maximumStep + ) { + failures.push(`${path} changed by more than ${maximumStep}`); + break; + } + } + } + } + + for (const [path, allowed] of Object.entries(rule.allowedCategoricalValues ?? {})) { + const values = readPathValues(record, path); + if (values.some((value) => typeof value !== "string" || !allowed.includes(value))) { + failures.push(`${path} must be one of ${allowed.join(", ")}`); + } + } + + for (const [path, expected] of Object.entries(rule.expectedUnits)) { + const values = readPathValues(record, path); + if (values.length === 0 || values.some((value) => value !== expected)) { + failures.push(`${path} must equal the configured unit ${expected}`); + } + } + + if (failures.length > 0) { + throw new InvalidFeedPayloadError(feedId, failures); + } + this.previousPayloads.set(feedId, payload); + } +} + +function asRecord(value: unknown): Record | null { + return value !== null && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : null; +} + +function readPathValues(value: Record, path: string): unknown[] { + let current: unknown[] = [value]; + for (const segment of path.split(".")) { + const expandsArray = segment.endsWith("[]"); + const key = expandsArray ? segment.slice(0, -2) : segment; + const next: unknown[] = []; + for (const item of current) { + const record = asRecord(item); + if (!record || !(key in record)) continue; + const property = record[key]; + if (expandsArray) { + if (Array.isArray(property)) next.push(...property); + } else { + next.push(property); + } + } + current = next; + } + return current.filter((entry) => entry !== undefined && entry !== null); +} diff --git a/src/oracle/oracle-publisher.service.spec.ts b/src/oracle/oracle-publisher.service.spec.ts index 530dea9..f11482e 100644 --- a/src/oracle/oracle-publisher.service.spec.ts +++ b/src/oracle/oracle-publisher.service.spec.ts @@ -2,8 +2,10 @@ import { ConfigService } from "@nestjs/config"; import { Account, Keypair, StrKey, TransactionBuilder, nativeToScVal, rpc, scValToNative, xdr } from "@stellar/stellar-sdk"; import { AppConfig } from "../config/configuration"; import { OraclePublisherService } from "./oracle-publisher.service"; +import { MetricsService } from "../metrics/metrics.service"; import { scaleOracleValue, coverageTypeToU32 } from "./oracle-encoding"; import { OracleReading } from "./oracle-reading"; +import { OracleTrustService } from "./oracle-trust.service"; const NETWORK_PASSPHRASE = "Test SDF Network ; September 2015"; @@ -100,6 +102,40 @@ describe("OraclePublisherService", () => { expect(getAccountSpy).not.toHaveBeenCalled(); }); + it("never publishes without a fresh feed observation satisfying quorum", async () => { + const trust = new OracleTrustService(); + const service = new OraclePublisherService(buildConfig(), undefined, trust); + const getAccountSpy = jest.spyOn(rpc.Server.prototype, "getAccount"); + + const result = await service.maybePublish( + reading({ feedId: "coingecko-usdc", deadlineAt: Date.now() + 10_000 }), + ); + + expect(result).toBeNull(); + expect(getAccountSpy).not.toHaveBeenCalled(); + }); + + it("aborts an expired feed budget before making an RPC call and records the breached stage", async () => { + const metrics = new MetricsService(); + const service = new OraclePublisherService(buildConfig(), metrics); + const getAccountSpy = jest.spyOn(rpc.Server.prototype, "getAccount"); + + const result = await service.maybePublish( + reading({ + feedId: "coingecko-usdc", + observedAt: Date.now() - 31_000, + deadlineAt: Date.now() - 1, + }), + ); + + expect(result).toBeNull(); + expect(getAccountSpy).not.toHaveBeenCalled(); + const output = await metrics.metrics(); + expect(output).toContain( + 'refract_oracle_deadline_breaches_total{feed="coingecko-usdc",stage="publication"} 1', + ); + }); + it("encodes update_reading(u32, i128) with the scaled value", async () => { const service = new OraclePublisherService(buildConfig()); @@ -121,6 +157,8 @@ describe("OraclePublisherService", () => { envelopeXdr: {} as never, resultXdr: {} as never, resultMetaXdr: {} as never, + events: { transactionEventsXdr: [], contractEventsXdr: [] }, + txHash: "oracle-tx", }); const result = await service.maybePublish(reading({ coverageType: "MarketCrash", value: -30.12 })); @@ -160,6 +198,8 @@ describe("OraclePublisherService", () => { envelopeXdr: {} as never, resultXdr: {} as never, resultMetaXdr: {} as never, + events: { transactionEventsXdr: [], contractEventsXdr: [] }, + txHash: "oracle-tx", }); const first = service.maybePublish(reading({ value: 0.99 })); @@ -190,6 +230,8 @@ describe("OraclePublisherService", () => { envelopeXdr: {} as never, resultXdr: {} as never, resultMetaXdr: {} as never, + events: { transactionEventsXdr: [], contractEventsXdr: [] }, + txHash: "oracle-tx", }); const result = await service.maybePublish(reading()); @@ -219,6 +261,8 @@ describe("OraclePublisherService", () => { envelopeXdr: {} as never, resultXdr: {} as never, resultMetaXdr: {} as never, + events: { transactionEventsXdr: [], contractEventsXdr: [] }, + txHash: "oracle-tx", }); await service.maybePublish(reading({ value: 1.0 })); diff --git a/src/oracle/oracle-publisher.service.ts b/src/oracle/oracle-publisher.service.ts index 2b34419..7016034 100644 --- a/src/oracle/oracle-publisher.service.ts +++ b/src/oracle/oracle-publisher.service.ts @@ -1,7 +1,9 @@ -import { Injectable, Logger } from "@nestjs/common"; +import { Injectable, Logger, Optional } from "@nestjs/common"; import { ConfigService } from "@nestjs/config"; import { BASE_FEE, Contract, Keypair, TransactionBuilder, nativeToScVal, rpc } from "@stellar/stellar-sdk"; import { AppConfig } from "../config/configuration"; +import { MetricsService } from "../metrics/metrics.service"; +import { OracleTrustService } from "./oracle-trust.service"; import { pollForConfirmation } from "../stellar/soroban-confirmation.util"; import { OracleReading } from "./oracle-reading"; import { coverageTypeToU32, scaleOracleValue } from "./oracle-encoding"; @@ -40,7 +42,11 @@ export class OraclePublisherService { private readonly inFlight = new Set(); private readonly auditLog: OraclePublishRecord[] = []; - constructor(private readonly configService: ConfigService) { + constructor( + private readonly configService: ConfigService, + @Optional() private readonly metricsService?: MetricsService, + @Optional() private readonly oracleTrustService?: OracleTrustService, + ) { const stellar = this.configService.get("stellar", { infer: true }); this.server = new rpc.Server(stellar.sorobanRpcUrl); this.networkPassphrase = stellar.networkPassphrase; @@ -50,6 +56,7 @@ export class OraclePublisherService { this.publishMode = publish.mode; this.minIntervalMs = publish.minIntervalMs; this.confirmationDefaults = this.configService.get("confirmation", { infer: true }); + this.oracleTrustService?.setPublisherConfigured(this.isConfigured()); } /** True once an oracle contract ID and relayer secret are configured. */ @@ -65,9 +72,24 @@ export class OraclePublisherService { * Decide whether to publish and, if so, submit. Never throws — degrades * with a clear log when unconfigured (same posture as ClaimSettlementService). */ - async maybePublish(reading: OracleReading): Promise { + async maybePublish( + reading: OracleReading, + deadlineAt: number | undefined = reading.deadlineAt, + ): Promise { + const feedId = reading.feedId ?? reading.coverageType; if (reading.degraded) { this.logger.warn(`Skipping on-chain publish for degraded ${reading.coverageType} reading`); + this.oracleTrustService?.recordPublication(feedId, false, "provider reading is degraded"); + return null; + } + + if (this.oracleTrustService && !this.oracleTrustService.canPublish(feedId)) { + this.logger.warn(`Skipping on-chain publish for untrusted feed ${feedId}`); + return null; + } + + if (this.deadlineExpired(feedId, "publication", deadlineAt)) { + this.oracleTrustService?.recordPublication(feedId, false, "publication deadline expired"); return null; } @@ -91,15 +113,14 @@ export class OraclePublisherService { const now = Date.now(); const lastAt = this.lastPublishedAt.get(reading.coverageType) ?? 0; - if (now - lastAt < this.minIntervalMs) { - return null; + const previousValue = this.lastScaled.get(reading.coverageType); + if (previousValue !== undefined && previousValue === scaled) { + this.oracleTrustService?.recordPublication(feedId, true); + if (this.publishMode === "on_change") return null; } - if (this.publishMode === "on_change") { - const prev = this.lastScaled.get(reading.coverageType); - if (prev !== undefined && prev === scaled) { - return null; - } + if (now - lastAt < this.minIntervalMs) { + return null; } if (this.inFlight.has(reading.coverageType)) { @@ -108,21 +129,37 @@ export class OraclePublisherService { } this.inFlight.add(reading.coverageType); + const publicationStartedAt = Date.now(); try { - const record = await this.publish(reading.coverageType, scaled); + const record = await this.publish(reading.coverageType, scaled, feedId, deadlineAt); this.auditLog.push(record); if (this.auditLog.length > 500) this.auditLog.shift(); if (record.confirmed) { this.lastScaled.set(reading.coverageType, scaled); this.lastPublishedAt.set(reading.coverageType, Date.now()); } + this.oracleTrustService?.recordPublication( + feedId, + record.confirmed, + record.error ?? record.outcome, + ); return record; } finally { + this.metricsService?.recordOracleStage( + feedId, + "publication", + (Date.now() - publicationStartedAt) / 1000, + ); this.inFlight.delete(reading.coverageType); } } - private async publish(coverageType: string, scaledValue: bigint): Promise { + private async publish( + coverageType: string, + scaledValue: bigint, + feedId: string, + deadlineAt?: number, + ): Promise { const attemptedAt = Date.now(); const base: OraclePublishRecord = { coverageType, @@ -132,7 +169,18 @@ export class OraclePublisherService { }; try { - const sourceAccount = await this.server.getAccount(this.relayerKeypair!.publicKey()); + if (this.deadlineExpired(feedId, "publication", deadlineAt)) { + return { ...base, error: "Feed publication deadline expired before RPC account lookup" }; + } + const sourceAccount = await this.withinDeadline( + feedId, + "publication", + deadlineAt, + () => this.server.getAccount(this.relayerKeypair!.publicKey()), + ); + if (this.deadlineExpired(feedId, "publication", deadlineAt)) { + return { ...base, error: "Feed publication deadline expired before transaction preparation" }; + } const contract = new Contract(this.oracleContractId); // RefractOracle.update_reading(coverage_type: u32, value: i128) const operation = contract.call( @@ -141,18 +189,42 @@ export class OraclePublisherService { nativeToScVal(scaledValue, { type: "i128" }) ); + const remainingMs = deadlineAt === undefined ? 30_000 : deadlineAt - Date.now(); + const transactionTimeoutSeconds = Math.min(30, Math.floor(remainingMs / 1000)); + if (transactionTimeoutSeconds < 1) { + this.metricsService?.recordOracleDeadlineBreach( + feedId, + "publication", + Math.max(0, (1_000 - remainingMs) / 1_000), + ); + return { ...base, error: "Insufficient latency budget remains for an on-chain transaction" }; + } + const builtTx = new TransactionBuilder(sourceAccount, { fee: BASE_FEE, networkPassphrase: this.networkPassphrase, }) .addOperation(operation) - .setTimeout(30) + .setTimeout(transactionTimeoutSeconds) .build(); - const preparedTx = await this.server.prepareTransaction(builtTx); + const preparedTx = await this.withinDeadline( + feedId, + "publication", + deadlineAt, + () => this.server.prepareTransaction(builtTx), + ); + if (this.deadlineExpired(feedId, "publication", deadlineAt)) { + return { ...base, error: "Feed publication deadline expired before transaction submission" }; + } preparedTx.sign(this.relayerKeypair!); - const sendResult = await this.server.sendTransaction(preparedTx); + const sendResult = await this.withinDeadline( + feedId, + "publication", + deadlineAt, + () => this.server.sendTransaction(preparedTx), + ); if (sendResult.status === "ERROR" || sendResult.status === "TRY_AGAIN_LATER") { const record = { ...base, @@ -163,13 +235,45 @@ export class OraclePublisherService { return record; } - const confirmation = await pollForConfirmation(this.server, sendResult.hash, { - initialIntervalMs: this.confirmationDefaults.initialIntervalMs, - backoffMultiplier: this.confirmationDefaults.backoffMultiplier, - maxIntervalMs: this.confirmationDefaults.maxIntervalMs, - jitterRatio: this.confirmationDefaults.jitterRatio, - deadlineMs: this.confirmationDefaults.settlementDeadlineMs, - }); + const confirmationStartedAt = Date.now(); + const confirmationDeadlineMs = deadlineAt === undefined + ? this.confirmationDefaults.settlementDeadlineMs + : Math.min(this.confirmationDefaults.settlementDeadlineMs, deadlineAt - Date.now()); + if (confirmationDeadlineMs <= 0) { + this.deadlineExpired(feedId, "confirmation", deadlineAt); + return { + ...base, + txHash: sendResult.hash, + error: "Feed publication was submitted but its confirmation deadline expired", + }; + } + + const confirmation = await this.withinDeadline( + feedId, + "confirmation", + deadlineAt, + () => + pollForConfirmation(this.server, sendResult.hash, { + initialIntervalMs: this.confirmationDefaults.initialIntervalMs, + backoffMultiplier: this.confirmationDefaults.backoffMultiplier, + maxIntervalMs: this.confirmationDefaults.maxIntervalMs, + jitterRatio: this.confirmationDefaults.jitterRatio, + deadlineMs: confirmationDeadlineMs, + }), + ); + this.metricsService?.recordOracleStage( + feedId, + "confirmation", + (Date.now() - confirmationStartedAt) / 1000, + ); + if (this.deadlineExpired(feedId, "confirmation", deadlineAt)) { + return { + ...base, + txHash: confirmation.txHash, + outcome: confirmation.outcome, + error: "Feed publication confirmation arrived after its deadline", + }; + } const record: OraclePublishRecord = { ...base, @@ -192,4 +296,49 @@ export class OraclePublisherService { return { ...base, error: message }; } } + + private deadlineExpired(feedId: string, stage: string, deadlineAt?: number): boolean { + if (deadlineAt === undefined) { + return false; + } + const overrunMs = Date.now() - deadlineAt; + if (overrunMs < 0) { + return false; + } + this.metricsService?.recordOracleDeadlineBreach(feedId, stage, Math.max(0, overrunMs / 1000)); + return true; + } + + private async withinDeadline( + feedId: string, + stage: string, + deadlineAt: number | undefined, + operation: () => Promise, + ): Promise { + if (deadlineAt === undefined) { + return operation(); + } + const remainingMs = deadlineAt - Date.now(); + if (remainingMs <= 0) { + this.deadlineExpired(feedId, stage, deadlineAt); + throw new Error(`Feed latency budget expired during ${stage}`); + } + + let timer: NodeJS.Timeout | undefined; + try { + return await Promise.race([ + operation(), + new Promise((_, reject) => { + timer = setTimeout(() => { + this.deadlineExpired(feedId, stage, deadlineAt); + reject(new Error(`Feed latency budget expired during ${stage}`)); + }, remainingMs); + }), + ]); + } finally { + if (timer) { + clearTimeout(timer); + } + } + } } diff --git a/src/oracle/oracle-reading.ts b/src/oracle/oracle-reading.ts index 7844ef5..242d2d5 100644 --- a/src/oracle/oracle-reading.ts +++ b/src/oracle/oracle-reading.ts @@ -11,8 +11,11 @@ export interface OracleReading { threshold: number; severity: Severity; message: string; + feedId?: string; source?: string; timestamp?: number; + deadlineAt?: number; + fetchDurationMs?: number; /** * True when this reading came from the fail-safe fallback (upstream outage). * OraclePublisherService must never publish degraded readings on-chain. diff --git a/src/oracle/oracle-trust.module.ts b/src/oracle/oracle-trust.module.ts new file mode 100644 index 0000000..f9988ea --- /dev/null +++ b/src/oracle/oracle-trust.module.ts @@ -0,0 +1,9 @@ +import { Global, Module } from "@nestjs/common"; +import { OracleTrustService } from "./oracle-trust.service"; + +@Global() +@Module({ + providers: [OracleTrustService], + exports: [OracleTrustService], +}) +export class OracleTrustModule {} diff --git a/src/oracle/oracle-trust.service.ts b/src/oracle/oracle-trust.service.ts new file mode 100644 index 0000000..107b733 --- /dev/null +++ b/src/oracle/oracle-trust.service.ts @@ -0,0 +1,181 @@ +import { Injectable, Optional } from "@nestjs/common"; +import { ConfigService } from "@nestjs/config"; +import { AppConfig } from "../config/configuration"; +import { + COVERAGE_REQUIRED_FEEDS, + ORACLE_FEEDS, + OracleFeedId, +} from "../config/oracle-feed.config"; + +export interface FeedTrustStatus { + feedId: string; + healthy: boolean; + confirmed: boolean; + quorumRequired: number; + quorumAvailable: number; + observedAt: number | null; + confirmedAt: number | null; + reason?: string; +} + +export interface CoverageAvailability { + available: boolean; + requiredFeeds: string[]; + unavailableFeeds: string[]; + reason?: string; +} + +@Injectable() +export class OracleTrustService { + private readonly feedStatuses = new Map(); + private publisherConfigured = false; + + constructor( + @Optional() private readonly configService?: ConfigService, + ) {} + + setPublisherConfigured(configured: boolean): void { + this.publisherConfigured = configured; + } + + recordObservation( + feedId: string, + healthy: boolean, + reason?: string, + observedAt: number = Date.now(), + ): void { + const configuration = ORACLE_FEEDS[feedId as OracleFeedId]; + if (!configuration) { + throw new Error(`No oracle feed configuration exists for ${feedId}`); + } + this.feedStatuses.set(feedId, { + feedId, + healthy, + confirmed: false, + quorumRequired: configuration.requiredQuorum, + quorumAvailable: healthy ? 1 : 0, + observedAt, + confirmedAt: null, + reason: healthy ? undefined : reason ?? "source is unreachable or payload was rejected", + }); + } + + recordPublication(feedId: string, confirmed: boolean, reason?: string): void { + const current = this.feedStatuses.get(feedId); + if (!current) { + return; + } + this.feedStatuses.set(feedId, { + ...current, + confirmed, + confirmedAt: confirmed ? Date.now() : null, + reason: confirmed ? undefined : reason ?? "on-chain publication is not confirmed", + }); + } + + feedStatus(feedId: string): FeedTrustStatus { + const existing = this.feedStatuses.get(feedId); + if (existing) return { ...existing }; + return { + feedId, + healthy: false, + confirmed: false, + quorumRequired: ORACLE_FEEDS[feedId as OracleFeedId]?.requiredQuorum ?? 1, + quorumAvailable: 0, + observedAt: null, + confirmedAt: null, + reason: "no accepted observation or on-chain confirmation", + }; + } + + canPublish(feedId: string): boolean { + const status = this.feedStatus(feedId); + const budget = ORACLE_FEEDS[feedId as OracleFeedId]?.latencyBudgetMs; + return ( + budget !== undefined && + status.healthy && + status.quorumAvailable >= status.quorumRequired && + status.observedAt !== null && + Date.now() - status.observedAt <= budget + ); + } + + getCoverageAvailability(coverageType: string): CoverageAvailability { + const requiredFeeds = + COVERAGE_REQUIRED_FEEDS[coverageType as keyof typeof COVERAGE_REQUIRED_FEEDS]; + if (!requiredFeeds) { + return { + available: false, + requiredFeeds: [], + unavailableFeeds: [], + reason: "coverage type has no configured feed dependencies", + }; + } + const unavailableFeeds = requiredFeeds.filter((feedId) => { + const status = this.feedStatus(feedId); + const budget = ORACLE_FEEDS[feedId]?.latencyBudgetMs ?? 0; + const isFresh = status.observedAt !== null && Date.now() - status.observedAt <= budget; + return ( + !status.healthy || + !status.confirmed || + status.quorumAvailable < status.quorumRequired || + !isFresh + ); + }); + return { + available: unavailableFeeds.length === 0, + requiredFeeds: [...requiredFeeds], + unavailableFeeds, + reason: + unavailableFeeds.length === 0 + ? undefined + : `required feed(s) unavailable or unconfirmed: ${unavailableFeeds.join(", ")}`, + }; + } + + readiness(): { + sourcesReachable: boolean; + quorumSatisfiable: boolean; + publisherConfigured: boolean; + feeds: FeedTrustStatus[]; + trustMode: "strict" | "quorum" | "development"; + } { + const trustMode = parseTrustMode( + this.configService?.get("oracleTrustMode", { infer: true }) ?? + process.env.ORACLE_TRUST_MODE, + ); + const entries = Object.entries(ORACLE_FEEDS).filter( + ([, config]) => trustMode !== "development" || config.sourceType !== "mock", + ); + const feeds = entries.map(([feedId, configuration]) => { + const status = this.feedStatus(feedId); + const stale = + status.observedAt === null || + Date.now() - status.observedAt > configuration.latencyBudgetMs; + return stale + ? { + ...status, + healthy: false, + quorumAvailable: 0, + reason: "source observation is missing or stale", + } + : status; + }); + const sourcesReachable = feeds.every((feed) => feed.healthy); + const quorumSatisfiable = feeds.every( + (feed) => feed.quorumAvailable >= feed.quorumRequired, + ); + return { + sourcesReachable, + quorumSatisfiable, + publisherConfigured: this.publisherConfigured, + feeds, + trustMode, + }; + } +} + +function parseTrustMode(value: string | undefined): "strict" | "quorum" | "development" { + if (value === "quorum" || value === "development") return value; + return "strict"; +} diff --git a/src/oracle/oracle.gateway.ts b/src/oracle/oracle.gateway.ts index 868c5e2..54c2078 100644 --- a/src/oracle/oracle.gateway.ts +++ b/src/oracle/oracle.gateway.ts @@ -90,26 +90,34 @@ export class OracleGateway * come from the worker via the ReadingBus (see OracleFeedRelay); this * process never polls upstreams itself. */ - broadcastAlert(alert: OracleReading): void { + broadcastAlert(reading: OracleReading): void { if (!this.server?.clients) { return; } - const payload = JSON.stringify({ ...alert, timestamp: Date.now() }); + const message = JSON.stringify({ + type: 'oracle_alert', + data: reading, + timestamp: new Date().toISOString(), + }); let sent = 0; for (const client of this.server.clients) { - if (client.readyState !== WebSocket.OPEN) continue; + if (client.readyState !== WebSocket.OPEN) { + continue; + } // send() can throw for a socket that is mid-close; one bad client must // not stop the broadcast to everyone else. try { - client.send(payload); + client.send(message, (error) => { + if (error) this.logger.warn(`Oracle WebSocket send failed: ${error.message}`); + }); sent++; - } catch (err) { + } catch (error) { this.logger.warn( - `Failed to send oracle alert to a client: ${err instanceof Error ? err.message : String(err)}`, + `Failed to send oracle alert: ${error instanceof Error ? error.message : String(error)}`, ); } } - this.logger.debug(`Broadcast ${alert.coverageType} alert to ${sent} client(s)`); + this.logger.debug(`Broadcast ${reading.coverageType} alert to ${sent} client(s)`); } handleDisconnect(client: WebSocket): void { @@ -117,23 +125,6 @@ export class OracleGateway this.logger.log('Oracle WebSocket client disconnected'); } - broadcastAlert(reading: OracleReading): void { - if (!this.server) return; - const payload = JSON.stringify(reading); - for (const client of this.server.clients) { - if (client.readyState !== WebSocket.OPEN) continue; - try { - client.send(payload, (error) => { - if (error) this.logger.warn(`Oracle WebSocket send failed: ${error.message}`); - }); - } catch (error) { - this.logger.warn( - `Oracle WebSocket send failed: ${error instanceof Error ? error.message : String(error)}`, - ); - } - } - } - /** Authenticated (ops) clients present `?token=` equal to OPS_WS_TOKEN. */ private isAuthed(req?: IncomingMessage): boolean { const expected = process.env.OPS_WS_TOKEN; diff --git a/src/oracle/oracle.module.ts b/src/oracle/oracle.module.ts index 9fe13a0..b71727c 100644 --- a/src/oracle/oracle.module.ts +++ b/src/oracle/oracle.module.ts @@ -3,6 +3,9 @@ import { ConfigModule, ConfigService } from "@nestjs/config"; import { DbModule } from "../db/db.module"; import { StellarModule } from "../stellar/stellar.module"; +import { MetricsModule } from "../metrics/metrics.module"; +import { OracleTrustModule } from "./oracle-trust.module"; +import { HealthModule } from "../health/health.module"; import { OracleController } from "./oracle.controller"; import { OracleEventRepository } from "./oracle-event.repository"; import { OracleGateway } from "./oracle.gateway"; @@ -73,7 +76,14 @@ const attestationVerifierProvider: Provider = { }; @Module({ - imports: [ConfigModule, DbModule, StellarModule], + imports: [ + ConfigModule, + DbModule, + StellarModule, + MetricsModule, + OracleTrustModule, + HealthModule, + ], controllers: [OracleController], providers: [ // Clock & calendar abstraction (issue #291) @@ -115,6 +125,8 @@ const attestationVerifierProvider: Provider = { OraclePublisherService, OracleMonitorService, FreshnessEnforcementService, + OracleGateway, + OraclePublisherService, ClockService, ORACLE_CLOCK, FeedCalendarResolver, diff --git a/src/oracle/oracle.scheduler.spec.ts b/src/oracle/oracle.scheduler.spec.ts index dda1bbe..a783d10 100644 --- a/src/oracle/oracle.scheduler.spec.ts +++ b/src/oracle/oracle.scheduler.spec.ts @@ -2,6 +2,7 @@ import { OracleScheduler } from "./oracle.scheduler"; import { OracleService } from "./oracle.service"; import { OracleGateway } from "./oracle.gateway"; import { OracleReading } from "./oracle-reading"; +import { HealthService } from "../health/health.service"; function buildReading(overrides: Partial = {}): OracleReading { return { @@ -16,7 +17,7 @@ function buildReading(overrides: Partial = {}): OracleReading { } function buildServices() { - const oracleService = { checkAll: jest.fn() } as unknown as jest.Mocked; + const oracleService = { checkAllFresh: jest.fn(), checkAll: jest.fn() } as unknown as jest.Mocked; const oracleGateway = { broadcastAlert: jest.fn() } as unknown as jest.Mocked; const oraclePublisher = { maybePublish: jest.fn().mockResolvedValue(null) } as unknown as jest.Mocked< import("./oracle-publisher.service").OraclePublisherService @@ -31,6 +32,7 @@ describe("OracleScheduler", () => { const low = buildReading({ coverageType: "StablecoinDepeg", severity: "low" }); const high = buildReading({ coverageType: "MarketCrash", severity: "high" }); const triggered = buildReading({ coverageType: "SmartContractRisk", severity: "triggered" }); + oracleService.checkAllFresh.mockResolvedValue([low, high, triggered]); oracleService.checkAll.mockResolvedValue([low, high, triggered]); const scheduler = new OracleScheduler(oracleService, oracleGateway, oraclePublisher); @@ -45,6 +47,7 @@ describe("OracleScheduler", () => { it("does not broadcast anything when every reading is 'low' severity", async () => { const { oracleService, oracleGateway, oraclePublisher } = buildServices(); + oracleService.checkAllFresh.mockResolvedValue([buildReading({ severity: "low" })]); oracleService.checkAll.mockResolvedValue([buildReading({ severity: "low" })]); const scheduler = new OracleScheduler(oracleService, oracleGateway, oraclePublisher); @@ -56,6 +59,7 @@ describe("OracleScheduler", () => { it("catches and logs an error from checkAll() instead of throwing", async () => { const { oracleService, oracleGateway, oraclePublisher } = buildServices(); + oracleService.checkAllFresh.mockRejectedValue(new Error("all sources down")); oracleService.checkAll.mockRejectedValue(new Error("all sources down")); const scheduler = new OracleScheduler(oracleService, oracleGateway, oraclePublisher); @@ -63,5 +67,46 @@ describe("OracleScheduler", () => { expect(oracleGateway.broadcastAlert).not.toHaveBeenCalled(); expect(oraclePublisher.maybePublish).not.toHaveBeenCalled(); }); + + it("aborts an expired consensus result without publishing or broadcasting it", async () => { + const { oracleService, oracleGateway, oraclePublisher } = buildServices(); + oracleService.checkAllFresh.mockResolvedValue([ + buildReading({ + severity: "triggered", + feedId: "coingecko-usdc", + observedAt: Date.now() - 2_000, + deadlineAt: Date.now() - 1, + }), + ]); + const scheduler = new OracleScheduler(oracleService, oracleGateway, oraclePublisher); + + await scheduler.pollOracles(); + + expect(oraclePublisher.maybePublish).not.toHaveBeenCalled(); + expect(oracleGateway.broadcastAlert).not.toHaveBeenCalled(); + }); + + it("does not publish while RPC or evidence storage is not ready", async () => { + const { oracleService, oracleGateway, oraclePublisher } = buildServices(); + oracleService.checkAllFresh.mockResolvedValue([buildReading()]); + const health = { + getInfrastructureReadiness: jest.fn().mockResolvedValue({ + ready: false, + checks: { rpcHealthy: false, evidenceStoreWritable: true, publisherConfigured: true }, + }), + } as unknown as HealthService; + const scheduler = new OracleScheduler( + oracleService, + oracleGateway, + oraclePublisher, + undefined, + health, + ); + + await scheduler.pollOracles(); + + expect(oraclePublisher.maybePublish).not.toHaveBeenCalled(); + expect(oracleGateway.broadcastAlert).not.toHaveBeenCalled(); + }); }); }); diff --git a/src/oracle/oracle.scheduler.ts b/src/oracle/oracle.scheduler.ts index fcd40ff..c2fc8d0 100644 --- a/src/oracle/oracle.scheduler.ts +++ b/src/oracle/oracle.scheduler.ts @@ -1,22 +1,35 @@ -import { Injectable, Logger, OnModuleInit, OnModuleDestroy, Optional } from '@nestjs/common'; -import { SchedulerRegistry } from '@nestjs/schedule'; -import { CronJob } from 'cron'; -import { OracleService } from './oracle.service'; -import { OracleGateway } from './oracle.gateway'; -import { OraclePublisherService } from './oracle-publisher.service'; -import { ProtocolParametersService } from '../parameters/protocol-parameters.service'; -import { ProtocolParameterKey } from '../parameters/protocol-parameter-key.enum'; -import { ReadingBus } from './reading-bus'; -import { OracleReading } from './oracle-reading'; -import { runsPipeline } from '../runtime/process-role'; -import { PipelineHeartbeat } from '../runtime/pipeline-heartbeat'; - -const ORACLE_REFRESH_JOB_NAME = 'oracle-refresh'; -export const ORACLE_PIPELINE_STAGE = 'oracle'; +import { Injectable, Logger, OnModuleInit, OnModuleDestroy, Optional } from "@nestjs/common"; +import { SchedulerRegistry } from "@nestjs/schedule"; +import { CronJob } from "cron"; +import { MetricsService } from "../metrics/metrics.service"; +import { HealthService } from "../health/health.service"; +import { OracleGateway } from "./oracle.gateway"; +import { OraclePublisherService } from "./oracle-publisher.service"; +import { OracleReading } from "./oracle-reading"; +import { OracleService } from "./oracle.service"; +import { ProtocolParametersService } from "../parameters/protocol-parameters.service"; +import { ProtocolParameterKey } from "../parameters/protocol-parameter-key.enum"; +import { ReadingBus } from "./reading-bus"; +import { runsPipeline } from "../runtime/process-role"; +import { PipelineHeartbeat } from "../runtime/pipeline-heartbeat"; + +const ORACLE_REFRESH_JOB_NAME = "oracle-refresh"; +export const ORACLE_PIPELINE_STAGE = "oracle"; + +const DEFAULT_POLL_INTERVAL_MS = 60_000; + +function pollIntervalMs(): number { + const value = Number(process.env.ORACLE_SCHEDULER_INTERVAL_MS ?? DEFAULT_POLL_INTERVAL_MS); + if (!Number.isInteger(value) || value < 1_000 || value > 3_600_000) { + throw new Error("ORACLE_SCHEDULER_INTERVAL_MS must be an integer between 1000 and 3600000"); + } + return value; +} @Injectable() export class OracleScheduler implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(OracleScheduler.name); + private polling = false; constructor( private readonly oracleService: OracleService, @@ -25,6 +38,8 @@ export class OracleScheduler implements OnModuleInit, OnModuleDestroy { private readonly schedulerRegistry: SchedulerRegistry, private readonly protocolParameters: ProtocolParametersService, private readonly readingBus: ReadingBus, + @Optional() private readonly metricsService?: MetricsService, + @Optional() private readonly healthService?: HealthService, @Optional() private readonly heartbeat?: PipelineHeartbeat, ) {} @@ -32,7 +47,7 @@ export class OracleScheduler implements OnModuleInit, OnModuleDestroy { // Polling and consensus belong to the worker tier. The API only relays // what the worker publishes (see OracleFeedRelay). if (!runsPipeline()) { - this.logger.log('Oracle polling disabled in this process (PROCESS_ROLE=api)'); + this.logger.log("Oracle polling disabled in this process (PROCESS_ROLE=api)"); return; } this.heartbeat?.expect(ORACLE_PIPELINE_STAGE); @@ -54,24 +69,81 @@ export class OracleScheduler implements OnModuleInit, OnModuleDestroy { } onModuleDestroy(): void { - if (this.schedulerRegistry.doesExist('cron', ORACLE_REFRESH_JOB_NAME)) { + if (this.schedulerRegistry.doesExist("cron", ORACLE_REFRESH_JOB_NAME)) { this.schedulerRegistry.deleteCronJob(ORACLE_REFRESH_JOB_NAME); } } private async refresh(): Promise { + if (this.polling) { + this.logger.warn("Skipping oracle poll because the previous round is still running"); + return; + } + this.polling = true; + const startedAt = Date.now(); + let outcome: "success" | "failure" = "success"; + try { const result: unknown = await this.oracleService.refreshPrices(); const readings = Array.isArray(result) ? (result as OracleReading[]) : []; + + const infrastructure = await this.healthService?.getInfrastructureReadiness(); + if (infrastructure && !infrastructure.ready) { + this.logger.warn("Skipping oracle publications while required infrastructure is not ready"); + return; + } + for (const reading of readings) { - await this.oraclePublisher.maybePublish(reading); - if (reading.severity !== 'low') this.oracleGateway.broadcastAlert(reading); + await this.processReading(reading); } + // Hand readings to the API tier for the WebSocket feed and reads. await this.readingBus.publishAll(readings); this.heartbeat?.beat(ORACLE_PIPELINE_STAGE); } catch (error) { - this.logger.error('Oracle refresh failed', error as Error); + outcome = "failure"; + this.logger.error("Oracle refresh failed", error instanceof Error ? error.stack : String(error)); + } finally { + this.polling = false; + this.metricsService?.recordSchedulerRun( + "oracle", + outcome, + (Date.now() - startedAt) / 1000, + ); + } + } + + private async processReading(reading: OracleReading): Promise { + const feed = reading.feedId ?? reading.coverageType; + const observedAt = reading.observedAt ?? Date.now(); + const deadlineAt = reading.deadlineAt ?? observedAt + 30_000; + if (reading.degraded) { + this.logger.warn(`Skipping degraded ${feed} reading`); + return; + } + if (reading.fetchDurationMs !== undefined) { + this.metricsService?.recordOracleStage(feed, "source", reading.fetchDurationMs / 1000); + } + + const consensusDuration = (Date.now() - observedAt) / 1000; + this.metricsService?.recordOracleStage(feed, "consensus", consensusDuration); + if (Date.now() >= deadlineAt) { + this.metricsService?.recordOracleDeadlineBreach( + feed, + "consensus", + (Date.now() - deadlineAt) / 1000, + ); + this.logger.warn(`Aborting stale ${feed} reading before publication`); + return; + } + + await this.oraclePublisher.maybePublish(reading, deadlineAt); + if (Date.now() >= deadlineAt) { + return; + } + + if (reading.severity !== "low") { + this.oracleGateway.broadcastAlert(reading); } } } diff --git a/src/oracle/oracle.service.ts b/src/oracle/oracle.service.ts index 2c65356..51c3213 100644 --- a/src/oracle/oracle.service.ts +++ b/src/oracle/oracle.service.ts @@ -1,9 +1,15 @@ -import { Injectable, Logger } from "@nestjs/common"; +import { Injectable, Logger, Optional } from "@nestjs/common"; import { ConfigService } from "@nestjs/config"; import { randomUUID } from "node:crypto"; import axios from "axios"; import { CacheService, ORACLE_TTL } from "../cache/cache.service"; import { AppConfig } from "../config/configuration"; +import { + COVERAGE_REQUIRED_FEEDS, + feedLatencyBudgetMs, +} from "../config/oracle-feed.config"; +import { FeedPayloadValidator } from "./feed-payload.validator"; +import { OracleTrustService } from "./oracle-trust.service"; import { OracleReading, Severity } from "./oracle-reading"; interface DefiLlamaTvlPoint { @@ -59,10 +65,12 @@ export class OracleService { private readonly horizonUrl: string; private readonly defiLlamaBaseUrl: string; private readonly defiLlamaProtocolSlug: string; + private readonly feedValidator = new FeedPayloadValidator(); constructor( private readonly configService: ConfigService, - private readonly cacheService: CacheService + private readonly cacheService: CacheService, + @Optional() private readonly oracleTrustService?: OracleTrustService ) { const oracles = this.configService.get("oracles", { infer: true }); this.timeoutMs = oracles.httpTimeoutMs; @@ -95,6 +103,17 @@ export class OracleService { */ async checkAllFresh(): Promise { const readings = await this.runAllChecks(); + for (const reading of readings) { + const feedId = reading.feedId ?? feedIdForCoverage(reading.coverageType); + if (feedId) { + this.oracleTrustService?.recordObservation( + feedId, + !reading.degraded, + reading.degraded ? reading.message : undefined, + reading.observedAt, + ); + } + } // Proactively refresh the cache so a subsequent HTTP call within the // same cycle gets the scheduler's fresh data immediately. await this.cacheService.set(CACHE_KEY_CHECK_ALL, readings, ORACLE_TTL); @@ -110,13 +129,20 @@ export class OracleService { } async checkStablecoinDepeg(): Promise { + const startedAt = Date.now(); const threshold = 0.95; // 5% depeg try { - const { data } = await axios.get>( + const response = await axios.get>( `${this.coingeckoBaseUrl}/simple/price`, { params: { ids: "usd-coin", vs_currencies: "usd" }, timeout: this.timeoutMs } ); + const observedAt = Date.now(); + const { data } = response; const usdcPrice = data["usd-coin"].usd; + this.feedValidator.validateAndRemember("coingecko-usdc", { + price: usdcPrice, + unit: "USD", + }); const deviation = (1 - usdcPrice) * 100; const severity: Severity = @@ -125,7 +151,7 @@ export class OracleService { : usdcPrice < 0.99 ? "medium" : "low"; - return { + return this.withFeedTiming({ readingId: randomUUID(), coverageType: "StablecoinDepeg", type: "oracle_update", @@ -135,13 +161,14 @@ export class OracleService { message: `USDC price: $${usdcPrice.toFixed(4)} (${deviation > 0 ? "-" : "+"}${Math.abs(deviation).toFixed(3)}% from peg) [CoinGecko]`, source: "coingecko", timestamp: Math.floor(Date.now() / 1000), - }; + }, "coingecko-usdc", startedAt, observedAt); } catch (err) { return this.degraded("StablecoinDepeg", threshold, "CoinGecko", err); } } async checkMarketCrash(): Promise { + const startedAt = Date.now(); const threshold = -30; // 30% crash triggers try { const priceRequest = axios.get>( @@ -150,7 +177,7 @@ export class OracleService { params: { ids: "stellar", vs_currencies: "usd", include_24hr_change: true }, timeout: this.timeoutMs, } - ); + ).then((response) => ({ ...response, observedAt: Date.now() })); // Best-effort: attach real chain context, but never let a Horizon // hiccup fail the whole check — the price signal comes from CoinGecko. const ledgerRequest = axios @@ -162,6 +189,12 @@ export class OracleService { const [priceRes, ledgerRes] = await Promise.all([priceRequest, ledgerRequest]); const change24h = priceRes.data.stellar.usd_24h_change; + this.feedValidator.validateAndRemember("coingecko-stellar", { + price: priceRes.data.stellar.usd, + change24h, + unit: "USD", + changeUnit: "percent", + }); const ledger = ledgerRes?.data?._embedded?.records?.[0]; const severity: Severity = @@ -174,7 +207,7 @@ export class OracleService { ? ` [CoinGecko; Horizon testnet ledger #${ledger.sequence} @ ${ledger.closed_at}]` : " [CoinGecko]"; - return { + return this.withFeedTiming({ readingId: randomUUID(), coverageType: "MarketCrash", type: "oracle_update", @@ -184,19 +217,26 @@ export class OracleService { message: `XLM 24h change: ${change24h.toFixed(2)}% (trigger at ${threshold}%)${chainContext}`, source: "coingecko", timestamp: Math.floor(Date.now() / 1000), - }; + }, "coingecko-stellar", startedAt, priceRes.observedAt); } catch (err) { return this.degraded("MarketCrash", threshold, "CoinGecko", err); } } async checkSmartContractRisk(): Promise { + const startedAt = Date.now(); const threshold = -50; // 50% TVL drop in 24h triggers try { - const { data } = await axios.get<{ name?: string; tvl?: DefiLlamaTvlPoint[] }>( + const response = await axios.get<{ name?: string; tvl?: DefiLlamaTvlPoint[] }>( `${this.defiLlamaBaseUrl}/protocol/${this.defiLlamaProtocolSlug}`, { timeout: this.timeoutMs } ); + const observedAt = Date.now(); + const { data } = response; + this.feedValidator.validateAndRemember("defillama-aave", { + tvl: data.tvl, + unit: "USD", + }); const points = data.tvl ?? []; if (points.length < 2) throw new Error("DeFiLlama returned insufficient TVL history"); @@ -217,7 +257,7 @@ export class OracleService { : pctChange < -10 ? "medium" : "low"; - return { + return this.withFeedTiming({ readingId: randomUUID(), coverageType: "SmartContractRisk", type: "oracle_update", @@ -227,7 +267,7 @@ export class OracleService { message: `${data.name ?? this.defiLlamaProtocolSlug} TVL 24h change: ${pctChange.toFixed(2)}% (trigger at ${threshold}%) [DeFiLlama]`, source: "defillama", timestamp: Math.floor(Date.now() / 1000), - }; + }, "defillama-aave", startedAt, observedAt); } catch (err) { return this.degraded("SmartContractRisk", threshold, "DeFiLlama", err); } @@ -242,8 +282,13 @@ export class OracleService { // sometimes — a ±0.05 spread around 0.92 never dips below 0.85. const collateralRatio = 0.92 + (Math.random() - 0.5) * 0.3; const threshold = 0.85; // shield triggers below 85% collateralization + this.feedValidator.validateAndRemember("nexus-liquidations", { + collateralRatio, + status: collateralRatio < threshold ? "liquidated" : "healthy", + unit: "ratio", + }); - return { + return this.withFeedTiming({ readingId: randomUUID(), coverageType: "LiquidationShield", type: "oracle_update", @@ -253,7 +298,8 @@ export class OracleService { message: `Collateral ratio ${(collateralRatio * 100).toFixed(1)}% (shield triggers below ${(threshold * 100).toFixed(0)}%) [mocked — no NEXUS Protocol integration]`, source: "mock", timestamp: Math.floor(Date.now() / 1000), - }; + degraded: true, + }, "nexus-liquidations", Date.now()); } async checkFlightDelay(flightNumber: string): Promise { @@ -262,8 +308,14 @@ export class OracleService { // source is available. See README's "Oracle data sources" section. const delayMinutes = Math.floor(Math.random() * 300); const threshold = 120; // 2h delay triggers + this.feedValidator.validateAndRemember("aviationstack", { + flightNumber, + delayMinutes, + status: delayMinutes >= threshold ? "delayed" : "scheduled", + unit: "minutes", + }); - return { + return this.withFeedTiming({ readingId: randomUUID(), coverageType: "FlightDelay", type: "oracle_update", @@ -273,6 +325,23 @@ export class OracleService { message: `Flight ${flightNumber}: ${delayMinutes}m delay (trigger at ${threshold}m) [mocked — AviationStack requires a paid key]`, source: "mock", timestamp: Math.floor(Date.now() / 1000), + degraded: true, + }, "aviationstack", Date.now()); + } + + private withFeedTiming( + reading: OracleReading, + feedId: string, + startedAt: number, + observedAt: number = Date.now(), + ): OracleReading { + const budgets = this.configService.get("oracles", { infer: true }).feedLatencyBudgetsMs; + return { + ...reading, + feedId, + observedAt, + fetchDurationMs: observedAt - startedAt, + deadlineAt: observedAt + feedLatencyBudgetMs(feedId, budgets), }; } @@ -296,9 +365,16 @@ export class OracleService { threshold, severity: "low", message: `${source} unavailable — degraded to a non-triggering reading`, + feedId: feedIdForCoverage(coverageType), source: source.toLowerCase(), timestamp: Math.floor(Date.now() / 1000), degraded: true, }; } } + +function feedIdForCoverage(coverageType: string): string | undefined { + return COVERAGE_REQUIRED_FEEDS[ + coverageType as keyof typeof COVERAGE_REQUIRED_FEEDS + ]?.[0]; +} diff --git a/src/policy/dto/buy-policy.dto.ts b/src/policy/dto/buy-policy.dto.ts index 312b512..0a872f4 100644 --- a/src/policy/dto/buy-policy.dto.ts +++ b/src/policy/dto/buy-policy.dto.ts @@ -42,7 +42,7 @@ class TriggerParamsConstraint implements ValidatorConstraintInterface { export class BuyPolicyDto { @IsString() - @IsNotEmpty() + @Length(56, 56) @IsStellarPublicKey() holder!: string; @@ -83,30 +83,4 @@ export class BuyPolicyDto { * condition (capacity, utilization, coverage bounds, duration, premium balance) * without building or returning an XDR. */ -export class PreflightPolicyDto { - @IsString() - @IsNotEmpty() - @IsStellarPublicKey() - @Length(56, 56) - holder!: string; - - @IsInt() - @Min(0) - @Max(4) - coverageType!: number; - - /** USDC amount in 1e7 base units, passed as a decimal string to avoid precision loss. */ - @IsString() - @Matches(/^\d+$/) - @MaxLength(38) - coverageAmount!: string; - - @IsInt() - @Min(1) - @Max(365) - durationDays!: number; - - @ValidateIf((obj) => obj.triggerParams !== undefined) - @Validate(TriggerParamsConstraint) - triggerParams?: Record; -} +export class PreflightPolicyDto extends BuyPolicyDto {} diff --git a/src/policy/policy.module.ts b/src/policy/policy.module.ts index 70592fc..3d97672 100644 --- a/src/policy/policy.module.ts +++ b/src/policy/policy.module.ts @@ -3,9 +3,10 @@ import { StellarModule } from "../stellar/stellar.module"; import { PolicyController } from "./policy.controller"; import { PolicyRepository } from "./policy.repository"; import { PolicyService } from "./policy.service"; +import { OracleTrustModule } from "../oracle/oracle-trust.module"; @Module({ - imports: [StellarModule], + imports: [StellarModule, OracleTrustModule], controllers: [PolicyController], providers: [PolicyRepository, PolicyService], exports: [PolicyService], diff --git a/src/policy/policy.service.spec.ts b/src/policy/policy.service.spec.ts index d9cb162..4f4f127 100644 --- a/src/policy/policy.service.spec.ts +++ b/src/policy/policy.service.spec.ts @@ -15,6 +15,7 @@ import { AppConfig } from "../config/configuration"; import { testStellarConfig } from "../config/test-stellar"; import { MetricsService } from "../metrics/metrics.service"; import { PolicyService } from "./policy.service"; +import { OracleTrustService } from "../oracle/oracle-trust.service"; import { BuyPolicyDto } from "./dto/buy-policy.dto"; const NETWORK_PASSPHRASE = "Test SDF Network ; September 2015"; @@ -405,5 +406,24 @@ describe("PolicyService", () => { it("returns all five coverage types", () => { expect(service.listTypes()).toHaveLength(5); }); + + it("marks products unavailable and rejects buying until their required feed is confirmed", async () => { + const trust = new OracleTrustService(); + const guardedService = new PolicyService( + buildConfig(), + new MetricsService(), + undefined, + trust, + ); + + expect(guardedService.listTypes()[0].available).toBe(false); + await expect(guardedService.buy(buildDto(holder))).rejects.toBeInstanceOf( + BadRequestException, + ); + + trust.recordObservation("coingecko-usdc", true); + trust.recordPublication("coingecko-usdc", true); + expect(guardedService.listTypes()[0].available).toBe(true); + }); }); }); diff --git a/src/policy/policy.service.ts b/src/policy/policy.service.ts index 77cb8b1..58a7417 100644 --- a/src/policy/policy.service.ts +++ b/src/policy/policy.service.ts @@ -1,5 +1,5 @@ +import { BadRequestException, Injectable, Optional } from "@nestjs/common"; import { randomUUID } from "node:crypto"; -import { BadRequestException, Injectable } from "@nestjs/common"; import { ConfigService } from "@nestjs/config"; import { Account, @@ -18,6 +18,7 @@ const uuidv4 = randomUUID; import { AppConfig } from "../config/configuration"; import { MetricsService } from "../metrics/metrics.service"; import { BuyPolicyDto } from "./dto/buy-policy.dto"; +import { OracleTrustService } from "../oracle/oracle-trust.service"; const FLIGHT_DELAY_COVERAGE_TYPE = 4; const COVERAGE_TYPE_VARIANTS = [ @@ -42,6 +43,9 @@ export interface CoverageTypeCatalogEntry { maxCoverage: number; trigger: string; icon: string; + available: boolean; + requiredFeeds: string[]; + unavailableReason?: string; } export interface StoredPolicy { @@ -81,6 +85,8 @@ const COVERAGE_TYPES: CoverageTypeCatalogEntry[] = [ maxCoverage: 100_000, trigger: "USDC price < $0.95 for 15+ minutes", icon: "🪙", + available: true, + requiredFeeds: [], }, { id: 1, @@ -92,6 +98,8 @@ const COVERAGE_TYPES: CoverageTypeCatalogEntry[] = [ maxCoverage: 50_000, trigger: "Market index 24h return < -30%", icon: "📉", + available: true, + requiredFeeds: [], }, { id: 2, @@ -103,6 +111,8 @@ const COVERAGE_TYPES: CoverageTypeCatalogEntry[] = [ maxCoverage: 200_000, trigger: "Collateral ratio drops below maintenance threshold", icon: "🛡️", + available: true, + requiredFeeds: [], }, { id: 3, @@ -114,6 +124,8 @@ const COVERAGE_TYPES: CoverageTypeCatalogEntry[] = [ maxCoverage: 500_000, trigger: "Covered protocol TVL drops >50% in <1 hour", icon: "🔐", + available: true, + requiredFeeds: [], }, { id: 4, @@ -125,6 +137,8 @@ const COVERAGE_TYPES: CoverageTypeCatalogEntry[] = [ maxCoverage: 2_000, trigger: "Flight delayed > 120 minutes per AviationStack data", icon: "✈️", + available: true, + requiredFeeds: [], }, ]; @@ -142,7 +156,8 @@ export class PolicyService { constructor( private readonly configService: ConfigService, private readonly metricsService: MetricsService, - private readonly parameters?: ProtocolParametersReader + private readonly parameters?: ProtocolParametersReader, + @Optional() private readonly oracleTrust?: OracleTrustService, ) { const stellar = this.configService.get("stellar", { infer: true }); this.server = new rpc.Server(stellar.sorobanRpcUrl); @@ -263,7 +278,15 @@ export class PolicyService { } listTypes(): CoverageTypeCatalogEntry[] { - return COVERAGE_TYPES; + return COVERAGE_TYPES.map((entry) => { + const availability = this.oracleTrust?.getCoverageAvailability(COVERAGE_TYPE_VARIANTS[entry.id]); + return { + ...entry, + available: availability?.available ?? true, + requiredFeeds: availability?.requiredFeeds ?? [], + unavailableReason: availability?.reason, + }; + }); } findByHolder(address: string): StoredPolicy[] { @@ -354,6 +377,17 @@ export class PolicyService { error: `coverageType must be between 0 and ${COVERAGE_TYPES.length - 1}`, }); } + const coverageAvailability = this.oracleTrust?.getCoverageAvailability( + COVERAGE_TYPE_VARIANTS[coverageType], + ); + if (coverageAvailability && !coverageAvailability.available) { + throw new BadRequestException({ + error: `${catalogEntry.name} coverage is unavailable`, + requiredFeeds: coverageAvailability.requiredFeeds, + unavailableFeeds: coverageAvailability.unavailableFeeds, + reason: coverageAvailability.reason, + }); + } if (coverageType === FLIGHT_DELAY_COVERAGE_TYPE && typeof triggerParams?.flightNumber !== "string") { throw new BadRequestException({ diff --git a/src/quote/quote.module.ts b/src/quote/quote.module.ts index 6bb392c..5427f28 100644 --- a/src/quote/quote.module.ts +++ b/src/quote/quote.module.ts @@ -1,10 +1,11 @@ import { Module } from "@nestjs/common"; import { PoolModule } from "../pool/pool.module"; +import { OracleTrustModule } from "../oracle/oracle-trust.module"; import { QuoteController } from "./quote.controller"; import { QuoteService } from "./quote.service"; @Module({ - imports: [PoolModule], + imports: [PoolModule, OracleTrustModule], controllers: [QuoteController], providers: [QuoteService], exports: [QuoteService], diff --git a/src/quote/quote.service.spec.ts b/src/quote/quote.service.spec.ts index b32910e..ede0aaf 100644 --- a/src/quote/quote.service.spec.ts +++ b/src/quote/quote.service.spec.ts @@ -2,6 +2,7 @@ import { BadRequestException } from "@nestjs/common"; import { QuoteService } from "./quote.service"; import { CoverageTypeName } from "./coverage-type"; import { CreateQuoteDto } from "./dto/create-quote.dto"; +import { OracleTrustService } from "../oracle/oracle-trust.service"; function buildDto(overrides: Partial = {}): CreateQuoteDto { return { @@ -23,6 +24,21 @@ describe("QuoteService", () => { it("returns all five coverage types", () => { expect(service.listCoverageTypes()).toHaveLength(5); }); + + it("marks a coverage unavailable until its required feed is healthy and confirmed", () => { + const trust = new OracleTrustService(); + const guardedService = new QuoteService(trust); + + expect(guardedService.listCoverageTypes()[0].available).toBe(false); + expect(() => + guardedService.createQuote(buildDto()), + ).toThrow(BadRequestException); + + trust.recordObservation("coingecko-usdc", true); + trust.recordPublication("coingecko-usdc", true); + expect(guardedService.listCoverageTypes()[0].available).toBe(true); + expect(() => guardedService.createQuote(buildDto())).not.toThrow(); + }); }); describe("createQuote", () => { diff --git a/src/quote/quote.service.ts b/src/quote/quote.service.ts index 3ab50b1..a9fcead 100644 --- a/src/quote/quote.service.ts +++ b/src/quote/quote.service.ts @@ -1,4 +1,4 @@ -import { BadRequestException, Injectable } from "@nestjs/common"; +import { BadRequestException, Injectable, Optional } from "@nestjs/common"; import { COVERAGE_TYPES, coverageTypeByName } from "../common/coverage-types"; import { CoverageTypeName } from "./coverage-type"; import { @@ -9,6 +9,7 @@ import { } from "./oracle-confidence"; import { CompareQuotesDto } from "./dto/compare-quotes.dto"; import { CreateQuoteDto } from "./dto/create-quote.dto"; +import { OracleTrustService } from "../oracle/oracle-trust.service"; export interface QuoteResult { coverageType: CoverageTypeName; @@ -49,6 +50,9 @@ export interface CoverageTypeInfo { riskLevel: string; maxDuration: number; riskMultiplier: number; + available: boolean; + requiredFeeds: string[]; + unavailableReason?: string; } const DEFAULT_THRESHOLDS: Record = { @@ -65,6 +69,8 @@ const BASE_RATE = 0.03; // 3% annual base premium export class QuoteService { private readonly confidenceConfig = oracleConfidenceConfigFromEnv(process.env); + constructor(@Optional() private readonly oracleTrust?: OracleTrustService) {} + private calcPremium(coverageAmount: number, riskMultiplier: number, durationDays: number): number { const annualPremium = coverageAmount * BASE_RATE * riskMultiplier; const dailyPremium = annualPremium / 365; @@ -87,6 +93,15 @@ export class QuoteService { if (!catalogEntry) { throw new BadRequestException({ error: `Unknown coverage type: ${coverageType}` }); } + const availability = this.oracleTrust?.getCoverageAvailability(coverageType); + if (availability && !availability.available) { + throw new BadRequestException({ + error: `${catalogEntry.name} coverage is unavailable`, + requiredFeeds: availability.requiredFeeds, + unavailableFeeds: availability.unavailableFeeds, + reason: availability.reason, + }); + } if (durationDays > catalogEntry.maxDuration) { throw new BadRequestException({ error: `${catalogEntry.name} coverage is limited to ${catalogEntry.maxDuration} day(s)`, @@ -163,14 +178,22 @@ export class QuoteService { } listCoverageTypes(): CoverageTypeInfo[] { - return COVERAGE_TYPES.map(({ key, name, description, trigger, riskLevel, maxDuration, riskMultiplier }) => ({ - id: key, - name, - description, - trigger, - riskLevel, - maxDuration, - riskMultiplier, - })); + return COVERAGE_TYPES.map( + ({ key, name, description, trigger, riskLevel, maxDuration, riskMultiplier }) => { + const availability = this.oracleTrust?.getCoverageAvailability(key); + return { + id: key, + name, + description, + trigger, + riskLevel, + maxDuration, + riskMultiplier, + available: availability?.available ?? true, + requiredFeeds: availability?.requiredFeeds ?? [], + unavailableReason: availability?.reason, + }; + }, + ); } }