feat(reku): storable checkpoint

refactor: optimize autocc, simplify polling, getBlock, ignoreLogs

Author: nom4dv3

packages/reku/README.md

@@ -26,7 +26,7 @@ i.e. It starts with 'realtime' mode by default.
 
 Options:
 - `store`?: the Store used to cache the <txhash, logindex> that already processed.
-- `storeKeyPrefix`?: set the prefix to all keys when set key-value to store (cache), e.g. key = prefix+'txHashList', prefix can be "project:app:" to form a "project:app:txHashList" redis key., defult: ''
+- `storeKeyPrefix`?: set the prefix to all keys when set key-value to store (cache), e.g. key = prefix+'txHashList', prefix can be "project:app:network:" to form a "project:app:network:txHashList" redis key., defult: ''
 - `storeTtl`?: the ttl for <txhash, logindex> record in store, defualt: no limit
 - `batchBlocksCount`?: how many blocks to get per `getLogs` check, in readtime mode it waits until the new block num >= `batchBlocksCount`.
 - `delayBlockFromLatest`?: mostly for realtime mode; each time cc wait until `latest height > toBlock + delayBlockFromLatest`, default: 1

packages/reku/event/crosschecker/autochecker.ts

@@ -8,7 +8,7 @@ import { BaseCrossChecker } from './basechecker'
 
 export class AutoCrossChecker extends BaseCrossChecker {
   cache: CrossCheckerCacheManager | undefined = undefined
-  checkpointBlockNumber = 0
+  checkpointBlockNumber: number | undefined
 
   constructor(
     provider: Providers,
@@ -41,6 +41,11 @@ export class AutoCrossChecker extends BaseCrossChecker {
     }
   }
 
+  async setCheckpoint(cp: number) {
+    this.checkpointBlockNumber = cp
+    await this.cache!.setCheckpoint(cp)
+  }
+
   // TODO: keep type hint of inside AutoCrossCheckParam
   /**
    * real time auto crosscheck, start from the lastest block
@@ -52,17 +57,20 @@ export class AutoCrossChecker extends BaseCrossChecker {
     this.cache = new CrossCheckerCacheManager(options?.store, { keyPrefix: options?.storeKeyPrefix, logger: this.logger, ttl: options?.storeTtl })
 
     const latestblocknum = await retryOnNull(async () => await this.provider.provider?.getBlockNumber())
+
+    // resume checkpoint priority: options.fromBlock > cache > latestblocknum + 1
+    const defaultInitCheckpoint = await this.cache.getCheckpoint() ?? latestblocknum + 1
+
     const {
-      fromBlock = latestblocknum + 1,
+      fromBlock = defaultInitCheckpoint,
       batchBlocksCount = 10,
       pollingInterval = 3000,
       blockInterval = ETH_BLOCK_INTERVAL,
       delayBlockFromLatest = 1,
-      toBlock, ignoreLogs,
+      toBlock,
     } = options
 
-    // init checkpoint block num
-    this.checkpointBlockNumber = fromBlock
+    await this.setCheckpoint(fromBlock)
 
     const ccrOptions: CrossCheckRangeParam = {
       ...options,
@@ -72,26 +80,12 @@ export class AutoCrossChecker extends BaseCrossChecker {
           await options.onMissingLog(log)
           this.cache!.addLogs([log])
         },
-      fromBlock: -1, // placeholder
-      toBlock: -1, // placeholder
-    }
-
-    if (ignoreLogs)
-      await this.cache.addLogs(ignoreLogs)
-
-    // initialize the ignore logs from redis
-    ccrOptions.ignoreLogs = await this.cache.getLogs()
-
-    const updateCCROptions = async (ccrOptions: any) => {
-      // iterate block range
-      ccrOptions.fromBlock = this.checkpointBlockNumber
-      // batchBlocksCount should > 0
-      ccrOptions.toBlock = ccrOptions.fromBlock + batchBlocksCount - 1
+      fromBlock,
+      toBlock: fromBlock + batchBlocksCount - 1,
     }
 
     const waitNextCrosscheck = async (): Promise<boolean> => {
-      // TODO: use blockNumber for performance
-      const latestblocknum = (await retryOnNull(async () => await this.provider.provider?.getBlock('latest'))).number
+      const latestblocknum = await retryOnNull(async () => await this.provider.provider?.getBlockNumber())
       this.logger.info('[*] ccrOptions: fromBlock', ccrOptions.fromBlock, ', toBlock', ccrOptions.toBlock, ', latestblocknum', latestblocknum)
       if (ccrOptions.toBlock + delayBlockFromLatest > latestblocknum) {
         // sleep until the toBlock
@@ -106,22 +100,28 @@ export class AutoCrossChecker extends BaseCrossChecker {
       ? () => { ccrOptions.toBlock = Math.min(ccrOptions.toBlock, toBlock); return true }
       : waitNextCrosscheck
 
+    const updateCCROptions = async (ccrOptions: any) => {
+      // only set after cc succ
+      await this.setCheckpoint(ccrOptions.toBlock + 1)
+      // iterate block range
+      ccrOptions.fromBlock = this.checkpointBlockNumber
+      // batchBlocksCount should > 0
+      ccrOptions.toBlock = ccrOptions.fromBlock + batchBlocksCount - 1
+    }
+
     const endingCondition = toBlock
       // ends on up to options.toBlock
-      ? () => this.checkpointBlockNumber > toBlock
+      ? () => this.checkpointBlockNumber! > toBlock
       // never ends if options.toBlock is not provided
       : () => false
 
     // TODO: replace polling with schedule cron
     await polling(async () => {
-      await updateCCROptions(ccrOptions)
-
       if (await waitOrUpdateToBlock()) {
         await this.crossCheckRange(ccrOptions)
-        // only set after cc succ
-        this.checkpointBlockNumber = ccrOptions.toBlock + 1
+        // only update options after cc succ
+        await updateCCROptions(ccrOptions)
       }
-
       return endingCondition()
     }, pollingInterval)
   }
@@ -130,7 +130,7 @@ export class AutoCrossChecker extends BaseCrossChecker {
     const newlogs = await super.diff(logs, ignoreLogs)
     const res: ethers.Log[] = []
     for (const log of newlogs) {
-      const key = this.cache!.encodeKey(log)
+      const key = this.cache!.encodeLogKey(log)
       const logExist = await this.cache!.has(key)
       if (!logExist)
         res.push(log)

packages/reku/event/crosschecker/cache/manager.ts

@@ -26,39 +26,13 @@ export class CrossCheckerCacheManager extends SimpleStoreManager {
     this.logger = options?.logger ?? logger
   }
 
-  /**
-   * @dev can add this.style: string = 'redis' when supporting other store type
-   * @param log
-   * @returns
-   */
-  encodeKey(log: SimpleLog): string {
-    const key = log.index && !this.noLogIndex ? `${this.storeKeyPrefix + log.transactionHash}:${log.index}` : this.storeKeyPrefix + log.transactionHash
-    logger.debug('cc-cm-encodeKey', key)
-    return key
-  }
-
-  decodeKey(key: string): SimpleLog {
-    logger.debug('cc-cm-decodeKey', key)
-    if (!key.startsWith(this.storeKeyPrefix))
-      throw new Error(`The prefix ${this.storeKeyPrefix} is not a prefix of ${key}`)
-
-    const _noprefix_key = key.slice(this.storeKeyPrefix.length)
-
-    const parts = _noprefix_key.split(':')
-    if (parts.length > 2)
-      throw new Error(`wrong key format when decoding, expecting ${this.storeKeyPrefix}+xx:xx, getting ${key}`)
-
-    const log = { transactionHash: parts[0], index: parts.length === 2 ? parseInt(parts[1]) : undefined }
-    return log
-  }
-
   /**
    * add log into store record that can indicate a log
    * @param log
    */
   async addLog(log: SimpleLog) {
     this.logger.debug('cache manager - addLog:', log.transactionHash, log.index)
-    const key = this.encodeKey(log)
+    const key = this.encodeLogKey(log)
     await this.set(key, true, this.ttl)
   }
 
@@ -83,8 +57,46 @@ export class CrossCheckerCacheManager extends SimpleStoreManager {
     this.logger.debug('cachemanager-getLogs:', keys)
     const logs: SimpleLog[] = []
     for (const key of keys)
-      logs.push(this.decodeKey(key))
+      logs.push(this.decodeLogKey(key))
 
     return logs
   }
+
+  /**
+   * @dev can add this.style: string = 'redis' when supporting other store type
+   * @param log
+   * @returns
+   */
+  encodeLogKey(log: SimpleLog): string {
+    const key = log.index && !this.noLogIndex ? `${this.storeKeyPrefix + log.transactionHash}:${log.index}` : this.storeKeyPrefix + log.transactionHash
+    logger.debug('cc-cm-encodeKey', key)
+    return key
+  }
+
+  decodeLogKey(key: string): SimpleLog {
+    logger.debug('cc-cm-decodeKey', key)
+    if (!key.startsWith(this.storeKeyPrefix))
+      throw new Error(`The prefix ${this.storeKeyPrefix} is not a prefix of ${key}`)
+
+    const _noprefix_key = key.slice(this.storeKeyPrefix.length)
+
+    const parts = _noprefix_key.split(':')
+    if (parts.length > 2)
+      throw new Error(`wrong key format when decoding, expecting ${this.storeKeyPrefix}+xx:xx, getting ${key}`)
+
+    const log = { transactionHash: parts[0], index: parts.length === 2 ? parseInt(parts[1]) : undefined }
+    return log
+  }
+
+  /**
+   * ttl: no limit
+   * @param checkpoint
+   */
+  async setCheckpoint(checkpoint: number) {
+    await this.set(`${this.storeKeyPrefix}checkpoint`, checkpoint)
+  }
+
+  async getCheckpoint(): Promise<number | undefined> {
+    return await this.get(`${this.storeKeyPrefix}checkpoint`)
+  }
 }

