Merge pull request #156 from nervosnetwork/cache-raw-l2tx

feat: add cache for executeRawL2Tx

README.md

@@ -29,6 +29,7 @@ REDIS_URL=redis://user:password@localhost:6379 <redis url, optional, default to
 PG_POOL_MAX=<pg pool max count, optional, default to 20>
 GAS_PRICE_CACHE_SECONDS=<seconds, optional, default to 0, and 0 means no cache>
 EXTRA_ESTIMATE_GAS=<eth_estimateGas will add this number to result, optional, default to 0>
+ENABLE_CACHE_POLY_EXECUTE_RAW_L2Tx=<optional, enable poly_executeRawL2Transaction cache, default to false>
 EOF
 
 $ yarn

packages/api-server/src/base/env-config.ts

@@ -25,6 +25,9 @@ export const envConfig = {
   sentryDns: getOptional("SENTRY_DNS"),
   sentryEnvironment: getOptional("SENTRY_ENVIRONMENT"),
   godwokenReadonlyJsonRpc: getOptional("GODWOKEN_READONLY_JSON_RPC"),
+  enableCachePolyExecuteRawL2Tx: getOptional(
+    "ENABLE_CACHE_POLY_EXECUTE_RAW_L2Tx"
+  ),
 };
 
 function getRequired(name: string): string {

packages/api-server/src/cache/data.ts

@@ -0,0 +1,230 @@
+import { createClient } from "redis";
+import { envConfig } from "../base/env-config";
+import crypto from "crypto";
+import { asyncSleep } from "../util";
+
+// init publisher redis client
+export const pubClient = createClient({
+  url: envConfig.redisUrl,
+});
+pubClient.connect();
+pubClient.on("error", (err) => console.log("Redis Client Error", err));
+
+// init subscriber redis client
+export const subClient = createClient({
+  url: envConfig.redisUrl,
+});
+subClient.connect();
+subClient.on("error", (err) => console.log("Redis Client Error", err));
+
+export const SUB_TIME_OUT_MS = 5 * 1000; // 5s;
+export const LOCK_KEY_EXPIRED_TIME_OUT_MS = 60 * 1000; // 60s, the max tolerate timeout for execute call
+export const DATA_KEY_EXPIRED_TIME_OUT_MS = 5 * 60 * 1000; // 5 minutes
+export const POLL_INTERVAL_MS = 50; // 50ms
+export const POLL_TIME_OUT_MS = 2 * 60 * 1000; // 2 minutes
+
+export const DEFAULT_PREFIX_NAME = "defaultDataCache";
+export const DEFAULT_IS_ENABLE_LOCK = true;
+
+export interface DataCacheConstructor {
+  rawDataKey: string;
+  executeCallResult: ExecuteCallResult;
+  prefixName?: string;
+  isLockEnable?: boolean;
+  lock?: Partial<RedisLock>;
+  dataKeyExpiredTimeOutMs?: number;
+}
+
+export type ExecuteCallResult = () => Promise<string>;
+
+export interface RedisLock {
+  key: LockKey;
+  subscribe: RedSubscribe;
+  pollIntervalMs: number;
+  pollTimeOutMs: number;
+}
+
+export interface LockKey {
+  name: string;
+  expiredTimeMs: number;
+}
+
+export interface RedSubscribe {
+  channel: string;
+  timeOutMs: number;
+}
+
+export class RedisDataCache {
+  public prefixName: string;
+  public rawDataKey: string; // unique part of dataKey
+  public dataKey: string; // real dataKey saved on redis combined from rawDataKey with prefix name and so on.
+  public lock: RedisLock | undefined;
+  public dataKeyExpiredTimeOut: number;
+  public executeCallResult: ExecuteCallResult;
+
+  constructor(args: DataCacheConstructor) {
+    this.prefixName = args.prefixName || DEFAULT_PREFIX_NAME;
+    this.rawDataKey = args.rawDataKey;
+    this.dataKey = `${this.prefixName}:key:${this.rawDataKey}`;
+    this.executeCallResult = args.executeCallResult;
+    this.dataKeyExpiredTimeOut =
+      args.dataKeyExpiredTimeOutMs || DATA_KEY_EXPIRED_TIME_OUT_MS;
+
+    const isLockEnable = args.isLockEnable ?? DEFAULT_IS_ENABLE_LOCK; // default is true;
+    if (isLockEnable) {
+      this.lock = {
+        key: {
+          name:
+            args.lock?.key?.name ||
+            `${this.prefixName}:lock:${this.rawDataKey}`,
+          expiredTimeMs:
+            args.lock?.key?.expiredTimeMs || LOCK_KEY_EXPIRED_TIME_OUT_MS,
+        },
+        subscribe: {
+          channel:
+            args.lock?.subscribe?.channel ||
+            `${this.prefixName}:channel:${this.rawDataKey}`,
+          timeOutMs: args.lock?.subscribe?.timeOutMs || SUB_TIME_OUT_MS,
+        },
+        pollIntervalMs: args.lock?.pollIntervalMs || POLL_INTERVAL_MS,
+        pollTimeOutMs: args.lock?.pollTimeOutMs || POLL_TIME_OUT_MS,
+      };
+    }
+  }
+
+  async get() {
+    const dataKey = this.dataKey;
+    const value = await pubClient.get(dataKey);
+    if (value !== null) {
+      console.debug(
+        `[${this.constructor.name}]: hit cache via Redis.Get, key: ${dataKey}`
+      );
+      return value;
+    }
+
+    const setDataKeyOptions = { PX: this.dataKeyExpiredTimeOut };
+
+    if (this.lock == undefined) {
+      const result = await this.executeCallResult();
+      // set data cache
+      await pubClient.set(dataKey, result, setDataKeyOptions);
+      return result;
+    }
+
+    // use redis-lock for data cache
+    const t1 = new Date();
+    const lockValue = getLockUniqueValue();
+    const setLockKeyOptions = {
+      NX: true,
+      PX: this.lock.key.expiredTimeMs,
+    };
+
+    const releaseLock = async (lockValue: string) => {
+      if (!this.lock) throw new Error("enable lock first!");
+
+      const value = await pubClient.get(this.lock.key.name);
+      if (value === lockValue) {
+        // only lock owner can delete the lock
+        const delNumber = await pubClient.del(this.lock.key.name);
+        console.debug(
+          `[${this.constructor.name}]: delete key ${this.lock.key.name}, result: ${delNumber}`
+        );
+      }
+    };
+
+    while (true) {
+      const value = await pubClient.get(dataKey);
+      if (value !== null) {
+        console.debug(
+          `[${this.constructor.name}]: hit cache via Redis.Get, key: ${dataKey}`
+        );
+        return value;
+      }
+
+      const isLockAcquired = await pubClient.set(
+        this.lock.key.name,
+        lockValue,
+        setLockKeyOptions
+      );
+
+      if (isLockAcquired) {
+        try {
+          const result = await this.executeCallResult();
+          // set data cache
+          await pubClient.set(dataKey, result, setDataKeyOptions);
+          // publish the result to channel
+          const totalSubs = await pubClient.publish(
+            this.lock.subscribe.channel,
+            result
+          );
+          console.debug(
+            `[${this.constructor.name}]: publish message ${result} on channel ${this.lock.subscribe.channel}, total subscribers: ${totalSubs}`
+          );
+          await releaseLock(lockValue);
+          return result;
+        } catch (error) {
+          console.debug(error);
+          await releaseLock(lockValue);
+        }
+      }
+
+      // if lock is not acquired
+      try {
+        const msg = await this.subscribe();
+        console.debug(
+          `[${this.constructor.name}]: hit cache via Redis.Subscribe, key: ${dataKey}`
+        );
+        return msg;
+      } catch (error: any) {
+        if (
+          !JSON.stringify(error).includes(
+            "subscribe channel for message time out"
+          )
+        ) {
+          console.debug(`[${this.constructor.name}]: subscribe err:`, error);
+        }
+      }
+
+      // check if poll time out
+      const t2 = new Date();
+      const diff = t1.getTime() - t2.getTime();
+      if (diff > this.lock.pollTimeOutMs) {
+        throw new Error(
+          `poll data value from cache layer time out ${this.lock.pollTimeOutMs}`
+        );
+      }
+
+      await asyncSleep(this.lock.pollIntervalMs);
+    }
+  }
+
+  async subscribe() {
+    if (this.lock == undefined) {
+      throw new Error(`enable redis lock first!`);
+    }
+
+    const p = new Promise((resolve) => {
+      subClient.subscribe(
+        this.lock!.subscribe.channel,
+        async (message: string) => {
+          await subClient.unsubscribe(this.lock!.subscribe.channel);
+          return resolve(message);
+        }
+      );
+    });
+
+    const t = new Promise((_resolve, reject) => {
+      setTimeout(() => {
+        return reject(
+          `subscribe channel for message time out ${this.lock?.subscribe.timeOutMs}`
+        );
+      }, this.lock?.subscribe.timeOutMs);
+    });
+
+    return (await Promise.race([p, t])) as Promise<string>;
+  }
+}
+
+export function getLockUniqueValue() {
+  return "0x" + crypto.randomBytes(20).toString("hex");
+}

packages/api-server/src/methods/modules/gw.ts

@@ -411,6 +411,15 @@ export class Gw {
       parseGwRpcError(error);
     }
   }
