Merge pull request #21 from ora-io/dev

Dev

Author: murongg

.github/workflows/ci.yml

@@ -87,4 +87,4 @@ jobs:
         run: nr build
 
       - name: Test
-        run: nr test
+        run: npx vitest

package.json

@@ -1,6 +1,6 @@
 {
   "name": "ora-stack",
-  "version": "0.2.0",
+  "version": "0.2.1",
   "private": true,
   "packageManager": "[email protected]",
   "description": "",
@@ -47,7 +47,7 @@
     "publish": "nr prepublishOnly && esno scripts/publish.ts",
     "release": "esno scripts/release.ts",
     "start": "esno src/index.ts",
-    "test": "vitest",
+    "test": "DEBUG=DEBUG:ora-stack:* vitest",
     "typecheck": "tsc --noEmit"
   },
   "devDependencies": {

packages/orap/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@ora-io/orap",
   "type": "module",
-  "version": "0.2.0",
+  "version": "0.2.1",
   "packageManager": "[email protected]",
   "description": "",
   "author": "Norman (nom4dv3), MuRong",

packages/orap/src/debug.ts

@@ -0,0 +1,5 @@
+import { debug as debugFn } from '@ora-io/utils'
+
+export function debug(formatter: any, ...args: any[]) {
+  return debugFn('orap')(formatter, ...args)
+}

packages/orap/src/flow/orap.ts

@@ -3,6 +3,7 @@ import type { Fn } from '@ora-io/utils'
 import type { Interface, InterfaceAbi } from 'ethers'
 import { OrapVerse } from '../verse/orap'
 import type { EventSignalRegisterParams } from '../signal'
+import { debug } from '../debug'
 import { EventFlow } from './event'
 import type { Flow, HandleFn } from './interface'
 
@@ -35,6 +36,7 @@ export class OrapFlow implements Flow {
 
     const eventFlow = new EventFlow(this, params, handler)
     this.subflows.event.push(eventFlow)
+    debug('The event %s is registered, address: %s', params.eventName, params.address)
     return eventFlow
   }
 

packages/orap/src/signal/event.ts

@@ -124,8 +124,7 @@ export class EventSignal implements Signal {
       && !(provider instanceof ethers.WebSocketProvider)
     )
       throw new Error('crosscheckProvider must be an instance of RekuProviderManager or ethers.JsonRpcProvider or ethers.WebSocketProvider')
-
     this.crosschecker = new AutoCrossChecker(provider)
-    await this.crosschecker.start(this.crosscheckerOptions)
+    this.crosschecker.start(this.crosscheckerOptions)
   }
 }

packages/reku/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@ora-io/reku",
   "type": "module",
-  "version": "0.2.0",
+  "version": "0.2.1",
   "packageManager": "[email protected]",
   "description": "",
   "author": "Norman (nom4dv3), MuRong",

packages/reku/src/debug.ts

@@ -0,0 +1,5 @@
+import { debug as debugFn } from '@ora-io/utils'
+
+export function debug(formatter: any, ...args: any[]) {
+  return debugFn('reku')(formatter, ...args)
+}

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

@@ -1,7 +1,8 @@
 import type { ethers } from 'ethers'
-import { polling, sleep, timeoutWithRetry } from '@ora-io/utils'
+import { polling, timeoutWithRetry } from '@ora-io/utils'
 import { ETH_BLOCK_INTERVAL } from '../../constants'
 import type { Providers } from '../../types/w3'
+import { debug } from '../../debug'
 import { CrossCheckerCacheManager } from './cache/manager'
 import type { AutoCrossCheckParam, CrossCheckRangeParam, SimpleLog } from './interface'
 import { BaseCrossChecker } from './basechecker'
