Merge pull request #136 from presidojay1/fix/event-poller-truncation-cursor-115

fix: don't advance indexer cursor past a truncated event batch

Author: prodbycorne

src/indexer/eventPoller.js

@@ -59,20 +59,37 @@ class EventPoller {
       limit: this.pollLimit,
     });
 
-    const parsedEvents = (response.events || [])
-      .map(parseContractEvent)
-      .filter(Boolean);
+    const rawEvents = response.events || [];
+    const parsedEvents = rawEvents.map(parseContractEvent).filter(Boolean);
 
     for (const event of parsedEvents) {
       await this.store.saveEvent(event);
     }
 
-    const latestIndexedLedger = Math.max(
-      response.latestLedger || previousLedger,
-      ...parsedEvents.map((event) => event.ledger)
-    );
+    // response.latestLedger is the chain's current tip, not how far this
+    // particular call actually got — if the RPC returned a full pollLimit
+    // batch, more matching events may exist beyond it. Jumping the cursor
+    // to the tip in that case permanently skips whatever wasn't returned,
+    // with no retry and no downstream reconciliation that could recover
+    // it (#115). Only safe to advance to the tip when this batch wasn't
+    // truncated; otherwise advance only past what was actually processed,
+    // so the next poll picks up right where this one left off.
+    const truncated = rawEvents.length >= this.pollLimit;
+    const eventLedgers = parsedEvents.map((event) => event.ledger);
+    const latestIndexedLedger = truncated
+      ? Math.max(previousLedger ?? 0, ...eventLedgers)
+      : Math.max(response.latestLedger || previousLedger || 0, ...eventLedgers);
+
     await this.store.setLastLedger(latestIndexedLedger);
 
+    if (truncated) {
+      this.logger.warn('SmartDrop event poll truncated by pollLimit; more events pending next cycle', {
+        pollLimit: this.pollLimit,
+        indexed_events: parsedEvents.length,
+        resumed_from_ledger: latestIndexedLedger + 1,
+      });
+    }
+
     this.latestLedger = response.latestLedger || null;
     this.lastRun = new Date().toISOString();
     this.lastError = null;
@@ -82,6 +99,7 @@ class EventPoller {
       start_ledger: startLedger,
       latest_ledger: response.latestLedger,
       indexed_events: parsedEvents.length,
+      truncated,
     };
   }
 

test/eventPoller.test.js

@@ -3,7 +3,7 @@
 const { nativeToScVal } = require('stellar-sdk');
 const { EventPoller } = require('../src/indexer/eventPoller');
 
-function contractEvent() {
+function contractEvent(overrides = {}) {
   return {
     id: 'evt-1',
     type: 'contract',
@@ -19,9 +19,30 @@ function contractEvent() {
       total_amount: 1000n,
       expiry_ledger: 500n,
     }),
+    ...overrides,
   };
 }
 
