From bf069ab54fe1e0634f58ca1d44edf26a89017727 Mon Sep 17 00:00:00 2001 From: Henry Ebubechukwu Date: Mon, 28 Sep 2026 16:16:31 +0100 Subject: [PATCH 1/2] feat(messaging): support message status reconciliation Reconcile the relayer's tracked message status against the destination chain so a hung executor, a dropped completion callback, or a receipt that never finalised cannot strand a message in flight forever. - MessageQueue lifecycle API for reconciliation: markSubmitted() records a broadcast transaction hash, fail() force-fails an in-flight message, getMessageStatus()/getInflightSnapshot() expose tracked state, and complete() now advances the message to confirmed and records a late duplicate result only once. Staleness is measured from the current delivery attempt (inflightAt, set at dequeue) so normal retries are never flagged stale. - New MessageStatusReconciler repairs drift: stale processing messages are re-queued (or failed once the retry budget is spent); submitted messages are confirmed when the receipt is final, failed on revert, and re-queued when no receipt appears after the stale threshold. Fresh deliveries and chains without a provider are left untouched. Emits message-reconciled, stale-message-detected, reconcile-error, reconciler-degraded/recovered and a per-run reconciliation-summary, and exports getStats(). - Ships EvmExecutorStatusProvider, which reuses EvmExecutor's own getTransactionStatus so reconciliation shares the live confirmation policy instead of defining a second notion of finality. - 42 new tests (121 total): queue lifecycle, reconciler unit coverage at 100% statements/functions/lines, and an integration pipeline across queue -> reconciler -> executor -> mocked RPC. - Docs: MESSAGE_STATUS_RECONCILIATION.md plus a cross-link from the duplicate-detection doc; CI now runs relayer-service tests. Closes #1164 --- .github/workflows/ci.yml | 3 + apps/relayer-service/src/index.ts | 15 + .../__tests__/message-queue-status.spec.ts | 138 ++++++ .../src/queue/message-queue.ts | 139 +++++- .../message-status-reconciler.spec.ts | 434 ++++++++++++++++++ .../__tests__/reconciliation-pipeline.spec.ts | 172 +++++++ .../reconciler/message-status-reconciler.ts | 402 ++++++++++++++++ apps/relayer-service/src/types.ts | 118 +++++ ...CROSS_CHAIN_DUPLICATE_MESSAGE_DETECTION.md | 2 + docs/MESSAGE_STATUS_RECONCILIATION.md | 104 +++++ 10 files changed, 1503 insertions(+), 24 deletions(-) create mode 100644 apps/relayer-service/src/queue/__tests__/message-queue-status.spec.ts create mode 100644 apps/relayer-service/src/reconciler/__tests__/message-status-reconciler.spec.ts create mode 100644 apps/relayer-service/src/reconciler/__tests__/reconciliation-pipeline.spec.ts create mode 100644 apps/relayer-service/src/reconciler/message-status-reconciler.ts create mode 100644 docs/MESSAGE_STATUS_RECONCILIATION.md diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b10b4915..a295561b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -49,6 +49,9 @@ jobs: - name: Build relayer service run: pnpm --filter @bridgewise/relayer-service build + - name: Test relayer service + run: pnpm --filter @bridgewise/relayer-service test + - uses: dtolnay/rust-toolchain@stable - name: Test Cargo Rules diff --git a/apps/relayer-service/src/index.ts b/apps/relayer-service/src/index.ts index 255d14ef..5bf6be85 100644 --- a/apps/relayer-service/src/index.ts +++ b/apps/relayer-service/src/index.ts @@ -2,6 +2,12 @@ export { MessageQueue } from './queue/message-queue'; export { EvmExecutor } from './executors/evm-executor'; export { SorobanExecutor } from './executors/soroban-executor'; export { CanaryRolloutRouter } from './executors/canary-rollout'; +export { + MessageStatusReconciler, + EvmExecutorStatusProvider, + DEFAULT_RECONCILIATION_CONFIG, +} from './reconciler/message-status-reconciler'; +export type { GetTransactionStatusResult } from './reconciler/message-status-reconciler'; export { DuplicateMessageDetector, computeMessageFingerprint, @@ -27,6 +33,15 @@ export type { ExecutorConfig, QueueConfig, MessageQueueItem, + InflightMessageSnapshot, + OnChainMessageStatus, + MessageStatusProvider, + ReconciliationConfig, + ReconciliationAction, + ReconciliationReason, + MessageReconciliation, + ReconciliationSummary, + ReconciliationStats, ChainType, DuplicateDetectorConfig, DuplicateReason, diff --git a/apps/relayer-service/src/queue/__tests__/message-queue-status.spec.ts b/apps/relayer-service/src/queue/__tests__/message-queue-status.spec.ts new file mode 100644 index 00000000..6990d1ba --- /dev/null +++ b/apps/relayer-service/src/queue/__tests__/message-queue-status.spec.ts @@ -0,0 +1,138 @@ +import { MessageQueue } from '../message-queue'; +import { CrossChainMessage } from '../../types'; + +function makeMessage(overrides: Partial = {}): CrossChainMessage { + return { + id: 'msg-1', + sourceChainId: 'ethereum', + destinationChainId: 'stellar', + sourceTxHash: '0x' + 'a'.repeat(64), + sourceBlockNumber: 10_000_000, + messageType: 'lock', + payload: '0xdeadbeef', + sender: '0xsender', + recipient: 'Grecipient', + createdAt: Date.now(), + status: 'pending', + retryCount: 0, + ...overrides, + }; +} + +describe('MessageQueue status lifecycle', () => { + let queue: MessageQueue; + + beforeEach(() => { + queue = new MessageQueue({ maxRetries: 3, retryDelayMs: 100, concurrency: 5 }); + }); + + afterEach(() => { + queue.removeAllListeners(); + }); + + it('tracks a message as submitted after markSubmitted', () => { + queue.enqueue(makeMessage({ id: 'm' })); + queue.dequeue(); + expect(queue.markSubmitted('m', '0xtx')).toBe(true); + expect(queue.getMessageStatus('m')).toBe('submitted'); + expect(queue.getSubmittedCount()).toBe(1); + expect(queue.getProcessingCount()).toBe(1); + }); + + it('refuses to mark a message that is not in flight', () => { + expect(queue.markSubmitted('nope', '0xtx')).toBe(false); + expect(queue.getMessageStatus('nope')).toBeNull(); + }); + + it('emits message-submitted with the transaction hash', () => { + const spy = jest.fn(); + queue.on('message-submitted', spy); + queue.enqueue(makeMessage({ id: 'm' })); + queue.dequeue(); + queue.markSubmitted('m', '0xtx'); + expect(spy).toHaveBeenCalledWith(expect.objectContaining({ messageId: 'm', txHash: '0xtx', destinationChainId: 'stellar' })); + }); + + it('advances the tracked status through the whole lifecycle', async () => { + queue.enqueue(makeMessage({ id: 'm' })); + expect(queue.getMessageStatus('m')).toBe('queued'); + queue.dequeue(); + expect(queue.getMessageStatus('m')).toBe('processing'); + queue.markSubmitted('m', '0xtx'); + expect(queue.getMessageStatus('m')).toBe('submitted'); + await queue.complete({ messageId: 'm', success: true, transactionHash: '0xtx', timestamp: Date.now() }); + expect(queue.getMessageStatus('m')).toBe('confirmed'); + }); + + it('reports queued, failed and unknown statuses', async () => { + queue.enqueue(makeMessage({ id: 'waiting' })); + expect(queue.getMessageStatus('waiting')).toBe('queued'); + + queue = new MessageQueue({ maxRetries: 0 }); + queue.enqueue(makeMessage({ id: 'dead' })); + queue.dequeue(); + await queue.complete({ messageId: 'dead', success: false, error: 'boom', timestamp: Date.now() }); + expect(queue.getMessageStatus('dead')).toBe('failed'); + + expect(queue.getMessageStatus('never-seen')).toBeNull(); + }); + + it('records the confirmed status on successful completion', async () => { + queue.enqueue(makeMessage({ id: 'ok' })); + queue.dequeue(); + await queue.complete({ messageId: 'ok', success: true, transactionHash: '0xreal', timestamp: Date.now() }); + expect(queue.getCompletedMessages()[0]).toMatchObject({ messageId: 'ok', success: true, transactionHash: '0xreal' }); + }); + + it('does not let a late reconciled duplicate overwrite the real completion', async () => { + queue.enqueue(makeMessage({ id: 'ok' })); + queue.dequeue(); + await queue.complete({ messageId: 'ok', success: true, transactionHash: '0xreal', timestamp: 1 }); + await queue.complete({ messageId: 'ok', success: false, error: 'late result', timestamp: 2 }); + expect(queue.getCompletedMessages()[0]).toMatchObject({ messageId: 'ok', success: true, transactionHash: '0xreal' }); + expect(queue.getFailedCount()).toBe(0); + }); + + it('records a completion for a message the queue never dequeued', async () => { + await queue.complete({ messageId: 'external', success: true, transactionHash: '0xext', timestamp: Date.now() }); + expect(queue.getMessageStatus('external')).toBe('confirmed'); + expect(queue.getCompletedCount()).toBe(1); + }); + + it('exposes a read-only in-flight snapshot', () => { + queue.enqueue(makeMessage({ id: 'snap' })); + queue.dequeue(); + queue.markSubmitted('snap', '0xtx'); + const snap = queue.getInflightSnapshot(); + expect(snap).toHaveLength(1); + expect(snap[0]).toMatchObject({ + messageId: 'snap', + status: 'submitted', + submittedTxHash: '0xtx', + destinationChainId: 'stellar', + }); + snap[0].message.payload = '0xmutated'; + snap[0].status = 'queued'; + expect(queue.getInflightSnapshot()[0].message.payload).toBe('0xdeadbeef'); + expect(queue.getInflightSnapshot()[0].status).toBe('submitted'); + }); + + it('forces an in-flight message to failed with fail()', () => { + queue.enqueue(makeMessage({ id: 'revert' })); + queue.dequeue(); + expect(queue.fail('revert', 'reverted on chain')).toBe(true); + expect(queue.getMessageStatus('revert')).toBe('failed'); + expect(queue.getFailedMessages()[0]).toMatchObject({ lastError: 'reverted on chain' }); + expect(queue.fail('not-inflight', 'x')).toBe(false); + }); + + it('resets lifecycle state on clear()', () => { + queue.enqueue(makeMessage({ id: 'a' })); + queue.dequeue(); + queue.markSubmitted('a', '0xtx'); + queue.clear(); + expect(queue.getProcessingCount()).toBe(0); + expect(queue.getSubmittedCount()).toBe(0); + expect(queue.getMessageStatus('a')).toBeNull(); + }); +}); \ No newline at end of file diff --git a/apps/relayer-service/src/queue/message-queue.ts b/apps/relayer-service/src/queue/message-queue.ts index 4a90f9f1..d77eed71 100644 --- a/apps/relayer-service/src/queue/message-queue.ts +++ b/apps/relayer-service/src/queue/message-queue.ts @@ -1,6 +1,7 @@ import { EventEmitter } from 'events'; import { CrossChainMessage, + InflightMessageSnapshot, MessageStatus, MessageQueueItem, QueueConfig, @@ -80,6 +81,7 @@ export class MessageQueue extends EventEmitter { this.queue.delete(oldest.message.id); this.processing.set(oldest.message.id, oldest); + oldest.inflightAt = Date.now(); oldest.message.status = 'processing'; this.emit('message-dequeued', { messageId: oldest.message.id }); @@ -88,36 +90,48 @@ export class MessageQueue extends EventEmitter { async complete(result: ExecutionResult): Promise { const item = this.processing.get(result.messageId) ?? null; + if (!item) { + // Completion for a message the queue no longer tracks, for example a late + // reconciled result racing the executor's own completion. Record it once + // so the real outcome is never overwritten. + if (!this.completed.has(result.messageId)) { + this.completed.set(result.messageId, result); + } + return; + } + this.processing.delete(result.messageId); + if (result.success) { + item.message.status = 'confirmed'; + } this.completed.set(result.messageId, result); if (result.success) { this.emit('message-completed', result); + return; + } + + item.attempts++; + if (item.attempts <= this.config.maxRetries) { + item.nextRetryAt = Date.now() + this.config.retryDelayMs * Math.pow(2, item.attempts - 1); + item.message.status = 'queued'; + item.message.lastError = result.error; + this.queue.set(result.messageId, item); + this.emit('message-retrying', { + messageId: result.messageId, + attempt: item.attempts, + maxRetries: this.config.maxRetries, + nextRetryAt: item.nextRetryAt, + error: result.error, + }); } else { - if (item) { - item.attempts++; - if (item.attempts <= this.config.maxRetries) { - item.nextRetryAt = Date.now() + this.config.retryDelayMs * Math.pow(2, item.attempts - 1); - item.message.status = 'queued'; - item.message.lastError = result.error; - this.queue.set(result.messageId, item); - this.emit('message-retrying', { - messageId: result.messageId, - attempt: item.attempts, - maxRetries: this.config.maxRetries, - nextRetryAt: item.nextRetryAt, - error: result.error, - }); - } else { - item.message.status = 'failed'; - this.failed.set(result.messageId, item); - this.emit('message-failed', { - messageId: result.messageId, - attempts: item.attempts, - lastError: result.error, - }); - } - } + item.message.status = 'failed'; + this.failed.set(result.messageId, item); + this.emit('message-failed', { + messageId: result.messageId, + attempts: item.attempts, + lastError: result.error, + }); } } @@ -129,6 +143,83 @@ export class MessageQueue extends EventEmitter { return this.processing.size; } + /** Number of in-flight messages whose transaction has been broadcast. */ + getSubmittedCount(): number { + let count = 0; + for (const item of this.processing.values()) { + if (item.message.status === 'submitted') count++; + } + return count; + } + + /** + * Records that a dequeued message was broadcast to its destination chain. + * The message stays in flight under the `submitted` status until it is + * completed, forced to `failed`, or reconciled. Returns `false` when the + * message is not actually in flight. + */ + markSubmitted(messageId: string, txHash: string): boolean { + const item = this.processing.get(messageId); + if (!item) return false; + item.submittedTxHash = txHash; + item.submittedAt = Date.now(); + item.message.status = 'submitted'; + this.emit('message-submitted', { + messageId, + txHash, + destinationChainId: item.message.destinationChainId, + }); + return true; + } + + /** The tracked status of a message, or `null` when the queue does not know it. */ + getMessageStatus(messageId: string): MessageStatus | null { + const pending = this.queue.get(messageId); + if (pending) return pending.message.status; + const inflight = this.processing.get(messageId); + if (inflight) return inflight.message.status; + if (this.failed.has(messageId)) return 'failed'; + if (this.completed.has(messageId)) return 'confirmed'; + return null; + } + + /** Read-only view of every message currently being delivered, for monitoring and reconciliation. */ + getInflightSnapshot(): InflightMessageSnapshot[] { + return Array.from(this.processing.values()).map((item) => ({ + messageId: item.message.id, + status: item.message.status, + attempts: item.attempts, + queuedAt: item.queuedAt, + inflightAt: item.inflightAt, + submittedAt: item.submittedAt, + submittedTxHash: item.submittedTxHash, + destinationChainId: item.message.destinationChainId, + message: { ...item.message }, + })); + } + + /** + * Marks an in-flight message failed immediately, with no further delivery + * attempt. Used when the destination chain proves the attempt cannot + * succeed, such as a reverted transaction. Returns `false` when the message + * is not in flight. + */ + fail(messageId: string, error?: string): boolean { + const item = this.processing.get(messageId); + if (!item) return false; + this.processing.delete(messageId); + item.message.status = 'failed'; + item.message.lastError = error; + this.failed.set(messageId, item); + this.emit('message-failed', { + messageId, + attempts: item.attempts, + lastError: error, + reason: 'forced', + }); + return true; + } + getCompletedCount(): number { return this.completed.size; } diff --git a/apps/relayer-service/src/reconciler/__tests__/message-status-reconciler.spec.ts b/apps/relayer-service/src/reconciler/__tests__/message-status-reconciler.spec.ts new file mode 100644 index 00000000..d78900bf --- /dev/null +++ b/apps/relayer-service/src/reconciler/__tests__/message-status-reconciler.spec.ts @@ -0,0 +1,434 @@ +import { MessageQueue } from '../../queue/message-queue'; +import { EvmExecutorStatusProvider, GetTransactionStatusResult, MessageStatusReconciler } from '../message-status-reconciler'; +import { CrossChainMessage, ExecutionResult, MessageStatusProvider, OnChainMessageStatus } from '../../types'; + +function makeMessage(overrides: Partial = {}): CrossChainMessage { + return { + id: 'msg-1', + sourceChainId: 'ethereum', + destinationChainId: 'stellar', + sourceTxHash: '0x' + 'a'.repeat(64), + sourceBlockNumber: 10_000_000, + messageType: 'lock', + payload: '0xdeadbeef', + sender: '0xsender', + recipient: 'Grecipient', + createdAt: 1_700_000_000_000, + status: 'pending', + retryCount: 0, + ...overrides, + }; +} + +interface ProviderApi { + provider: MessageStatusProvider; + calls: Array<{ messageId: string; txHash: string }>; +} + +function programmableProvider(chainId: string, responses: Array): ProviderApi { + const calls: Array<{ messageId: string; txHash: string }> = []; + const provider: MessageStatusProvider = { + chainId, + async getMessageStatus(messageId, txHash) { + calls.push({ messageId, txHash }); + const next = responses.shift(); + if (next instanceof Error) throw next; + return next ?? { kind: 'unknown', error: 'provider responses exhausted' }; + }, + }; + return { provider, calls }; +} + +function flushMicrotasks(): Promise { + return new Promise((resolve) => queueMicrotask(resolve)); +} + +describe('MessageStatusReconciler', () => { + let queue: MessageQueue; + + beforeEach(() => { + queue = new MessageQueue({ maxRetries: 3, retryDelayMs: 100, concurrency: 5 }); + }); + + afterEach(() => { + queue.removeAllListeners(); + jest.useRealTimers(); + }); + + function inFlightMessage(id: string, destinationChainId = 'stellar'): void { + queue.enqueue(makeMessage({ id, destinationChainId })); + queue.dequeue(); + } + + function reconciler( + config: Partial<{ + staleAfterMs: number; + intervalMs: number; + reconcileOnStart: boolean; + maxConsecutiveErrorRuns: number; + }> = {}, + providers: MessageStatusProvider[] = [], + ): MessageStatusReconciler { + return new MessageStatusReconciler({ + queue, + provider: providers, + config: { staleAfterMs: 10_000, intervalMs: 0, reconcileOnStart: false, maxConsecutiveErrorRuns: 2, ...config }, + }); + } + + it('leaves a fresh in-flight message alone', async () => { + inFlightMessage('fresh'); + const r = reconciler(); + const summary = await r.reconcile(); + expect(queue.getMessageStatus('fresh')).toBe('processing'); + expect(summary.transitions).toEqual([ + expect.objectContaining({ messageId: 'fresh', action: 'deferred', reason: 'not-stale' }), + ]); + expect(r.getStats()).toMatchObject({ runs: 1, scanned: 1, corrections: 0, deferred: 1 }); + }); + + it('re-queues a stale processing message for another attempt', async () => { + jest.useFakeTimers(); + inFlightMessage('hung'); + jest.advanceTimersByTime(10_000); + const r = reconciler(); + const summary = await r.reconcile(); + expect(queue.getMessageStatus('hung')).toBe('queued'); + expect(queue.getPendingCount()).toBe(1); + expect(summary.transitions[0]).toMatchObject({ + messageId: 'hung', + action: 'requeued', + reason: 'stale-processing', + before: 'processing', + after: 'queued', + }); + expect(r.getStats()).toMatchObject({ requeued: 1, corrections: 1 }); + }); + + it('moves a stale processing message to failed once the retry budget is spent', async () => { + jest.useFakeTimers(); + queue = new MessageQueue({ maxRetries: 0 }); + inFlightMessage('spent'); + jest.advanceTimersByTime(30_000); + const r = reconciler(); + const summary = await r.reconcile(); + expect(queue.getMessageStatus('spent')).toBe('failed'); + expect(queue.getFailedCount()).toBe(1); + expect(summary.transitions[0]).toMatchObject({ action: 'failed', reason: 'stale-processing', after: 'failed' }); + }); + + it('confirms a submitted message that is final on-chain', async () => { + inFlightMessage('sub'); + queue.markSubmitted('sub', '0xabc'); + const { provider } = programmableProvider('stellar', [{ kind: 'confirmed', blockNumber: 12345 }]); + const r = reconciler({}, [provider]); + const summary = await r.reconcile(); + expect(queue.getMessageStatus('sub')).toBe('confirmed'); + expect(queue.getCompletedMessages()[0]).toMatchObject({ + messageId: 'sub', + success: true, + transactionHash: '0xabc', + blockNumber: 12345, + }); + expect(summary.transitions[0]).toMatchObject({ action: 'confirmed', reason: 'confirmed-on-chain' }); + expect(r.getStats()).toMatchObject({ confirmed: 1, corrections: 1 }); + }); + + it('fails a submitted message whose transaction reverted, without retrying', async () => { + inFlightMessage('rev'); + queue.markSubmitted('rev', '0xrev'); + const retrying = jest.fn(); + queue.on('message-retrying', retrying); + const { provider } = programmableProvider('stellar', [{ kind: 'reverted', error: 'execution reverted' }]); + const r = reconciler({}, [provider]); + const summary = await r.reconcile(); + expect(queue.getMessageStatus('rev')).toBe('failed'); + expect(queue.getFailedCount()).toBe(1); + expect(retrying).not.toHaveBeenCalled(); + expect(summary.transitions[0]).toMatchObject({ action: 'failed', reason: 'reverted-on-chain' }); + }); + + it('does not resubmit a submitted message right after broadcast', async () => { + inFlightMessage('w'); + queue.markSubmitted('w', '0xw'); + const { provider } = programmableProvider('stellar', [{ kind: 'not-found' }]); + const r = reconciler({}, [provider]); + const summary = await r.reconcile(); + expect(queue.getMessageStatus('w')).toBe('submitted'); + expect(summary.transitions[0]).toMatchObject({ action: 'deferred', reason: 'not-stale' }); + }); + + it('re-queues a submitted message with no receipt past the stale threshold', async () => { + jest.useFakeTimers(); + inFlightMessage('lost'); + queue.markSubmitted('lost', '0xlost'); + jest.advanceTimersByTime(20_000); + const { provider } = programmableProvider('stellar', [{ kind: 'not-found' }]); + const r = reconciler({}, [provider]); + const summary = await r.reconcile(); + expect(queue.getMessageStatus('lost')).toBe('queued'); + expect(queue.getPendingCount()).toBe(1); + expect(summary.transitions[0]).toMatchObject({ action: 'requeued', reason: 'not-found-on-chain' }); + }); + + it('leaves a message untouched and counts an error when the provider throws', async () => { + inFlightMessage('boom'); + queue.markSubmitted('boom', '0xboom'); + const errorSpy = jest.fn(); + const { provider } = programmableProvider('stellar', [new Error('rpc down')]); + const r = reconciler({}, [provider]); + r.on('reconcile-error', errorSpy); + const summary = await r.reconcile(); + expect(queue.getMessageStatus('boom')).toBe('submitted'); + expect(summary.transitions[0]).toMatchObject({ action: 'deferred', reason: 'provider-error' }); + expect(errorSpy).toHaveBeenCalledWith(expect.objectContaining({ messageId: 'boom', error: 'rpc down' })); + expect(r.getStats()).toMatchObject({ errors: 1, deferred: 1, corrections: 0 }); + }); + + it('treats an unknown on-chain status as an error', async () => { + inFlightMessage('unknown'); + queue.markSubmitted('unknown', '0xu'); + const { provider } = programmableProvider('stellar', [{ kind: 'unknown', error: 'node syncing' }]); + const r = reconciler({}, [provider]); + const summary = await r.reconcile(); + expect(queue.getMessageStatus('unknown')).toBe('submitted'); + expect(summary.transitions[0]).toMatchObject({ action: 'deferred', reason: 'provider-error' }); + expect(r.getStats()).toMatchObject({ errors: 1 }); + }); + + it('reports a submitted message without a transaction hash as an error', async () => { + inFlightMessage('nohash'); + queue.markSubmitted('nohash', '0xignored'); + const snapshot = queue.getInflightSnapshot()[0]; + snapshot.submittedTxHash = undefined; + snapshot.submittedAt = undefined; + queue.getInflightSnapshot = (() => [snapshot]) as MessageQueue['getInflightSnapshot']; + + const r = reconciler(); + const summary = await r.reconcile(); + expect(summary.transitions).toEqual([ + expect.objectContaining({ messageId: 'nohash', action: 'deferred', reason: 'provider-error' }), + ]); + expect(r.getStats()).toMatchObject({ errors: 1 }); + }); + + it('defers messages when no provider covers the destination chain', async () => { + inFlightMessage('nowhere', 'near'); + queue.markSubmitted('nowhere', '0xn'); + const r = reconciler(); + const summary = await r.reconcile(); + expect(queue.getMessageStatus('nowhere')).toBe('submitted'); + expect(summary.transitions[0]).toMatchObject({ action: 'deferred', reason: 'provider-unavailable' }); + expect(r.getStats()).toMatchObject({ providerUnavailable: 1 }); + }); + + it('consults only the provider for the message destination chain', async () => { + inFlightMessage('evm-msg', 'ethereum'); + queue.markSubmitted('evm-msg', '0xe'); + const { provider: ethereumP, calls: ethCalls } = programmableProvider('ethereum', [{ kind: 'confirmed' }]); + const { provider: stellarP } = programmableProvider('stellar', [{ kind: 'confirmed' }]); + const r = reconciler({}, [ethereumP, stellarP]); + await r.reconcile(); + expect(queue.getMessageStatus('evm-msg')).toBe('confirmed'); + expect(ethCalls).toEqual([{ messageId: 'evm-msg', txHash: '0xe' }]); + }); + + it('emits message-reconciled only for actual transitions', async () => { + inFlightMessage('transition'); + queue.markSubmitted('transition', '0x2'); + const { provider } = programmableProvider('stellar', [{ kind: 'confirmed' }]); + const r = reconciler({}, [provider]); + const reconciled = jest.fn(); + r.on('message-reconciled', reconciled); + await r.reconcile(); + expect(reconciled).toHaveBeenCalledTimes(1); + expect(reconciled).toHaveBeenCalledWith( + expect.objectContaining({ messageId: 'transition', action: 'confirmed', reason: 'confirmed-on-chain' }), + ); + }); + + it('emits stale-message-detected when it acts on a stale message', async () => { + jest.useFakeTimers(); + inFlightMessage('stale'); + jest.advanceTimersByTime(10_000); + const r = reconciler(); + const stale = jest.fn(); + r.on('stale-message-detected', stale); + await r.reconcile(); + expect(stale).toHaveBeenCalledWith(expect.objectContaining({ messageId: 'stale', status: 'processing' })); + }); + + it('degrades after repeated provider errors and recovers on a correction', async () => { + let attempts = 0; + const flaky: MessageStatusProvider = { + chainId: 'stellar', + async getMessageStatus() { + attempts++; + if (attempts <= 2) throw new Error('rpc down'); + return { kind: 'confirmed' }; + }, + }; + inFlightMessage('x'); + queue.markSubmitted('x', '0xx'); + const r = reconciler({ maxConsecutiveErrorRuns: 2 }, [flaky]); + const degraded = jest.fn(); + const recovered = jest.fn(); + r.on('reconciler-degraded', degraded); + r.on('reconciler-recovered', recovered); + + await r.reconcile(); + expect(r.getStats().consecutiveErrorRuns).toBe(1); + expect(degraded).not.toHaveBeenCalled(); + + await r.reconcile(); + expect(r.getStats().consecutiveErrorRuns).toBe(2); + expect(degraded).toHaveBeenCalledTimes(1); + + await r.reconcile(); + expect(attempts).toBe(3); + expect(queue.getMessageStatus('x')).toBe('confirmed'); + expect(recovered).toHaveBeenCalledTimes(1); + expect(r.getStats().consecutiveErrorRuns).toBe(0); + expect(r.getStats()).toMatchObject({ runs: 3, errors: 2, confirmed: 1 }); + }); + + it('start() runs one pass on start when configured, then on schedule', async () => { + jest.useFakeTimers({ doNotFake: ['queueMicrotask'] }); + inFlightMessage('m'); + const summaries = jest.fn(); + const r = reconciler({ reconcileOnStart: true, intervalMs: 1000 }, []); + r.on('reconciliation-summary', summaries); + r.start(); + await flushMicrotasks(); + expect(summaries).toHaveBeenCalledTimes(1); + jest.advanceTimersByTime(1000); + await flushMicrotasks(); + expect(summaries).toHaveBeenCalledTimes(2); + r.stop(); + jest.advanceTimersByTime(3000); + expect(summaries).toHaveBeenCalledTimes(2); + }); + + it('start() is a no-op while already started', async () => { + jest.useFakeTimers(); + const started = jest.fn(); + const r = reconciler({ reconcileOnStart: false, intervalMs: 1000 }, []); + r.on('reconciler-started', started); + r.start(); + r.start(); + expect(started).toHaveBeenCalledTimes(1); + r.stop(); + }); + + it('emits stop events and cleans up listeners with dispose()', async () => { + const r = reconciler(); + const stopped = jest.fn(); + r.on('reconciler-stopped', stopped); + r.stop(); + expect(stopped).toHaveBeenCalledTimes(1); + r.dispose(); + expect(r.listenerCount('reconciliation-summary')).toBe(0); + }); + + it('shares a single run across concurrent callers', async () => { + inFlightMessage('m'); + queue.markSubmitted('m', '0xm'); + const { provider } = programmableProvider('stellar', [{ kind: 'confirmed' }]); + const r = reconciler({}, [provider]); + const [a, b] = await Promise.all([r.reconcile(), r.reconcile()]); + expect(a.runId).toBe(b.runId); + expect(r.getStats().runs).toBe(1); + }); + + it('EvmExecutorStatusProvider maps the executor chain view', async () => { + const executor: { getTransactionStatus(txHash: string): Promise } = { + getTransactionStatus: (_txHash: string) => Promise.resolve({ confirmed: true, blockNumber: 9 }), + }; + const provider = new EvmExecutorStatusProvider('ethereum', executor); + expect(provider.chainId).toBe('ethereum'); + expect(await provider.getMessageStatus('m', '0x1')).toEqual({ kind: 'confirmed', blockNumber: 9 }); + + executor.getTransactionStatus = () => Promise.resolve({ confirmed: false }); + expect(await provider.getMessageStatus('m', '0x2')).toEqual({ kind: 'not-found' }); + }); + + it('treats an unknown on-chain status with no detail as an error', async () => { + inFlightMessage('silent'); + queue.markSubmitted('silent', '0xs'); + const { provider } = programmableProvider('stellar', [{ kind: 'unknown' }]); + const r = reconciler({}, [provider]); + const errorSpy = jest.fn(); + r.on('reconcile-error', errorSpy); + const summary = await r.reconcile(); + expect(queue.getMessageStatus('silent')).toBe('submitted'); + expect(summary.transitions[0]).toMatchObject({ action: 'deferred', reason: 'provider-error' }); + expect(errorSpy).toHaveBeenCalledWith(expect.objectContaining({ messageId: 'silent', error: 'on-chain status unknown' })); + }); + + it('normalises a non-Error provider failure', async () => { + inFlightMessage('str'); + queue.markSubmitted('str', '0xstr'); + const provider: MessageStatusProvider = { + chainId: 'stellar', + getMessageStatus: () => Promise.reject('gremlin'), + }; + const r = reconciler({}, [provider]); + const errorSpy = jest.fn(); + r.on('reconcile-error', errorSpy); + await r.reconcile(); + expect(queue.getMessageStatus('str')).toBe('submitted'); + expect(errorSpy).toHaveBeenCalledWith(expect.objectContaining({ error: 'gremlin' })); + }); + + it('accepts a single provider instead of an array', async () => { + inFlightMessage('one'); + queue.markSubmitted('one', '0x1'); + const { provider } = programmableProvider('stellar', [{ kind: 'confirmed' }]); + const r = new MessageStatusReconciler({ queue, provider, config: { intervalMs: 0 } }); + await r.reconcile(); + expect(queue.getMessageStatus('one')).toBe('confirmed'); + }); + + describe('defensive branches for mid-run races', () => { + it('defers an untracked in-flight status', async () => { + inFlightMessage('m'); + const snapshot = queue.getInflightSnapshot()[0]; + snapshot.status = 'queued'; + snapshot.inflightAt = undefined; + queue.getInflightSnapshot = (() => [snapshot]) as MessageQueue['getInflightSnapshot']; + const r = reconciler(); + const summary = await r.reconcile(); + expect(summary.transitions[0]).toMatchObject({ action: 'deferred', reason: 'already-resolved' }); + }); + + it('defers a confirmation whose completion was overwritten mid-run', async () => { + inFlightMessage('m'); + queue.markSubmitted('m', '0xm'); + const { provider } = programmableProvider('stellar', [{ kind: 'confirmed' }]); + const r = reconciler({}, [provider]); + queue.complete = jest.fn((_result: ExecutionResult) => Promise.resolve()) as MessageQueue['complete']; + const summary = await r.reconcile(); + expect(summary.transitions[0]).toMatchObject({ action: 'deferred', reason: 'already-resolved' }); + }); + + it('defers a requeue whose retry path resolved differently', async () => { + jest.useFakeTimers(); + inFlightMessage('m'); + jest.advanceTimersByTime(20_000); + const r = reconciler(); + queue.complete = jest.fn((_result: ExecutionResult) => Promise.resolve()) as MessageQueue['complete']; + const summary = await r.reconcile(); + expect(summary.transitions[0]).toMatchObject({ action: 'deferred', reason: 'already-resolved' }); + }); + + it('defers a forced failure that did not take effect', async () => { + inFlightMessage('m'); + queue.markSubmitted('m', '0xm'); + const { provider } = programmableProvider('stellar', [{ kind: 'reverted' }]); + const r = reconciler({}, [provider]); + queue.fail = jest.fn(() => false) as MessageQueue['fail']; + const summary = await r.reconcile(); + expect(summary.transitions[0]).toMatchObject({ action: 'deferred', reason: 'already-resolved' }); + }); + }); +}); \ No newline at end of file diff --git a/apps/relayer-service/src/reconciler/__tests__/reconciliation-pipeline.spec.ts b/apps/relayer-service/src/reconciler/__tests__/reconciliation-pipeline.spec.ts new file mode 100644 index 00000000..887be3b2 --- /dev/null +++ b/apps/relayer-service/src/reconciler/__tests__/reconciliation-pipeline.spec.ts @@ -0,0 +1,172 @@ +/** + * Integration coverage for message status reconciliation: the queue, the + * reconciler, the executor-backed status provider and a mocked destination + * chain work together so a broadcast message is either confirmed, failed, or + * returned to the queue for another attempt - and never dropped or lost. + */ +jest.mock('axios', () => ({ __esModule: true, default: { post: jest.fn() } })); + +import axios from 'axios'; +import { MessageQueue } from '../../queue/message-queue'; +import { EvmExecutor } from '../../executors/evm-executor'; +import { EvmExecutorStatusProvider, MessageStatusReconciler } from '../message-status-reconciler'; +import { CrossChainMessage, MessageStatusProvider, OnChainMessageStatus } from '../../types'; + +const post = axios.post as jest.Mock; + +function makeMessage(overrides: Partial = {}): CrossChainMessage { + return { + id: 'msg-1', + sourceChainId: 'ethereum', + destinationChainId: 'ethereum', + sourceTxHash: '0x' + 'a'.repeat(64), + sourceBlockNumber: 10_000_000, + messageType: 'lock', + payload: '0xdeadbeef', + sender: '0x' + '1'.repeat(40), + recipient: '0x' + '2'.repeat(40), + amount: '1000', + createdAt: 1_700_000_000_000, + status: 'pending', + retryCount: 0, + ...overrides, + }; +} + +interface ProviderApi { + provider: MessageStatusProvider; + calls: Array<{ messageId: string; txHash: string }>; +} + +function programmableProvider(chainId: string, responses: Array): ProviderApi { + const calls: Array<{ messageId: string; txHash: string }> = []; + const provider: MessageStatusProvider = { + chainId, + async getMessageStatus(messageId, txHash) { + calls.push({ messageId, txHash }); + const next = responses.shift(); + if (next instanceof Error) throw next; + return next ?? { kind: 'unknown', error: 'provider responses exhausted' }; + }, + }; + return { provider, calls }; +} + +describe('message status reconciliation pipeline', () => { + afterEach(() => { + jest.useRealTimers(); + post.mockReset(); + }); + + it('confirms a broadcast message through the executor-backed provider and real RPC shapes', async () => { + post.mockImplementation(async (_url: string, body: { method: string }) => { + if (body.method === 'eth_getTransactionReceipt') { + return { data: { result: { blockNumber: '0x64', status: '0x1', gasUsed: '0x5208' } } }; + } + if (body.method === 'eth_blockNumber') { + return { data: { result: '0x65' } }; + } + return { data: { error: { message: `unexpected RPC call ${body.method}` } } }; + }); + + const queue = new MessageQueue({ concurrency: 5 }); + queue.enqueue(makeMessage({ id: 'peer', destinationChainId: 'ethereum' })); + queue.dequeue(); + queue.markSubmitted('peer', '0xabc'); + + const executor = new EvmExecutor({ + chainId: 'ethereum', + chainType: 'evm', + rpcUrl: 'http://localhost:8545', + confirmationBlocks: 1, + confirmationPollIntervalMs: 1, + }); + const r = new MessageStatusReconciler({ + queue, + provider: new EvmExecutorStatusProvider('ethereum', executor), + config: { intervalMs: 0 }, + }); + const reconciled = jest.fn(); + r.on('message-reconciled', reconciled); + + const summary = await r.reconcile(); + + expect(queue.getMessageStatus('peer')).toBe('confirmed'); + expect(queue.getCompletedMessages()[0]).toMatchObject({ messageId: 'peer', success: true, transactionHash: '0xabc' }); + expect(r.getStats()).toMatchObject({ confirmed: 1, corrections: 1, runs: 1 }); + expect(reconciled).toHaveBeenCalledWith( + expect.objectContaining({ messageId: 'peer', action: 'confirmed', reason: 'confirmed-on-chain' }), + ); + expect(post).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ method: 'eth_getTransactionReceipt' }), + expect.anything(), + ); + expect(summary.transitions).toHaveLength(1); + }); + + it('recovers a lost submission: requeues after no receipt, then confirms on resubmission', async () => { + jest.useFakeTimers(); + const { provider, calls } = programmableProvider('ethereum', [ + { kind: 'not-found' }, + { kind: 'confirmed', blockNumber: 42 }, + ]); + const queue = new MessageQueue({ maxRetries: 5, retryDelayMs: 0 }); + queue.enqueue(makeMessage({ id: 'lost', destinationChainId: 'ethereum' })); + queue.dequeue(); + queue.markSubmitted('lost', '0xtx1'); + + const r = new MessageStatusReconciler({ queue, provider, config: { staleAfterMs: 10_000, intervalMs: 0 } }); + jest.advanceTimersByTime(15_000); + await r.reconcile(); + expect(queue.getMessageStatus('lost')).toBe('queued'); + expect(queue.getPendingCount()).toBe(1); + + // The relayer dequeues the recovered message and broadcasts it again. + queue.dequeue(); + queue.markSubmitted('lost', '0xtx2'); + await r.reconcile(); + + expect(queue.getMessageStatus('lost')).toBe('confirmed'); + expect(queue.getCompletedMessages()[0]).toMatchObject({ messageId: 'lost', success: true, transactionHash: '0xtx2' }); + expect(calls).toEqual([ + { messageId: 'lost', txHash: '0xtx1' }, + { messageId: 'lost', txHash: '0xtx2' }, + ]); + expect(r.getStats()).toMatchObject({ requeued: 1, confirmed: 1, corrections: 2 }); + }); + + it('fails a reverted submission and never sends it again', async () => { + const { provider } = programmableProvider('ethereum', [{ kind: 'reverted', error: 'TRANSFER_FROM_FAILED' }]); + const queue = new MessageQueue({ maxRetries: 50, retryDelayMs: 0 }); + queue.enqueue(makeMessage({ id: 'rev', destinationChainId: 'ethereum' })); + queue.dequeue(); + queue.markSubmitted('rev', '0xrev'); + + const r = new MessageStatusReconciler({ queue, provider, config: { intervalMs: 0 } }); + const failed = jest.fn(); + queue.on('message-failed', failed); + await r.reconcile(); + + expect(queue.getMessageStatus('rev')).toBe('failed'); + expect(queue.getFailedCount()).toBe(1); + expect(queue.getPendingCount()).toBe(0); + expect(queue.getFailedMessages()[0]).toMatchObject({ lastError: expect.stringContaining('TRANSFER_FROM_FAILED') }); + expect(failed).toHaveBeenCalledWith(expect.objectContaining({ messageId: 'rev', reason: 'forced' })); + }); + + it('leaves healthy in-flight deliveries untouched while reconciliation runs', async () => { + const queue = new MessageQueue({ concurrency: 5 }); + queue.enqueue(makeMessage({ id: 'healthy', destinationChainId: 'ethereum' })); + queue.dequeue(); + const r = new MessageStatusReconciler({ queue, config: { staleAfterMs: 60_000, intervalMs: 0 } }); + const summary = await r.reconcile(); + + expect(queue.getMessageStatus('healthy')).toBe('processing'); + expect(summary.transitions[0]).toMatchObject({ action: 'deferred', reason: 'not-stale' }); + + await queue.complete({ messageId: 'healthy', success: true, transactionHash: '0xok', timestamp: Date.now() }); + expect(queue.getMessageStatus('healthy')).toBe('confirmed'); + expect(r.getStats()).toMatchObject({ corrections: 0, confirmed: 0 }); + }); +}); \ No newline at end of file diff --git a/apps/relayer-service/src/reconciler/message-status-reconciler.ts b/apps/relayer-service/src/reconciler/message-status-reconciler.ts new file mode 100644 index 00000000..3e5d6f77 --- /dev/null +++ b/apps/relayer-service/src/reconciler/message-status-reconciler.ts @@ -0,0 +1,402 @@ +import { EventEmitter } from 'events'; +import { MessageQueue } from '../queue/message-queue'; +import { + ExecutionResult, + InflightMessageSnapshot, + MessageReconciliation, + MessageStatusProvider, + OnChainMessageStatus, + ReconciliationAction, + ReconciliationConfig, + ReconciliationReason, + ReconciliationStats, + ReconciliationSummary, +} from '../types'; + +export const DEFAULT_RECONCILIATION_CONFIG: ReconciliationConfig = { + staleAfterMs: 5 * 60_000, + intervalMs: 60_000, + reconcileOnStart: true, + maxConsecutiveErrorRuns: 3, +}; + +export interface MessageStatusReconcilerOptions { + queue: MessageQueue; + /** Chain status providers. A message is only reconciled when a provider exists for its destination chain. */ + provider?: MessageStatusProvider | MessageStatusProvider[]; + config?: Partial; +} + +function emptyStats(): ReconciliationStats { + return { + runs: 0, + scanned: 0, + corrections: 0, + requeued: 0, + confirmed: 0, + failed: 0, + deferred: 0, + errors: 0, + providerUnavailable: 0, + consecutiveErrorRuns: 0, + lastRunAt: null, + lastRunDurationMs: null, + }; +} + +function errMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +/** + * Repairs drift between the queue's tracked message status and the truth on + * the destination chain: + * + * - a `processing` message that has been in flight past `staleAfterMs` + * (an executor that hung without reporting) is returned to `queued` for + * another attempt, or moved to `failed` once the retry budget is spent; + * - a `submitted` message whose transaction is confirmed on-chain is + * completed as `confirmed`; + * - a `submitted` message whose transaction reverted is failed immediately + * (a revert is treated as deterministic, so retrying would only burn gas); + * - a `submitted` message with no destination receipt after `staleAfterMs` + * is returned to `queued` for resubmission; + * - when the destination chain cannot be reached, the message is left in + * flight and the run is counted as an error. + * + * Reconciliation never touches a freshly dequeued message (`staleAfterMs` + * grace), a `queued` or `failed` message, or a message without a provider, so + * a healthy pipeline is unaffected and nothing is silently resurrected. + */ +export class MessageStatusReconciler extends EventEmitter { + private readonly queue: MessageQueue; + private readonly config: ReconciliationConfig; + private readonly providers: Map; + private readonly stats: ReconciliationStats = emptyStats(); + private timer: ReturnType | null = null; + private nextRunId = 0; + private currentRun: Promise | null = null; + + constructor(options: MessageStatusReconcilerOptions) { + super(); + this.queue = options.queue; + const providers = options.provider + ? Array.isArray(options.provider) + ? options.provider + : [options.provider] + : []; + this.providers = new Map(providers.map((provider) => [provider.chainId, provider])); + this.config = { ...DEFAULT_RECONCILIATION_CONFIG, ...options.config }; + } + + /** Runs one reconciliation pass. Concurrent calls share a single in-flight run. */ + reconcile(): Promise { + if (!this.currentRun) { + this.currentRun = this.run().finally(() => { + this.currentRun = null; + }); + } + return this.currentRun; + } + + /** Starts periodic reconciliation. Also runs one pass when `reconcileOnStart` is set. */ + start(): void { + if (this.timer) return; + this.emit('reconciler-started', { config: { ...this.config } }); + if (this.config.reconcileOnStart) { + void this.reconcile(); + } + if (this.config.intervalMs > 0) { + this.timer = setInterval(() => void this.reconcile(), this.config.intervalMs); + } + } + + /** Stops periodic reconciliation. A run already in flight is allowed to finish. */ + stop(): void { + if (this.timer) { + clearInterval(this.timer); + this.timer = null; + } + this.emit('reconciler-stopped'); + } + + dispose(): void { + this.stop(); + this.removeAllListeners(); + } + + getStats(): ReconciliationStats { + return { ...this.stats }; + } + + private async run(): Promise { + const startedAt = Date.now(); + const runId = ++this.nextRunId; + const snapshot = this.queue.getInflightSnapshot(); + this.stats.runs++; + this.stats.scanned += snapshot.length; + + const degraded = (count: number) => count > 0 && count >= this.config.maxConsecutiveErrorRuns; + const wasDegraded = degraded(this.stats.consecutiveErrorRuns); + + const transitions: MessageReconciliation[] = []; + for (const item of snapshot) { + transitions.push(await this.reconcileItem(item, startedAt)); + } + + const durationMs = Date.now() - startedAt; + this.stats.lastRunAt = startedAt; + this.stats.lastRunDurationMs = durationMs; + + const hadErrors = transitions.some((t) => t.reason === 'provider-error'); + const hadCorrections = transitions.some( + (t) => t.action === 'requeued' || t.action === 'confirmed' || t.action === 'failed', + ); + this.stats.consecutiveErrorRuns = hadErrors && !hadCorrections ? this.stats.consecutiveErrorRuns + 1 : 0; + + const isDegraded = degraded(this.stats.consecutiveErrorRuns); + if (isDegraded && !wasDegraded) { + this.emit('reconciler-degraded', { + runId, + consecutiveErrorRuns: this.stats.consecutiveErrorRuns, + message: + 'repeated provider errors with no corrections; check the chain status providers before the queue fills with stuck submitted messages', + }); + } else if (!isDegraded && wasDegraded) { + this.emit('reconciler-recovered', { runId }); + } + + const summary: ReconciliationSummary = { + runId, + startedAt, + durationMs, + scanned: transitions.length, + transitions, + }; + this.emit('reconciliation-summary', summary); + return summary; + } + + private async reconcileItem(item: InflightMessageSnapshot, now: number): Promise { + const ageMs = now - (item.inflightAt ?? item.queuedAt); + + if (item.status === 'processing') { + if (ageMs < this.config.staleAfterMs) { + return this.defer(item, 'not-stale', `in flight for ${ageMs}ms, within the stale threshold`); + } + this.emit('stale-message-detected', { + messageId: item.messageId, + status: item.status, + ageMs, + staleAfterMs: this.config.staleAfterMs, + }); + return this.requeueOrFail(item, 'stale-processing', `no completion reported within ${ageMs}ms`); + } + + if (item.status === 'submitted') { + if (!item.submittedTxHash) { + this.stats.errors++; + this.emit('reconcile-error', { + messageId: item.messageId, + chainId: item.destinationChainId, + error: 'submitted message has no transaction hash to reconcile against', + }); + return this.defer(item, 'provider-error', 'submitted message has no transaction hash'); + } + + const provider = this.providers.get(item.destinationChainId); + if (!provider) { + this.stats.providerUnavailable++; + return this.defer( + item, + 'provider-unavailable', + `no MessageStatusProvider for chain ${item.destinationChainId}`, + ); + } + + return this.reconcileSubmitted(item, provider, ageMs); + } + + return this.defer(item, 'already-resolved', `untracked in-flight status ${item.status}`); + } + + private async reconcileSubmitted( + item: InflightMessageSnapshot, + provider: MessageStatusProvider, + ageMs: number, + ): Promise { + let onChain: OnChainMessageStatus; + try { + onChain = await provider.getMessageStatus(item.messageId, item.submittedTxHash as string); + } catch (error) { + const message = errMessage(error); + this.stats.errors++; + this.emit('reconcile-error', { messageId: item.messageId, chainId: item.destinationChainId, error: message }); + return this.defer(item, 'provider-error', message); + } + + switch (onChain.kind) { + case 'confirmed': + return this.confirm(item, onChain.blockNumber); + case 'reverted': + return this.forceFail(item, 'reverted-on-chain', onChain.error ?? 'destination transaction reverted'); + case 'not-found': + if (ageMs < this.config.staleAfterMs) { + return this.defer(item, 'not-stale', `no destination receipt after ${ageMs}ms`); + } + this.emit('stale-message-detected', { + messageId: item.messageId, + status: item.status, + ageMs, + staleAfterMs: this.config.staleAfterMs, + }); + return this.requeueOrFail(item, 'not-found-on-chain', `no destination receipt after ${ageMs}ms`); + case 'unknown': + default: + this.stats.errors++; + this.emit('reconcile-error', { + messageId: item.messageId, + chainId: item.destinationChainId, + error: onChain.error ?? 'on-chain status unknown', + }); + return this.defer(item, 'provider-error', onChain.error ?? 'on-chain status unknown'); + } + } + + private async confirm(item: InflightMessageSnapshot, blockNumber?: number): Promise { + const result: ExecutionResult = { + messageId: item.messageId, + success: true, + transactionHash: item.submittedTxHash, + blockNumber, + timestamp: Date.now(), + }; + await this.queue.complete(result); + + const after = this.queue.getMessageStatus(item.messageId); + if (after !== 'confirmed') { + return this.defer(item, 'already-resolved', 'message was resolved by another path during this run'); + } + this.stats.corrections++; + this.stats.confirmed++; + return this.concluded(item, 'confirmed', 'confirmed-on-chain', this.confirmDetail(blockNumber)); + } + + private async requeueOrFail( + item: InflightMessageSnapshot, + reason: ReconciliationReason, + detail?: string, + ): Promise { + await this.queue.complete({ + messageId: item.messageId, + success: false, + error: this.reasonError(reason, detail), + timestamp: Date.now(), + }); + + const after = this.queue.getMessageStatus(item.messageId); + if (after === 'queued') { + this.stats.corrections++; + this.stats.requeued++; + return this.concluded(item, 'requeued', reason, detail); + } + if (after === 'failed') { + this.stats.corrections++; + this.stats.failed++; + return this.concluded(item, 'failed', reason, detail); + } + return this.defer(item, 'already-resolved', `resolved to ${after} by another path during this run`); + } + + private async forceFail( + item: InflightMessageSnapshot, + reason: ReconciliationReason, + detail?: string, + ): Promise { + this.queue.fail(item.messageId, this.reasonError(reason, detail)); + + const after = this.queue.getMessageStatus(item.messageId); + if (after !== 'failed') { + return this.defer(item, 'already-resolved', 'message was resolved by another path during this run'); + } + this.stats.corrections++; + this.stats.failed++; + return this.concluded(item, 'failed', reason, detail); + } + + private concluded( + item: InflightMessageSnapshot, + action: ReconciliationAction, + reason: ReconciliationReason, + detail?: string, + ): MessageReconciliation { + const transition: MessageReconciliation = { + messageId: item.messageId, + destinationChainId: item.destinationChainId, + before: item.status, + after: this.queue.getMessageStatus(item.messageId), + action, + reason, + detail, + }; + this.emit('message-reconciled', transition); + return transition; + } + + private defer(item: InflightMessageSnapshot, reason: ReconciliationReason, detail?: string): MessageReconciliation { + this.stats.deferred++; + return { + messageId: item.messageId, + destinationChainId: item.destinationChainId, + before: item.status, + after: item.status, + action: 'deferred', + reason, + detail, + }; + } + + private confirmDetail(blockNumber?: number): string { + if (blockNumber === undefined) return 'receipt is final on the destination chain'; + return `receipt is final on the destination chain in block ${blockNumber}`; + } + + private reasonError(reason: ReconciliationReason, detail?: string): string { + return `reconcile: ${reason}${detail ? ` (${detail})` : ''}`; + } +} + +export interface GetTransactionStatusResult { + confirmed: boolean; + blockNumber?: number; +} + +/** + * Reference `MessageStatusProvider` for EVM destinations. It reuses the + * executor's own chain view (`EvmExecutor.getTransactionStatus`), so + * reconciliation applies the same confirmation-depth policy as live delivery. + * `confirmed: false` from the executor — a missing receipt, or one not yet + * deep enough — maps to `not-found`, which reconciliation only acts on once + * the message is past its stale threshold. This avoids thrash right after a + * broadcast. + */ +export class EvmExecutorStatusProvider implements MessageStatusProvider { + readonly chainId: string; + private readonly executor: { getTransactionStatus(txHash: string): Promise }; + + constructor( + chainId: string, + executor: { getTransactionStatus(txHash: string): Promise }, + ) { + this.chainId = chainId; + this.executor = executor; + } + + async getMessageStatus(_messageId: string, txHash: string): Promise { + const result = await this.executor.getTransactionStatus(txHash); + if (result.confirmed) { + return { kind: 'confirmed', blockNumber: result.blockNumber }; + } + return { kind: 'not-found' }; + } +} \ No newline at end of file diff --git a/apps/relayer-service/src/types.ts b/apps/relayer-service/src/types.ts index 6ce8295d..aecc0343 100644 --- a/apps/relayer-service/src/types.ts +++ b/apps/relayer-service/src/types.ts @@ -109,4 +109,122 @@ export interface MessageQueueItem { queuedAt: number; nextRetryAt: number; attempts: number; + /** Wall-clock time the current delivery attempt started (when the message left the queue). */ + inflightAt?: number; + /** Destination transaction hash once the message has been broadcast, if any. */ + submittedTxHash?: string; + /** Wall-clock time the message entered the `submitted` state. */ + submittedAt?: number; +} + +/** Read-only view of a message currently being delivered, for monitoring and reconciliation. */ +export interface InflightMessageSnapshot { + messageId: string; + status: MessageStatus; + attempts: number; + queuedAt: number; + inflightAt?: number; + submittedAt?: number; + submittedTxHash?: string; + destinationChainId: string; + message: CrossChainMessage; +} + +/** + * The actual state of a transaction on its destination chain. + * `not-found` means no receipt exists at the queried state; `unknown` means the + * provider could not reach the chain or could not answer. + */ +export type OnChainMessageStatus = + | { kind: 'confirmed'; blockNumber?: number } + | { kind: 'reverted'; error?: string } + | { kind: 'not-found' } + | { kind: 'unknown'; error?: string }; + +/** + * Resolves whether a broadcast message actually landed on its destination + * chain. Implementations are chain-specific: the relayer ships an EVM adapter + * backed by `EvmExecutor.getTransactionStatus`, and adapters for other chains + * can be built to the same shape. + */ +export interface MessageStatusProvider { + /** Chain this provider answers for; matched against `CrossChainMessage.destinationChainId`. */ + chainId: string; + getMessageStatus(messageId: string, txHash: string): Promise; +} + +export interface ReconciliationConfig { + /** An in-flight message older than this (measured from when it left the queue) is reconciled. */ + staleAfterMs: number; + /** Automatic poll interval. `0` disables the timer; `start()` stays an explicit call. */ + intervalMs: number; + /** Run one `reconcile()` pass when `start()` is called, before the first interval tick. */ + reconcileOnStart: boolean; + /** + * A run that only sees provider errors and makes no corrections increments a + * consecutive counter. When the counter reaches this threshold the + * reconciler emits `reconciler-degraded` (and `reconciler-recovered` once it + * makes a correction again). Runs always keep trying, so a recovered + * provider resumes reconciliation without operator action. + */ + maxConsecutiveErrorRuns: number; +} + +export type ReconciliationAction = + /** The tracked status already matches the source of truth; nothing was moved. */ + | 'no-op' + /** Moved back to `queued` for another delivery attempt. */ + | 'requeued' + /** Marked `confirmed` because the destination receipt is final. */ + | 'confirmed' + /** Marked `failed` and left alone until an operator intervenes. */ + | 'failed' + /** Left in flight because the truth could not be established during this run. */ + | 'deferred'; + +export type ReconciliationReason = + | 'not-stale' + | 'stale-processing' + | 'stale-submitted' + | 'confirmed-on-chain' + | 'reverted-on-chain' + | 'not-found-on-chain' + | 'provider-unavailable' + | 'provider-error' + | 'already-resolved'; + +export interface MessageReconciliation { + messageId: string; + destinationChainId: string; + /** Tracked status before the reconciliation run. */ + before: MessageStatus; + /** Tracked status afterwards; unchanged for `no-op` and `deferred`. */ + after: MessageStatus | null; + action: ReconciliationAction; + reason: ReconciliationReason; + detail?: string; +} + +export interface ReconciliationSummary { + runId: number; + startedAt: number; + durationMs: number; + scanned: number; + transitions: MessageReconciliation[]; +} + +export interface ReconciliationStats { + runs: number; + scanned: number; + /** Successful corrections: requeued, confirmed or failed transitions. */ + corrections: number; + requeued: number; + confirmed: number; + failed: number; + deferred: number; + errors: number; + providerUnavailable: number; + consecutiveErrorRuns: number; + lastRunAt: number | null; + lastRunDurationMs: number | null; } diff --git a/docs/CROSS_CHAIN_DUPLICATE_MESSAGE_DETECTION.md b/docs/CROSS_CHAIN_DUPLICATE_MESSAGE_DETECTION.md index 097a04e7..56c9c848 100644 --- a/docs/CROSS_CHAIN_DUPLICATE_MESSAGE_DETECTION.md +++ b/docs/CROSS_CHAIN_DUPLICATE_MESSAGE_DETECTION.md @@ -2,6 +2,8 @@ The relayer (`apps/relayer-service`) rejects cross-chain messages it has already accepted, so a re-delivered message cannot be executed twice on the destination chain. This is the relayer's first line of defence. It does not replace on-chain replay protection (see [Dependencies](#dependencies)). +Once a message is accepted it is tracked through delivery. If the queue's local view of a message's state ever diverges from the destination chain - a hung executor, a lost completion callback, a receipt that never finalised - see [Message status reconciliation](MESSAGE_STATUS_RECONCILIATION.md), which repairs that divergence. + ## What counts as a duplicate `MessageQueue.enqueue()` checks every message before queueing it and returns `false` when it is rejected. diff --git a/docs/MESSAGE_STATUS_RECONCILIATION.md b/docs/MESSAGE_STATUS_RECONCILIATION.md new file mode 100644 index 00000000..ef47172b --- /dev/null +++ b/docs/MESSAGE_STATUS_RECONCILIATION.md @@ -0,0 +1,104 @@ +# Message status reconciliation + +The relayer (`apps/relayer-service`) keeps a local view of every message it is delivering. `MessageQueue` tracks each message as `queued` → `processing` → `submitted` → `confirmed`/`failed`. That view is only trustworthy while every component reports back. The `MessageStatusReconciler` periodically compares the tracked status of in-flight messages with the truth on the destination chain and repairs the divergence, so a relayer restart, a hung executor, or a dropped callback cannot leave a message stuck in flight forever. + +Reconciliation is the relayer's parsing/validation layer (see [Cross-chain message validation](CROSS_CHAIN_MESSAGE_VALIDATION.md)) extended forward: validation keeps bad messages out at intake, and reconciliation keeps good messages from dying in the middle. + +## What gets reconciled + +`reconcile()` examines only in-flight messages (`processing` and `submitted`, from `queue.getInflightSnapshot()`). A message younger than `staleAfterMs` is never touched. `queued` and `failed` messages are never touched automatically. + +| Tracked state | On-chain truth | Action | +| --- | --- | --- | +| `processing` | unknown (executor never reported) | After `staleAfterMs`, returned to `queued` for another attempt. Once the retry budget is spent it moves to `failed` | +| `submitted` | receipt is final | Completed as `confirmed` with the transaction hash and block | +| `submitted` | receipt shows a revert | Moved straight to `failed` — a revert is treated as deterministic, so retrying would only burn gas | +| `submitted` | no receipt | Left in flight while inside `staleAfterMs`; past the threshold it is returned to `queued` so it can be broadcast again | +| `submitted` | provider unavailable / unknown / error | Left in flight (`deferred`), counted as an error; retried on the next run | + +Reconciliation never fabricates a transaction. `confirmed` and `not-found` decisions come only from the configured `MessageStatusProvider` for the message's destination chain; a message with no provider for its chain is always left in flight. + +## Ordering and safety + +- Reconciliation runs on `queue.complete()` and `queue.fail()`, so its transitions reuse the queue's normal retry/backoff accounting and emit the same lifecycle events a live delivery would. +- A `submitted` message with no destination receipt is only resubmitted once it is past `staleAfterMs`. Configure `staleAfterMs` larger than the destination chain's expected time to finality, and larger than `deduplication.windowMs`, so a receipt that is merely slow is never mistaken for a lost one. +- `processing` staleness is measured from the start of the current delivery attempt (`inflightAt`, set at dequeue), not from the original enqueue, so a normal retry loop is never flagged stale. +- A completion that arrives twice (the live path and the reconciler) is recorded once; the first outcome wins, so a late reconciled result cannot overwrite the real one. + +## Events + +| Event | Payload | +| --- | --- | +| `message-submitted` (queue) | `{ messageId, txHash, destinationChainId }` | +| `message-reconciled` | `{ messageId, destinationChainId, before, after, action, reason, detail? }` | +| `stale-message-detected` | `{ messageId, status, ageMs, staleAfterMs }` | +| `reconcile-error` | `{ messageId, chainId, error }` | +| `reconciler-degraded` | `{ runId, consecutiveErrorRuns, message }` | +| `reconciler-recovered` | `{ runId }` | +| `reconciliation-summary` | `ReconciliationSummary` with the full `transitions` list | + +`message-reconciled` fires only for actual transitions (requeued / confirmed / failed). `no-op` and `deferred` outcomes appear in the summary but are not emitted individually. + +## Metrics + +`reconciler.getStats()` returns `{ runs, scanned, corrections, requeued, confirmed, failed, deferred, errors, providerUnavailable, consecutiveErrorRuns, lastRunAt, lastRunDurationMs }`. + +Operationally: + +- A rising `requeued`/`not-found-on-chain` rate usually means the destination RPC is lagging behind the coordinator's estimate of finality — widen `staleAfterMs`. +- `deferred` with `provider-unavailable` means a destination chain has no provider registered; wire one up or accept that messages on that chain are never auto-recovered. +- A climb in `consecutiveErrorRuns` ends in a `reconciler-degraded` alert; alert on it. Runs keep trying, so the first run that makes a correction emits `reconciler-recovered` automatically. + +## Configuration + +```ts +import { + MessageQueue, + MessageStatusReconciler, + EvmExecutorStatusProvider, + EvmExecutor, +} from '@bridgewise/relayer-service'; + +const queue = new MessageQueue({ deduplication: { windowMs: 10 * 60_000 } }); + +const executor = new EvmExecutor({ + chainId: 'ethereum', + chainType: 'evm', + rpcUrl: process.env.ETH_RPC_URL, + // ... +}); + +const reconciler = new MessageStatusReconciler({ + queue, + provider: new EvmExecutorStatusProvider('ethereum', executor), // or an array, one per chain + config: { + staleAfterMs: 5 * 60_000, // default: larger than the chain's time to finality + intervalMs: 60_000, // default; 0 disables the timer + reconcileOnStart: true, // default: run one pass at start() + maxConsecutiveErrorRuns: 3, // default: alert threshold for the degraded event + }, +}); + +reconciler.on('message-reconciled', (event) => metrics.observe(event)); +reconciler.on('reconciler-degraded', () => pager.ping('reconciliation degraded')); +reconciler.start(); // explicit: reconciliation is opt-in per deployment +``` + +### Message status provider + +A `MessageStatusProvider` is a single function, `getMessageStatus(messageId, txHash)`, returning one of `confirmed` (with an optional block), `reverted` (with an optional error), `not-found`, or `unknown` (with an optional error). The relayer ships `EvmExecutorStatusProvider`, which reuses `EvmExecutor.getTransactionStatus` — the same RPC and confirmation-depth policy as live delivery, so there is exactly one definition of "final" per chain. Soroban and Solana adapters can be built to the same shape. + +## Dependencies + +- Upstream: none hard; the reconciler talks to the queue it is given. +- Downstream: reconciliation depends on a reachable, honest `MessageStatusProvider` per destination chain. A provider that answers `not-found` too early (before finality) is the main correctness risk — it is why the stale threshold exists. +- `deduplication.windowMs` must exceed `staleAfterMs`: a resubmission after an erroneous `not-found` would otherwise be rejected at the queue as a repeat of the original broadcast. +- Logging of a `message-reconciled` event carries no user content; the `errors` payloads from providers are the RPC error text and must be treated as untrusted log data. + +## Troubleshooting + +- **Messages are `failed` with `stale-processing` and the retry budget is spent.** The executor's `execute()` either hangs or throws without a useful result. Check executor logs and RPC health; raise `maxRetries` after fixing the root cause and use `queue.retryFailed(messageId)` to resume. +- **`submitted` messages stay `deferred` forever with `provider-unavailable`.** No provider matches `destinationChainId`. Register a provider for that chain (or this chain's messages are deliberately outside reconciliation). +- **`reconciler-degraded` fires repeatedly.** Status providers are erroring. Verify RPC nodes, then confirm the first successful run that corrects a message flips the counter back to `0` and emits `reconciler-recovered`. +- **A message was resubmitted and landed twice.** `staleAfterMs` is too small relative to destination finality, or `deduplication.windowMs` is smaller than `staleAfterMs`. Both assumptions are documented above; correct them before enabling reconciliation on a network with slow finality. +- **Manual intervention.** Call `reconciler.reconcile()` directly for an on-demand pass; calls are coalesced, so concurrent callers share one run. \ No newline at end of file From 7b120351a1851018189765ffba6e14a82b122cef Mon Sep 17 00:00:00 2001 From: Henry Ebubechukwu Date: Mon, 28 Sep 2026 16:21:38 +0100 Subject: [PATCH 2/2] test(reconciler): use distinct source chain in fixtures after validation merge Merging upstream main brought in malformed-message validation (PR #1241), which rejects messages whose sourceChainId equals destinationChainId. Our reconciler and pipeline fixtures used 'ethereum' for both. Point the fixtures at an 'arbitrum' source so they stay valid under the new intake validation while still exercising the EVM destination path. Full suite: 295 tests passing (11 suites). --- .../src/reconciler/__tests__/message-status-reconciler.spec.ts | 2 +- .../src/reconciler/__tests__/reconciliation-pipeline.spec.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/relayer-service/src/reconciler/__tests__/message-status-reconciler.spec.ts b/apps/relayer-service/src/reconciler/__tests__/message-status-reconciler.spec.ts index d78900bf..bae80350 100644 --- a/apps/relayer-service/src/reconciler/__tests__/message-status-reconciler.spec.ts +++ b/apps/relayer-service/src/reconciler/__tests__/message-status-reconciler.spec.ts @@ -5,7 +5,7 @@ import { CrossChainMessage, ExecutionResult, MessageStatusProvider, OnChainMessa function makeMessage(overrides: Partial = {}): CrossChainMessage { return { id: 'msg-1', - sourceChainId: 'ethereum', + sourceChainId: 'arbitrum', destinationChainId: 'stellar', sourceTxHash: '0x' + 'a'.repeat(64), sourceBlockNumber: 10_000_000, diff --git a/apps/relayer-service/src/reconciler/__tests__/reconciliation-pipeline.spec.ts b/apps/relayer-service/src/reconciler/__tests__/reconciliation-pipeline.spec.ts index 887be3b2..3a96a905 100644 --- a/apps/relayer-service/src/reconciler/__tests__/reconciliation-pipeline.spec.ts +++ b/apps/relayer-service/src/reconciler/__tests__/reconciliation-pipeline.spec.ts @@ -17,7 +17,7 @@ const post = axios.post as jest.Mock; function makeMessage(overrides: Partial = {}): CrossChainMessage { return { id: 'msg-1', - sourceChainId: 'ethereum', + sourceChainId: 'arbitrum', destinationChainId: 'ethereum', sourceTxHash: '0x' + 'a'.repeat(64), sourceBlockNumber: 10_000_000,