add global concurrency control and HTTP timeouts for Bitcoin and CKB

[fghdotio]

No description provided.

[gemini-code-assist]

Code Review

This pull request introduces global concurrency limits and timeout controls for both Bitcoin and CKB JSON-RPC traffic. It adds a new CkbRpcCaller service to coordinate CKB RPC and Indexer requests, auto-splitting oversized batches and wrapping indexer methods in a proxy to enforce concurrency limits. Additionally, it updates Bitcoin and SPV clients to enforce HTTP timeouts and concurrency limits. The review feedback highlights a critical issue where Lumos's CKBCellCollector bypasses the proxy limiter, along with potential slot leakage due to missing indexer timeouts. It also suggests refactoring CkbRpcCaller to accept configuration parameters via its constructor to maintain proper dependency injection.

[gemini-code-assist]

There are two critical issues with the current wrapIndexer implementation:\n\n1. Indexer Collector Bypass: Lumos's CKBCellCollector retrieves the raw terminableCellFetcher from the indexer instance (e.g., this.terminableCellFetcher = indexer.terminableCellFetcher). Since 'terminableCellFetcher' is not in limitedMethods, the proxy returns the raw, unwrapped object. Consequently, all internal getCells pagination calls made by the collector bypass the proxy and the concurrency limiter entirely.\n2. Hanging Request Slot Leakage: Lumos's Indexer does not support a native timeout option. If an indexer request hangs indefinitely, it will permanently occupy a slot in p-limit. If this happens CKB_RPC_MAX_CONCURRENCY times, the entire application's CKB RPC traffic will freeze.\n\nWe can solve both issues by:\n- Recursively wrapping the terminableCellFetcher and terminableTransactionFetcher properties in the proxy.\n- Implementing a withTimeout helper to wrap all proxied indexer calls, ensuring slots are released if a request hangs.

  private wrapIndexer(indexer: Indexer): Indexer {\n    const limitedMethods = new Set(['tip', 'getCells', 'getTransactions']);\n    const withTimeout = <T>(promise: Promise<T>, ms: number, errorMsg: string): Promise<T> => {\n      let timer: NodeJS.Timeout;\n      const timeout = new Promise<never>((_, reject) => {\n        timer = setTimeout(() => reject(new Error(errorMsg)), ms);\n      });\n      return Promise.race([promise, timeout]).finally(() => clearTimeout(timer));\n    };\n    const wrap = (obj: any): any => {\n      return new Proxy(obj, {\n        get: (target, prop, receiver) => {\n          const value = Reflect.get(target, prop, receiver);\n          if (typeof value === 'function' && typeof prop === 'string' && limitedMethods.has(prop)) {\n            return (...args: unknown[]) =>\n              this.limit(() =>\n                withTimeout(\n                  (value as (...a: unknown[]) => unknown).apply(target, args) as Promise<unknown>,\n                  this.timeoutMs,\n                  `CKB Indexer method ${prop} timed out after ${this.timeoutMs}ms`\n                )\n              );\n          }\n          if (value && typeof value === 'object' && (prop === 'terminableCellFetcher' || prop === 'terminableTransactionFetcher')) {\n            return wrap(value);\n          }\n          return value;\n        },\n      });\n    };\n    return wrap(indexer);\n  }

[gemini-code-assist]

The CkbRpcCaller class currently imports the global env object directly from ../env. This violates the dependency injection pattern used throughout the codebase (where configuration is retrieved from the cradle container) and makes unit testing or mocking configuration values difficult.\n\nWe should refactor the constructor to accept a configuration object instead of relying on the static env import.

export class CkbRpcCaller {\n  private readonly limit: pLimit.Limit;\n  private readonly batchMaxSize: number;\n  private readonly timeoutMs: number;\n  public readonly rpc: RPC;\n  public readonly indexer: Indexer;\n\n  constructor(\n    url: string,\n    config: {\n      maxConcurrency: number;\n      batchMaxSize: number;\n      timeoutMs: number;\n    }\n  ) {\n    this.limit = pLimit(config.maxConcurrency);\n    this.batchMaxSize = config.batchMaxSize;\n    this.timeoutMs = config.timeoutMs;\n\n    const limitedFetch = ((input: any, init?: any) => this.limit(() => fetch(input, init))) as typeof fetch;\n\n    this.rpc = new RPC(url, {\n      timeout: config.timeoutMs,\n      fetch: limitedFetch,\n    });\n\n    this.indexer = this.wrapIndexer(new Indexer(url));\n  }

[gemini-code-assist]

Update the instantiation of CkbRpcCaller to pass the configuration object from cradle.env to align with the new constructor signature and maintain proper dependency injection.

  constructor(private cradle: Cradle) {\n    this.caller = new CkbRpcCaller(cradle.env.CKB_RPC_URL, {\n      maxConcurrency: cradle.env.CKB_RPC_MAX_CONCURRENCY,\n      batchMaxSize: cradle.env.CKB_RPC_BATCH_MAX_SIZE,\n      timeoutMs: cradle.env.CKB_HTTP_TIMEOUT_MS,\n    });\n    this.rpc = this.caller.rpc;\n    this.indexer = this.caller.indexer;