From 18ed7873bdb5ed681b5509f334cff8370e6cde1f Mon Sep 17 00:00:00 2001 From: TCROWN10 Date: Mon, 27 Jul 2026 19:23:58 +0100 Subject: [PATCH] fix(backend): serialize triggerPoll behind activeBatch mutex (#843) Prevent concurrent poll/replay from overlapping fetchAndProcessEvents so cursor writes cannot regress, and make waitForDrain cover replay batches. Co-authored-by: Cursor --- backend/src/workers/soroban-event-worker.ts | 48 ++++++++++++--- backend/tests/soroban-event-worker.test.ts | 68 +++++++++++++++++++++ 2 files changed, 108 insertions(+), 8 deletions(-) diff --git a/backend/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts index c2ac91ab..c2f886d6 100644 --- a/backend/src/workers/soroban-event-worker.ts +++ b/backend/src/workers/soroban-event-worker.ts @@ -77,7 +77,13 @@ export class SorobanEventWorker { private isRunning = false; private pollTimer: NodeJS.Timeout | undefined; + /** + * In-flight fetch/process work from either the scheduler or `triggerPoll`. + * `waitForDrain()` awaits this so graceful shutdown covers replay batches too. + */ private activeBatch: Promise | null = null; + /** Promise chain that serializes all `fetchAndProcessEvents` invocations. */ + private batchMutex: Promise = Promise.resolve(); constructor() { const rpcUrl = @@ -118,17 +124,21 @@ export class SorobanEventWorker { logger.info("[SorobanWorker] Stopped."); } - /** Wait for the currently-running poll batch to finish (no-op if idle). */ + /** Wait for the currently-running poll/replay batch to finish (no-op if idle). */ async waitForDrain(): Promise { if (this.activeBatch) await this.activeBatch; } - /** Trigger an immediate poll cycle (used for replay and manual updates). */ + /** + * Trigger an immediate poll cycle (used for replay and manual updates). + * Serialized with the scheduled poll via `runExclusive` so two cursor writes + * cannot overlap and regress `lastCursor` (#843). + */ async triggerPoll(): Promise { if (!this.isRunning) return; try { - await this.fetchAndProcessEvents(); + await this.runExclusive(() => this.fetchAndProcessEvents()); } catch (err) { logger.error("[SorobanWorker] Manual poll error:", err); } @@ -136,6 +146,28 @@ export class SorobanEventWorker { // ─── Internal ────────────────────────────────────────────────────────────── + /** + * Run `fn` exclusively with any other poll/replay batch. + * Registers the work on `activeBatch` so `waitForDrain` awaits replays too. + */ + private runExclusive(fn: () => Promise): Promise { + const run = this.batchMutex.then(fn); + // Keep the mutex chain alive even when a batch rejects. + const gate = run.then( + () => undefined, + () => undefined, + ); + this.batchMutex = gate; + this.activeBatch = gate; + // Clear only if nothing newer registered on activeBatch after this gate. + void gate.then(() => { + if (this.activeBatch === gate) { + this.activeBatch = null; + } + }); + return run; + } + private scheduleNext(): void { if (!this.isRunning) return; this.pollTimer = setTimeout(() => this.poll(), this.pollIntervalMs); @@ -172,13 +204,13 @@ export class SorobanEventWorker { } private async poll(): Promise { - this.activeBatch = this.fetchAndProcessEvents().catch((err) => { - logger.error("[SorobanWorker] Unhandled error during poll:", err); - }); try { - await this.activeBatch; + await this.runExclusive(() => + this.fetchAndProcessEvents().catch((err) => { + logger.error("[SorobanWorker] Unhandled error during poll:", err); + }), + ); } finally { - this.activeBatch = null; this.scheduleNext(); } } diff --git a/backend/tests/soroban-event-worker.test.ts b/backend/tests/soroban-event-worker.test.ts index 772017f8..cffda849 100644 --- a/backend/tests/soroban-event-worker.test.ts +++ b/backend/tests/soroban-event-worker.test.ts @@ -631,4 +631,72 @@ describe('SorobanEventWorker', () => { ); }); }); + + describe('poll / triggerPoll serialization (#843)', () => { + it('does not run fetchAndProcessEvents concurrently for overlapping poll and triggerPoll', async () => { + let concurrent = 0; + let maxConcurrent = 0; + const releases: Array<() => void> = []; + + const fetchSpy = vi + .spyOn(worker as any, 'fetchAndProcessEvents') + .mockImplementation(async () => { + concurrent += 1; + maxConcurrent = Math.max(maxConcurrent, concurrent); + await new Promise((resolve) => { + releases.push(resolve); + }); + concurrent -= 1; + }); + + // Avoid scheduling real timers from poll()'s finally + vi.spyOn(worker as any, 'scheduleNext').mockImplementation(() => {}); + (worker as any).isRunning = true; + + const pollPromise = (worker as any).poll(); + const triggerPromise = worker.triggerPoll(); + + await vi.waitFor(() => expect(releases.length).toBe(1)); + expect(concurrent).toBe(1); + + releases[0]!(); + await vi.waitFor(() => expect(releases.length).toBe(2)); + expect(concurrent).toBe(1); + + releases[1]!(); + await Promise.all([pollPromise, triggerPromise]); + + expect(maxConcurrent).toBe(1); + expect(fetchSpy).toHaveBeenCalledTimes(2); + }); + + it('waitForDrain awaits an in-flight triggerPoll batch', async () => { + let resolveFetch!: () => void; + vi.spyOn(worker as any, 'fetchAndProcessEvents').mockImplementation( + () => + new Promise((resolve) => { + resolveFetch = resolve; + }), + ); + vi.spyOn(worker as any, 'scheduleNext').mockImplementation(() => {}); + (worker as any).isRunning = true; + + const triggerPromise = worker.triggerPoll(); + // activeBatch is registered synchronously in runExclusive + expect((worker as any).activeBatch).not.toBeNull(); + + let drained = false; + const drainPromise = worker.waitForDrain().then(() => { + drained = true; + }); + + await Promise.resolve(); + expect(drained).toBe(false); + + resolveFetch(); + await Promise.all([triggerPromise, drainPromise]); + expect(drained).toBe(true); + expect((worker as any).activeBatch).toBeNull(); + }); + }); });