Added the reconcilation worker to run for deploymenment

src/reconciliation/INVARIANTS.md

@@ -50,11 +50,14 @@ Config: `RECON_DRIFT_WINDOW_MS` (default 1h), `RECON_DRIFT_CAP_MINOR`
 > Auto-repair of a `missing_credit` is idempotent — retries never double-credit.
 
 The resolver re-attempts an open mismatch on every pass until it succeeds or hits
-`maxResolutionAttempts`, and the worker fires passes on a fixed interval with no
-overlap guard — so the repair effect can be invoked many times for one mismatch,
-including from two passes at once. `applyIdempotentRepair` (`repair.ts`) keys the
-repair by a deterministic, mismatch-derived key (`repairKey`) and runs the
-credit-posting effect **at most once per key**, even under concurrency:
+`maxResolutionAttempts`. The worker fires passes on a fixed interval with two
+overlap guards:
+- **Intra-process guard**: a slow pass cannot overlap itself (checks `processingPromise` before scheduling)
+- **Cluster-wide leader election**: a Postgres advisory lock (`pg_try_advisory_lock`) ensures only one replica runs passes at a time
+
+These guards prevent concurrent passes from invoking the repair effect for the same mismatch.
+`applyIdempotentRepair` (`repair.ts`) keys the repair by a deterministic, mismatch-derived key
+(`repairKey`) and runs the credit-posting effect **at most once per key**, even under concurrency:
 
 - a completed key short-circuits (the `applied` cache is bounded by size and TTL,
   so a long-running worker cannot grow it without limit);

src/reconciliation/store.ts

@@ -11,6 +11,34 @@ import type {
 } from './types.js'
 import { SLA_HOURS_BY_CLASS } from './types.js'
 
