From 24bfe751790f97c6671e66980e06db0b9ceb6489 Mon Sep 17 00:00:00 2001 From: ponmileleke54-dev Date: Mon, 31 Aug 2026 11:35:22 +0000 Subject: [PATCH 1/2] fix: stop poison-pill batches from freezing indexer cursor --- backend/src/workers/soroban-event-worker.ts | 22 ++++++++++----------- backend/tests/soroban-event-worker.test.ts | 19 +++++------------- 2 files changed, 16 insertions(+), 25 deletions(-) diff --git a/backend/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts index 495e2539..82119942 100644 --- a/backend/src/workers/soroban-event-worker.ts +++ b/backend/src/workers/soroban-event-worker.ts @@ -342,7 +342,7 @@ export class SorobanEventWorker { let lastCursor: string | null = state.lastCursor; let lastLedger: number = state.lastLedger; - let hasError = false; + let sawSuccess = false; // Sort events so that 'stream_created' events are processed first in the batch. // This ensures that subsequent events (like 'fee_collected') that depend on @@ -363,13 +363,12 @@ export class SorobanEventWorker { await this.processEvent(event); this.eventsProcessed += 1; this.recordOutcome(true); - if (!hasError) { - // Use the event ID as the cursor if pagingToken is not available - lastCursor = event.id; - lastLedger = event.ledger; - } + sawSuccess = true; + // Advance the cursor to the most recent event that was successfully processed. + // This keeps a single malformed event from pinning the entire batch forever. + lastCursor = event.id; + lastLedger = event.ledger; } catch (err) { - hasError = true; this.eventsFailed += 1; this.lastErrorAt = new Date(); this.recordOutcome(false); @@ -381,10 +380,11 @@ export class SorobanEventWorker { } } - // Use the response's final cursor if provided and no error occurred, otherwise the last valid event's ID - const finalCursor = hasError - ? lastCursor - : ((response as any).latestCursor || lastCursor); + // If we successfully processed any events in the batch, advance to the last + // successful event so a single poison-pill failure cannot freeze the cursor. + const finalCursor = sawSuccess + ? ((response as any).latestCursor || lastCursor) + : lastCursor; await prisma.indexerState.upsert({ where: { id: INDEXER_STATE_ID }, diff --git a/backend/tests/soroban-event-worker.test.ts b/backend/tests/soroban-event-worker.test.ts index 8d7cf386..f414b924 100644 --- a/backend/tests/soroban-event-worker.test.ts +++ b/backend/tests/soroban-event-worker.test.ts @@ -687,7 +687,7 @@ describe('SorobanEventWorker', () => { expect(typeof capturedEventUpsert?.create?.streamId).toBe('bigint'); }); - it('cursor_does_not_advance_past_failed_event_in_mixed_batch', async () => { + it('cursor_advances_past_valid_events_after_an_earlier_failed_event_in_same_batch', async () => { // Setup initial state: lastCursor is 'cursor-initial' (prisma.indexerState.findUnique as ReturnType).mockResolvedValue({ id: 'singleton', @@ -721,7 +721,6 @@ describe('SorobanEventWorker', () => { } as any, }; - // Event 2: Valid admin_transferred event const event2: rpc.Api.EventResponse = { id: 'cursor-event-2', type: 'contract', @@ -743,7 +742,6 @@ describe('SorobanEventWorker', () => { } as any, }; - // Event 3: Valid admin_transferred event const event3: rpc.Api.EventResponse = { id: 'cursor-event-3', type: 'contract', @@ -765,12 +763,10 @@ describe('SorobanEventWorker', () => { } as any, }; - // Mock getEvents on worker.server vi.spyOn((worker as any).server, 'getEvents').mockResolvedValue({ events: [event1, event2, event3], }); - // Track upserted stream events const upsertedStreamEvents: any[] = []; const mockTx = { user: { upsert: vi.fn().mockResolvedValue({}) }, @@ -786,31 +782,26 @@ describe('SorobanEventWorker', () => { (prisma.$transaction as ReturnType).mockImplementation((cb) => cb(mockTx)); - // Run fetchAndProcessEvents await (worker as any).fetchAndProcessEvents(); - // Assert successful later events (event2 and event3) were written exactly once each const event1Writes = upsertedStreamEvents.filter( - (e) => e.create?.transactionHash === 'tx-failed-1' + (e) => e.create?.transactionHash === 'tx-failed-1', ); const event2Writes = upsertedStreamEvents.filter( - (e) => e.create?.transactionHash === 'tx-success-2' + (e) => e.create?.transactionHash === 'tx-success-2', ); const event3Writes = upsertedStreamEvents.filter( - (e) => e.create?.transactionHash === 'tx-success-3' + (e) => e.create?.transactionHash === 'tx-success-3', ); expect(event1Writes.length).toBe(0); expect(event2Writes.length).toBe(1); expect(event3Writes.length).toBe(1); - // Assert: persisted IndexerState.lastCursor is NOT advanced past the failed event's position - // (i.e. it must not be set to 'cursor-event-2' or 'cursor-event-3' after a failure in event 1) const indexerUpsertCalls = (prisma.indexerState.upsert as ReturnType).mock.calls; const lastSaveCall = indexerUpsertCalls[indexerUpsertCalls.length - 1]![0]; - expect(lastSaveCall.update.lastCursor).not.toBe('cursor-event-2'); - expect(lastSaveCall.update.lastCursor).not.toBe('cursor-event-3'); + expect(lastSaveCall.update.lastCursor).toBe('cursor-event-3'); }); }); From 1abb0e63ceb054b46f7ccbcbfaf94ccb0c7118a0 Mon Sep 17 00:00:00 2001 From: ponmileleke54-dev Date: Fri, 4 Sep 2026 10:08:56 +0000 Subject: [PATCH 2/2] fix: remove unused event test helpers --- backend/tests/soroban-event-worker.test.ts | 49 ---------------------- 1 file changed, 49 deletions(-) diff --git a/backend/tests/soroban-event-worker.test.ts b/backend/tests/soroban-event-worker.test.ts index 8078a881..ed2037a8 100644 --- a/backend/tests/soroban-event-worker.test.ts +++ b/backend/tests/soroban-event-worker.test.ts @@ -67,55 +67,6 @@ const mockContractAddr = () => ({ address: { type: 'scAddressTypeContract', cont const mockMapEntry = (keyName: string, val: any) => ({ key: mockSym(keyName), val } as any); const mockMapValue = (entries: any[]) => ({ map: entries } as any); -/** Build a valid admin_transferred event (processes without throwing). */ -function makeAdminTransferredEvent( - id: string, - ledger: number, - txHash: string, -): rpc.Api.EventResponse { - return { - id, - type: 'contract', - ledger, - ledgerClosedAt: '2024-01-01T00:00:00Z', - txHash, - transactionIndex: 0, - operationIndex: 0, - inSuccessfulContractCall: true, - topic: [mockSym('admin_transferred')], - value: { - type: 'scvMap', - map: [ - mockMapEntry('previous_admin', mockAccountAddr()), - mockMapEntry('new_admin', mockAccountAddr()), - ] as any, - } as any, - }; -} - -/** Build a malformed fee_config_updated event (missing body fields → throws). */ -function makeMalformedEvent( - id: string, - ledger: number, - txHash: string, -): rpc.Api.EventResponse { - return { - id, - type: 'contract', - ledger, - ledgerClosedAt: '2024-01-01T00:00:00Z', - txHash, - transactionIndex: 0, - operationIndex: 0, - inSuccessfulContractCall: true, - topic: [mockSym('fee_config_updated')], - value: { - type: 'scvMap', - map: [] as any, - } as any, - }; -} - // Standard stream fields map used across most tests const streamFields = (overrides?: { withdrawn_amount?: string; isActive?: boolean; is_active_value?: boolean }) => [ mockMapEntry('sender', mockAccountAddr()),