feat(reku): enhance BaseCrossChecker to support multiple addresses and add retry options for log retrieval

Author: murongg

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

@@ -1,4 +1,5 @@
 import type { ethers } from 'ethers'
+import type { ContractAddress } 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'
@@ -112,18 +113,46 @@ export class BaseCrossChecker {
     // get period logs
     const { fromBlock, toBlock, address, topics } = options
     debug('start crosscheck from %d to %d', fromBlock, toBlock)
-    const params = {
-      fromBlock,
-      toBlock,
-      ...(address && { address }),
-      ...(topics && { topics }),
-    }
+
     if (this.provider.provider) {
-      const logs = await timeoutWithRetry(() => {
-        if (!this.provider || !this.provider.provider)
-          throw new Error('provider not ready')
-        return this.provider?.provider?.getLogs(params)
-      }, 15 * 1000, 3)
+      const addresses: ContractAddress[][] = []
+      if (Array.isArray(address)) {
+        if (options.addressGroupLimit) {
+          for (let i = 0; i < address.length; i += options.addressGroupLimit)
+            addresses.push(address.slice(i, i + options.addressGroupLimit))
+        }
+        else {
+          addresses.push(address)
+        }
+      }
+      else {
+        addresses.push([address])
+      }
+
+      const requests = addresses.map((address) => {
+        const params = {
+          fromBlock,
+          toBlock,
+          ...(address && { address }),
+          ...(topics && { topics }),
+        }
+
+        const fn = async () => {
+          if (!this.provider || !this.provider.provider)
+            throw new Error('provider not ready')
+          return this.provider?.provider?.getLogs(params)
+        }
+        if (options.retryOptions)
+          return timeoutWithRetry(fn, options.retryOptions.timeout || 15 * 1000, options.retryOptions.retries || 3)
+
+        else
+          return fn()
+      })
+
+      const logs = (await Promise.all(requests)).reduce((acc, curr) => {
+        return acc.concat(curr)
+      }, [])
+
       // get ignoreLogs keys
       const ignoreLogs = options.ignoreLogs
 

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

@@ -6,7 +6,7 @@ export type FnOnMissingLog = (log: ethers.Log) => Awaitable<void>
 export interface SimpleLog { transactionHash: string; index?: number }
 
 export interface LogFilterParam {
-  address: ContractAddress
+  address: ContractAddress | ContractAddress[]
   topics: string[]
   fromBlock?: number
   toBlock?: number
@@ -15,6 +15,11 @@ export interface LogFilterParam {
 export interface BaseCrossCheckParam extends LogFilterParam {
   onMissingLog: FnOnMissingLog
   ignoreLogs?: SimpleLog[]
+  addressGroupLimit?: number // how many addresses to get logs at most
+  retryOptions?: {
+    timeout?: Milliseconds
+    retries?: number
+  }
 }
 
 export interface CrossCheckFromParam extends BaseCrossCheckParam {

packages/reku/tests/basechecker.test.ts

@@ -312,7 +312,7 @@ describe('BaseCrossChecker', () => {
       expect(mockProvider.provider.getLogs).toHaveBeenCalledWith({
         fromBlock: 1000,
         toBlock: 2000,
-        address: '0xcontract1',
+        address: ['0xcontract1'],
         topics: ['0xtopic1'],
       })
       expect(onMissingLogSpy).toHaveBeenCalledTimes(2)
@@ -324,7 +324,7 @@ describe('BaseCrossChecker', () => {
       const options: CrossCheckRangeParam = {
         fromBlock: 1000,
         toBlock: 2000,
-        address: '0xcontract1',
+        address: ['0xcontract1'],
         topics: ['0xtopic1'],
         onMissingLog: onMissingLogSpy,
       }
@@ -348,7 +348,7 @@ describe('BaseCrossChecker', () => {
       expect(mockProvider.provider.getLogs).toHaveBeenCalledWith({
         fromBlock: 1000,
         toBlock: 2000,
-        address: '0xcontract1',
+        address: ['0xcontract1'],
         topics: ['0xtopic1'],
       })
     })
@@ -368,5 +368,75 @@ describe('BaseCrossChecker', () => {
 
       expect(onMissingLogSpy).toHaveBeenCalledTimes(3)
     })
+
+    it('should handle addressGroupLimit', async () => {
+      mockProvider.provider.getLogs.mockResolvedValue(mockLogs)
+
+      const options: CrossCheckRangeParam = {
+        fromBlock: 1000,
+        toBlock: 2000,
+        address: ['0xcontract1', '0xcontract2'],
+        addressGroupLimit: 1,
+        onMissingLog: onMissingLogSpy,
+        topics: [],
+      }
+
+      await (baseChecker as any)._crossCheck(options)
+
+      expect(mockProvider.provider.getLogs).toHaveBeenCalledTimes(2)
+    })
+
+    it('should handle addressGroupLimit with single address', async () => {
+      mockProvider.provider.getLogs.mockResolvedValue(mockLogs)
+
+      const options: CrossCheckRangeParam = {
+        fromBlock: 1000,
+        toBlock: 2000,
+        address: '0xcontract1',
+        addressGroupLimit: 1,
+        onMissingLog: onMissingLogSpy,
+        topics: [],
+      }
+
+      await (baseChecker as any)._crossCheck(options)
+
+      expect(mockProvider.provider.getLogs).toHaveBeenCalledTimes(1)
+
+      expect(mockProvider.provider.getLogs).toHaveBeenCalledWith({
+        fromBlock: 1000,
+        toBlock: 2000,
+        address: ['0xcontract1'],
+        topics: [],
+      })
+    })
+
+    it.only('should handle retryOptions', async () => {
+      mockProvider.provider.getLogs.mockRejectedValue(Error('provider not ready'))
+
+      const options: CrossCheckRangeParam = {
+        fromBlock: 1000,
+        toBlock: 2000,
+        address: '0xcontract1',
+        topics: ['0xtopic1'],
+        onMissingLog: onMissingLogSpy,
+        retryOptions: {
+          timeout: 1000,
+          retries: 1,
+        },
+      }
+
+      const defaultCallTimes = 1
+
+      try {
+        await (baseChecker as any)._crossCheck(options)
+      }
+      catch (error) {
+        expect(error).toBeInstanceOf(Error)
+        expect((error as Error).message).toBe('provider not ready')
+      }
+
+      expect(mockProvider.provider.getLogs).toHaveBeenCalledTimes(defaultCallTimes + options.retryOptions!.retries!)
+      expect(onMissingLogSpy).toHaveBeenCalledTimes(0)
+    })
   })
 })