diff --git a/src/reconciliation/INVARIANTS.md b/src/reconciliation/INVARIANTS.md index 1699bdc..6a1c97f 100644 --- a/src/reconciliation/INVARIANTS.md +++ b/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); diff --git a/src/reconciliation/store.ts b/src/reconciliation/store.ts index 86253b4..a822a6c 100644 --- a/src/reconciliation/store.ts +++ b/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 { + 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 { + 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): LedgerEvent { diff --git a/src/reconciliation/worker.test.ts b/src/reconciliation/worker.test.ts new file mode 100644 index 0000000..6ce86fb --- /dev/null +++ b/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(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) + }) +}) diff --git a/src/reconciliation/worker.ts b/src/reconciliation/worker.ts index c128e1c..9ebd2b0 100644 --- a/src/reconciliation/worker.ts +++ b/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') + } } } }