From e220673d3a299c17cca34e5c64bd8c4bb1078ff7 Mon Sep 17 00:00:00 2001 From: mode Date: Sat, 25 Jul 2026 17:50:41 +0100 Subject: [PATCH] fix(contract-sync): close reliability gaps in Soroban event indexer --- __tests__/contract-sync/listener.test.ts | 112 +++++++++++++ __tests__/contract-sync/queue.test.ts | 38 +++++ __tests__/contract-sync/service.test.ts | 157 ++++++++++++++++++ docs/sync-flow.md | 33 +++- lib/contract-sync/listener.ts | 53 +++++- lib/contract-sync/queue.ts | 16 +- lib/contract-sync/service.ts | 83 ++++++++- lib/contract-sync/types.ts | 12 ++ .../007_contract_sync_durability.sql | 23 +++ 9 files changed, 501 insertions(+), 26 deletions(-) create mode 100644 __tests__/contract-sync/service.test.ts create mode 100644 lib/db/migrations/007_contract_sync_durability.sql diff --git a/__tests__/contract-sync/listener.test.ts b/__tests__/contract-sync/listener.test.ts index 701c1e9..6a3e59d 100644 --- a/__tests__/contract-sync/listener.test.ts +++ b/__tests__/contract-sync/listener.test.ts @@ -80,4 +80,116 @@ describe('SorobanEventListener', () => { await listener.start() expect(mockGetLatestLedger).toHaveBeenCalledTimes(1) }) + + describe('checkpoint resumption', () => { + it('resumes from an initialLedger option instead of fetching the chain tip', async () => { + const onCheckpoint = vi.fn() + const resumable = new SorobanEventListener({ + rpcUrl: 'https://soroban-testnet.stellar.org', + networkPassphrase: 'Test SDF Network ; September 2015', + contractAddresses: ['CA1234'], + pollIntervalMs: 1000, + initialLedger: 900, + onCheckpoint, + }) + resumable.setCallback(callback as any) + + await resumable.start() + + expect(mockGetLatestLedger).not.toHaveBeenCalled() + resumable.stop() + }) + + it('setInitialLedger seeds the resume point before start()', async () => { + listener.setInitialLedger(900) + await listener.start() + + expect(mockGetLatestLedger).not.toHaveBeenCalled() + }) + + it('setInitialLedger is a no-op once already running', async () => { + mockGetLatestLedger.mockResolvedValue({ sequence: 500 }) + await listener.start() + + listener.setInitialLedger(900) + mockGetEvents.mockResolvedValue({ events: [] }) + mockGetLatestLedger.mockResolvedValue({ sequence: 600 }) + + await vi.advanceTimersByTimeAsync(1000) + + // startSeq should derive from the original 500 checkpoint (501), not 900 + expect(mockGetEvents).toHaveBeenCalledWith( + expect.objectContaining({ startLedger: 501 }) + ) + }) + + it('invokes onCheckpoint with the new high-water-mark after each poll', async () => { + const onCheckpoint = vi.fn() + listener = new SorobanEventListener({ + rpcUrl: 'https://soroban-testnet.stellar.org', + networkPassphrase: 'Test SDF Network ; September 2015', + contractAddresses: ['CA1234'], + pollIntervalMs: 1000, + onCheckpoint, + }) + listener.setCallback(callback as any) + + mockGetLatestLedger.mockResolvedValueOnce({ sequence: 500 }) + await listener.start() + + mockGetEvents.mockResolvedValue({ events: [] }) + mockGetLatestLedger.mockResolvedValue({ sequence: 505 }) + await vi.advanceTimersByTimeAsync(1000) + + expect(onCheckpoint).toHaveBeenCalledWith(505) + }) + + it('awaits the callback for each event before advancing the checkpoint', async () => { + // Uses real timers and calls the private poll() directly so the + // ordering can be observed deterministically without racing fake-timer + // microtask flushing against a manually-controlled promise. + vi.useRealTimers() + + const order: string[] = [] + const onCheckpoint = vi.fn((ledger: number) => order.push(`checkpoint:${ledger}`)) + let resolveCallback!: () => void + const slowCallback = vi.fn( + () => + new Promise((resolve) => { + resolveCallback = () => { + order.push('callback-resolved') + resolve() + } + }) + ) + + const realtimeListener = new SorobanEventListener({ + rpcUrl: 'https://soroban-testnet.stellar.org', + networkPassphrase: 'Test SDF Network ; September 2015', + contractAddresses: ['CA1234'], + pollIntervalMs: 1000, + initialLedger: 500, + onCheckpoint, + }) + realtimeListener.setCallback(slowCallback) + + mockGetEvents.mockResolvedValue({ + events: [{ topic: ['fund'], value: [], ledger: 501, txHash: 'tx1' }], + }) + mockGetLatestLedger.mockResolvedValue({ sequence: 505 }) + + const pollPromise = (realtimeListener as any).poll() + await Promise.resolve() // let poll() reach the awaited callback + await Promise.resolve() + expect(slowCallback).toHaveBeenCalled() + expect(onCheckpoint).not.toHaveBeenCalled() + + resolveCallback() + await pollPromise + + expect(order).toEqual(['callback-resolved', 'checkpoint:505']) + + vi.useFakeTimers() + }) + }) }) diff --git a/__tests__/contract-sync/queue.test.ts b/__tests__/contract-sync/queue.test.ts index da3f245..9dc1e10 100644 --- a/__tests__/contract-sync/queue.test.ts +++ b/__tests__/contract-sync/queue.test.ts @@ -107,6 +107,44 @@ describe('SyncQueue', () => { expect(deadLetters[0].retryCount).toBeGreaterThanOrEqual(3) }) + it('calls onRetry with the item and error message when a retry is scheduled', async () => { + const onRetry = vi.fn() + const retryQueue = new SyncQueue({ maxRetries: 3, concurrency: 2, pollIntervalMs: 100, onRetry }) + retryQueue.setHandler(async () => { + throw new Error('transient failure') + }) + + retryQueue.enqueue(makePayload({ txHash: 'retry001' })) + retryQueue.start() + await vi.advanceTimersByTimeAsync(100) + retryQueue.stop() + + expect(onRetry).toHaveBeenCalledTimes(1) + const [item, error] = onRetry.mock.calls[0] + expect(item.id).toBe('retry001:fund:0') + expect(item.retryCount).toBe(1) + expect(error).toBe('transient failure') + }) + + it('calls onDeadLetter once retries are exhausted, and does not call onRetry for that attempt', async () => { + const onRetry = vi.fn() + const onDeadLetter = vi.fn() + const dlQueue = new SyncQueue({ maxRetries: 1, concurrency: 2, pollIntervalMs: 100, onRetry, onDeadLetter }) + dlQueue.setHandler(async () => { + throw new Error('fatal') + }) + + dlQueue.enqueue(makePayload({ txHash: 'dead002' })) + dlQueue.start() + await vi.advanceTimersByTimeAsync(100) + dlQueue.stop() + + expect(onRetry).not.toHaveBeenCalled() + expect(onDeadLetter).toHaveBeenCalledTimes(1) + expect(onDeadLetter.mock.calls[0][0].id).toBe('dead002:fund:0') + expect(onDeadLetter.mock.calls[0][0].status).toBe('dead_letter') + }) + it('returns empty dead letters when all succeed', async () => { handler.mockResolvedValue(undefined) diff --git a/__tests__/contract-sync/service.test.ts b/__tests__/contract-sync/service.test.ts new file mode 100644 index 0000000..109b7a5 --- /dev/null +++ b/__tests__/contract-sync/service.test.ts @@ -0,0 +1,157 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest' + +vi.mock('@/lib/db', () => ({ + sql: Object.assign(vi.fn(), { unsafe: vi.fn() }), +})) + +const { mockGetLatestLedger, mockGetEvents, MockSorobanServer } = vi.hoisted(() => { + const mGetLatestLedger = vi.fn().mockResolvedValue({ sequence: 100 }) + const mGetEvents = vi.fn().mockResolvedValue({ events: [] }) + + const MServer = class { + getLatestLedger = mGetLatestLedger + getEvents = mGetEvents + } + + return { + mockGetLatestLedger: mGetLatestLedger, + mockGetEvents: mGetEvents, + MockSorobanServer: MServer, + } +}) + +vi.mock('@stellar/stellar-sdk', () => ({ + default: MockSorobanServer, +})) + +import { sql } from '@/lib/db' +import { ContractSyncService } from '@/lib/contract-sync/service' +import type { SorobanEventPayload } from '@/lib/contract-sync/types' + +function makePayload(overrides: Partial = {}): SorobanEventPayload { + return { + event: 'fund', + contractAddress: 'CA1234', + ledgerSequence: 1000, + timestamp: Date.now(), + txHash: 'abc123', + data: [], + ...overrides, + } +} + +describe('ContractSyncService', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + describe('onEvent (duplicate protection)', () => { + it('skips enqueueing an event that was already synced successfully', async () => { + const service = new ContractSyncService({ contractAddresses: ['CA1'] }) + const enqueueSpy = vi.spyOn(service.getQueue(), 'enqueue') + + ;(sql as unknown as ReturnType).mockResolvedValueOnce([{ '?column?': 1 }]) + + await (service as any).onEvent(makePayload()) + + expect(enqueueSpy).not.toHaveBeenCalled() + expect(sql).toHaveBeenCalledTimes(1) + }) + + it('enqueues and logs an event that has not been synced before', async () => { + const service = new ContractSyncService({ contractAddresses: ['CA1'] }) + const enqueueSpy = vi.spyOn(service.getQueue(), 'enqueue') + + ;(sql as unknown as ReturnType) + .mockResolvedValueOnce([]) // isAlreadySynced -> not found + .mockResolvedValueOnce([]) // createSyncLog insert + + await (service as any).onEvent(makePayload({ txHash: 'new-tx' })) + + expect(enqueueSpy).toHaveBeenCalledWith(expect.objectContaining({ txHash: 'new-tx' })) + expect(sql).toHaveBeenCalledTimes(2) + }) + }) + + describe('checkpoint persistence', () => { + it('loadCheckpoint returns null when no row exists', async () => { + const service = new ContractSyncService({ contractAddresses: ['CA1'] }) + ;(sql as unknown as ReturnType).mockResolvedValueOnce([]) + + const result = await (service as any).loadCheckpoint() + expect(result).toBeNull() + }) + + it('loadCheckpoint returns the persisted ledger as a number', async () => { + const service = new ContractSyncService({ contractAddresses: ['CA1'] }) + ;(sql as unknown as ReturnType).mockResolvedValueOnce([{ last_ledger: '4242' }]) + + const result = await (service as any).loadCheckpoint() + expect(result).toBe(4242) + }) + + it('persistCheckpoint issues an upsert', async () => { + const service = new ContractSyncService({ contractAddresses: ['CA1'] }) + ;(sql as unknown as ReturnType).mockResolvedValueOnce([]) + + await (service as any).persistCheckpoint(555) + expect(sql).toHaveBeenCalledTimes(1) + }) + + it('start() seeds the listener from a persisted checkpoint instead of the chain tip', async () => { + const service = new ContractSyncService({ contractAddresses: ['CA1'] }) + ;(sql as unknown as ReturnType).mockResolvedValueOnce([{ last_ledger: '777' }]) + + await service.start() + service.stop() + + expect(mockGetLatestLedger).not.toHaveBeenCalled() + }) + }) + + describe('failure / dead-letter audit logging', () => { + it('records a failed sync attempt with the error message on retry', async () => { + const service = new ContractSyncService({ contractAddresses: ['CA1'], maxRetries: 5 }) + const updateSpy = vi.spyOn(service as any, 'updateSyncLog').mockResolvedValue(undefined) + + const queue = service.getQueue() + queue.setHandler(async () => { + throw new Error('boom') + }) + + vi.useFakeTimers() + queue.enqueue(makePayload({ txHash: 'fail-tx' })) + queue.start() + await vi.advanceTimersByTimeAsync(1000) + queue.stop() + vi.useRealTimers() + + expect(updateSpy).toHaveBeenCalledWith( + 'fail-tx:fund:0', + expect.objectContaining({ status: 'failed', errorMessage: 'boom' }) + ) + }) + + it('records dead_letter status once retries are exhausted', async () => { + const service = new ContractSyncService({ contractAddresses: ['CA1'], maxRetries: 1 }) + const updateSpy = vi.spyOn(service as any, 'updateSyncLog').mockResolvedValue(undefined) + + const queue = service.getQueue() + queue.setHandler(async () => { + throw new Error('permanent failure') + }) + + vi.useFakeTimers() + queue.enqueue(makePayload({ txHash: 'dead-tx' })) + queue.start() + await vi.advanceTimersByTimeAsync(1000) + queue.stop() + vi.useRealTimers() + + expect(updateSpy).toHaveBeenCalledWith( + 'dead-tx:fund:0', + expect.objectContaining({ status: 'dead_letter', errorMessage: 'permanent failure' }) + ) + }) + }) +}) diff --git a/docs/sync-flow.md b/docs/sync-flow.md index 92e4e04..e16b9f1 100644 --- a/docs/sync-flow.md +++ b/docs/sync-flow.md @@ -75,15 +75,17 @@ In-memory queue that processes events with: - **Configurable concurrency** (default: 3) - **Exponential backoff**: `min(1000 * 2^retry, 60000)` ms - **Dead letter** after 5 failed retries -- Thread-safe item tracking by `txHash:event:milestoneId` +- Thread-safe item tracking by `txHash:event:milestoneId` (see `buildSyncDedupeKey` in `types.ts`) +- `onRetry` / `onDeadLetter` hooks so every retry and terminal failure is mirrored into `contract_sync_log` — not just successes ### 4. Soroban Event Listener (`lib/contract-sync/listener.ts`) Polls the Soroban RPC endpoint for contract events: - Uses `@stellar/stellar-sdk` `SorobanRpc.Server.getEvents()` -- Tracks the latest processed ledger to avoid re-processing - Configurable poll interval (default: 10s) - Parses Soroban event topics and data into typed payloads +- Resumes from a persisted ledger checkpoint on restart (`initialLedger` / `setInitialLedger`) instead of jumping to the chain tip, so events emitted during downtime are still picked up +- The checkpoint only advances once the event callback for a batch has resolved (`onCheckpoint` fires after the awaited callback), so a crash mid-processing can't advance past an event that was never durably logged ### 5. Sync Service (`lib/contract-sync/service.ts`) @@ -110,11 +112,21 @@ contract_sync_log ( error_message TEXT, retry_count INTEGER DEFAULT 0, raw_payload JSONB, + dedupe_key TEXT, -- unique per event; see Durability section created_at TIMESTAMPTZ, updated_at TIMESTAMPTZ ) ``` +## Durability & Restart Behavior + +Two mechanisms guarantee "exactly once" processing and a complete audit trail across process restarts, not just within a single run (added in `lib/db/migrations/007_contract_sync_durability.sql`): + +1. **`dedupe_key` + unique index on `contract_sync_log`** — every event's identity (`txHash:event:milestoneId`) is checked against the audit log *before* enqueueing. If a row already exists with `status = 'success'`, the event is skipped entirely. The insert itself also uses `ON CONFLICT (dedupe_key) DO NOTHING` as a race-safety net. This prevents duplicate DB writes (and duplicate audit rows) if the same event is redelivered after a restart, when the in-memory queue has been wiped. +2. **`contract_sync_checkpoint` table** — a single-row table storing the last ledger sequence the listener has fully processed. `ContractSyncService.start()` loads it and seeds the listener via `setInitialLedger()`, so polling resumes from where it left off instead of starting from the chain's current tip (which would silently skip any events emitted while the service was down). + +Combined, these mean: on restart, events from the downtime window are re-fetched (via the checkpoint) and safely deduplicated against `contract_sync_log` (via `dedupe_key`) rather than either being dropped or double-processed. + ## Running the Sync Worker ### Prerequisites @@ -127,10 +139,11 @@ contract_sync_log ( SOROBAN_CONTRACT_ADDRESSES=CA...ID1,CA...ID2 ``` -2. Database migration applied: +2. Database migrations applied: ```bash - # Run the SQL migration in your Neon console - lib/db/migrations/005_contract_sync_log.sql + npm run migrate + # applies lib/db/migrations/005_contract_sync_log.sql and + # lib/db/migrations/007_contract_sync_durability.sql, among others ``` ### Start the Worker @@ -180,10 +193,10 @@ const deadLetters = syncService.getQueue().getDeadLetters() ## Error Handling -- **Transient errors** (network issues, RPC timeouts): Retried with exponential backoff -- **Permanent errors** (contract not found, invalid event data): Move to dead letter after max retries +- **Transient errors** (network issues, RPC timeouts): Retried with exponential backoff; each retry updates `contract_sync_log` to `status = 'failed'` with the error message and current retry count via the queue's `onRetry` hook +- **Permanent errors** (contract not found, invalid event data): Move to dead letter after max retries; the queue's `onDeadLetter` hook records the final `status = 'dead_letter'` row - **DB constraint violations**: Logged and moved to dead letter (require manual intervention) -- **All failures** are recorded in `contract_sync_log` with the error message and retry count +- **All failures** — not just successes — are recorded in `contract_sync_log` with the error message and retry count ## Testing @@ -193,10 +206,12 @@ npm test -- --reporter=verbose Unit tests cover: - Event mapping correctness for all 10 event types -- Queue enqueue/dequeue/retry/dead letter logic +- Queue enqueue/dequeue/retry/dead letter logic, including `onRetry`/`onDeadLetter` hooks - Backoff delay calculation - Soroban event parsing - DB update generation +- Listener checkpoint resumption (`initialLedger`, `setInitialLedger`, `onCheckpoint`) and callback-before-checkpoint ordering +- Service-level duplicate protection (`isAlreadySynced`) and checkpoint load/persist ## Adding New Contract Events diff --git a/lib/contract-sync/listener.ts b/lib/contract-sync/listener.ts index 94421cc..382fb73 100644 --- a/lib/contract-sync/listener.ts +++ b/lib/contract-sync/listener.ts @@ -1,7 +1,8 @@ import Server from '@stellar/stellar-sdk' import type { SorobanContractEvent, SorobanEventPayload } from './types' -export type EventCallback = (payload: SorobanEventPayload) => void +export type EventCallback = (payload: SorobanEventPayload) => void | Promise +export type CheckpointCallback = (ledgerSequence: number) => void export interface SorobanListenerOptions { rpcUrl: string @@ -9,6 +10,15 @@ export interface SorobanListenerOptions { contractAddresses: string[] pollIntervalMs?: number maxLedgerOffset?: number + /** + * Ledger sequence to resume polling from (e.g. the last checkpoint + * persisted before a restart). When omitted, polling starts from the + * chain's current latest ledger, which means any events emitted while the + * listener was down are permanently skipped. + */ + initialLedger?: number + /** Invoked after every successful poll with the new high-water-mark ledger. */ + onCheckpoint?: CheckpointCallback } export class SorobanEventListener { @@ -18,6 +28,7 @@ export class SorobanEventListener { private readonly pollIntervalMs: number private readonly maxLedgerOffset: number private callback: EventCallback | null = null + private readonly onCheckpoint: CheckpointCallback | null private timer: ReturnType | null = null private lastLedger: number = 0 private running = false @@ -33,25 +44,45 @@ export class SorobanEventListener { this.contractAddresses = options.contractAddresses this.pollIntervalMs = options.pollIntervalMs ?? 10_000 this.maxLedgerOffset = options.maxLedgerOffset ?? 100 + this.onCheckpoint = options.onCheckpoint ?? null + if (options.initialLedger && options.initialLedger > 0) { + this.lastLedger = options.initialLedger + } } setCallback(cb: EventCallback): void { this.callback = cb } + /** + * Seeds the resume point before `start()` is called (e.g. from a + * checkpoint persisted by a previous run). No-op once the listener is + * already running. + */ + setInitialLedger(ledgerSequence: number): void { + if (!this.running && ledgerSequence > 0) { + this.lastLedger = ledgerSequence + } + } + async start(): Promise { if (this.running) return this.running = true - try { - const info = await this.server.getLatestLedger() - this.lastLedger = info.sequence - } catch (err) { - console.warn('[SorobanListener] Could not get latest ledger, starting from 0') + // Only fall back to "start from now" when we have no persisted checkpoint + // to resume from — resuming from a checkpoint means events emitted while + // the listener was offline still get picked up on the first poll. + if (this.lastLedger === 0) { + try { + const info = await this.server.getLatestLedger() + this.lastLedger = info.sequence + } catch (err) { + console.warn('[SorobanListener] Could not get latest ledger, starting from 0') + } } this.timer = setInterval(() => this.poll(), this.pollIntervalMs) - console.log(`[SorobanListener] Started polling ${this.contractAddresses.length} contract(s) every ${this.pollIntervalMs}ms`) + console.log(`[SorobanListener] Started polling ${this.contractAddresses.length} contract(s) every ${this.pollIntervalMs}ms (resuming from ledger ${this.lastLedger})`) } stop(): void { @@ -75,12 +106,14 @@ export class SorobanEventListener { if (this.lastLedger === 0) { this.lastLedger = latestSeq + this.onCheckpoint?.(latestSeq) return } const startSeq = Math.max(this.lastLedger + 1, latestSeq - this.maxLedgerOffset) if (startSeq >= latestSeq) { this.lastLedger = latestSeq + this.onCheckpoint?.(latestSeq) return } @@ -89,6 +122,7 @@ export class SorobanEventListener { } this.lastLedger = latestSeq + this.onCheckpoint?.(latestSeq) } catch (err) { console.error('[SorobanListener] Poll error:', err) } @@ -116,7 +150,10 @@ export class SorobanEventListener { for (const event of events.events) { const parsed = this.parseSorobanEvent(event, contractAddress) if (parsed) { - this.callback!(parsed) + // Awaited so the checkpoint only advances once the event is + // durably enqueued/logged — otherwise a crash between "advance + // checkpoint" and "persist event" would permanently drop it. + await this.callback!(parsed) } } diff --git a/lib/contract-sync/queue.ts b/lib/contract-sync/queue.ts index 6f146d8..fa5852a 100644 --- a/lib/contract-sync/queue.ts +++ b/lib/contract-sync/queue.ts @@ -1,12 +1,18 @@ import type { SorobanEventPayload, SyncQueueItem, SyncStatus } from './types' -import { getDefaultMaxRetries, getBackoffDelay } from './types' +import { getDefaultMaxRetries, getBackoffDelay, buildSyncDedupeKey } from './types' export type QueueHandler = (item: SyncQueueItem) => Promise +export type QueueRetryHandler = (item: SyncQueueItem, error: string) => void +export type QueueDeadLetterHandler = (item: SyncQueueItem) => void export interface SyncQueueOptions { maxRetries?: number concurrency?: number pollIntervalMs?: number + /** Called every time an item fails but will still be retried. */ + onRetry?: QueueRetryHandler + /** Called once an item exhausts all retries and is moved to the dead letter queue. */ + onDeadLetter?: QueueDeadLetterHandler } export class SyncQueue { @@ -17,16 +23,20 @@ export class SyncQueue { private readonly maxRetries: number private readonly concurrency: number private readonly pollIntervalMs: number + private readonly onRetry: QueueRetryHandler | null + private readonly onDeadLetter: QueueDeadLetterHandler | null private deadLetters: SyncQueueItem[] = [] constructor(options: SyncQueueOptions = {}) { this.maxRetries = options.maxRetries ?? getDefaultMaxRetries() this.concurrency = options.concurrency ?? 3 this.pollIntervalMs = options.pollIntervalMs ?? 1_000 + this.onRetry = options.onRetry ?? null + this.onDeadLetter = options.onDeadLetter ?? null } enqueue(payload: SorobanEventPayload): string { - const id = `${payload.txHash}:${payload.event}:${payload.milestoneId ?? 0}` + const id = buildSyncDedupeKey(payload) if (this.items.has(id)) return id const item: SyncQueueItem = { @@ -124,9 +134,11 @@ export class SyncQueue { item.status = 'dead_letter' this.deadLetters.push({ ...item }) this.items.delete(item.id) + this.onDeadLetter?.({ ...item }) } else { item.status = 'pending' item.nextRetryAt = Date.now() + getBackoffDelay(item.retryCount) + this.onRetry?.({ ...item }, message) } } finally { this.processing.delete(item.id) diff --git a/lib/contract-sync/service.ts b/lib/contract-sync/service.ts index 4e4ccae..1e1901e 100644 --- a/lib/contract-sync/service.ts +++ b/lib/contract-sync/service.ts @@ -2,7 +2,8 @@ import { sql } from '@/lib/db' import { SorobanEventListener } from './listener' import { SyncQueue } from './queue' import { mapEventToAction } from './mapper' -import type { SorobanEventPayload, ContractSyncLog, SyncStatus, SorobanContractEvent } from './types' +import { buildSyncDedupeKey } from './types' +import type { SorobanEventPayload, SyncStatus, SorobanContractEvent } from './types' export interface SyncServiceOptions { rpcUrl?: string @@ -22,6 +23,20 @@ export class ContractSyncService { this.queue = new SyncQueue({ maxRetries: options.maxRetries, concurrency: options.queueConcurrency, + onRetry: (item, error) => { + this.updateSyncLog(item.id, { + status: 'failed', + errorMessage: error, + retryCount: item.retryCount, + }).catch((err) => console.error('[ContractSyncService] Failed to persist retry status:', err)) + }, + onDeadLetter: (item) => { + this.updateSyncLog(item.id, { + status: 'dead_letter', + errorMessage: item.lastError ?? undefined, + retryCount: item.retryCount, + }).catch((err) => console.error('[ContractSyncService] Failed to persist dead-letter status:', err)) + }, }) this.listener = new SorobanEventListener({ @@ -29,6 +44,11 @@ export class ContractSyncService { networkPassphrase: options.networkPassphrase ?? process.env.STELLAR_NETWORK_PASSPHRASE ?? 'Test SDF Network ; September 2015', contractAddresses: options.contractAddresses ?? [], pollIntervalMs: options.pollIntervalMs, + onCheckpoint: (ledgerSequence) => { + this.persistCheckpoint(ledgerSequence).catch((err) => + console.error('[ContractSyncService] Failed to persist sync checkpoint:', err) + ) + }, }) this.listener.setCallback((payload) => this.onEvent(payload)) @@ -39,6 +59,11 @@ export class ContractSyncService { if (this.started) return this.started = true + const checkpoint = await this.loadCheckpoint() + if (checkpoint != null) { + this.listener.setInitialLedger(checkpoint) + } + this.queue.start() await this.listener.start() @@ -61,10 +86,21 @@ export class ContractSyncService { } private async onEvent(payload: SorobanEventPayload): Promise { + const dedupeKey = buildSyncDedupeKey(payload) + + // Guards against reprocessing an event that was already synced in a + // previous process lifetime (the in-memory queue only dedupes within a + // single run — this check survives restarts). + if (await this.isAlreadySynced(dedupeKey)) { + console.log(`[ContractSyncService] Skipping already-synced event (dedupe_key=${dedupeKey})`) + return + } + const id = this.queue.enqueue(payload) console.log(`[ContractSyncService] Enqueued event: ${payload.event} @ ${payload.contractAddress} (id=${id})`) await this.createSyncLog({ + dedupeKey, eventType: payload.event, txHash: payload.txHash, ledgerSequence: payload.ledgerSequence, @@ -73,6 +109,34 @@ export class ContractSyncService { }) } + private async isAlreadySynced(dedupeKey: string): Promise { + const rows = (await sql` + SELECT 1 FROM contract_sync_log + WHERE dedupe_key = ${dedupeKey} + AND status = 'success' + LIMIT 1 + `) as unknown[] + return rows.length > 0 + } + + private async loadCheckpoint(): Promise { + const rows = (await sql` + SELECT last_ledger FROM contract_sync_checkpoint WHERE id = 1 + `) as { last_ledger: number | string }[] + if (rows.length === 0) return null + return Number(rows[0].last_ledger) + } + + private async persistCheckpoint(ledgerSequence: number): Promise { + await sql` + INSERT INTO contract_sync_checkpoint (id, last_ledger) + VALUES (1, ${ledgerSequence}) + ON CONFLICT (id) DO UPDATE + SET last_ledger = ${ledgerSequence}, + updated_at = NOW() + ` + } + private async processSync(item: { id: string; payload: SorobanEventPayload }): Promise { const { payload } = item const action = mapEventToAction(payload.event, payload) @@ -199,6 +263,7 @@ export class ContractSyncService { } private async createSyncLog(params: { + dedupeKey: string eventType: SorobanContractEvent txHash: string ledgerSequence: number @@ -207,6 +272,7 @@ export class ContractSyncService { }): Promise { await sql` INSERT INTO contract_sync_log ( + dedupe_key, event_type, tx_hash, ledger_sequence, @@ -214,28 +280,31 @@ export class ContractSyncService { raw_payload ) VALUES ( + ${params.dedupeKey}, ${params.eventType}::contract_sync_event_type, ${params.txHash}, ${params.ledgerSequence}, ${params.status}::sync_status, ${JSON.stringify(params.rawPayload)}::jsonb ) + ON CONFLICT (dedupe_key) DO NOTHING ` } + // Note: `itemId`/`dedupeKey` are the same value — the queue's item id is + // built from `buildSyncDedupeKey`, which is also stored as the audit row's + // dedupe_key, so both sides can be reconciled without any lookup. private async updateSyncLog( - itemId: string, - params: { status: SyncStatus; errorMessage?: string } + dedupeKey: string, + params: { status: SyncStatus; errorMessage?: string; retryCount?: number } ): Promise { - const [txHash] = itemId.split(':') await sql` UPDATE contract_sync_log SET status = ${params.status}::sync_status, error_message = COALESCE(${params.errorMessage ?? null}, error_message), - retry_count = retry_count + 1, + retry_count = COALESCE(${params.retryCount ?? null}, retry_count), updated_at = NOW() - WHERE tx_hash = ${txHash} - AND status = 'pending' + WHERE dedupe_key = ${dedupeKey} ` } diff --git a/lib/contract-sync/types.ts b/lib/contract-sync/types.ts index 6c660ba..08e51d5 100644 --- a/lib/contract-sync/types.ts +++ b/lib/contract-sync/types.ts @@ -56,4 +56,16 @@ export function getBackoffDelay(retryCount: number): number { return Math.min(1000 * Math.pow(2, retryCount), 60_000) } +/** + * Stable identity for a single on-chain event, shared by the in-memory queue + * and the `contract_sync_log` audit table. Used to guarantee at-most-once + * processing even across process restarts (the queue alone only dedupes + * while an item is in memory). + */ +export function buildSyncDedupeKey( + payload: Pick +): string { + return `${payload.txHash}:${payload.event}:${payload.milestoneId ?? 0}` +} + export const ESCROW_EVENT_TOPIC_PREFIX = 'escrow_event' diff --git a/lib/db/migrations/007_contract_sync_durability.sql b/lib/db/migrations/007_contract_sync_durability.sql new file mode 100644 index 0000000..6eaee80 --- /dev/null +++ b/lib/db/migrations/007_contract_sync_durability.sql @@ -0,0 +1,23 @@ +-- Durability improvements for the Soroban contract-sync indexer: +-- 1. dedupe_key + unique index — guarantees an on-chain event is recorded +-- exactly once in the audit trail, even if the same event is redelivered +-- after a service restart (the in-memory queue alone can't guarantee this +-- across process boundaries). +-- 2. contract_sync_checkpoint — persists the last ledger sequence the +-- listener has fully processed, so a restart resumes polling from where +-- it left off instead of jumping to "now" and silently skipping any +-- events emitted while the service was down. + +ALTER TABLE contract_sync_log + ADD COLUMN IF NOT EXISTS dedupe_key TEXT; + +CREATE UNIQUE INDEX IF NOT EXISTS uq_contract_sync_log_dedupe_key + ON contract_sync_log (dedupe_key) + WHERE dedupe_key IS NOT NULL; + +CREATE TABLE IF NOT EXISTS contract_sync_checkpoint ( + id SMALLINT PRIMARY KEY DEFAULT 1, + last_ledger BIGINT NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT single_row CHECK (id = 1) +);