feat(messaging): support message status reconciliation

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,
@@ -44,6 +50,15 @@ export type {
   ExecutorConfig,
   QueueConfig,
   MessageQueueItem,
+  InflightMessageSnapshot,
+  OnChainMessageStatus,
+  MessageStatusProvider,
+  ReconciliationConfig,
+  ReconciliationAction,
+  ReconciliationReason,
+  MessageReconciliation,
+  ReconciliationSummary,
+  ReconciliationStats,
   ChainType,
   DuplicateDetectorConfig,
   DuplicateReason,

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> = {}): 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();
+  });
+});

apps/relayer-service/src/queue/message-queue.ts

@@ -1,6 +1,7 @@
 import { EventEmitter } from 'events';
 import {
   CrossChainMessage,
+  InflightMessageSnapshot,
   MessageStatus,
   MessageQueueItem,
   MessageValidationOptions,
@@ -98,6 +99,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 });
@@ -106,36 +108,48 @@ export class MessageQueue extends EventEmitter {
 
   async complete(result: ExecutionResult): Promise<void> {
     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,
+      });
     }
   }
 
@@ -147,6 +161,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;
   }