Skip to content

fix: stop poison-pill batches from freezing indexer cursor #1422

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ponmileleke54-dev wants to merge 3 commits into
base: main
Choose a base branch
from
Open

fix: stop poison-pill batches from freezing indexer cursor #1422

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
24bfe75
fix: stop poison-pill batches from freezing indexer cursor
ponmileleke54-dev Aug 31, 2026
5fbb691
Merge branch 'main' into fix/issue-1292
ponmileleke54-dev Sep 2, 2026
70cf433
Merge branch 'main' into fix/issue-1292
ponmileleke54-dev Sep 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 9 additions & 11 deletions backend/src/workers/soroban-event-worker.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -368,7 +368,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
Expand All @@ -389,15 +389,12 @@ export class SorobanEventWorker {
await this.processEvent(event);
this.eventsProcessed += 1;
this.recordOutcome(true);
// Always advance the cursor past a successfully processed event,
// even when an earlier event in this batch failed. The previous
// `!hasError` guard froze the cursor after the first failure, so a
// single always-failing event (e.g. malformed body) reprocessed
// every later event on every poll indefinitely.
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);
Expand All @@ -421,10 +418,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 },
Expand Down
184 changes: 4 additions & 180 deletions backend/tests/soroban-event-worker.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -746,7 +746,7 @@ describe('SorobanEventWorker', () => {
expect(typeof capturedEventUpsert?.create?.streamId).toBe('bigint');
});

