fix(backend): add graceful degradation for RPC outages (#1162)

.github/workflows/backend-governance.yml

@@ -49,6 +49,10 @@ jobs:
           cache: npm
           cache-dependency-path: backend/package-lock.json
 
+      - name: Build shared API schemas
+        run: npm ci && npm run build
+        working-directory: packages/api-schemas
+
       - name: Install dependencies
         run: npm ci
 

.github/workflows/rust-security.yml

@@ -49,7 +49,7 @@ jobs:
         working-directory: frontend
 
       - name: Install cargo-audit
-        run: cargo install cargo-audit
+        run: cargo install cargo-audit --version 0.22.1 --locked
 
       - name: Run cargo audit
         run: |
@@ -96,7 +96,7 @@ jobs:
         uses: dtolnay/rust-toolchain@nightly
 
       - name: Install cargo-fuzz
-        run: cargo install cargo-fuzz --locked
+        run: cargo +nightly install cargo-fuzz --locked
 
       - name: Run vault share-price fuzz (60s)
         run: |

backend/package.json

@@ -30,18 +30,18 @@
   "author": "",
   "license": "MIT",
   "dependencies": {
-    "@yieldvault/api-schemas": "file:../packages/api-schemas",
     "@aws-sdk/client-s3": "^3.1058.0",
-    "@opentelemetry/exporter-trace-otlp-http": "^0.218.0",
+    "@opentelemetry/exporter-trace-otlp-http": "^0.221.0",
     "@opentelemetry/instrumentation-express": "^0.66.0",
-    "@opentelemetry/instrumentation-http": "^0.218.0",
+    "@opentelemetry/instrumentation-http": "^0.221.0",
     "@opentelemetry/resources": "^2.7.1",
-    "@opentelemetry/sdk-node": "^0.218.0",
+    "@opentelemetry/sdk-node": "^0.221.0",
     "@opentelemetry/semantic-conventions": "^1.41.1",
     "@prisma/client": "^5.10.0",
     "@prisma/instrumentation": "^7.8.0",
     "@stellar/stellar-base": "^13.1.0",
     "@stellar/stellar-sdk": "^13.0.0",
+    "@yieldvault/api-schemas": "file:../packages/api-schemas",
     "cors": "^2.8.6",
     "decimal.js": "^10.6.0",
     "dotenv": "^16.3.1",

backend/prisma/migrations/20260824204600_add_event_outbox/migration.sql

@@ -0,0 +1,88 @@
+-- CreateTable
+CREATE TABLE "AdminConfigChange" (
+    "id" TEXT NOT NULL PRIMARY KEY,
+    "configType" TEXT NOT NULL,
+    "action" TEXT NOT NULL,
+    "actor" TEXT NOT NULL,
+    "ipAddress" TEXT,
+    "userAgent" TEXT,
+    "preChangeSnapshot" TEXT NOT NULL,
+    "postChangeSnapshot" TEXT NOT NULL,
+    "metadata" TEXT NOT NULL,
+    "createdAt" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+-- CreateTable
+CREATE TABLE "FeatureFlagOverride" (
+    "id" TEXT NOT NULL PRIMARY KEY,
+    "flagName" TEXT NOT NULL,
+    "enabled" BOOLEAN NOT NULL,
+    "scopeType" TEXT NOT NULL,
+    "scopeValue" TEXT,
+    "expiresAt" DATETIME NOT NULL,
+    "actor" TEXT NOT NULL,
+    "createdAt" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+-- CreateTable
+CREATE TABLE "EventOutbox" (
+    "id" TEXT NOT NULL PRIMARY KEY,
+    "eventType" TEXT NOT NULL,
+    "payload" TEXT NOT NULL,
+    "status" TEXT NOT NULL DEFAULT 'pending',
+    "aggregateType" TEXT NOT NULL,
+    "aggregateId" TEXT NOT NULL,
+    "attemptCount" INTEGER NOT NULL DEFAULT 0,
+    "maxAttempts" INTEGER NOT NULL DEFAULT 3,
+    "lastError" TEXT,
+    "lockedAt" DATETIME,
+    "lockedBy" TEXT,
+    "createdAt" DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" DATETIME NOT NULL,
+    "relayedAt" DATETIME
+);
+
+-- AlterTable: add optimistic-concurrency version column.
+-- A plain ADD COLUMN with a constant DEFAULT is fully supported by SQLite
+-- and preserves the existing table (and its indexes) in place, unlike
+-- Prisma's default drop-and-rebuild diff for SQLite.
+ALTER TABLE "BulkExportJob" ADD COLUMN "version" INTEGER DEFAULT 1 NOT NULL;
+
+-- AlterTable: add optimistic-concurrency version column (see above).
+ALTER TABLE "VaultState" ADD COLUMN "version" INTEGER DEFAULT 1 NOT NULL;
+
+-- CreateIndex
+CREATE INDEX "AdminConfigChange_configType_idx" ON "AdminConfigChange"("configType");
+
+-- CreateIndex
+CREATE INDEX "AdminConfigChange_createdAt_idx" ON "AdminConfigChange"("createdAt");
+
+-- CreateIndex
+CREATE INDEX "AdminConfigChange_actor_idx" ON "AdminConfigChange"("actor");
+
+-- CreateIndex
+CREATE INDEX "FeatureFlagOverride_flagName_idx" ON "FeatureFlagOverride"("flagName");
+
+-- CreateIndex
+CREATE INDEX "FeatureFlagOverride_scopeType_scopeValue_idx" ON "FeatureFlagOverride"("scopeType", "scopeValue");
+
+-- CreateIndex
+CREATE INDEX "FeatureFlagOverride_expiresAt_idx" ON "FeatureFlagOverride"("expiresAt");
+
+-- CreateIndex
+CREATE INDEX "FeatureFlagOverride_actor_idx" ON "FeatureFlagOverride"("actor");
+
+-- CreateIndex
+CREATE INDEX "EventOutbox_status_idx" ON "EventOutbox"("status");
+
+-- CreateIndex
+CREATE INDEX "EventOutbox_status_createdAt_idx" ON "EventOutbox"("status", "createdAt");
+
+-- CreateIndex
+CREATE INDEX "EventOutbox_aggregateType_aggregateId_idx" ON "EventOutbox"("aggregateType", "aggregateId");
+
+-- CreateIndex
+CREATE INDEX "EventOutbox_lockedAt_idx" ON "EventOutbox"("lockedAt");
+
+-- CreateIndex
+CREATE INDEX "EventOutbox_createdAt_idx" ON "EventOutbox"("createdAt");

backend/schema-snapshots/get-_api_v1_vault_summary.json

@@ -2,10 +2,13 @@
   "type": "object",
   "properties": {
     "totalAssets": {
-      "type": "number"
+      "type": "string"
     },
     "totalShares": {
-      "type": "number"
+      "type": "string"
+    },
+    "sharePrice": {
+      "type": "string"
     },
     "apy": {
       "type": "number"
@@ -17,6 +20,7 @@
   "required": [
     "totalAssets",
     "totalShares",
+    "sharePrice",
     "apy",
     "timestamp"
   ],

backend/schema-snapshots/get-_health.json

@@ -13,6 +13,9 @@
     "environment": {
       "type": "string"
     },
+    "lastIndexedLedger": {
+      "type": "number"
+    },
     "checks": {
       "type": "object",
       "properties": {
@@ -117,6 +120,7 @@
     "timestamp",
     "uptime",
     "environment",
+    "lastIndexedLedger",
     "checks",
     "sorobanCircuitBreaker"
   ],

backend/scripts/check-migrations.js

@@ -80,14 +80,17 @@ function checkFile(file) {
   // Safe opt-out: add `-- migration-safety: canary-safe` at the top of the
   // file to acknowledge the risk (e.g. zero-downtime step 2 of 3).
   //
-  // The regex looks for ADD COLUMN ... NOT NULL within a 300-char window and
-  // checks that DEFAULT does NOT appear in that window.
-  const addColumnNotNullPattern = /\badd\s+column\b[^;]{0,300}\bnot\s+null\b/gi;
+  // The regex captures the whole ADD COLUMN statement (up to 300 chars or the
+  // next semicolon) and checks that whole window for NOT NULL and DEFAULT,
+  // since either can come first — "NOT NULL DEFAULT x" (Prisma's own
+  // convention) and "DEFAULT x NOT NULL" are equally valid SQL.
+  const addColumnStatementPattern = /\badd\s+column\b[^;]{0,300}/gi;
   let match;
-  while ((match = addColumnNotNullPattern.exec(content)) !== null) {
+  while ((match = addColumnStatementPattern.exec(content)) !== null) {
     const snippet = match[0];
+    const hasNotNull = /\bnot\s+null\b/i.test(snippet);
     const hasDefault = /\bdefault\b/i.test(snippet);
-    if (!hasDefault && !isCanarySafeOptIn) {
+    if (hasNotNull && !hasDefault && !isCanarySafeOptIn) {
       results.push({
         file,
         severity: 'error',

backend/src/__tests__/adminFeatures.test.ts

@@ -5,6 +5,7 @@ import { resetWebhookState } from '../webhookDelivery';
 import { resetAuditLogs } from '../auditLog';
 import { resetTransactionBackfillJobsForTests } from '../transactionBackfill';
 import { resetExportManifestsForTests } from '../exportManifest';
+import { eventOutboxService } from '../eventOutbox';
 
 describe('Admin backend features', () => {
   const adminKey = 'admin-feature-test-key';
@@ -52,7 +53,7 @@ describe('Admin backend features', () => {
       .send({
         amount: '125.00',
         asset: 'USDC',
-        walletAddress: 'GABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz234567',
+        walletAddress: 'G234567ABCDEFGHIJKLMNOPQRSTUVWXYZ234567ABCDEFGHIJKLMNOPQ',
       });
 
     if (typeof previousAllowlistEnabled === 'string') {
@@ -63,7 +64,11 @@ describe('Admin backend features', () => {
 
     expect(depositResponse.status).toBe(201);
 
+    // The deposit handler writes the event to the outbox fire-and-forget, so
+    // wait for that write to land before explicitly draining it — the
+    // background poller isn't guaranteed to run within a fixed test delay.
     await new Promise((resolve) => setTimeout(resolve, 30));
+    await eventOutboxService.processOutbox(10);
 
     const deliveriesResponse = await request(app)
       .get('/admin/webhooks/deliveries')

backend/src/__tests__/api.test.ts

@@ -1,3 +1,11 @@
+// This file has its own dedicated adaptive-throttle escalation test and
+// already resets that middleware's state before every test (below), so it
+// needs the real threshold restored here — setup.ts globally raises
+// ADAPTIVE_THROTTLE_SCORE_THRESHOLD for every other integration test file,
+// most of which have no such per-test reset. Must run before importing
+// '../index', which reads this value once at module-load time.
+process.env.ADAPTIVE_THROTTLE_SCORE_THRESHOLD = '6';
+
 import request from 'supertest';
 import app from '../index';
 import { resetAdaptiveThrottleStateForTests } from '../middleware/adaptiveThrottle';
@@ -141,12 +149,21 @@ describe('Backend API', () => {
     it('should adaptively throttle repeated 4xx abuse patterns per IP', async () => {
       const clientIp = '198.51.100.33';
 
-      for (let i = 0; i < 8; i++) {
+      // 404s are scored at half weight (0.5) vs. other 4xx (1) in
+      // scoreForStatus — deliberately less suspicious than e.g. repeated
+      // 401/403s — so crossing the default threshold of 6 needs at least 12
+      // hits, not 8.
+      for (let i = 0; i < 20; i++) {
         await request(app)
           .get('/api/not-found-abuse')
           .set('x-forwarded-for', clientIp);
       }
 
+      // The middleware scores each response in a fire-and-forget res.on('finish')
+      // handler, not before the response is sent, so the priming loop above can
+      // outrun its own bookkeeping. Give it a beat to settle before relying on it.
+      await new Promise((resolve) => setTimeout(resolve, 100));
+
       const throttled = await request(app)
         .get('/api/not-found-abuse')
         .set('x-forwarded-for', clientIp);

backend/src/__tests__/eventOutbox.test.ts

@@ -448,7 +448,7 @@ describe('EventOutboxService', () => {
 
   describe('end-to-end flow', () => {
     it('completes the full outbox lifecycle', async () => {
-      let deliveredPayloads: unknown[] = [];
+      const deliveredPayloads: unknown[] = [];
       global.fetch = jest.fn(async (_url, init) => {
         if (init?.body && String(init.body).includes('webhook.verification')) {
           const body = JSON.parse(String(init.body));

backend/src/__tests__/eventPollingService.test.ts

@@ -329,6 +329,50 @@ describe('EventPollingService', () => {
       expect(true).toBe(true);
     });
 
+    it('starts the continuous polling loop even when startup replay fails outright', async () => {
+      (prisma.eventCursor.findUnique as jest.Mock).mockResolvedValue({
+        id: 1,
+        lastLedgerSeq: 1000,
+      });
+
+      // getLatestLedger succeeds (there are missed events to replay), but the
+      // subsequent getEvents call for the replay fails outright — simulating
+      // an RPC outage mid-replay rather than a clean "nothing missed" case.
+      (fetch as jest.Mock).mockImplementation((_url, options) => {
+        const body = JSON.parse((options as RequestInit).body as string);
+        if (body.method === 'getLatestLedger') {
+          return Promise.resolve({
+            json: async () => ({ result: { sequence: 1050 } }),
+          });
+        }
+        return Promise.reject(new Error('RPC unavailable'));
+      });
+
+      // start() must not reject: a failed replay should degrade gracefully
+      // instead of aborting before the continuous polling loop is armed.
+      await expect(service.start()).resolves.toBeUndefined();
+
+      // The continuous polling loop should now be armed for this leader —
+      // subsequent successful polls pick up where the failed replay left off.
+      (fetch as jest.Mock).mockReset();
+      (fetch as jest.Mock).mockImplementation((_url, options) => {
+        const body = JSON.parse((options as RequestInit).body as string);
+        if (body.method === 'getLatestLedger') {
+          return Promise.resolve({ json: async () => ({ result: { sequence: 1050 } }) });
+        }
+        return Promise.resolve({ json: async () => ({ result: { events: [] } }) });
+      });
+      (prisma.eventCursor.upsert as jest.Mock).mockResolvedValue({});
+
+      await (service as any).pollEvents();
+
+      expect(prisma.eventCursor.upsert).toHaveBeenCalledWith({
+        where: { id: 1 },
+        update: { lastLedgerSeq: 1050 },
+        create: { id: 1, lastLedgerSeq: 1050 },
+      });
+    });
+
     it('should continue polling after transient errors', async () => {
       (prisma.eventCursor.findUnique as jest.Mock).mockResolvedValue({
         id: 1,