+// ── Leader election (Postgres advisory lock) ───────────────────────────────────
+
+const RECONCILIATION_LEADER_LOCK_KEY = 123456789 // Fixed key for reconciliation leader election
+
+/**
+ * Try to acquire the reconciliation leader lock using Postgres advisory lock.
+ * Returns true if lock was acquired (this instance is now the leader), false otherwise.
+ * The lock is session-level and is automatically released when the session ends or
+ * explicitly via releaseLeaderLock().
+ */
+export async function tryAcquireLeaderLock(): Promise<boolean> {
+  const pool = await getPool()
+  if (!pool) return false
+  const { rows } = await pool.query('SELECT pg_try_advisory_lock($1) AS acquired', [RECONCILIATION_LEADER_LOCK_KEY])
+  return rows[0].acquired as boolean
+}
+
+/**
+ * Release the reconciliation leader lock.
+ * Returns true if lock was released, false if this session did not hold the lock.
+ */
+export async function releaseLeaderLock(): Promise<boolean> {
+  const pool = await getPool()
+  if (!pool) return false
+  const { rows } = await pool.query('SELECT pg_advisory_unlock($1) AS released', [RECONCILIATION_LEADER_LOCK_KEY])
+  return rows[0].released as boolean
+}
+
 // ── helpers ───────────────────────────────────────────────────────────────────
 
 function rowToLedger(row: Record<string, unknown>): LedgerEvent {

src/reconciliation/worker.test.ts

@@ -0,0 +1,227 @@
+/**
+ * Tests for ReconciliationWorker concurrency safety.
+ * 
+ * Tests cover:
+ * - Intra-process overlap guard (slow pass skips next tick)
+ * - Advisory lock (other instances no-op when lock held)
+ * - Leader takeover after holder stops
+ * - Concurrent passes not double-absorbing drift past cap
+ * - Concurrent passes not double-escalating mismatch
+ */
+
+import { describe, it, expect, beforeEach, vi, afterEach } from 'vitest'
+import { ReconciliationWorker } from './worker.js'
+import { tryAcquireLeaderLock, releaseLeaderLock } from './store.js'
+import { resetDrift, configureDrift, tryAbsorbDrift } from './drift.js'
+import { persistMismatch } from './store.js'
+import type { ToleranceRule } from './types.js'
+
+// Mock store functions
+vi.mock('./store.js', () => ({
+  tryAcquireLeaderLock: vi.fn().mockResolvedValue(true),
+  releaseLeaderLock: vi.fn().mockResolvedValue(true),
+  listPendingLedgerEvents: vi.fn().mockResolvedValue([]),
+  persistMismatch: vi.fn().mockResolvedValue({}),
+  markLedgerEventStatus: vi.fn().mockResolvedValue(),
+}))
+
+// Mock engine functions
+vi.mock('./engine.js', () => ({
+  runReconciliationPass: vi.fn().mockResolvedValue({ matched: 0, mismatches: 0, skipped: 0 }),
+}))
+
+// Mock resolver functions
+vi.mock('./resolver.js', () => ({
+  runResolutionPass: vi.fn().mockResolvedValue({ resolved: 0, escalated: 0 }),
+}))
+
+// Mock chain reconciliation
+vi.mock('./chain-reconciliation.js', () => ({
+  reconcileChainPositions: vi.fn().mockResolvedValue({ matched: 0, mismatches: 0, skipped: 0, unknown: 0 }),
+}))
+
+// Mock metrics
+vi.mock('../metrics.js', () => ({
+  recordReconciliationPending: vi.fn(),
+  recordReconciliationProcessed: vi.fn(),
+  recordReconciliationProcessingDuration: vi.fn(),
+  recordToleranceAbsorbed: vi.fn(),
+  recordDriftCapBreach: vi.fn(),
+}))
+
+// Mock Soroban adapter
+vi.mock('../soroban/index.js', () => ({
+  createSorobanAdapter: vi.fn().mockReturnValue({
+    getOnChainPosition: vi.fn().mockResolvedValue({ balanceMinor: 0n, contractType: 'staking_pool', contractId: 'test', account: 'test' }),
+  }),
+}))
+
+vi.mock('../soroban/client.js', () => ({
+  getSorobanConfigFromEnv: vi.fn().mockReturnValue({}),
+}))
+
+import * as store from './store.js'
+import * as engine from './engine.js'
+import * as resolver from './resolver.js'
+
+// Disable chain reconciliation for tests
+const originalEnv = process.env.RECON_CHAIN_ENABLED
+
+beforeEach(() => {
+  process.env.RECON_CHAIN_ENABLED = 'false'
+  vi.clearAllMocks()
+  resetDrift()
+  configureDrift({ windowMs: 3_600_000, capMinor: 100_000n })
+  // Reset mocks to default state
+  vi.mocked(store.tryAcquireLeaderLock).mockResolvedValue(true)
+  vi.mocked(store.releaseLeaderLock).mockResolvedValue(true)
+  vi.mocked(engine.runReconciliationPass).mockResolvedValue({ matched: 0, mismatches: 0, skipped: 0 })
+  vi.mocked(resolver.runResolutionPass).mockResolvedValue({ resolved: 0, escalated: 0 })
+})
+
+afterEach(() => {
+  process.env.RECON_CHAIN_ENABLED = originalEnv
+  vi.restoreAllMocks()
+})
+
+// ── Intra-process overlap guard ─────────────────────────────────────────────
+
+describe('intra-process overlap guard', () => {
+  it('skips tick if previous pass is still running', async () => {
+    const worker = new ReconciliationWorker()
+
+    // Mock a slow pass that takes longer than interval
+    let resolvePass: () => void
+    const slowPass = new Promise<void>(resolve => {
+      resolvePass = resolve
+    })
+    vi.mocked(engine.runReconciliationPass).mockReturnValue(slowPass as any)
+    vi.mocked(resolver.runResolutionPass).mockReturnValue(Promise.resolve({ resolved: 0, escalated: 0 }))
+
+    // Start worker with short interval
+    worker.start(50)
+
+    // Wait for first tick to start
+    await new Promise(resolve => setTimeout(resolve, 60))
+
+    // First tick should have started
+    expect(engine.runReconciliationPass).toHaveBeenCalledTimes(1)
+
+    // Wait for second tick (should be skipped due to guard)
+    await new Promise(resolve => setTimeout(resolve, 60))
+
+    // Should still be 1 (second tick was skipped)
+    expect(engine.runReconciliationPass).toHaveBeenCalledTimes(1)
+
+    // Complete the slow pass
+    resolvePass!()
+    await slowPass
+
+    await worker.stop()
+  })
+})
+
+// ── Advisory lock (cluster-wide leader election) ───────────────────────────
+
+describe('advisory lock', () => {
+  it('skips pass when lock is held by another instance', async () => {
+    // Mock lock acquisition failure (another instance holds lock)
+    vi.mocked(store.tryAcquireLeaderLock).mockResolvedValue(false)
+
+    const worker = new ReconciliationWorker()
+    await worker.poll()
+
+    // Should not run reconciliation pass
+    expect(engine.runReconciliationPass).not.toHaveBeenCalled()
+    expect(resolver.runResolutionPass).not.toHaveBeenCalled()
+  })
+
+  it('runs pass when lock is acquired', async () => {
+    // Mock lock acquisition success
+    vi.mocked(store.tryAcquireLeaderLock).mockResolvedValue(true)
+
+    const worker = new ReconciliationWorker()
+    await worker.poll()
+
+    // Should run reconciliation pass
+    expect(engine.runReconciliationPass).toHaveBeenCalledTimes(1)
+    expect(resolver.runResolutionPass).toHaveBeenCalledTimes(1)
+    expect(store.releaseLeaderLock).toHaveBeenCalledTimes(1)
+  })
+
+  it('releases lock even if pass fails', async () => {
+    vi.mocked(store.tryAcquireLeaderLock).mockResolvedValue(true)
+    vi.mocked(engine.runReconciliationPass).mockRejectedValue(new Error('DB error'))
+
+    const worker = new ReconciliationWorker()
+    await worker.poll()
+
+    // Should still release lock despite error
+    expect(store.releaseLeaderLock).toHaveBeenCalledTimes(1)
+  })
+})
+
+// ── Leader takeover ─────────────────────────────────────────────────────────
+
+describe('leader takeover', () => {
+  it('another instance can acquire lock after holder releases', async () => {
+    // First instance acquires lock
+    vi.mocked(store.tryAcquireLeaderLock).mockResolvedValueOnce(true)
+
+    const worker1 = new ReconciliationWorker()
+    await worker1.poll()
+
+    expect(store.releaseLeaderLock).toHaveBeenCalledTimes(1)
+
+    // Second instance can now acquire lock
+    vi.mocked(store.tryAcquireLeaderLock).mockResolvedValueOnce(true)
+
+    const worker2 = new ReconciliationWorker()
+    await worker2.poll()
+
+    expect(engine.runReconciliationPass).toHaveBeenCalledTimes(2) // Both instances ran
+  })
+})
+
+// ── Drift cap under concurrency ───────────────────────────────────────────────
+
+describe('drift cap under concurrency', () => {
+  it('concurrent passes cannot absorb drift past cap', async () => {
+    const rule: ToleranceRule = { rail: 'paystack', toleranceMinor: 100n, maxDelaySeconds: 3600, maxResolutionAttempts: 3 }
+
+    // Configure drift with small cap
+    configureDrift({ windowMs: 60_000, capMinor: 200n })
+
+    // Simulate two concurrent passes trying to absorb drift
+    const pass1 = tryAbsorbDrift('paystack', 'NGN', 150n, rule)
+    const pass2 = tryAbsorbDrift('paystack', 'NGN', 150n, rule)
+
+    const [result1, result2] = await Promise.all([pass1, pass2])
+
+    // At most one should succeed (total absorbed <= 200n cap)
+    const totalAbsorbed = (result1 ? 150n : 0n) + (result2 ? 150n : 0n)
+    expect(totalAbsorbed).toBeLessThanOrEqual(200n)
+  })
+})
+
+// ── Mismatch escalation under concurrency ───────────────────────────────────
+
+describe('mismatch escalation under concurrency', () => {
+  it('concurrent passes do not double-escalate a single mismatch', async () => {
+    // The advisory lock ensures only one instance runs at a time
+    // So concurrent passes cannot both escalate the same mismatch
+    vi.mocked(store.tryAcquireLeaderLock).mockResolvedValueOnce(true).mockResolvedValueOnce(false)
+
+    const worker1 = new ReconciliationWorker()
+    const worker2 = new ReconciliationWorker()
+
+    // Both try to run passes
+    const pass1 = worker1.poll()
+    const pass2 = worker2.poll()
+
+    await Promise.all([pass1, pass2])
+
+    // Only one pass should have run (the one that acquired the lock)
+    expect(engine.runReconciliationPass).toHaveBeenCalledTimes(1)
+  })
+})

src/reconciliation/worker.ts

@@ -2,7 +2,7 @@ import { logger } from '../utils/logger.js'
 import { runReconciliationPass } from './engine.js'
 import { runResolutionPass } from './resolver.js'
 import { reconcileChainPositions } from './chain-reconciliation.js'
-import { listPendingLedgerEvents } from './store.js'
+import { listPendingLedgerEvents, tryAcquireLeaderLock, releaseLeaderLock } from './store.js'
 import type { ToleranceRule } from './types.js'
 import { DEFAULT_TOLERANCE_RULES } from './types.js'
 import {
@@ -27,6 +27,11 @@ export class ReconciliationWorker {
     if (this.interval) return
     logger.info('[ReconciliationWorker] Starting', { intervalMs, batchSize: RECON_BATCH_SIZE })
     this.interval = setInterval(() => {
+      // Skip tick if previous pass is still running (intra-process overlap guard)
+      if (this.processingPromise) {
+        logger.warn('[ReconciliationWorker] Previous pass still in progress, skipping tick')
+        return
+      }
       this.processingPromise = this.poll().finally(() => {
         this.processingPromise = null
       })
@@ -47,16 +52,25 @@ export class ReconciliationWorker {
 
   async poll() {
     const startTime = Date.now()
+    let lockAcquired = false
+
     try {
+      // Try to acquire leader lock (cluster-wide single-leader guarantee)
+      lockAcquired = await tryAcquireLeaderLock()
+      if (!lockAcquired) {
+        logger.debug('[ReconciliationWorker] Another instance holds the leader lock, skipping pass')
+        return
+      }
+
+      logger.debug('[ReconciliationWorker] Leader lock acquired, starting pass')
       const reconResult = await runReconciliationPass(this.toleranceRules, RECON_BATCH_SIZE)
       logger.info('[ReconciliationWorker] Reconciliation pass done', reconResult)
 
       // Record pending count (matched + mismatches + skipped = total processed)
       recordReconciliationPending(reconResult.matched + reconResult.mismatches + reconResult.skipped)
 
-      // Record processed counts by status
-      recordReconciliationProcessed('matched')
-      for (let i = 1; i < reconResult.matched; i++) {
+      // Record processed counts by status (exact counts, including 0)
+      for (let i = 0; i < reconResult.matched; i++) {
         recordReconciliationProcessed('matched')
       }
       for (let i = 0; i < reconResult.mismatches; i++) {
@@ -128,6 +142,12 @@ export class ReconciliationWorker {
       logger.error('[ReconciliationWorker] Poll failed', {
         error: err instanceof Error ? err.message : String(err),
       })
+    } finally {
+      // Release leader lock if we acquired it
+      if (lockAcquired) {
+        await releaseLeaderLock()
+        logger.debug('[ReconciliationWorker] Leader lock released')
+      }
     }
   }
 }