diff --git a/listener/src/database/schema.sql b/listener/src/database/schema.sql index 56304e12..a1c0d4a9 100644 --- a/listener/src/database/schema.sql +++ b/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 ); diff --git a/listener/src/migrations/006-database-data-integrity.ts b/listener/src/migrations/006-database-data-integrity.ts index 1ecfbbe5..8b2152ca 100644 --- a/listener/src/migrations/006-database-data-integrity.ts +++ b/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)`, }, { diff --git a/listener/src/services/concurrent-event-processing.test.ts b/listener/src/services/concurrent-event-processing.test.ts index ef99d337..6cb9545a 100644 --- a/listener/src/services/concurrent-event-processing.test.ts +++ b/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)', () => { diff --git a/listener/src/services/event-deduplication-service.ts b/listener/src/services/event-deduplication-service.ts index b2c572d9..fd3f271c 100644 --- a/listener/src/services/event-deduplication-service.ts +++ b/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 { + 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. diff --git a/listener/src/services/event-subscriber.ts b/listener/src/services/event-subscriber.ts index 67c4612d..f5fbb592 100644 --- a/listener/src/services/event-subscriber.ts +++ b/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,