it('advances cursor past a failed event when later events in the batch succeed', 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<typeof vi.fn>).mockResolvedValue({
id: 'singleton',
Expand Down Expand Up @@ -787,7 +787,6 @@ describe('SorobanEventWorker', () => {
} as any,
};

// Event 2: Valid admin_transferred event
const event2: rpc.Api.EventResponse = {
id: 'cursor-event-2',
type: 'contract',
Expand All @@ -809,7 +808,6 @@ describe('SorobanEventWorker', () => {
} as any,
};

// Event 3: Valid admin_transferred event
const event3: rpc.Api.EventResponse = {
id: 'cursor-event-3',
type: 'contract',
Expand All @@ -831,12 +829,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({}) },
Expand All @@ -852,198 +848,26 @@ describe('SorobanEventWorker', () => {

(prisma.$transaction as ReturnType<typeof vi.fn>).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: the failed event was dead-lettered with its raw payload so it
// can be triaged manually.
expect(prisma.indexerDeadLetterEvent.upsert).toHaveBeenCalledWith(
expect.objectContaining({
where: { eventId: 'cursor-event-1' },
create: expect.objectContaining({
eventId: 'cursor-event-1',
ledger: 101,
transactionHash: 'tx-failed-1',
attempts: 1,
errorMessage: expect.any(String),
}),
}),
);

// Assert: the persisted IndexerState.lastCursor DID advance past the
// failed event's position, because events 2 and 3 processed
// successfully. A single bad event must not freeze the indexer.
const indexerUpsertCalls = (prisma.indexerState.upsert as ReturnType<typeof vi.fn>).mock.calls;
const lastSaveCall = indexerUpsertCalls[indexerUpsertCalls.length - 1]![0];

expect(lastSaveCall.update.lastCursor).toBe('cursor-event-3');
expect(lastSaveCall.update.lastLedger).toBe(103);
});

it('processes the other four events and advances the cursor when one event in a batch of five is malformed', async () => {
(prisma.indexerState.findUnique as ReturnType<typeof vi.fn>).mockResolvedValue({
id: 'singleton',
lastLedger: 100,
lastCursor: 'batch-cursor-0',
updatedAt: new Date(),
});
(prisma.indexerState.upsert as ReturnType<typeof vi.fn>).mockResolvedValue({
id: 'singleton',
lastLedger: 100,
lastCursor: 'batch-cursor-0',
updatedAt: new Date(),
});
(prisma.indexerDeadLetterEvent.upsert as ReturnType<typeof vi.fn>).mockResolvedValue({
id: 'dl-batch',
eventId: 'batch-bad-3',
attempts: 1,
});

// One deliberately malformed event (position 3) in a batch of five.
const events = [
makeAdminTransferredEvent('batch-ok-1', 101, 'tx-batch-1'),
makeAdminTransferredEvent('batch-ok-2', 102, 'tx-batch-2'),
makeMalformedEvent('batch-bad-3', 103, 'tx-batch-3'),
makeAdminTransferredEvent('batch-ok-4', 104, 'tx-batch-4'),
makeAdminTransferredEvent('batch-ok-5', 105, 'tx-batch-5'),
];

vi.spyOn((worker as any).server, 'getEvents').mockResolvedValue({ 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<typeof vi.fn>).mockImplementation((cb) => cb(mockTx));

await (worker as any).fetchAndProcessEvents();

// All four valid events processed exactly once; malformed one never written.
for (const txHash of ['tx-batch-1', 'tx-batch-2', 'tx-batch-4', 'tx-batch-5']) {
expect(
upsertedStreamEvents.filter((e) => e.create?.transactionHash === txHash).length,
).toBe(1);
}
expect(
upsertedStreamEvents.filter((e) => e.create?.transactionHash === 'tx-batch-3').length,
).toBe(0);

// Malformed event dead-lettered with its raw payload for manual triage.
expect(prisma.indexerDeadLetterEvent.upsert).toHaveBeenCalledWith(
expect.objectContaining({
where: { eventId: 'batch-bad-3' },
create: expect.objectContaining({
transactionHash: 'tx-batch-3',
rawPayload: expect.stringContaining('batch-bad-3'),
attempts: 1,
}),
}),
);

// Cursor advances past the malformed event to the last processed event.
const lastSaveCall =
(prisma.indexerState.upsert as ReturnType<typeof vi.fn>).mock.calls.at(-1)![0];
expect(lastSaveCall.update.lastCursor).toBe('batch-ok-5');
expect(lastSaveCall.update.lastLedger).toBe(105);

// Counters reflect 4 processed, 1 failed.
const counters = worker.getEventCounters();
expect(counters.eventsProcessed).toBe(4);
expect(counters.eventsFailed).toBe(1);
expect(counters.lastErrorAt).not.toBeNull();
});

it('abandons an always-failing event after the retry cap and advances past it', async () => {
process.env.INDEXER_DEAD_LETTER_MAX_RETRIES = '2';
worker = new SorobanEventWorker();

(prisma.indexerState.findUnique as ReturnType<typeof vi.fn>).mockResolvedValue({
id: 'singleton',
lastLedger: 100,
lastCursor: 'cap-cursor-0',
updatedAt: new Date(),
});
(prisma.indexerState.upsert as ReturnType<typeof vi.fn>).mockResolvedValue({
id: 'singleton',
lastLedger: 100,
lastCursor: 'cap-cursor-0',
updatedAt: new Date(),
});

const okEvent = makeAdminTransferredEvent('cap-ok-1', 101, 'tx-cap-1');
const badEvent = makeMalformedEvent('cap-bad-2', 102, 'tx-cap-2');

// Poll 1: both events. Poll 2: only the bad tail event is re-fetched.
// Poll 3: cursor moved past it — nothing left to fetch.
const getEvents = vi
.spyOn((worker as any).server, 'getEvents')
.mockResolvedValueOnce({ events: [okEvent, badEvent] })
.mockResolvedValueOnce({ events: [badEvent] })
.mockResolvedValueOnce({ events: [] });

// Dead-letter attempts: 1 (below cap) then 2 (cap reached → abandon).
(prisma.indexerDeadLetterEvent.upsert as ReturnType<typeof vi.fn>)
.mockResolvedValueOnce({ id: 'dl-1', eventId: 'cap-bad-2', attempts: 1 })
.mockResolvedValueOnce({ id: 'dl-2', eventId: 'cap-bad-2', attempts: 2 });

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().mockResolvedValue({ id: 'event-id' }),
},
};
(prisma.$transaction as ReturnType<typeof vi.fn>).mockImplementation((cb) => cb(mockTx));

// Poll 1: ok event processed, bad event fails but stays below the cap.
await (worker as any).fetchAndProcessEvents();
let lastSaveCall =
(prisma.indexerState.upsert as ReturnType<typeof vi.fn>).mock.calls.at(-1)![0];
expect(lastSaveCall.update.lastCursor).toBe('cap-ok-1');

// Poll 2: bad event retried, hits the cap → abandoned, cursor advances past it.
await (worker as any).fetchAndProcessEvents();
lastSaveCall =
(prisma.indexerState.upsert as ReturnType<typeof vi.fn>).mock.calls.at(-1)![0];
expect(lastSaveCall.update.lastCursor).toBe('cap-bad-2');
expect(logger.warn).toHaveBeenCalledWith(
expect.stringContaining('exceeded 2 attempts — abandoning'),
);

// Poll 3: cursor is past the bad event — no more re-processing.
await (worker as any).fetchAndProcessEvents();
expect(getEvents).toHaveBeenCalledTimes(3);
expect((prisma.indexerState.upsert as ReturnType<typeof vi.fn>).mock.calls.length).toBe(2);

const counters = worker.getEventCounters();
expect(counters.eventsProcessed).toBe(1);
expect(counters.eventsFailed).toBe(2);

delete process.env.INDEXER_DEAD_LETTER_MAX_RETRIES;
});
});

Expand Down
Loading