+// N distinct events at consecutive ledgers starting at `startLedger`, used
+// to simulate a burst/backlog large enough to fill a batch of size `n`.
+function contractEvents(n, startLedger) {
+  return Array.from({ length: n }, (_, i) => {
+    const ledger = startLedger + i;
+    return contractEvent({
+      id: `evt-${ledger}`,
+      ledger,
+      pagingToken: `${ledger}-1`,
+      value: nativeToScVal({
+        airdrop_id: `drop-${ledger}`,
+        creator: 'GCREATOR',
+        token: 'USDC',
+        total_amount: 1000n,
+        expiry_ledger: 500n,
+      }),
+    });
+  });
+}
+
 describe('EventPoller', () => {
   test('polls Soroban RPC, stores parsed events, and advances last ledger', async () => {
     const server = {
@@ -95,4 +116,122 @@ describe('EventPoller', () => {
 
     await expect(poller.pollOnce()).resolves.toMatchObject({ skipped: true });
   });
+
+  describe('truncated batch (#115)', () => {
+    test('advances last_ledger only to the last processed event, not to the chain tip', async () => {
+      const pollLimit = 5;
+      // Simulates a real burst/backlog: exactly pollLimit events returned,
+      // last event's ledger (24) is far behind the chain tip (500).
+      const events = contractEvents(pollLimit, 20);
+      const server = {
+        getEvents: jest.fn(async () => ({ latestLedger: 500, events })),
+      };
+      const store = {
+        getLastLedger: jest.fn(async () => null),
+        saveEvent: jest.fn(async () => {}),
+        setLastLedger: jest.fn(async () => {}),
+      };
+      const logger = { info: jest.fn(), warn: jest.fn(), error: jest.fn(), debug: jest.fn() };
+
+      const poller = new EventPoller({
+        enabled: true,
+        contractId: 'CCONTRACT',
+        startLedger: 10,
+        pollLimit,
+        server,
+        store,
+        logger,
+      });
+
+      const result = await poller.pollOnce();
+
+      // Not 500 (response.latestLedger) — that would permanently skip
+      // whatever exists between ledger 24 and the tip.
+      expect(store.setLastLedger).toHaveBeenCalledWith(24);
+      expect(result).toMatchObject({ truncated: true, indexed_events: pollLimit });
+      expect(logger.warn).toHaveBeenCalledWith(
+        'SmartDrop event poll truncated by pollLimit; more events pending next cycle',
+        expect.objectContaining({ pollLimit, indexed_events: pollLimit, resumed_from_ledger: 25 }),
+      );
+    });
+
+    test('the next poll resumes from the last processed event, picking up previously-skippable events', async () => {
+      const pollLimit = 5;
+      const firstBatch = contractEvents(pollLimit, 20); // ledgers 20-24
+      // Events that would have been silently skipped pre-fix: they sit
+      // between the last processed ledger (24) and the previous poll's
+      // chain-tip snapshot (500).
+      const skippableRangeEvents = contractEvents(2, 100); // ledgers 100-101
+
+      const server = { getEvents: jest.fn() };
+      server.getEvents
+        .mockImplementationOnce(async () => ({ latestLedger: 500, events: firstBatch }))
+        .mockImplementationOnce(async () => ({ latestLedger: 500, events: skippableRangeEvents }));
+
+      // Stateful store, so the second pollOnce() actually reads back what
+      // the first one wrote — proves resumption across ticks, not just
+      // within one call.
+      let lastLedger = null;
+      const store = {
+        getLastLedger: jest.fn(async () => lastLedger),
+        saveEvent: jest.fn(async () => {}),
+        setLastLedger: jest.fn(async (ledger) => {
+          lastLedger = ledger;
+        }),
+      };
+
+      const poller = new EventPoller({
+        enabled: true,
+        contractId: 'CCONTRACT',
+        startLedger: 10,
+        pollLimit,
+        server,
+        store,
+        logger: { info: jest.fn(), warn: jest.fn(), error: jest.fn(), debug: jest.fn() },
+      });
+
+      const first = await poller.pollOnce();
+      expect(first.truncated).toBe(true);
+      expect(lastLedger).toBe(24);
+
+      const second = await poller.pollOnce();
+
+      // Resumed from 25 (last processed + 1), not 501 (tip + 1) — the
+      // pre-fix bug would have started here at 501, skipping ledgers
+      // 25-499 (including skippableRangeEvents) forever.
+      expect(server.getEvents.mock.calls[1][0].startLedger).toBe(25);
+      expect(second.indexed_events).toBe(2);
+      expect(store.saveEvent).toHaveBeenCalledTimes(pollLimit + 2);
+    });
+
+    test('a batch smaller than pollLimit still advances to the chain tip and logs no warning', async () => {
+      const pollLimit = 100;
+      const events = contractEvents(3, 20);
+      const server = {
+        getEvents: jest.fn(async () => ({ latestLedger: 500, events })),
+      };
+      const store = {
+        getLastLedger: jest.fn(async () => null),
+        saveEvent: jest.fn(async () => {}),
+        setLastLedger: jest.fn(async () => {}),
+      };
+      const logger = { info: jest.fn(), warn: jest.fn(), error: jest.fn(), debug: jest.fn() };
+
+      const poller = new EventPoller({
+        enabled: true,
+        contractId: 'CCONTRACT',
+        startLedger: 10,
+        pollLimit,
+        server,
+        store,
+        logger,
+      });
+
+      const result = await poller.pollOnce();
+
+      expect(store.setLastLedger).toHaveBeenCalledWith(500);
+      expect(result.truncated).toBe(false);
+      expect(logger.warn).not.toHaveBeenCalled();
+    });
+  });
 });