fix(listener): prevent concurrent processing of the same event

listener/src/database/schema.sql

@@ -252,7 +252,7 @@ CREATE TABLE IF NOT EXISTS processed_events (
   last_redetected_at DATETIME,              -- When the event was last detected again (for reorg monitoring)
 
   -- Status and metadata
-  status VARCHAR(20) NOT NULL DEFAULT 'PROCESSED' CHECK (status IN ('PROCESSED', 'SKIPPED', 'ERROR')),
+  status VARCHAR(20) NOT NULL DEFAULT 'PROCESSED' CHECK (status IN ('PROCESSED', 'PROCESSING', 'SKIPPED', 'ERROR')),
   notification_sent BOOLEAN NOT NULL DEFAULT 0 CHECK (notification_sent IN (0, 1)),
   error_reason TEXT                         -- If status is ERROR, what went wrong
 );

listener/src/migrations/006-database-data-integrity.ts

@@ -73,7 +73,7 @@ const tables = [
       is_reorg_duplicate BOOLEAN NOT NULL DEFAULT 0 CHECK (is_reorg_duplicate IN (0, 1)),
       reorg_detection_count INTEGER NOT NULL DEFAULT 0 CHECK (reorg_detection_count >= 0),
       last_redetected_at DATETIME,
-      status VARCHAR(20) NOT NULL DEFAULT 'PROCESSED' CHECK (status IN ('PROCESSED', 'SKIPPED', 'ERROR')),
+      status VARCHAR(20) NOT NULL DEFAULT 'PROCESSED' CHECK (status IN ('PROCESSED', 'PROCESSING', 'SKIPPED', 'ERROR')),
       notification_sent BOOLEAN NOT NULL DEFAULT 0 CHECK (notification_sent IN (0, 1)),
       error_reason TEXT
     )`,
@@ -217,7 +217,7 @@ const audits: Array<{ table: string; sql: string }> = [
         OR ledger_number IS NULL OR ledger_number < 0 OR event_type IS NULL
         OR is_reorg_duplicate IS NULL OR is_reorg_duplicate NOT IN (0, 1)
         OR reorg_detection_count IS NULL OR reorg_detection_count < 0
-        OR status IS NULL OR status NOT IN ('PROCESSED', 'SKIPPED', 'ERROR')
+        OR status IS NULL OR status NOT IN ('PROCESSED', 'PROCESSING', 'SKIPPED', 'ERROR')
         OR notification_sent IS NULL OR notification_sent NOT IN (0, 1)`,
   },
   {

listener/src/services/concurrent-event-processing.test.ts

@@ -58,6 +58,23 @@ jest.mock('../utils/event-utils', () => ({
   matchesEventFilter: jest.fn().mockReturnValue(true),
 }));
 
+jest.mock('./event-normalization', () => ({
+  normalizeSorobanEvent: jest.fn().mockImplementation((event: any, ctx: any) => ({
+    ok: true,
+    event: {
+      eventId: event.id,
+      contractAddress: ctx.contractAddress,
+      eventName: 'TaskCreated',
+      eventType: event.type,
+      ledger: event.ledger,
+      topic: [],
+      value: '',
+      txHash: event.txHash,
+      payloadVersion: null,
+    },
+  })),
+}));
+
 jest.mock('../store/event-registry', () => ({
   eventRegistry: {
     addFromInput: jest.fn().mockImplementation((input: { eventId: string }) => ({
@@ -71,7 +88,7 @@ jest.mock('../store/event-registry', () => ({
 // ── Helpers ─────────────────────────────────────────────────────────────────
 
 const CONTRACT = 'CCONCURRENCYTESTCONTRACT';
-const contractConfig: ContractConfig = { address: CONTRACT, name: 'Concurrency' } as ContractConfig;
+const contractConfig: ContractConfig = { address: CONTRACT, name: 'Concurrency', events: ['*'] } as ContractConfig;
 
 function makeConfig(): Config {
   return {
@@ -244,6 +261,17 @@ describe('concurrent event processing', () => {
       const second = await service.claimEvent('evt-1', CONTRACT, 100, 'tx', 'contract');
       expect(second.claimed).toBe(false);
     });
+
+    it('allows immediate reclaim after a claim is explicitly released', async () => {
+      const service = new EventDeduplicationService(dbA);
+      const first = await service.claimEvent('evt-released', CONTRACT, 100, 'tx', 'contract');
+      expect(first.claimed).toBe(true);
+
+      await service.releaseClaim('evt-released', CONTRACT);
+
+      const second = await service.claimEvent('evt-released', CONTRACT, 100, 'tx', 'contract');
+      expect(second.claimed).toBe(true);
+    });
   });
 
   describe('atomic counters (no lost updates)', () => {

listener/src/services/event-deduplication-service.ts

@@ -128,6 +128,23 @@ export class EventDeduplicationService {
     }
   }
 
+  /**
+   * Release an in-flight claim when processing is aborted or fails before completion,
+   * allowing another worker or subsequent poll cycle to claim it immediately.
+   */
+  async releaseClaim(eventId: string, contractAddress: string): Promise<void> {
+    const fingerprint = generateFingerprint(eventId, contractAddress);
+    try {
+      await this.db.run(
+        `DELETE FROM processed_events WHERE fingerprint = ? AND status = 'PROCESSING'`,
+        [fingerprint],
+      );
+      logger.info('Event claim released', { eventId, contractAddress, fingerprint });
+    } catch (error) {
+      logger.error('Error releasing event claim', { eventId, contractAddress, error });
+    }
+  }
+
   /**
    * Finalise a claimed event. A single UPDATE, so the outcome written by the
    * claiming worker cannot interleave with another writer's read-modify-write.

listener/src/services/event-subscriber.ts

@@ -514,14 +514,17 @@ export class EventSubscriber {
     }
     const isDryRun = this.config.dryRun === true;
 
-    // Check persistent deduplication first (to catch reorg duplicates)
+    // Atomically claim event for processing to prevent concurrent workers from processing simultaneously
     if (this.deduplicationService && !isDryRun) {
-      const duplicate = await this.deduplicationService.isDuplicate(
+      const claim = await this.deduplicationService.claimEvent(
         event.id,
         contractConfig.address,
+        event.ledger,
+        event.txHash,
+        event.type,
       );
-      if (duplicate.isDuplicate) {
-        logger.warn('Skipping event: already processed (persistent deduplication)', {
+      if (!claim.claimed) {
+        logger.warn('Skipping event: already claimed or processed by another worker', {
           requestId: correlationId,
           correlationId,
           eventId: event.id,