@@ -52,20 +53,21 @@ export class AutoCrossChecker extends BaseCrossChecker {
    * @param options
    */
   async start(options: AutoCrossCheckParam) {
+    debug('auto crosscheck start with options: %O', options)
     this.validate(options)
 
     this.cache = new CrossCheckerCacheManager(options?.store, { keyPrefix: options?.storeKeyPrefix, ttl: options?.storeTtl })
 
-    const latestblocknum = await timeoutWithRetry(() => this.provider.provider?.getBlockNumber(), 15 * 1000, 3)
+    let latestblocknum = await timeoutWithRetry(() => this.provider.provider?.getBlockNumber(), 15 * 1000, 3)
 
     // resume checkpoint priority: options.fromBlock > cache > latestblocknum + 1
-    const defaultInitCheckpoint = await this.cache.getCheckpoint() ?? (latestblocknum + 1)
+    const defaultInitCheckpoint = await this.cache.getCheckpoint() ?? (latestblocknum)
 
     const {
       fromBlock = defaultInitCheckpoint,
       batchBlocksCount = 10,
       pollingInterval = 3000,
-      blockInterval = ETH_BLOCK_INTERVAL,
+      // blockInterval = ETH_BLOCK_INTERVAL,
       delayBlockFromLatest = 1,
       toBlock,
     } = options
@@ -85,17 +87,20 @@ export class AutoCrossChecker extends BaseCrossChecker {
     }
 
     const waitNextCrosscheck = async (): Promise<boolean> => {
-      const latestblocknum = await timeoutWithRetry(() => this.provider.provider?.getBlockNumber(), 15 * 1000, 3)
+      latestblocknum = await timeoutWithRetry(() => this.provider.provider?.getBlockNumber(), 15 * 1000, 3)
       if (ccrOptions.toBlock + delayBlockFromLatest > latestblocknum) {
         // sleep until the toBlock
-        await sleep((ccrOptions.toBlock + delayBlockFromLatest - latestblocknum) * blockInterval)
+        // await sleep((ccrOptions.toBlock + delayBlockFromLatest - latestblocknum) * blockInterval)
         return false
       }
       return true
     }
 
     const waitOrUpdateToBlock = toBlock
-      ? () => { ccrOptions.toBlock = Math.min(ccrOptions.toBlock, toBlock); return true }
+      ? () => {
+          ccrOptions.toBlock = Math.min(ccrOptions.toBlock, toBlock)
+          return true
+        }
       : waitNextCrosscheck
 
     const updateCCROptions = async (ccrOptions: CrossCheckRangeParam) => {
@@ -113,13 +118,20 @@ export class AutoCrossChecker extends BaseCrossChecker {
       // never ends if options.toBlock is not provided
       : () => false
 
+    debug('crosscheck running')
+
     // TODO: replace polling with schedule cron
     await polling(async () => {
-      if (await waitOrUpdateToBlock()) {
+      const wait = await waitOrUpdateToBlock()
+      debug('polling interval: %d, wait: %s, from block: %d, to block: %d', pollingInterval, wait, ccrOptions.fromBlock, ccrOptions.toBlock)
+      if (wait) {
         await this.crossCheckRange(ccrOptions)
         // only update options after cc succ
         await updateCCROptions(ccrOptions)
       }
+      else {
+        debug('Because the latest block %d is too old, skip this cross check', latestblocknum)
+      }
       return endingCondition()
     }, pollingInterval)
   }

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

@@ -2,6 +2,7 @@ import type { ethers } from 'ethers'
 import { timeoutWithRetry } from '@ora-io/utils'
 import { ETH_BLOCK_COUNT_ONE_HOUR } from '../../constants'
 import type { Providers } from '../../types/w3'
+import { debug } from '../../debug'
 import type { CrossCheckFromParam, CrossCheckRangeParam, CrossCheckRetroParam, SimpleLog } from './interface'
 
 export class BaseCrossChecker {
@@ -77,6 +78,13 @@ 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
@@ -105,6 +113,7 @@ export class BaseCrossChecker {
   async _crossCheck(options: CrossCheckRangeParam) {
     // get period logs
     const { fromBlock, toBlock, address, topics } = options
+    debug('start crosscheck from %d to %d', fromBlock, toBlock)
     const params = {
       fromBlock,
       toBlock,
@@ -120,8 +129,10 @@ export class BaseCrossChecker {
       const missingLogs = ignoreLogs ? await this.diff(logs, ignoreLogs) : logs
 
       // callback on missing logs
-      for (const log of missingLogs)
+      for (const log of missingLogs) {
+        debug('missing tx hash: %s', log.transactionHash)
         await options.onMissingLog(log)
+      }
     }
   }
 }