From f0c523c5156b49f004127449b4a02f6b3f4257f4 Mon Sep 17 00:00:00 2001
From: Dhvani Patel <dhvani.patel@uwaterloo.ca>
Date: Thu, 13 Jun 2024 12:41:21 +0100
Subject: [PATCH] fix: indexer skipping blocks

---
 packages/block-logs-stream/src/fetchLogs.ts | 13 ++++++++++-
 packages/store-sync/src/createStoreSync.ts  | 24 ++++++++++++++-------
 2 files changed, 28 insertions(+), 9 deletions(-)

diff --git a/packages/block-logs-stream/src/fetchLogs.ts b/packages/block-logs-stream/src/fetchLogs.ts
index d6de091619..34fd9d5657 100644
--- a/packages/block-logs-stream/src/fetchLogs.ts
+++ b/packages/block-logs-stream/src/fetchLogs.ts
@@ -90,7 +90,18 @@ export async function* fetchLogs<TAbiEvents extends readonly AbiEvent[]>({
     try {
       const toBlock = fromBlock + blockRange;
       debug("getting logs", { fromBlock, toBlock });
-      const logs = await publicClient.getLogs({ ...getLogsOpts, fromBlock, toBlock, strict: true });
+
+      const [latestBlockNumber, logs] = await Promise.all([
+        publicClient.getBlockNumber({ cacheTime: 0 }),
+        publicClient.getLogs({ ...getLogsOpts, fromBlock, toBlock, strict: true }),
+      ]);
+      if (latestBlockNumber < toBlock) {
+        const blockTimeInSeconds = 2;
+        const seconds = Number(toBlock - latestBlockNumber) * blockTimeInSeconds;
+        debug(`latest block number ${latestBlockNumber} is less than toBlock ${toBlock}, retrying in ${seconds}s`);
+        await wait(1000 * seconds);
+        continue;
+      }
       yield { fromBlock, toBlock, logs };
       fromBlock = toBlock + 1n;
       blockRange = bigIntMin(maxBlockRange, getLogsOpts.toBlock - fromBlock);
diff --git a/packages/store-sync/src/createStoreSync.ts b/packages/store-sync/src/createStoreSync.ts
index 53606d154c..acac95d0b0 100644
--- a/packages/store-sync/src/createStoreSync.ts
+++ b/packages/store-sync/src/createStoreSync.ts
@@ -188,6 +188,10 @@ export async function createStoreSync<config extends StoreConfig = StoreConfig>(
     tap((startBlock) => debug("starting sync from block", startBlock)),
   );
 
+  let startBlock: bigint | null = null;
+  let endBlock: bigint | null = null;
+  let lastBlockNumberProcessed: bigint | null = null;
+
   const latestBlock$ = createBlockStream({ publicClient, blockTag: followBlockTag }).pipe(shareReplay(1));
   const latestBlockNumber$ = latestBlock$.pipe(
     map((block) => block.number),
@@ -198,13 +202,12 @@ export async function createStoreSync<config extends StoreConfig = StoreConfig>(
       count: undefined, // ie unlimited
       delay: 3000, // poll the RPC every 3 seconds
     }),
+    filter((blockNumber) => {
+      return lastBlockNumberProcessed == null || blockNumber > lastBlockNumberProcessed;
+    }),
     shareReplay(1),
   );
 
-  let startBlock: bigint | null = null;
-  let endBlock: bigint | null = null;
-  let lastBlockNumberProcessed: bigint | null = null;
-
   const storedBlock$ = combineLatest([startBlock$, latestBlockNumber$]).pipe(
     map(([startBlock, endBlock]) => ({ startBlock, endBlock })),
     tap((range) => {
@@ -212,15 +215,20 @@ export async function createStoreSync<config extends StoreConfig = StoreConfig>(
       endBlock = range.endBlock;
     }),
     concatMap((range) => {
+      const fromBlock = lastBlockNumberProcessed
+        ? bigIntMax(range.startBlock, lastBlockNumberProcessed + 1n)
+        : range.startBlock;
+      const toBlock = range.endBlock;
+      if (toBlock < fromBlock) {
+        throw new Error(`toBlock ${toBlock} is less than fromBlock ${fromBlock}`);
+      }
       const storedBlocks = fetchAndStoreLogs({
         publicClient,
         address,
         events: storeEventsAbi,
         maxBlockRange,
-        fromBlock: lastBlockNumberProcessed
-          ? bigIntMax(range.startBlock, lastBlockNumberProcessed + 1n)
-          : range.startBlock,
-        toBlock: range.endBlock,
+        fromBlock: fromBlock,
+        toBlock: toBlock,
         storageAdapter,
         logFilter,
       });