feat(orap): add ttl for tasks

fix(reku): sync checkpoint ttl with log records

Author: nom4dv3

packages/orap/mock/demo/app.ts

@@ -29,6 +29,7 @@ export function startDemo(options: ListenOptions, storeConfig?: any) {
     .crosscheck({
       store,
       storeKeyPrefix: 'ora-stack:orap:demo:cc:',
+      storeTtl: config.CROSSCHECKER_CACHE_TTL,
       pollingInterval: 3000,
       batchBlocksCount: 1,
       blockInterval: 12000,

packages/orap/mock/demo/config.ts

@@ -3,6 +3,9 @@ import { Logger } from '@ora-io/utils'
 export const config = {
   MAINNET_USDT_ADDR: '0xdAC17F958D2ee523a2206206994597C13D831ec7',
   TRANSFER_EVENT_NAME: 'Transfer',
+  TASK_TTL: 60000,
+  TASK_DONE_TTL: 60000,
+  CROSSCHECKER_CACHE_TTL: 60000,
 }
 
 const logLevel = 'debug'

packages/orap/mock/demo/taskTransfer.ts

@@ -1,10 +1,12 @@
 import type { Constructor } from '@ora-io/utils'
 import { TaskStorable } from '../../index'
-import { logger } from './config'
+import { config, logger } from './config'
 
 export class TransferTask extends TaskStorable {
   static readonly taskPrefix: string = 'ora-stack:orap:demo:TransferTask:'
   static readonly taskPrefixDone: string = 'ora-stack:orap:demo:Done-TransferTask:'
+  static readonly taskTtl: number | undefined = config.TASK_TTL
+  static readonly taskTtlDone: number | undefined = config.TASK_DONE_TTL
 
   constructor(
     public id?: string,

packages/orap/task/storable.ts

@@ -6,6 +6,8 @@ export abstract class TaskStorable extends TaskBase {
   // overwrite these for store key customize
   static readonly taskPrefix: string = 'Task:'
   static readonly taskPrefixDone: string = 'Done-Task:'
+  static readonly taskTtl: number | undefined = undefined
+  static readonly taskTtlDone: number | undefined = undefined
 
   getTaskPrefix(): string {
     return (this.constructor as typeof TaskStorable).taskPrefix
@@ -15,10 +17,18 @@ export abstract class TaskStorable extends TaskBase {
     return (this.constructor as typeof TaskStorable).taskPrefixDone
   }
 
+  getTaskTtl(): number | undefined {
+    return (this.constructor as typeof TaskStorable).taskTtl
+  }
+
+  getTaskTtlDone(): number | undefined {
+    return (this.constructor as typeof TaskStorable).taskTtlDone
+  }
+
   static async _load<T extends TaskStorable>(this: Constructor<T>, sm: StoreManager): Promise<T> {
     const instance = new this()
     // get all task keys
-    const keys = await sm.keys(instance.getTaskPrefix() + '*', true)
+    const keys = await sm.keys(`${instance.getTaskPrefix()}*`, true)
     // get the first task (del when finish)
     const serializedTask: string = (await sm.get(keys[0]))! // never undefined ensured by keys isWait=true
 
@@ -30,15 +40,15 @@ export abstract class TaskStorable extends TaskBase {
   }
 
   async save(sm: StoreManager) {
-    await sm.set(this.getTaskPrefix() + this.toKey(), this.toString())
+    await sm.set(this.getTaskPrefix() + this.toKey(), this.toString(), this.getTaskTtl())
   }
 
   async remove(sm: StoreManager) {
     await sm.del(this.getTaskPrefix() + this.toKey())
   }
 
   async done(sm: StoreManager) {
-    await sm.set(this.getTaskPrefixDone() + this.toKey(), this.toString())
+    await sm.set(this.getTaskPrefixDone() + this.toKey(), this.toString(), this.getTaskTtlDone())
     await this.remove(sm)
   }
 }

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

@@ -89,11 +89,11 @@ export class CrossCheckerCacheManager extends SimpleStoreManager {
   }
 
   /**
-   * ttl: no limit
+   * ttl: should be consist with log records, otherwise will do redundant work
    * @param checkpoint
    */
   async setCheckpoint(checkpoint: number) {
-    await this.set(`${this.storeKeyPrefix}checkpoint`, checkpoint)
+    await this.set(`${this.storeKeyPrefix}checkpoint`, checkpoint, this.ttl)
   }
 
   async getCheckpoint(): Promise<number | undefined> {