fix: indexer skipping blocks

tenetxyz · Jun 13, 2024 · f0c523c · f0c523c
1 parent 4fc754d
commit f0c523c
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 9 deletions.
diff --git 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
@@ -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,29 +202,33 @@ 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) => {
       startBlock = range.startBlock;
       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,
       });