Feature/oracle provenance worker

README.md

@@ -165,15 +165,43 @@ copy `.env.example` to `.env`, install Node.js 22 dependencies, apply
 
 Stop the stack with `docker compose down`. Persistent database and Redis data
 remain in Docker volumes; `docker compose down -v` removes them.
+
+### API and worker tiers
+
+The backend runs as two separately deployable processes:
+
+| Process | Entrypoint | `PROCESS_ROLE` | Does |
+|---|---|---|---|
+| API | `dist/main.js` | `api` | REST reads + the `/oracle` WebSocket feed |
+| Worker | `dist/worker.js` | `worker` | oracle polling, consensus, on-chain publishing, claim settlement |
+
+The worker publishes readings to Redis (`REDIS_URL` is required for a split
+deployment); the API relays them to WebSocket clients and serves the latest
+reading per coverage type. Each tier can be scaled and restarted on its own,
+and a burst of API traffic cannot delay a payout.
+
+- Run **one** worker per relayer key — the relayer's sequence numbers are
+  tracked in-process.
+- The worker exposes `GET /health` (503 when a pipeline stage stops
+  succeeding for `WORKER_STALE_AFTER_MS`, default 10 min) and `GET /metrics`
+  on `WORKER_HEALTH_PORT` (default 4002).
+- Leaving `PROCESS_ROLE` unset runs both tiers in one process (`all`). That
+  is the default so existing deployments keep paying out, and it is handy
+  locally, but deployed environments should split.
+
+`docker compose up` starts both tiers.
 ## Scripts
 
 | Command | Purpose |
 |---|---|
 | `npm run dev` | Hot-reloading dev server (ts-node + nodemon) |
 | `npm run build` | Compile TypeScript to `dist/` |
 | `npm start` | Run the compiled server |
+| `npm run start:worker` / `start:worker:prod` | Run the worker tier (dev / compiled) |
 | `npm run typecheck` | `tsc --noEmit` |
 | `npm run lint` | ESLint over `src/` |
+| `npm test` | Jest unit tests (includes property-based tests) |
+| `npm run test:property` | Only the property-based / fuzz suites (`*.property.spec.ts`); set `FC_NUM_RUNS=20000` for a deeper soak, `FC_SEED=<seed>` to replay a failure |
 
 ## Operator CLI
 

docker-compose.yml

