From 55d980487ee46e2ee15d911591ff0ee0af1f7619 Mon Sep 17 00:00:00 2001 From: Damilola Ogunrotimi <98775983+Fury03@users.noreply.github.com> Date: Sun, 30 Aug 2026 20:05:36 +0000 Subject: [PATCH 1/2] Add integration test for resetIndexer/replayFromLedger racing concurrent poll MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds an integration test that exercises the race condition described in Functional Edge Case #19 (issue #1293): resetIndexer and replayFromLedger bypass the SorobanEventWorker's batchMutex, so their DB cursor writes can be overwritten by a concurrent poll's stale upsert. The test deliberately fails against current code and will pass once the race is fixed (the poll must re-read the cursor before writing it back, or resetIndexer/replayFromLedger must go through the batchMutex). 🤖 Generated with Codebuff Co-Authored-By: Codebuff --- .../integration/reset-replay-race.test.ts | 429 ++++++++++++++++++ 1 file changed, 429 insertions(+) create mode 100644 backend/tests/integration/reset-replay-race.test.ts diff --git a/backend/tests/integration/reset-replay-race.test.ts b/backend/tests/integration/reset-replay-race.test.ts new file mode 100644 index 00000000..6c61ecb3 --- /dev/null +++ b/backend/tests/integration/reset-replay-race.test.ts @@ -0,0 +1,429 @@ +/** + * Integration test for resetIndexer / replayFromLedger racing a concurrent + * scheduled poll — Functional Edge Case #19 from the second-wave audit (#1293). + * + * Both `resetIndexer` and `replayFromLedger` are tested in isolation elsewhere, + * but never interleaved with SorobanEventWorker's mutex-protected poll cycle. + * This file fills that gap by simulating the exact scenario operators hit during + * incident recovery: a live, running indexer whose scheduled poll is mid-flight + * when an operator resets or replays. + * + * Acceptance criteria: + * - New test fails against current code (the race is real). + * - Once Functional Edge Case #19 is fixed, the test passes. + * + * The race (Functional Edge Case #19): + * 1. SorobanEventWorker.poll() -> runExclusive -> fetchAndProcessEvents() + * 2. fetchAndProcessEvents reads IndexerState.lastLedger (e.g. 200) + * 3. fetchAndProcessEvents awaits server.getEvents (async network I/O) + * 4. ^ WINDOW: resetIndexer(50) is called, writing lastLedger=50 + * 5. fetchAndProcessEvents finishes, upserts lastLedger=200 (stale value + * captured in step 2, because processEvent error prevents ledger advance) + * 6. Reset is lost - cursor is 200 instead of 50. + * + * Root cause: resetIndexer bypasses the worker's batchMutex, so its DB write + * can be overwritten by a concurrent poll's cursor upsert. + */ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; + +// ─── Deferred promise (for controlling async timing in tests) ───────────────── + +interface Deferred { + promise: Promise; + resolve: (value: T) => void; + reject: (reason?: unknown) => void; +} + +function defer(): Deferred { + let resolve!: (value: T) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +// ─── Shared mutable DB state ───────────────────────────────────────────────── +// +// Both the worker (via ensureIndexerState + upsert) and resetIndexer +// (via upsert) write to this shared state, simulating a real Postgres +// where concurrent upserts actually race. + +let dbIndexerState: { + lastLedger: number; + lastCursor: string | null; + updatedAt: Date; +}; + +// Track all upsert calls in order for forensic assertions. +const upsertLog: Array<{ lastLedger: number; lastCursor: string | null; caller: string }> = []; + +// ─── Hoisted mock factories ────────────────────────────────────────────────── + +const { mockPrisma, mockSseService, mockLogger } = vi.hoisted(() => { + const mockPrisma = { + indexerState: { + upsert: vi.fn(async (args: any) => { + const update = args.update ?? {}; + dbIndexerState = { + ...dbIndexerState, + ...update, + updatedAt: new Date(), + }; + const caller = update.lastCursor === null && update.lastLedger !== undefined + ? 'resetIndexer' + : 'worker'; + upsertLog.push({ + lastLedger: dbIndexerState.lastLedger, + lastCursor: dbIndexerState.lastCursor, + caller, + }); + return { ...dbIndexerState }; + }), + }, + $disconnect: vi.fn(), + }; + + return { + mockPrisma, + mockSseService: { + broadcastToStream: vi.fn(), + broadcastToAdmin: vi.fn(), + }, + mockLogger: { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }, + }; +}); + +vi.mock('../../src/lib/prisma.js', () => ({ + prisma: mockPrisma, +})); + +vi.mock('../../src/lib/indexer-state.js', async (importOriginal) => { + const original = await importOriginal(); + return { + ...original, + ensureIndexerState: vi.fn(async (_startLedger: number) => { + return { + id: 'singleton' as const, + lastLedger: dbIndexerState.lastLedger, + lastCursor: dbIndexerState.lastCursor, + createdAt: new Date(), + updatedAt: dbIndexerState.updatedAt, + }; + }), + }; +}); + +vi.mock('../../src/services/sse.service.js', () => ({ + sseService: mockSseService, +})); + +vi.mock('../../src/logger.js', () => ({ + default: mockLogger, + requestContext: vi.fn(() => ({})), +})); + +// ─── Imports (after mocks) ─────────────────────────────────────────────────── + +import { SorobanEventWorker } from '../../src/workers/soroban-event-worker.js'; +import { resetIndexer, replayFromLedger } from '../../src/services/indexerService.js'; + +/** + * Build a minimal Soroban EventResponse that the worker can decode. + * We use `stream_created` with the minimum required body fields so + * `fetchAndProcessEvents` reaches the final cursor upsert. + */ +function fakeStreamCreatedEvent(overrides: { + id: string; + txHash: string; + ledger: number; +}) { + return { + id: overrides.id, + type: 'contract' as const, + ledger: overrides.ledger, + ledgerClosedAt: new Date().toISOString(), + txHash: overrides.txHash, + transactionIndex: 0, + operationIndex: 0, + inSuccessfulContractCall: true, + topic: [ + { + switch: () => ({ value: 0 }), + sym: () => 'stream_created', + }, + { + switch: () => ({ value: 1 }), + u64: () => ({ toString: () => '42' }), + }, + ], + value: { + switch: () => ({ value: 4 }), + map: () => [ + { + key: () => ({ sym: () => 'sender' }), + val: () => ({ + address: () => ({ + switch: () => ({ value: 0 }), + accountId: () => ({ + ed25519: () => Buffer.alloc(32), + }), + }), + }), + }, + { + key: () => ({ sym: () => 'recipient' }), + val: () => ({ + address: () => ({ + switch: () => ({ value: 0 }), + accountId: () => ({ + ed25519: () => Buffer.alloc(32), + }), + }), + }), + }, + { + key: () => ({ sym: () => 'token_address' }), + val: () => ({ + address: () => ({ + switch: () => ({ value: 1 }), + contractId: () => Buffer.alloc(32), + }), + }), + }, + { + key: () => ({ sym: () => 'rate_per_second' }), + val: () => ({ + i128: () => ({ + hi: () => ({ toString: () => '0' }), + lo: () => ({ toString: () => '100' }), + }), + }), + }, + { + key: () => ({ sym: () => 'deposited_amount' }), + val: () => ({ + i128: () => ({ + hi: () => ({ toString: () => '0' }), + lo: () => ({ toString: () => '86400' }), + }), + }), + }, + { + key: () => ({ sym: () => 'start_time' }), + val: () => ({ + u64: () => ({ toString: () => '1700000000' }), + }), + }, + ], + } as any, + } as any; +} + +// ─── Tests ─────────────────────────────────────────────────────────────────── + +describe('Reset/replay race with concurrent poll (Edge Case #19 - issue #1293)', () => { + let worker: SorobanEventWorker; + let getEventsDeferred: Deferred<{ events: any[]; latestCursor?: string | null }>; + + beforeEach(() => { + vi.clearAllMocks(); + upsertLog.length = 0; + + // Initialize shared DB state. + dbIndexerState = { + lastLedger: 0, + lastCursor: null, + updatedAt: new Date(), + }; + + worker = new SorobanEventWorker(); + (worker as any).contractId = 'CTESTCONTRACT'; + (worker as any).pollIntervalMs = 600_000; + + // Wire up deferred getEvents mock. + getEventsDeferred = defer(); + const server = (worker as any).server as { getEvents: ReturnType }; + server.getEvents = vi.fn(() => getEventsDeferred.promise); + }); + + afterEach(() => { + worker.stop(); + }); + + it( + 'resetIndexer cursor must survive a concurrent scheduled poll - ' + + 'FAILS before fix (poll overwrites reset)', + async () => { + // ── Arrange ────────────────────────────────────────────────────── + + // Simulate: indexer is at ledger 200, poll starts reading from there. + dbIndexerState = { + lastLedger: 200, + lastCursor: 'cursor-old', + updatedAt: new Date(), + }; + + // ── Act ────────────────────────────────────────────────────────── + + // 1. Start the worker. This calls poll() -> runExclusive -> + // fetchAndProcessEvents -> ensureIndexerState (reads lastLedger=200) + // -> server.getEvents (suspended on deferred promise). + void worker.start(); + await new Promise((r) => setTimeout(r, 0)); + + // 2. While the poll is suspended on getEvents, the operator calls + // resetIndexer(50). This directly upserts the DB - no mutex. + await resetIndexer(50); + + // Verify the DB now has lastLedger=50 (the reset). + expect(dbIndexerState.lastLedger).toBe(50); + + // 3. Resolve getEvents with a real event so the poll reaches the + // cursor upsert at the end of fetchAndProcessEvents. + // The poll captured lastLedger=200 before the reset, so it will + // upsert lastLedger=200, overwriting the reset's 50. + getEventsDeferred.resolve({ + events: [ + fakeStreamCreatedEvent({ id: 'e1', txHash: 'tx1', ledger: 210 }), + ], + latestCursor: 'cursor-new', + }); + + // Wait for the poll to fully complete. + await worker.waitForDrain(); + + // ── Assert ─────────────────────────────────────────────────────── + + // The upsert log tells the story: + // 1. resetIndexer writes lastLedger=50 (the operator's reset) + // 2. Worker's fetchAndProcessEvents writes lastLedger=200 (stale!) + // + // The poll's upsert overwrites the reset because it captured + // lastLedger=200 at the start and never re-read. + expect(upsertLog.length).toBe(2); + + // First upsert: resetIndexer. + expect(upsertLog[0]!.caller).toBe('resetIndexer'); + expect(upsertLog[0]!.lastLedger).toBe(50); + + // Second upsert: worker's stale cursor write. + expect(upsertLog[1]!.caller).toBe('worker'); + expect(upsertLog[1]!.lastLedger).toBe(200); + + // Final DB state: the reset value (50) must survive. + // This assertion asserts the DESIRED behavior. Before the fix, it + // fails because the poll's stale upsert (200) overwrites the reset. + // After the fix, the poll must respect the externally-set cursor. + expect(dbIndexerState.lastLedger).toBe(50); + }, + ); + + it( + 'replayFromLedger cursor must survive a concurrent scheduled poll - ' + + 'FAILS before fix (poll overwrites replay reset)', + async () => { + // ── Arrange ────────────────────────────────────────────────────── + + dbIndexerState = { + lastLedger: 300, + lastCursor: 'cursor-abc', + updatedAt: new Date(), + }; + + // ── Act ────────────────────────────────────────────────────────── + + // 1. Start the worker - first poll reads lastLedger=300, awaits getEvents. + void worker.start(); + await new Promise((r) => setTimeout(r, 0)); + + // 2. Resolve the first poll's getEvents so it finishes. + getEventsDeferred.resolve({ + events: [ + fakeStreamCreatedEvent({ id: 'e2', txHash: 'tx2', ledger: 310 }), + ], + latestCursor: 'cursor-first', + }); + + await worker.waitForDrain(); + + // 3. Set up a new deferred for the SECOND poll. + getEventsDeferred = defer(); + const server = (worker as any).server as { getEvents: ReturnType }; + server.getEvents = vi.fn(() => getEventsDeferred.promise); + + // 4. Start a second poll that will be mid-flight when we call + // replayFromLedger. + void worker.start(); + await new Promise((r) => setTimeout(r, 0)); + + // Second poll is now suspended at getEvents, having read lastLedger=300. + + // 5. Operator calls replayFromLedger(100): + // a) resetIndexer(100) -> upserts lastLedger=100 + // b) triggerPoll() -> queued behind the second poll via mutex + const replayPromise = replayFromLedger(100); + + // Yield to let resetIndexer(100) execute. + await new Promise((r) => setTimeout(r, 0)); + + // Verify resetIndexer wrote lastLedger=100. + expect(dbIndexerState.lastLedger).toBe(100); + + // 6. Resolve the second poll's getEvents with a real event. + // The second poll will finish, upserting its stale lastLedger=300, + // overwriting the reset's 100. + getEventsDeferred.resolve({ + events: [ + fakeStreamCreatedEvent({ id: 'e3', txHash: 'tx3', ledger: 320 }), + ], + latestCursor: 'cursor-poll2', + }); + + // The replay's triggerPoll will run after the second poll finishes. + // Set up yet another deferred for the replay's poll. + getEventsDeferred = defer(); + server.getEvents = vi.fn(() => getEventsDeferred.promise); + + // Yield so the replay's poll starts and suspends at getEvents. + await new Promise((r) => setTimeout(r, 0)); + + // Resolve the replay's poll with empty events (no more work to do). + getEventsDeferred.resolve({ + events: [], + latestCursor: 'cursor-replay', + }); + + // Wait for replay to fully complete. + await replayPromise; + await worker.waitForDrain(); + + // ── Assert ─────────────────────────────────────────────────────── + + // The upsert log tells the story: + // 1. First poll writes 300 (event ledger, normal advancement) + // 2. resetIndexer writes 100 (the operator's reset) + // 3. Second poll writes 300 (stale - overwrites the reset!) + // 4. Replay's poll writes whatever it read + // + // The key bug: step 3 overwrites step 2. + expect(upsertLog.length).toBeGreaterThanOrEqual(3); + + // Find the resetIndexer upsert. + const resetEntry = upsertLog.find((e) => e.caller === 'resetIndexer'); + expect(resetEntry).toBeDefined(); + expect(resetEntry!.lastLedger).toBe(100); + + // Final DB state: the reset value (100) must survive. + // This assertion asserts the DESIRED behavior. Before the fix, it + // fails because the second poll's stale upsert (300) overwrites the reset. + // After the fix, the poll must respect the externally-set cursor. + expect(dbIndexerState.lastLedger).toBe(100); + }, + ); +}); From d0f2d0a384545ffcc2da5978d5216832882aaa81 Mon Sep 17 00:00:00 2001 From: Fury03 Date: Sun, 6 Sep 2026 07:18:56 +0100 Subject: [PATCH 2/2] test(indexer): align reset/replay race test with the batchMutex fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI was red on both backend jobs: the two tests asserted the pre-fix race and the logger mock stubbed `requestContext` as a plain vi.fn(), so replayFromLedger threw "requestContext.getStore is not a function" as an unhandled rejection. Since this branch was opened, #1221 landed the production fix — resetIndexer now writes inside `sorobanEventWorker.runExclusive`, the same mutex that serialises poll/replay batches. Rework the tests to pin that behaviour down: - Drive the exported `sorobanEventWorker` singleton instead of a fresh `new SorobanEventWorker()`. That singleton is the instance resetIndexer and replayFromLedger lock against, so a separate instance shares no mutex with them and the ordering under test would not exist. - Assert the guarantee rather than the bug: the operator action is held until the in-flight batch releases the mutex, the poll's cursor write lands first, and the reset/replay cursor is the one that survives. - Give the logger mock a real AsyncLocalStorage for `requestContext`. Both tests still fail if resetIndexer's `runExclusive` wrapper is removed, so they guard the fix rather than merely passing alongside it. Co-Authored-By: Claude Opus 5 --- .../integration/reset-replay-race.test.ts | 332 ++++++++---------- 1 file changed, 140 insertions(+), 192 deletions(-) diff --git a/backend/tests/integration/reset-replay-race.test.ts b/backend/tests/integration/reset-replay-race.test.ts index 6c61ecb3..0a30220e 100644 --- a/backend/tests/integration/reset-replay-race.test.ts +++ b/backend/tests/integration/reset-replay-race.test.ts @@ -5,24 +5,28 @@ * Both `resetIndexer` and `replayFromLedger` are tested in isolation elsewhere, * but never interleaved with SorobanEventWorker's mutex-protected poll cycle. * This file fills that gap by simulating the exact scenario operators hit during - * incident recovery: a live, running indexer whose scheduled poll is mid-flight - * when an operator resets or replays. + * incident recovery: a live indexer whose scheduled poll is mid-flight when an + * operator resets or replays. * - * Acceptance criteria: - * - New test fails against current code (the race is real). - * - Once Functional Edge Case #19 is fixed, the test passes. - * - * The race (Functional Edge Case #19): + * The race (Functional Edge Case #19), as it behaved before #1221: * 1. SorobanEventWorker.poll() -> runExclusive -> fetchAndProcessEvents() * 2. fetchAndProcessEvents reads IndexerState.lastLedger (e.g. 200) * 3. fetchAndProcessEvents awaits server.getEvents (async network I/O) - * 4. ^ WINDOW: resetIndexer(50) is called, writing lastLedger=50 - * 5. fetchAndProcessEvents finishes, upserts lastLedger=200 (stale value - * captured in step 2, because processEvent error prevents ledger advance) - * 6. Reset is lost - cursor is 200 instead of 50. + * 4. ^ WINDOW: resetIndexer(50) writes lastLedger=50 with no lock held + * 5. fetchAndProcessEvents resumes and upserts the cursor it captured in + * step 2, rolling lastLedger forward past the reset + * 6. The reset is lost — the recovery action silently did nothing. + * + * resetIndexer now takes the worker's batchMutex (`runExclusive`), so its write + * is ordered after any in-flight batch and cannot be clobbered by one. These + * tests pin that ordering down end to end: they drive a real + * SorobanEventWorker poll to the point where it is suspended on RPC, inject the + * operator action, and assert the operator's cursor is the one that survives. * - * Root cause: resetIndexer bypasses the worker's batchMutex, so its DB write - * can be overwritten by a concurrent poll's cursor upsert. + * They exercise the exported `sorobanEventWorker` singleton on purpose — that + * is the instance `resetIndexer`/`replayFromLedger` lock against, so a fresh + * `new SorobanEventWorker()` would share no mutex with them and the ordering + * under test would not exist. */ import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; @@ -123,14 +127,18 @@ vi.mock('../../src/services/sse.service.js', () => ({ sseService: mockSseService, })); -vi.mock('../../src/logger.js', () => ({ - default: mockLogger, - requestContext: vi.fn(() => ({})), -})); +vi.mock('../../src/logger.js', async () => { + const { AsyncLocalStorage } = await import('async_hooks'); + return { + default: mockLogger, + // Real ALS: both the worker and replayFromLedger call getStore()/run(). + requestContext: new AsyncLocalStorage<{ requestId: string }>(), + }; +}); // ─── Imports (after mocks) ─────────────────────────────────────────────────── -import { SorobanEventWorker } from '../../src/workers/soroban-event-worker.js'; +import { sorobanEventWorker } from '../../src/workers/soroban-event-worker.js'; import { resetIndexer, replayFromLedger } from '../../src/services/indexerService.js'; /** @@ -228,9 +236,12 @@ function fakeStreamCreatedEvent(overrides: { // ─── Tests ─────────────────────────────────────────────────────────────────── describe('Reset/replay race with concurrent poll (Edge Case #19 - issue #1293)', () => { - let worker: SorobanEventWorker; + const worker = sorobanEventWorker; let getEventsDeferred: Deferred<{ events: any[]; latestCursor?: string | null }>; + /** Let queued microtasks and one timer tick run. */ + const flush = () => new Promise((r) => setTimeout(r, 0)); + beforeEach(() => { vi.clearAllMocks(); upsertLog.length = 0; @@ -242,188 +253,125 @@ describe('Reset/replay race with concurrent poll (Edge Case #19 - issue #1293)', updatedAt: new Date(), }; - worker = new SorobanEventWorker(); (worker as any).contractId = 'CTESTCONTRACT'; (worker as any).pollIntervalMs = 600_000; - // Wire up deferred getEvents mock. + // getEvents resolves whichever deferred is current when it is *called*, + // so a test can queue the next batch's response by reassigning. getEventsDeferred = defer(); - const server = (worker as any).server as { getEvents: ReturnType }; - server.getEvents = vi.fn(() => getEventsDeferred.promise); + (worker as any).server = { + getEvents: vi.fn(() => getEventsDeferred.promise), + }; }); - afterEach(() => { + afterEach(async () => { worker.stop(); + await worker.waitForDrain().catch(() => undefined); }); - it( - 'resetIndexer cursor must survive a concurrent scheduled poll - ' + - 'FAILS before fix (poll overwrites reset)', - async () => { - // ── Arrange ────────────────────────────────────────────────────── - - // Simulate: indexer is at ledger 200, poll starts reading from there. - dbIndexerState = { - lastLedger: 200, - lastCursor: 'cursor-old', - updatedAt: new Date(), - }; + it('resetIndexer cursor survives a poll that is already mid-flight', async () => { + // ── Arrange ──────────────────────────────────────────────────────── + // The indexer is at ledger 200; the poll about to start reads from there. + dbIndexerState = { + lastLedger: 200, + lastCursor: 'cursor-old', + updatedAt: new Date(), + }; - // ── Act ────────────────────────────────────────────────────────── - - // 1. Start the worker. This calls poll() -> runExclusive -> - // fetchAndProcessEvents -> ensureIndexerState (reads lastLedger=200) - // -> server.getEvents (suspended on deferred promise). - void worker.start(); - await new Promise((r) => setTimeout(r, 0)); - - // 2. While the poll is suspended on getEvents, the operator calls - // resetIndexer(50). This directly upserts the DB - no mutex. - await resetIndexer(50); - - // Verify the DB now has lastLedger=50 (the reset). - expect(dbIndexerState.lastLedger).toBe(50); - - // 3. Resolve getEvents with a real event so the poll reaches the - // cursor upsert at the end of fetchAndProcessEvents. - // The poll captured lastLedger=200 before the reset, so it will - // upsert lastLedger=200, overwriting the reset's 50. - getEventsDeferred.resolve({ - events: [ - fakeStreamCreatedEvent({ id: 'e1', txHash: 'tx1', ledger: 210 }), - ], - latestCursor: 'cursor-new', - }); - - // Wait for the poll to fully complete. - await worker.waitForDrain(); - - // ── Assert ─────────────────────────────────────────────────────── - - // The upsert log tells the story: - // 1. resetIndexer writes lastLedger=50 (the operator's reset) - // 2. Worker's fetchAndProcessEvents writes lastLedger=200 (stale!) - // - // The poll's upsert overwrites the reset because it captured - // lastLedger=200 at the start and never re-read. - expect(upsertLog.length).toBe(2); - - // First upsert: resetIndexer. - expect(upsertLog[0]!.caller).toBe('resetIndexer'); - expect(upsertLog[0]!.lastLedger).toBe(50); - - // Second upsert: worker's stale cursor write. - expect(upsertLog[1]!.caller).toBe('worker'); - expect(upsertLog[1]!.lastLedger).toBe(200); - - // Final DB state: the reset value (50) must survive. - // This assertion asserts the DESIRED behavior. Before the fix, it - // fails because the poll's stale upsert (200) overwrites the reset. - // After the fix, the poll must respect the externally-set cursor. - expect(dbIndexerState.lastLedger).toBe(50); - }, - ); - - it( - 'replayFromLedger cursor must survive a concurrent scheduled poll - ' + - 'FAILS before fix (poll overwrites replay reset)', - async () => { - // ── Arrange ────────────────────────────────────────────────────── - - dbIndexerState = { - lastLedger: 300, - lastCursor: 'cursor-abc', - updatedAt: new Date(), - }; + // ── Act ──────────────────────────────────────────────────────────── + // 1. start() -> poll() -> runExclusive -> fetchAndProcessEvents, which + // reads lastLedger=200 and then suspends on server.getEvents. + void worker.start(); + await flush(); + + // 2. Mid-flight, the operator resets to ledger 50. This is the whole + // point of the test: the call is issued while the poll still holds + // the batch mutex, so it must not be applied yet. + const resetPromise = resetIndexer(50); + await flush(); + expect(upsertLog).toHaveLength(0); + + // 3. Release the poll. It processes its event and writes its cursor + // (ledger 210) — that write is legitimate, it just must not be the + // last word. + getEventsDeferred.resolve({ + events: [fakeStreamCreatedEvent({ id: 'e1', txHash: 'tx1', ledger: 210 })], + latestCursor: 'cursor-new', + }); + + await resetPromise; + await worker.waitForDrain(); + + // ── Assert ───────────────────────────────────────────────────────── + // The poll's cursor write lands first, the reset second — the ordering + // the batch mutex guarantees. Before #1221 the reset slipped in ahead of + // the poll and was then overwritten by it. + expect(upsertLog.map((e) => e.caller)).toEqual(['worker', 'resetIndexer']); + expect(upsertLog[0]!.lastLedger).toBe(210); + expect(upsertLog[1]!.lastLedger).toBe(50); + + // Final DB state: the operator's reset is what survives. + expect(dbIndexerState.lastLedger).toBe(50); + expect(dbIndexerState.lastCursor).toBeNull(); + }); - // ── Act ────────────────────────────────────────────────────────── - - // 1. Start the worker - first poll reads lastLedger=300, awaits getEvents. - void worker.start(); - await new Promise((r) => setTimeout(r, 0)); - - // 2. Resolve the first poll's getEvents so it finishes. - getEventsDeferred.resolve({ - events: [ - fakeStreamCreatedEvent({ id: 'e2', txHash: 'tx2', ledger: 310 }), - ], - latestCursor: 'cursor-first', - }); - - await worker.waitForDrain(); - - // 3. Set up a new deferred for the SECOND poll. - getEventsDeferred = defer(); - const server = (worker as any).server as { getEvents: ReturnType }; - server.getEvents = vi.fn(() => getEventsDeferred.promise); - - // 4. Start a second poll that will be mid-flight when we call - // replayFromLedger. - void worker.start(); - await new Promise((r) => setTimeout(r, 0)); - - // Second poll is now suspended at getEvents, having read lastLedger=300. - - // 5. Operator calls replayFromLedger(100): - // a) resetIndexer(100) -> upserts lastLedger=100 - // b) triggerPoll() -> queued behind the second poll via mutex - const replayPromise = replayFromLedger(100); - - // Yield to let resetIndexer(100) execute. - await new Promise((r) => setTimeout(r, 0)); - - // Verify resetIndexer wrote lastLedger=100. - expect(dbIndexerState.lastLedger).toBe(100); - - // 6. Resolve the second poll's getEvents with a real event. - // The second poll will finish, upserting its stale lastLedger=300, - // overwriting the reset's 100. - getEventsDeferred.resolve({ - events: [ - fakeStreamCreatedEvent({ id: 'e3', txHash: 'tx3', ledger: 320 }), - ], - latestCursor: 'cursor-poll2', - }); - - // The replay's triggerPoll will run after the second poll finishes. - // Set up yet another deferred for the replay's poll. - getEventsDeferred = defer(); - server.getEvents = vi.fn(() => getEventsDeferred.promise); - - // Yield so the replay's poll starts and suspends at getEvents. - await new Promise((r) => setTimeout(r, 0)); - - // Resolve the replay's poll with empty events (no more work to do). - getEventsDeferred.resolve({ - events: [], - latestCursor: 'cursor-replay', - }); - - // Wait for replay to fully complete. - await replayPromise; - await worker.waitForDrain(); - - // ── Assert ─────────────────────────────────────────────────────── - - // The upsert log tells the story: - // 1. First poll writes 300 (event ledger, normal advancement) - // 2. resetIndexer writes 100 (the operator's reset) - // 3. Second poll writes 300 (stale - overwrites the reset!) - // 4. Replay's poll writes whatever it read - // - // The key bug: step 3 overwrites step 2. - expect(upsertLog.length).toBeGreaterThanOrEqual(3); - - // Find the resetIndexer upsert. - const resetEntry = upsertLog.find((e) => e.caller === 'resetIndexer'); - expect(resetEntry).toBeDefined(); - expect(resetEntry!.lastLedger).toBe(100); - - // Final DB state: the reset value (100) must survive. - // This assertion asserts the DESIRED behavior. Before the fix, it - // fails because the second poll's stale upsert (300) overwrites the reset. - // After the fix, the poll must respect the externally-set cursor. - expect(dbIndexerState.lastLedger).toBe(100); - }, - ); + it('replayFromLedger cursor survives a poll that is already mid-flight', async () => { + // ── Arrange ──────────────────────────────────────────────────────── + dbIndexerState = { + lastLedger: 300, + lastCursor: 'cursor-abc', + updatedAt: new Date(), + }; + + // ── Act ──────────────────────────────────────────────────────────── + // 1. First poll runs to completion normally, advancing to ledger 310. + void worker.start(); + await flush(); + getEventsDeferred.resolve({ + events: [fakeStreamCreatedEvent({ id: 'e2', txHash: 'tx2', ledger: 310 })], + latestCursor: 'cursor-first', + }); + await worker.waitForDrain(); + expect(dbIndexerState.lastLedger).toBe(310); + + // 2. Start a second poll and leave it suspended on getEvents. + getEventsDeferred = defer(); + void worker.start(); + await flush(); + + // 3. Operator replays from ledger 100 while that poll is in flight: + // resetIndexer(100) queues behind it on the mutex, and the replay's + // own triggerPoll queues behind the reset. + const replayPromise = replayFromLedger(100); + await flush(); + + // Nothing applied yet — the second poll still holds the mutex. + const beforeRelease = upsertLog.length; + expect(dbIndexerState.lastLedger).toBe(310); + + // 4. Release the second poll (it writes ledger 320), and hand the + // replay's own poll an empty batch so it makes no cursor write. + const secondPollDeferred = getEventsDeferred; + getEventsDeferred = defer(); + getEventsDeferred.resolve({ events: [], latestCursor: 'cursor-replay' }); + secondPollDeferred.resolve({ + events: [fakeStreamCreatedEvent({ id: 'e3', txHash: 'tx3', ledger: 320 })], + latestCursor: 'cursor-poll2', + }); + + await replayPromise; + await worker.waitForDrain(); + + // ── Assert ───────────────────────────────────────────────────────── + // The second poll's stale write lands, then the replay's reset — and + // nothing after it, because the replay's own poll found no events. + const afterRelease = upsertLog.slice(beforeRelease); + expect(afterRelease.map((e) => e.caller)).toEqual(['worker', 'resetIndexer']); + expect(afterRelease[0]!.lastLedger).toBe(320); + expect(afterRelease[1]!.lastLedger).toBe(100); + + // Final DB state: the replay's cursor is what survives. + expect(dbIndexerState.lastLedger).toBe(100); + expect(dbIndexerState.lastCursor).toBeNull(); + }); });