+
+  async get_mem_pool_state_root(args: any[]) {
+    try {
+      const result = await this.readonlyRpc.gw_get_mem_pool_state_root(...args);
+      return result;
+    } catch (error) {
+      parseGwRpcError(error);
+    }
+  }
 }
 
 function formatHexNumber(

packages/api-server/src/methods/modules/poly.ts

@@ -18,8 +18,10 @@ import {
   L2TransactionWithAddressMapping,
   RawL2TransactionWithAddressMapping,
 } from "@polyjuice-provider/godwoken/lib/addressTypes";
-import { GodwokenClient } from "@godwoken-web3/godwoken";
+import { GodwokenClient, RunResult } from "@godwoken-web3/godwoken";
 import { parseGwRpcError } from "../gw-error";
+import { keccakFromHexString } from "ethereumjs-util";
+import { DataCacheConstructor, RedisDataCache } from "../../cache/data";
 
 export class Poly {
   private query: Query;
@@ -87,15 +89,46 @@ export class Poly {
 
   async executeRawL2Transaction(args: any[]) {
     try {
-      const data = args[0];
-      const txWithAddressMapping: RawL2TransactionWithAddressMapping =
-        deserializeRawL2TransactionWithAddressMapping(data);
-      const rawL2Tx = txWithAddressMapping.raw_tx;
+      const serializeRawL2Tx = args[0];
+
+      const executeCallResult = async () => {
+        const txWithAddressMapping: RawL2TransactionWithAddressMapping =
+          deserializeRawL2TransactionWithAddressMapping(serializeRawL2Tx);
+        const rawL2Tx = txWithAddressMapping.raw_tx;
+        const jsonResult = await this.rpc.executeRawL2Transaction(rawL2Tx);
+        // if result is fine, then tx is legal, we can start thinking to store the address mapping
+        await saveAddressMapping(this.query, this.rpc, txWithAddressMapping);
+        const stringResult = JSON.stringify(jsonResult);
+        return stringResult;
+      };
 
-      const result = await this.rpc.executeRawL2Transaction(rawL2Tx);
-      // if result is fine, then tx is legal, we can start thinking to store the address mapping
-      await saveAddressMapping(this.query, this.rpc, txWithAddressMapping);
-      return result;
+      // using cache
+      if (envConfig.enableCachePolyExecuteRawL2Tx === "true") {
+        // calculate raw data cache key
+        const [tipBlockHash, memPollStateRoot] = await Promise.all([
+          this.rpc.getTipBlockHash(),
+          this.rpc.getMemPoolStateRoot(),
+        ]);
+        const rawDataKey = getPolyExecRawL2TxCacheKey(
+          serializeRawL2Tx,
+          tipBlockHash,
+          memPollStateRoot
+        );
+
+        const prefixName = `${this.constructor.name}:${this.executeRawL2Transaction.name}`;
+        const constructArgs: DataCacheConstructor = {
+          prefixName,
+          rawDataKey,
+          executeCallResult,
+        };
+        const dataCache = new RedisDataCache(constructArgs);
+        const stringResult = await dataCache.get();
+        return JSON.parse(stringResult) as RunResult;
+      }
+
+      // not using cache
+      const stringResult = await executeCallResult();
+      return JSON.parse(stringResult) as RunResult;
     } catch (error) {
       parseGwRpcError(error);
     }
@@ -405,3 +438,20 @@ function containsAddressType(abiItem: AbiItem) {
 
   return true;
 }
+
+// key: tipBlockHash first 8 bytes + memPollStateRoot first 8 bytes + dataHash first 8 bytes
+function getPolyExecRawL2TxCacheKey(
+  serializeRawL2Transaction: HexString,
+  tipBlockHash: HexString,
+  memPoolStateRoot: HexString
+) {
+  const hash =
+    "0x" + keccakFromHexString(serializeRawL2Transaction).toString("hex");
+  const id = `0x${tipBlockHash.slice(2, 18)}${memPoolStateRoot.slice(
+    2,
+    18
+  )}${hash.slice(2, 18)}`;
+  return id;
+  // const key = `${POLY_RPC_KEY}:executeRawL2Transaction:${id}`;
+  // return key;
+}

packages/api-server/src/util.ts

@@ -73,3 +73,7 @@ export function validateHexString(hex: string): boolean {
 export function validateHexNumber(hex: string): boolean {
   return /^0x(0|[0-9a-fA-F]+)$/.test(hex);
 }
+
+export function asyncSleep(ms = 0) {
+  return new Promise((r) => setTimeout(() => r("ok"), ms));
+}

packages/godwoken/src/client.ts

@@ -141,6 +141,14 @@ export class GodwokenClient {
     return await this.rpcCall("get_transaction_receipt", hash);
   }
 
+  public async getTipBlockHash(): Promise<HexString> {
+    return await this.rpcCall("get_tip_block_hash");
+  }
+
+  public async getMemPoolStateRoot(): Promise<HexString> {
+    return await this.rpcCall("get_mem_pool_state_root");
+  }
+
   private async rpcCall(methodName: string, ...args: any[]): Promise<any> {
     const name = "gw_" + methodName;
     try {