feat(reku): enhance AutoCrossChecker with start/stop functionality and improve error handling in BaseCrossChecker

Author: murongg

packages/reku/src/event/crosschecker/autochecker.ts

@@ -10,6 +10,7 @@ import { BaseCrossChecker } from './basechecker'
 export class AutoCrossChecker extends BaseCrossChecker {
   cache: CrossCheckerCacheManager | undefined = undefined
   checkpointBlockNumber: number | undefined
+  playing = false
 
   constructor(
     provider: Providers,
@@ -140,10 +141,14 @@ export class AutoCrossChecker extends BaseCrossChecker {
       // never ends if options.toBlock is not provided
       : () => false
 
+    this.playing = true
     debug('crosscheck running')
 
     // TODO: replace polling with schedule cron
-    await polling(async () => {
+
+    const pollingFn = async () => {
+      if (!this.playing)
+        return false
       try {
         debug('start polling')
         const wait = await waitOrUpdateToBlock()
@@ -165,7 +170,13 @@ export class AutoCrossChecker extends BaseCrossChecker {
         debug('polling error', error)
         return false
       }
-    }, pollingInterval)
+    }
+
+    polling(pollingFn, pollingInterval)
+  }
+
+  stop() {
+    this.playing = false
   }
 
   async diff(logs: ethers.Log[], ignoreLogs: SimpleLog[]): Promise<ethers.Log[]> {

packages/reku/src/event/crosschecker/basechecker.ts

@@ -1,5 +1,5 @@
 import type { ethers } from 'ethers'
-import { timeoutWithRetry } from '@ora-io/utils'
+import { timeoutWithRetry, to } from '@ora-io/utils'
 import { ETH_BLOCK_COUNT_ONE_HOUR } from '../../constants'
 import type { Providers } from '../../types/w3'
 import { debug } from '../../debug'
@@ -35,11 +35,15 @@ export class BaseCrossChecker {
 
     // define from, to
     // TODO: use blockNumber for performance
-    const block = await timeoutWithRetry(() => {
+    const [err, block] = await to(timeoutWithRetry(() => {
       if (!this.provider || !this.provider.provider)
         throw new Error('provider not ready')
       return this.provider.provider.getBlock('latest')
-    }, 15 * 1000, 3)
+    }, 15 * 1000, 3))
+    if (err) {
+      console.warn('crosscheck failed to get latest block', err)
+      return
+    }
     if (!block) {
       console.warn('crosscheck failed to get latest block')
       return
@@ -62,11 +66,15 @@ export class BaseCrossChecker {
     ccfOptions: CrossCheckFromParam,
   ) {
     // TODO: use blockNumber for performance
-    const block = await timeoutWithRetry(() => {
+    const [err, block] = await to(timeoutWithRetry(() => {
       if (!this.provider || !this.provider.provider)
         throw new Error('provider not ready')
       return this.provider.provider.getBlock('latest')
-    }, 15 * 1000, 3)
+    }, 15 * 1000, 3))
+    if (err) {
+      console.warn('crosscheck failed to get latest block', err)
+      return
+    }
     if (!block) {
       console.warn('crosscheck failed to get latest block')
       return
@@ -86,24 +94,6 @@ export class BaseCrossChecker {
     return block.number
   }
 
-  /**
-   * @deprecated
-   * @param logs
-   * @param txHashList
-   * @param logIndexList
-   * @returns
-   */
-  async diff_old(logs: ethers.Log[], txHashList: string[], logIndexList: number[][]): Promise<ethers.Log[]> {
-    const missing = (log: ethers.Log) => {
-      const txIndex = txHashList?.indexOf(log.transactionHash) || -1
-      // 1. tx missing, or 2. if passed in logIndexList, event idx missing
-      return txIndex === -1 || (logIndexList && !logIndexList[txIndex].includes(log.index))
-    }
-    // filter missing logs
-    const missingLogs = logs.filter((log) => { return missing(log) })//
-    return missingLogs
-  }
-
   async diff(logs: ethers.Log[], ignoreLogs: SimpleLog[]): Promise<ethers.Log[]> {
     const missing = (logToFind: ethers.Log) => {
       const logIndex = ignoreLogs.findIndex(
@@ -132,7 +122,7 @@ export class BaseCrossChecker {
       const logs = await timeoutWithRetry(() => {
         if (!this.provider || !this.provider.provider)
           throw new Error('provider not ready')
-        return this.provider.provider.getLogs(params)
+        return this.provider?.provider?.getLogs(params)
       }, 15 * 1000, 3)
       // get ignoreLogs keys
       const ignoreLogs = options.ignoreLogs