packages/reku/tests/.env.example

@@ -0,0 +1,12 @@
+############### General config ###############
+
+# general - provider url
+MAINNET_WSS=""
+MAINNET_HTTP=""
+SEPOLIA_WSS=""
+SEPOLIA_HTTP=""
+
+# general - redis
+REDIS_HOST="localhost"
+REDIS_PORT=6379
+REDIS_TTL=7200000 # 2 hour in ms

packages/reku/tests/crosscheck.test.ts

@@ -2,19 +2,23 @@
 import { ethers } from 'ethers'
 import { describe, expect, it, vi } from 'vitest'
 import { Logger, setLogger } from '@ora-io/utils'
+import dotenv from 'dotenv'
 import { AutoCrossChecker } from '../event/crosschecker/autochecker'
 import { BaseCrossChecker } from '../event/crosschecker/basechecker'
 import type { BaseCrossCheckParam } from '../event/crosschecker/interface'
+dotenv.config({ path: './packages/reku/tests/.env' })
 
-const ETHEREUM_RPC_URL = 'https://rpc.ankr.com/eth'
 const CONTRACT_ADDRESS = '0xdAC17F958D2ee523a2206206994597C13D831ec7'
 
+const chain = 'mainnet'
+
+const httpProvider = new ethers.JsonRpcProvider(
+  process.env[`${chain.toUpperCase()}_HTTP`]!,
+)
+
 setLogger(new Logger('debug', 'reku-tests'))
 
 export async function crossCheckerTest() {
-  const rpcUrl = ETHEREUM_RPC_URL
-  const provider = new ethers.JsonRpcProvider(rpcUrl)
-
   const onMissingLog = async (log: any) => {
     console.log('onMissingLog', log)
   }
@@ -33,7 +37,7 @@ export async function crossCheckerTest() {
     '0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef',
   ]
 
-  const cc = new BaseCrossChecker(provider)
+  const cc = new BaseCrossChecker(httpProvider)
   await cc.crossCheckRange({
     onMissingLog,
     ignoreLogs,
@@ -44,7 +48,7 @@ export async function crossCheckerTest() {
   })
 
   // choose catchup mode / realtime mode / catchup then realtime mode by fromBlock & toBlock
-  const acc = new AutoCrossChecker(provider)
+  const acc = new AutoCrossChecker(httpProvider)
   await acc.start({
     onMissingLog,
     ignoreLogs,