@@ -1,3 +1,20 @@
+x-backend-env: &backend-env
+  FRONTEND_URL: ${FRONTEND_URL:-http://localhost:3000}
+  DATABASE_URL: postgresql://refract:${POSTGRES_PASSWORD:-refract_dev}@postgres:5432/refract
+  REDIS_URL: redis://redis:6379
+  STELLAR_NETWORK: ${STELLAR_NETWORK:-testnet}
+  SOROBAN_RPC_URL: ${SOROBAN_RPC_URL:-https://soroban-testnet.stellar.org}
+  STELLAR_NETWORK_PASSPHRASE: ${STELLAR_NETWORK_PASSPHRASE:-Test SDF Network ; September 2015}
+  REFRACT_POOL_CONTRACT_ID: ${REFRACT_POOL_CONTRACT_ID:-}
+  REFRACT_POLICY_CONTRACT_ID: ${REFRACT_POLICY_CONTRACT_ID:-}
+  REFRACT_ORACLE_CONTRACT_ID: ${REFRACT_ORACLE_CONTRACT_ID:-}
+  ORACLE_RELAYER_SECRET: ${ORACLE_RELAYER_SECRET:-}
+  COINGECKO_BASE_URL: ${COINGECKO_BASE_URL:-https://api.coingecko.com/api/v3}
+  STELLAR_HORIZON_URL: ${STELLAR_HORIZON_URL:-https://horizon-testnet.stellar.org}
+  DEFILLAMA_BASE_URL: ${DEFILLAMA_BASE_URL:-https://api.llama.fi}
+  DEFILLAMA_PROTOCOL_SLUG: ${DEFILLAMA_PROTOCOL_SLUG:-aave}
+  ORACLE_HTTP_TIMEOUT_MS: ${ORACLE_HTTP_TIMEOUT_MS:-5000}
+
 services:
   postgres:
     image: postgres:16-alpine
@@ -61,22 +78,11 @@ services:
     # before Docker sends SIGKILL (the default is only 10s).
     stop_grace_period: 40s
     environment:
+      <<: *backend-env
       PORT: 4001
-      FRONTEND_URL: ${FRONTEND_URL:-http://localhost:3000}
-      DATABASE_URL: postgresql://refract:${POSTGRES_PASSWORD:-refract_dev}@postgres:5432/refract
-      REDIS_URL: redis://redis:6379
-      STELLAR_NETWORK: ${STELLAR_NETWORK:-testnet}
-      SOROBAN_RPC_URL: ${SOROBAN_RPC_URL:-https://soroban-testnet.stellar.org}
-      STELLAR_NETWORK_PASSPHRASE: ${STELLAR_NETWORK_PASSPHRASE:-Test SDF Network ; September 2015}
-      REFRACT_POOL_CONTRACT_ID: ${REFRACT_POOL_CONTRACT_ID:-}
-      REFRACT_POLICY_CONTRACT_ID: ${REFRACT_POLICY_CONTRACT_ID:-}
-      REFRACT_ORACLE_CONTRACT_ID: ${REFRACT_ORACLE_CONTRACT_ID:-}
-      ORACLE_RELAYER_SECRET: ${ORACLE_RELAYER_SECRET:-}
-      COINGECKO_BASE_URL: ${COINGECKO_BASE_URL:-https://api.coingecko.com/api/v3}
-      STELLAR_HORIZON_URL: ${STELLAR_HORIZON_URL:-https://horizon-testnet.stellar.org}
-      DEFILLAMA_BASE_URL: ${DEFILLAMA_BASE_URL:-https://api.llama.fi}
-      DEFILLAMA_PROTOCOL_SLUG: ${DEFILLAMA_PROTOCOL_SLUG:-aave}
-      ORACLE_HTTP_TIMEOUT_MS: ${ORACLE_HTTP_TIMEOUT_MS:-5000}
+      # API tier: serves reads and the WebSocket feed only. The payout
+      # pipeline runs in the `worker` service below.
+      PROCESS_ROLE: api
       SHUTDOWN_GRACE_PERIOD_MS: ${SHUTDOWN_GRACE_PERIOD_MS:-30000}
       ORACLE_SHUTDOWN_DRAIN_MS: ${ORACLE_SHUTDOWN_DRAIN_MS:-20000}
       ORACLE_EVIDENCE_SPOOL_DIR: /var/lib/refract/oracle-evidence
@@ -98,6 +104,32 @@ services:
       retries: 3
       start_period: 10s
 
+  worker:
+    # Oracle polling, consensus, on-chain publishing and claim settlement.
+    # Runs exactly one replica per relayer key (sequence numbers are managed
+    # in-process). Restarts independently of the API.
+    build:
+      context: .
+      dockerfile: Dockerfile
+    command: ["node", "dist/worker.js"]
+    environment:
+      <<: *backend-env
+      PROCESS_ROLE: worker
+      WORKER_HEALTH_PORT: 4002
+    ports:
+      - "127.0.0.1:4002:4002"
+    depends_on:
+      postgres:
+        condition: service_healthy
+      redis:
+        condition: service_healthy
+    healthcheck:
+      test: ["CMD", "node", "-e", "fetch('http://127.0.0.1:4002/health').then(r=>process.exit(r.ok?0:1)).catch(()=>process.exit(1))"]
+      interval: 30s
+      timeout: 3s
+      retries: 3
+      start_period: 30s
+
 volumes:
   postgres-data:
   redis-data:

docs/oracle-provenance.md

@@ -0,0 +1,69 @@
+# Oracle reading provenance
+
+Each oracle reading the relayer publishes carries a packed `u128` next to the
+value, so contracts and auditors can tell how well-supported the value is
+without leaving the ledger:
+
+```
+update_reading(coverage_type: u32, value: i128, provenance: u128)
+```
+
+The backend side lives in `src/oracle/oracle-encoding.ts`
+(`packProvenance`, `unpackProvenance`, `provenanceFromDecision`,
+`buildUpdateReadingArgs`). The contract must accept the third argument
+before the relayer switches to it.
+
+## Layout
+
+Most significant bits first:
+
+| bits     | width | field                                     |
+|----------|-------|-------------------------------------------|
+| 127..124 | 4     | layout version (currently `1`)            |
+| 123..120 | 4     | trust tier                                |
+| 119..112 | 8     | agreeing source count                     |
+| 111..104 | 8     | reserved, always `0`                      |
+| 103..64  | 40    | source timestamp, Unix seconds            |
+| 63..0    | 64    | consensus decision id                     |
+
+A provenance of `0` (version 0) means "no provenance" and is never produced.
+
+### Trust tiers
+
+| value | tier           | meaning                                                         |
+|-------|----------------|-----------------------------------------------------------------|
+| 0     | `Unverified`   | no consensus — never published                                  |
+| 1     | `SingleSource` | exactly one source agreed (only if policy allows one source)    |
+| 2     | `Corroborated` | two or more sources agreed and the weighted quorum was met      |
+| 3     | `Unanimous`    | three or more sources, all fresh, valid and in agreement        |
+
+The encoder rejects inconsistent combinations (e.g. `Corroborated` with one
+source), so consumers can rely on `tier >= 2` meaning "at least two sources".
+
+### Source timestamp
+
+The earliest observation time among the agreeing sources, floored to whole
+seconds. Using the earliest one means the value never looks fresher than its
+oldest supporting input.
+
+### Decision id
+
+The first 64 bits of a SHA-256 over the round's sorted inputs, the consensus
+policy, the outcome and the evaluation time (`computeDecisionId` in
+`src/oracle/consensus/consensus.ts`). Re-running consensus over the logged
+inputs reproduces the id, which ties an on-chain value back to the exact
+observations behind it.
+
+## Decoding in a contract (Rust sketch)
+
+```rust
+let version  = (p >> 124) & 0xF;
+let tier     = (p >> 120) & 0xF;
+let sources  = (p >> 112) & 0xFF;
+let ts_secs  = ((p >> 64) & 0xFF_FFFF_FFFF) as u64;
+let decision = (p & 0xFFFF_FFFF_FFFF_FFFF) as u64;
+assert!(version == 1);
+```
+
+A consumer that only wants corroborated data can require `tier >= 2`; one
+that cares about freshness can compare `ts_secs` with the ledger timestamp.

package.json

@@ -1,27 +1,41 @@
 {
-  "name": "refract",
+  "name": "refract-protocol-backend",
   "version": "0.1.0",
-  "private": true,
-  "description": "Refract oracle network monorepo",
+  "description": "Refract Protocol backend \u2014 oracle monitoring, claim processing, premium quoting",
   "main": "dist/main.js",
+  "private": true,
   "workspaces": [
     "refract-backend",
     "refract-frontend",
     "refract-contracts"
   ],
   "scripts": {
-    "build": "npm run build --workspaces --if-present",
-    "dev": "nodemon --exec ts-node src/main.ts",
-    "start": "node dist/main.js",
-    "typecheck": "tsc -p tsconfig.json --noEmit",
-    "test": "npm run test:unit && npm run test:contracts",
+    "build": "nest build",
+    "start": "nest start",
+    "start:dev": "nest start --watch",
+    "dev": "nest start --watch",
+    "start:prod": "node dist/main",
+    "start:worker": "nest start --entryFile worker",
+    "start:worker:dev": "nest start --watch --entryFile worker",
+    "start:worker:prod": "node dist/worker",
+    "lint": "eslint \"src/**/*.ts\"",
+    "typecheck": "tsc --noEmit",
+    "test": "jest",
+    "test:watch": "jest --watch",
+    "test:cov": "jest --coverage",
+    "test:property": "jest --testRegex '.*\\.property\\.spec\\.ts$'",
+    "test:integration": "jest --config test/integration/jest.integration.config.ts --runInBand",
     "test:unit": "npm run test --workspaces --if-present",
     "test:contracts": "node test/contracts/verify.js",
     "test:contracts:update": "node test/contracts/verify.js --update",
     "attack-cost": "node scripts/attack-cost.js",
     "attack-cost:reference": "node scripts/attack-cost.js --fixture test/fixtures/attack-cost/reference.json",
-    "lint": "npm run lint --workspaces --if-present",
-    "ci": "npm run lint && npm run test"
+    "ci": "npm run lint && npm run test",
+    "migrate:up": "node-pg-migrate up --migrations-dir migrations --config-file database.json",
+    "migrate:down": "node-pg-migrate down --migrations-dir migrations --config-file database.json",
+    "migrate:create": "node-pg-migrate create --migrations-dir migrations --config-file database.json",
+    "migrate:status": "node-pg-migrate status --migrations-dir migrations --config-file database.json",
+    "regenerate:xdr-fixtures": "ts-node scripts/regenerate-xdr-fixtures.ts"
   },
   "engines": {
     "node": ">=18"
@@ -89,16 +103,28 @@
     "testMatch": [
       "**/src/**/*.spec.ts"
     ],
-    "moduleFileExtensions": ["js", "json", "ts"],
+    "moduleFileExtensions": [
+      "js",
+      "json",
+      "ts"
+    ],
     "transform": {
-      "^.+\\.(t|j)s$": "ts-jest",
       "^.+\\.ts$": "ts-jest",
-      "^.+\\.js$": "babel-jest"
+      "^.+\\.js$": [
+        "babel-jest",
+        {
+          "plugins": [
+            "@babel/plugin-transform-modules-commonjs"
+          ]
+        }
+      ]
     },
     "transformIgnorePatterns": [
-      "/node_modules/(?!(@stellar|@noble|@scure|stellar-base|bignumber.js|crc|tweetnacl|base32.js|uint8array-extras|@adraffy|@noble/curves)/)"
+      "^(?!.*/node_modules/(@noble|uint8array-extras)/).*/node_modules/"
+    ],
+    "collectCoverageFrom": [
+      "**/*.(t|j)s"
     ],
-    "collectCoverageFrom": ["**/*.(t|j)s"],
     "coverageDirectory": "../coverage"
   }
 }

scripts/regenerate-xdr-fixtures.ts

@@ -19,6 +19,7 @@ import {
   encodeProvideCapitalArgs,
   encodeWithdrawCapitalArgs,
 } from "../src/stellar/scval-encoders";
+import { TrustTier, buildUpdateReadingArgs, encodeReadingForChain } from "../src/oracle/oracle-encoding";
 
 const outDir = path.join(__dirname, "../src/stellar/xdr-fixtures");
 
@@ -439,6 +440,16 @@ writeB64("oracle-publish-feed.xdr.b64", feed);
 writeB64("oracle-publish-value.xdr.b64", value);
 writeB64("oracle-publish-timestamp.xdr.b64", ts);
 
+const [, , provenance] = buildUpdateReadingArgs(
+  encodeReadingForChain("StablecoinDepeg", 0.9998, {
+    trustTier: TrustTier.Corroborated,
+    agreeingSources: 3,
+    decisionId: "0123456789abcdef",
+    sourceTimestamp: 1_700_000_000,
+  })
+);
+writeB64("oracle-update-reading-provenance.xdr.b64", provenance, "u128 provenance: Corroborated, 3 sources, ts=1700000000");
+
 // silence unused import if tree-shaken oddly
 void Address;
 void nativeToScVal;

src/app.module.ts

@@ -29,6 +29,8 @@ import { AuthModule } from './auth/auth.module';
 import { ApiKeyGuard } from './auth/api-key.guard';
 import { SecurityReportsModule } from './security-reports/security-reports.module';
 import { DigestModule } from './digest/digest.module';
+import { ReadingBusModule } from './oracle/reading-bus.module';
+import { OracleFeedModule } from './oracle/oracle-feed.module';
 import { OracleIngestionModule } from './oracle/sources/oracle-ingestion.module';
 import { OracleRegionsModule } from './oracle/regions/oracle-regions.module';
 import { ReevaluationModule } from './claim/reevaluation/reevaluation.module';
@@ -87,6 +89,9 @@ export const isSandboxMode = (): boolean =>
     // Global: per-adapter endpoint lists with health-based failover.
     ProviderEndpointsModule,
     OracleModule,
+    // Read side of the oracle: relays readings the worker publishes.
+    ReadingBusModule,
+    OracleFeedModule,
     OracleBootstrapModule,
     OracleAdminModule,
     ConsensusModule,

src/claim/claim.scheduler.ts

@@ -1,9 +1,13 @@
-import { Injectable, Logger, OnApplicationShutdown, OnModuleInit } from '@nestjs/common';
+import { Injectable, Logger, OnApplicationShutdown, OnModuleInit, Optional } from '@nestjs/common';
 import { Cron, CronExpression } from '@nestjs/schedule';
 import { ClaimService, ScanOutcome } from './claim.service';
 import { ProtocolParametersService } from '../parameters/protocol-parameters.service';
+import { runsPipeline } from '../runtime/process-role';
+import { PipelineHeartbeat } from '../runtime/pipeline-heartbeat';
 import { LeaderElector } from '../leader/leader-elector.service';
 
+export const CLAIM_PIPELINE_STAGE = 'claim-settlement';
+
 /**
  * Upper bound (ms) we are willing to wait for an in-flight settlement run to
  * finish before letting the process shut down. Kept in sync with the
@@ -42,17 +46,27 @@ export class ClaimScheduler implements OnModuleInit, OnApplicationShutdown {
   constructor(
     private readonly claimService: ClaimService,
     private readonly protocolParameters: ProtocolParametersService,
+    @Optional() private readonly heartbeat?: PipelineHeartbeat,
     private readonly leaderElector: LeaderElector,
   ) {}
 
   async onModuleInit(): Promise<void> {
+    if (!runsPipeline()) {
+      this.logger.log('Claim settlement disabled in this process (PROCESS_ROLE=api)');
+      return;
+    }
+    this.heartbeat?.expect(CLAIM_PIPELINE_STAGE);
     const interval = await this.getIntervalSeconds();
     this.logger.log(`Claim scheduler initialized with interval ${interval}s`);
   }
 
   async onApplicationShutdown(signal?: string): Promise<void> {
     this.shuttingDown = true;
 
+    if (!this.inFlight) {
+      return;
+    }
+
     this.logger.log(
       `ClaimScheduler shutdown (${signal ?? 'unknown'}): waiting up to ${SHUTDOWN_DRAIN_TIMEOUT_MS}ms for in-flight settlement run to finish.`,
     );
@@ -103,6 +117,11 @@ export class ClaimScheduler implements OnModuleInit, OnApplicationShutdown {
 
   @Cron(CronExpression.EVERY_MINUTE)
   async handleClaimSettlement(): Promise<void> {
+    // Settlement triggers payouts, so it only runs in the worker tier.
+    if (!runsPipeline()) {
+      return;
+    }
+
     if (this.shuttingDown) {
       this.logger.log(
         'Skipping claim settlement run: scheduler is shutting down.',
@@ -150,6 +169,7 @@ export class ClaimScheduler implements OnModuleInit, OnApplicationShutdown {
       if (outcome.settled.length > 0) {
         this.logger.log(`Settled ${outcome.settled.length} claims this cycle`);
       }
+      this.heartbeat?.beat(CLAIM_PIPELINE_STAGE);
     } catch (error) {
       this.logger.error("Claim scan failed", error instanceof Error ? error.stack : String(error));
     }