fix(backend): serialize indexer triggerPoll with activeBatch mutex (#843)

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<void> | null = null;
+  /** Promise chain that serializes all `fetchAndProcessEvents` invocations. */
+  private batchMutex: Promise<void> = Promise.resolve();
 
   constructor() {
     const rpcUrl =
@@ -118,24 +124,50 @@ 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<void> {
     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<void> {
     if (!this.isRunning) return;
 
     try {
-      await this.fetchAndProcessEvents();
+      await this.runExclusive(() => this.fetchAndProcessEvents());
     } catch (err) {
       logger.error("[SorobanWorker] Manual poll error:", err);
     }
   }
 
   // ─── Internal ──────────────────────────────────────────────────────────────
 
+  /**
+   * Run `fn` exclusively with any other poll/replay batch.
+   * Registers the work on `activeBatch` so `waitForDrain` awaits replays too.
+   */
+  private runExclusive(fn: () => Promise<void>): Promise<void> {
+    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<void> {
-    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();
     }
   }

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<void>((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<void>((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();
+    });
+  });
 });