fix(indexer): prevent overlapping poll executions

indexer/index.mjs

@@ -1,4 +1,5 @@
 import { appendFileSync, existsSync, readFileSync, writeFileSync } from "node:fs";
+import { pathToFileURL } from "node:url";
 import { rpc, scValToNative } from "@stellar/stellar-sdk";
 import { withRateLimitBackoff } from "./rpc-backoff.mjs";
 
@@ -75,7 +76,7 @@ function cursorLedger(cursor) {
 
 let isPolling = false;
 let shutdownRequested = false;
-let intervalId;
+let pollingTimerId;
 let backoffTimeoutId;
 let resumeBackoff;
 
@@ -104,8 +105,9 @@ function rpcCall(operation) {
 function handleShutdown(signal) {
   console.log(`Received ${signal}. Shutting down gracefully...`);
   shutdownRequested = true;
-  if (intervalId) {
-    clearInterval(intervalId);
+  if (pollingTimerId) {
+    clearTimeout(pollingTimerId);
+    pollingTimerId = undefined;
   }
   if (backoffTimeoutId) {
     clearTimeout(backoffTimeoutId);
@@ -173,6 +175,34 @@ async function poll() {
   }
 }
 
-console.log(`indexing ${CONTRACT_ID} from ${RPC_URL} every ${POLL_MS}ms`);
-await poll();
-intervalId = setInterval(() => poll().catch((e) => console.error(e.message ?? e)), POLL_MS);
+export async function runPollingLoop({
+  pollFn = poll,
+  pollMs = POLL_MS,
+  schedule = (callback, delay) => {
+    pollingTimerId = setTimeout(callback, delay);
+    return pollingTimerId;
+  },
+  logger = (message) => console.error(message),
+} = {}) {
+  try {
+    await pollFn();
+  } catch (error) {
+    logger(error.message ?? error);
+  }
+
+  if (shutdownRequested) {
+    return;
+  }
+
+  schedule(() => {
+    void runPollingLoop({ pollFn, pollMs, schedule, logger });
+  }, pollMs);
+}
+
+const shouldStartPolling =
+  process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href;
+
+if (shouldStartPolling) {
+  console.log(`indexing ${CONTRACT_ID} from ${RPC_URL} every ${POLL_MS}ms`);
+  void runPollingLoop();
+}

indexer/index.test.mjs

@@ -1,6 +1,6 @@
 import test from 'node:test';
 import assert from 'node:assert/strict';
-import { validateConfig } from './index.mjs';
+import { validateConfig, runPollingLoop } from './index.mjs';
 
 test('validateConfig rejects missing required env values', () => {
   const result = validateConfig({
@@ -23,3 +23,75 @@ test('validateConfig accepts populated env values', () => {
   assert.equal(result.value.CONTRACT_ID, 'CC123');
   assert.equal(result.value.RPC_URL, 'https://example.com');
 });
+
+test('runPollingLoop serializes polls and reschedules after failure', async () => {
+  let releaseFirstPoll;
+  let activePolls = 0;
+  let maxActivePolls = 0;
+  let pollCount = 0;
+  const scheduled = [];
+  const loggerMessages = [];
+
+  const flush = async () => {
+    await Promise.resolve();
+    await Promise.resolve();
+  };
+
+  const pollFn = async () => {
+    pollCount += 1;
+    activePolls += 1;
+    maxActivePolls = Math.max(maxActivePolls, activePolls);
+
+    try {
+      if (pollCount === 1) {
+        await new Promise((resolve) => {
+          releaseFirstPoll = resolve;
+        });
+      } else if (pollCount === 2) {
+        throw new Error('boom');
+      }
+    } finally {
+      activePolls -= 1;
+    }
+  };
+
+  const schedule = (callback, delay) => {
+    const entry = { callback, delay };
+    scheduled.push(entry);
+    return entry;
+  };
+
+  const logger = (message) => loggerMessages.push(message);
+
+  runPollingLoop({ pollFn, pollMs: 50, schedule, logger });
+
+  assert.equal(activePolls, 1);
+  assert.equal(maxActivePolls, 1);
+  assert.equal(scheduled.length, 0);
+
+  releaseFirstPoll();
+  await flush();
+
+  assert.equal(activePolls, 0);
+  assert.equal(scheduled.length, 1);
+  assert.equal(scheduled[0].delay, 50);
+
+  scheduled[0].callback();
+  await flush();
+
+  assert.equal(pollCount, 2);
+  assert.equal(activePolls, 0);
+  assert.deepEqual(loggerMessages, ['boom']);
+  assert.equal(maxActivePolls, 1);
+  assert.equal(scheduled.length, 2);
+
+  assert.equal(pollCount, 2);
+  scheduled[1].callback();
+  await flush();
+
+  assert.equal(pollCount, 3);
+  assert.equal(activePolls, 0);
+  assert.equal(maxActivePolls, 1);
+  assert.equal(scheduled.length, 3);
+  assert.equal(scheduled[2].delay, 50);
+});