From 3f33038cc991f9ecbb097b42cf29a8d9de9702e9 Mon Sep 17 00:00:00 2001 From: extolkom Date: Tue, 28 Jul 2026 16:54:40 -1200 Subject: [PATCH] test: add regression test for indexer cursor advancing past failed event (#846) --- backend/src/workers/soroban-event-worker.ts | 16 +- backend/tests/soroban-event-worker.test.ts | 188 ++++++++++++++++---- 2 files changed, 161 insertions(+), 43 deletions(-) diff --git a/backend/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts index 730c0a5f..9b17a74c 100644 --- a/backend/src/workers/soroban-event-worker.ts +++ b/backend/src/workers/soroban-event-worker.ts @@ -313,6 +313,7 @@ export class SorobanEventWorker { let lastCursor: string | null = state.lastCursor; let lastLedger: number = state.lastLedger; + let hasError = 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 @@ -333,10 +334,13 @@ export class SorobanEventWorker { await this.processEvent(event); this.eventsProcessed += 1; this.recordOutcome(true); - // Use the event ID as the cursor if pagingToken is not available - lastCursor = event.id; - lastLedger = event.ledger; + if (!hasError) { + // Use the event ID as the cursor if pagingToken is not available + lastCursor = event.id; + lastLedger = event.ledger; + } } catch (err) { + hasError = true; this.eventsFailed += 1; this.lastErrorAt = new Date(); this.recordOutcome(false); @@ -348,8 +352,10 @@ export class SorobanEventWorker { } } - // Use the response's final cursor if provided, otherwise the last event's ID - const finalCursor = (response as any).latestCursor || lastCursor; + // 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); 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 a75640e0..0a20f435 100644 --- a/backend/tests/soroban-event-worker.test.ts +++ b/backend/tests/soroban-event-worker.test.ts @@ -1,46 +1,32 @@ import { describe, it, expect, vi, beforeEach } from 'vitest'; import { rpc } from '@stellar/stellar-sdk'; -// Mock prisma before importing the worker -vi.mock('../src/lib/prisma.js', () => ({ - default: { - indexerState: { - upsert: vi.fn(), - }, - user: { - upsert: vi.fn(), - }, - stream: { - upsert: vi.fn(), - findUniqueOrThrow: vi.fn(), - }, - streamEvent: { - findUnique: vi.fn(), - upsert: vi.fn(), - create: vi.fn(), - }, - $transaction: vi.fn((cb) => cb({ streamEvent: { findUnique: vi.fn(), upsert: vi.fn() }, user: { upsert: vi.fn() }, stream: { upsert: vi.fn(), update: vi.fn() } })), - $disconnect: vi.fn(), +const mockPrismaObj = vi.hoisted(() => ({ + indexerState: { + findUnique: vi.fn(), + create: vi.fn(), + upsert: vi.fn(), }, - prisma: { - indexerState: { - upsert: vi.fn(), - }, - user: { - upsert: vi.fn(), - }, - stream: { - upsert: vi.fn(), - findUniqueOrThrow: vi.fn(), - }, - streamEvent: { - findUnique: vi.fn(), - upsert: vi.fn(), - create: vi.fn(), - }, - $transaction: vi.fn((cb) => cb({ streamEvent: { findUnique: vi.fn(), upsert: vi.fn() }, user: { upsert: vi.fn() }, stream: { upsert: vi.fn(), update: vi.fn() } })), - $disconnect: vi.fn(), + user: { + upsert: vi.fn(), }, + stream: { + upsert: vi.fn(), + findUniqueOrThrow: vi.fn(), + }, + streamEvent: { + findUnique: vi.fn(), + upsert: vi.fn(), + create: vi.fn(), + }, + $transaction: vi.fn((cb) => cb({ streamEvent: { findUnique: vi.fn(), upsert: vi.fn() }, user: { upsert: vi.fn() }, stream: { upsert: vi.fn(), update: vi.fn() } })), + $disconnect: vi.fn(), +})); + +// Mock prisma before importing the worker +vi.mock('../src/lib/prisma.js', () => ({ + default: mockPrismaObj, + prisma: mockPrismaObj, })); // Mock SSE service @@ -696,6 +682,132 @@ describe('SorobanEventWorker', () => { expect(capturedEventUpsert?.create?.streamId).toBe(streamId); expect(typeof capturedEventUpsert?.create?.streamId).toBe('bigint'); }); + + it('cursor_does_not_advance_past_failed_event_in_mixed_batch', async () => { + // Setup initial state: lastCursor is 'cursor-initial' + (prisma.indexerState.findUnique as ReturnType).mockResolvedValue({ + id: 'singleton', + lastLedger: 100, + lastCursor: 'cursor-initial', + updatedAt: new Date(), + }); + (prisma.indexerState.upsert as ReturnType).mockResolvedValue({ + id: 'singleton', + lastLedger: 100, + lastCursor: 'cursor-initial', + updatedAt: new Date(), + }); + + // Event 1: Missing required body fields for fee_config_updated -> handleFeeConfigUpdated throws + const event1: rpc.Api.EventResponse = { + id: 'cursor-event-1', + type: 'contract', + ledger: 101, + ledgerClosedAt: '2024-01-01T00:00:00Z', + txHash: 'tx-failed-1', + transactionIndex: 0, + operationIndex: 0, + inSuccessfulContractCall: true, + topic: [ + { switch: () => ({ value: 0 }), sym: () => 'fee_config_updated' } as any, + ], + value: { + switch: () => ({ value: 4 }), + map: () => [] as any, + } as any, + }; + + // Event 2: Valid admin_transferred event + const event2: rpc.Api.EventResponse = { + id: 'cursor-event-2', + type: 'contract', + ledger: 102, + ledgerClosedAt: '2024-01-01T00:00:00Z', + txHash: 'tx-success-2', + transactionIndex: 0, + operationIndex: 0, + inSuccessfulContractCall: true, + topic: [ + { switch: () => ({ value: 0 }), sym: () => 'admin_transferred' } as any, + ], + value: { + switch: () => ({ value: 4 }), + map: () => [ + { key: () => ({ sym: () => 'previous_admin' }), val: () => ({ address: () => ({ switch: () => ({ value: 0 }), accountId: () => ({ ed25519: () => Buffer.alloc(32) }) }) }) }, + { key: () => ({ sym: () => 'new_admin' }), val: () => ({ address: () => ({ switch: () => ({ value: 0 }), accountId: () => ({ ed25519: () => Buffer.alloc(32) }) }) }) }, + ] as any, + } as any, + }; + + // Event 3: Valid admin_transferred event + const event3: rpc.Api.EventResponse = { + id: 'cursor-event-3', + type: 'contract', + ledger: 103, + ledgerClosedAt: '2024-01-01T00:00:00Z', + txHash: 'tx-success-3', + transactionIndex: 0, + operationIndex: 0, + inSuccessfulContractCall: true, + topic: [ + { switch: () => ({ value: 0 }), sym: () => 'admin_transferred' } as any, + ], + value: { + switch: () => ({ value: 4 }), + map: () => [ + { key: () => ({ sym: () => 'previous_admin' }), val: () => ({ address: () => ({ switch: () => ({ value: 0 }), accountId: () => ({ ed25519: () => Buffer.alloc(32) }) }) }) }, + { key: () => ({ sym: () => 'new_admin' }), val: () => ({ address: () => ({ switch: () => ({ value: 0 }), accountId: () => ({ ed25519: () => Buffer.alloc(32) }) }) }) }, + ] as any, + } 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({}) }, + stream: { upsert: vi.fn().mockResolvedValue({ streamId: 0n, isActive: false }) }, + streamEvent: { + findUnique: vi.fn().mockResolvedValue(null), + upsert: vi.fn().mockImplementation((args) => { + upsertedStreamEvents.push(args); + return Promise.resolve({ id: 'event-id' }); + }), + }, + }; + + (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' + ); + const event2Writes = upsertedStreamEvents.filter( + (e) => e.create?.transactionHash === 'tx-success-2' + ); + const event3Writes = upsertedStreamEvents.filter( + (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'); + }); }); describe('poll / triggerPoll serialization (#843)', () => {