feat: on-chain vs persisted vault reconciliation job

Add reconciliation job to detect drift between persisted vault state and on-chain state from Soroban contract.

- Add get_vault function to soroban.ts for reading on-chain vault state
- Add reconcileVaults() to transactionETL.ts for batched vault comparison
- Add vault.reconcile job type and handler
- Integrate drift anomaly reporting via abuse-monitor taxonomy
- Document reconciliation process in docs/etl.md
- Add comprehensive test coverage for reconciliation functionality

The reconciliation job is read-only, batched, idempotent, and resumable.
It reports drift without auto-correcting to allow manual intervention.

Author: Simongodw

docs/etl.md

@@ -119,3 +119,111 @@ npx jest --testPathPattern="etlWorker|transactionETL" --run
 # With coverage
 npx jest --testPathPattern="etlWorker|transactionETL" --coverage --run
 ```
+
+---
+
+## Vault State Reconciliation
+
+### Overview
+
+The vault reconciliation job compares persisted vault state in the database with on-chain state from the Soroban contract to detect drift caused by missed Horizon events or failed ETL batches. This is a **read-only** operation that reports anomalies without auto-correcting them.
+
+### Reconciliation Job
+
+**Job Type**: `vault.reconcile`
+
+**Handler**: `TransactionETLService.reconcileVaults()`
+
+**Schedule**: Run periodically (e.g., daily or hourly) via the job system.
+
+### How It Works
+
+1. **Fetches persisted vaults** from the `vaults` table (all vaults or a subset via `vaultIds` option)
+2. **Reads on-chain state** via `soroban.getVault()` for each vault in batches
+3. **Compares key fields**:
+   - `status` (normalized for case-insensitive comparison)
+   - `amount`
+   - `verifier` address
+   - `success_destination` address
+   - `failure_destination` address
+4. **Reports drift** via structured logs using the abuse-monitor taxonomy:
+   - `vault.vault_missing_onchain` - vault exists in DB but not on-chain
+   - `vault.vault_state_drift` - one or more fields differ between DB and on-chain
+   - `vault.vault_reconciliation_error` - error during reconciliation (e.g., RPC timeout)
+
+### Job Payload
+
+```typescript
+interface VaultReconcileJobPayload {
+  vaultIds?: string[]    // Optional: specific vault IDs to reconcile
+  batchSize?: number     // Optional: batch size for RPC calls (default: 50)
+}
+```
+
+### Example Usage
+
+```ts
+// Enqueue a reconciliation job for all vaults
+await enqueueJob('vault.reconcile', { batchSize: 50 })
+
+// Enqueue a reconciliation job for specific vaults
+await enqueueJob('vault.reconcile', { vaultIds: ['vault-1', 'vault-2'] })
+```
+
+### Result
+
+The reconciliation returns:
+
+```typescript
+{
+  totalVaults: number      // Total vaults to reconcile
+  checked: number          // Successfully checked vaults
+  driftDetected: number    // Vaults with state drift
+  missingOnChain: number   // Vaults missing on-chain
+  errors: number           // Reconciliation errors
+}
+```
+
+### Idempotency and Resumability
+
+- **Idempotent**: Running the same reconciliation multiple times produces consistent results
+- **Resumable**: If aborted via `AbortSignal`, the job can be re-run; no partial state is persisted
+- **Bounded**: Processes vaults in configurable batches to avoid RPC rate limiting
+
+### Configuration
+
+Requires Soroban configuration (same as vault submission):
+
+- `SOROBAN_CONTRACT_ID`
+- `SOROBAN_NETWORK_PASSPHRASE`
+- `SOROBAN_SOURCE_ACCOUNT`
+- `SOROBAN_RPC_URL`
+- `SOROBAN_SECRET_KEY`
+
+If Soroban is not configured, the job skips reconciliation and returns zero counts.
+
+### Drift Response
+
+When drift is detected:
+
+1. **Log the anomaly** with full context (vault ID, drifted fields, persisted vs on-chain values)
+2. **Do NOT auto-correct** - manual intervention is required to replay missed events
+3. **Alert operators** via monitoring systems that consume the structured logs
+
+### Testing
+
+```bash
+# Run reconciliation tests
+bun test src/tests/transactionETL.reconcile.test.ts
+
+# With coverage
+bun test src/tests/transactionETL.reconcile.test.ts --coverage
+```
+
+Test coverage includes:
+- Status drift detection
+- Missing on-chain vault
+- RPC timeout handling
+- Fully consistent run (zero drift)
+- Batch processing with abort signal
+

src/jobs/handlers.ts

@@ -2,6 +2,7 @@ import { NotificationService } from '../services/notifications/factory.js'
 import { processJob as processExportJob } from '../services/exportQueue.js'
 import type { JobHandler, JobType } from './types.js'
 import { markVaultExpiries } from '../services/vault.js'
+import { TransactionETLService } from '../services/transactionETL.js'
 
 type JobHandlerRegistry = {
   [K in JobType]: JobHandler<K>
@@ -59,4 +60,21 @@ export const defaultJobHandlers: JobHandlerRegistry = {
       `exportJobId=${payload.exportJobId} attempt=${context.attempt}`,
     )
   },
+  'vault.reconcile': async (payload, context) => {
+    const etlConfig = {
+      horizonUrl: process.env.HORIZON_URL || 'https://horizon-testnet.stellar.org',
+      networkPassphrase: process.env.STELLAR_NETWORK_PASSPHRASE || 'Test SDF Network ; September 2015',
+      batchSize: payload.batchSize || 50,
+      maxRetries: 3,
+    }
+    const etlService = new TransactionETLService(etlConfig)
+    const result = await etlService.reconcileVaults({
+      vaultIds: payload.vaultIds,
+      batchSize: payload.batchSize,
+    })
+    logJob(
+      'vault.reconcile',
+      `vaultIds=${payload.vaultIds?.length || 'all'} batchSize=${payload.batchSize || 50} checked=${result.checked}/${result.totalVaults} drift=${result.driftDetected} missing=${result.missingOnChain} errors=${result.errors} attempt=${context.attempt}`,
+    )
+  },
 }

src/jobs/types.ts

@@ -4,6 +4,7 @@ export const JOB_TYPES = [
   'oracle.call',
   'analytics.recompute',
   'export.generate',
+  'vault.reconcile',
 ] as const
 
 export type JobType = (typeof JOB_TYPES)[number]
@@ -36,12 +37,18 @@ export interface ExportGenerateJobPayload {
   exportJobId: string
 }
 
+export interface VaultReconcileJobPayload {
+  vaultIds?: string[]
+  batchSize?: number
+}
+
 export interface JobPayloadByType {
   'notification.send': NotificationJobPayload
   'deadline.check': DeadlineCheckJobPayload
   'oracle.call': OracleCallJobPayload
   'analytics.recompute': AnalyticsRecomputeJobPayload
   'export.generate': ExportGenerateJobPayload
+  'vault.reconcile': VaultReconcileJobPayload
 }
 
 export interface JobContext {
@@ -114,6 +121,11 @@ export const isPayloadForJobType = (
       )
     case 'export.generate':
       return isNonEmptyString(payload.exportJobId)
+    case 'vault.reconcile':
+      return (
+        (payload.vaultIds === undefined || Array.isArray(payload.vaultIds)) &&
+        (payload.batchSize === undefined || typeof payload.batchSize === 'number')
+      )
     default:
       return false
   }

src/security/abuse-monitor.ts

@@ -193,6 +193,20 @@ export function __resetSecurityMonitorForTests(): void {
   processedEvents = 0
 }
 
+// ─── Vault drift anomaly logging ───────────────────────────────────────────
+
+type VaultDriftEventType =
+  | 'vault_missing_onchain'
+  | 'vault_state_drift'
+  | 'vault_reconciliation_error'
+
+export function logVaultDriftAnomaly(
+  event: VaultDriftEventType,
+  data: Record<string, unknown>,
+): void {
+  logSecurityEvent(`vault.${event}`, data)
+}
+
 function getIpState(ip: string, now: number): IpState {
   const existing = ipStates.get(ip)
   if (existing) {

src/services/soroban.ts

@@ -48,6 +48,19 @@ export interface SorobanClient {
     config: SorobanConfig,
     args: Record<string, unknown>,
   ): Promise<{ txHash: string }>
+  getVault(
+    config: SorobanConfig,
+    vaultId: string,
+  ): Promise<OnChainVaultState | null>
+}
+
+export interface OnChainVaultState {
+  vault_id: string
+  amount: string
+  verifier: string
+  success_destination: string
+  failure_destination: string
+  status: 'active' | 'completed' | 'failed' | 'cancelled'
 }
 
 /**
@@ -116,6 +129,41 @@ export const defaultSorobanClient: SorobanClient = {
 
     return { txHash: response.hash }
   },
+  async getVault(config, vaultId) {
+    const {
+      Contract,
+      rpc: SorobanRpc,
+      nativeToScVal,
+      scValToNative,
+    } = await import('@stellar/stellar-sdk')
+
+    const server = new SorobanRpc.Server(config.rpcUrl)
+    const contract = new Contract(config.contractId)
+
+    try {
+      const callOp = contract.call('get_vault', nativeToScVal(vaultId, { type: 'string' }))
+
+      const result = await server.simulateTransaction(callOp)
+
+      if (result.result === undefined || result.result === null) {
+        return null
+      }
+
+      const decoded = scValToNative(result.result)
+
+      return {
+        vault_id: decoded.vault_id || vaultId,
+        amount: decoded.amount || '0',
+        verifier: decoded.verifier || '',
+        success_destination: decoded.success_destination || '',
+        failure_destination: decoded.failure_destination || '',
+        status: decoded.status || 'active',
+      }
+    } catch (error) {
+      log('error', 'soroban.get_vault_error', { vaultId, error: error instanceof Error ? error.message : 'Unknown error' })
+      return null
+    }
+  },
 }
 
 // Allow overriding the client (for tests)
@@ -129,6 +177,8 @@ export const resetSorobanClient = (): void => {
   _client = defaultSorobanClient
 }
 
+export const getSorobanClient = (): SorobanClient => _client
+
 // ─── Structured logging helper (no PII) ─────────────────────────────────────
 
 const log = (level: 'info' | 'warn' | 'error', event: string, data: Record<string, unknown> = {}): void => {