Implement switching endpoints after slow eth_getLogs RPC requests (#…

…525) * Switch upstream endpoint if getLogs requests are too slow * Refactor methods for switching client to indexer * Update codegen indexer template * Add dummy methods in graph-node test Indexer * Upgrade package versions to 0.2.101 --------- Co-authored-by: Prathamesh Musale <[email protected]>
cerc-io · Jun 20, 2024 · b9a899a · b9a899a
1 parent ff471da
commit b9a899a
Show file tree

Hide file tree

Showing 24 changed files with 162 additions and 108 deletions.
diff --git a/lerna.json b/lerna.json
@@ -2,7 +2,7 @@
   "packages": [
     "packages/*"
   ],
-  "version": "0.2.100",
+  "version": "0.2.101",
   "npmClient": "yarn",
   "useWorkspaces": true,
   "command": {

diff --git a/packages/cache/package.json b/packages/cache/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/cache",
-  "version": "0.2.100",
+  "version": "0.2.101",
   "description": "Generic object cache",
   "main": "dist/index.js",
   "scripts": {

diff --git a/packages/cli/package.json b/packages/cli/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/cli",
-  "version": "0.2.100",
+  "version": "0.2.101",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "scripts": {
@@ -15,13 +15,13 @@
   },
   "dependencies": {
     "@apollo/client": "^3.7.1",
-    "@cerc-io/cache": "^0.2.100",
-    "@cerc-io/ipld-eth-client": "^0.2.100",
+    "@cerc-io/cache": "^0.2.101",
+    "@cerc-io/ipld-eth-client": "^0.2.101",
     "@cerc-io/libp2p": "^0.42.2-laconic-0.1.4",
     "@cerc-io/nitro-node": "^0.1.15",
-    "@cerc-io/peer": "^0.2.100",
-    "@cerc-io/rpc-eth-client": "^0.2.100",
-    "@cerc-io/util": "^0.2.100",
+    "@cerc-io/peer": "^0.2.101",
+    "@cerc-io/rpc-eth-client": "^0.2.101",
+    "@cerc-io/util": "^0.2.101",
     "@ethersproject/providers": "^5.4.4",
     "@graphql-tools/utils": "^9.1.1",
     "@ipld/dag-cbor": "^8.0.0",

diff --git a/packages/cli/src/base.ts b/packages/cli/src/base.ts
@@ -96,7 +96,7 @@ export class BaseCmd {
     this._jobQueue = new JobQueue({ dbConnectionString, maxCompletionLag: maxCompletionLagInSecs });
     await this._jobQueue.start();
 
-    const { ethClient, ethProvider } = await initClients(this._config);
+    const { ethClient, ethProvider } = await initClients(this._config.upstream);
     this._ethProvider = ethProvider;
     this._clients = { ethClient, ...clients };
   }

diff --git a/packages/cli/src/job-runner.ts b/packages/cli/src/job-runner.ts
@@ -7,8 +7,6 @@ import { hideBin } from 'yargs/helpers';
 import 'reflect-metadata';
 import assert from 'assert';
 import { ConnectionOptions } from 'typeorm';
-import { errors } from 'ethers';
-import debug from 'debug';
 
 import { JsonRpcProvider } from '@ethersproject/providers';
 import {
@@ -22,15 +20,10 @@ import {
   GraphWatcherInterface,
   startMetricsServer,
   Config,
-  UpstreamConfig,
-  NEW_BLOCK_MAX_RETRIES_ERROR,
-  setActiveUpstreamEndpointMetric
+  UpstreamConfig
 } from '@cerc-io/util';
 
 import { BaseCmd } from './base';
-import { initClients } from './utils/index';
-
-const log = debug('vulcanize:job-runner');
 
 interface Arguments {
   configFile: string;
@@ -40,10 +33,6 @@ export class JobRunnerCmd {
   _argv?: Arguments;
   _baseCmd: BaseCmd;
 
-  _currentEndpointIndex = {
-    rpcProviderEndpoint: 0
-  };
-
   constructor () {
     this._baseCmd = new BaseCmd();
   }
@@ -124,26 +113,8 @@ export class JobRunnerCmd {
     const jobRunner = new JobRunner(
       config.jobQueue,
       indexer,
-      jobQueue,
-      async (error: any) => {
-        // Check if it is a server error or timeout from ethers.js
-        // https://docs.ethers.org/v5/api/utils/logger/#errors--server-error
-        // https://docs.ethers.org/v5/api/utils/logger/#errors--timeout
-        if (error.code === errors.SERVER_ERROR || error.code === errors.TIMEOUT || error.message === NEW_BLOCK_MAX_RETRIES_ERROR) {
-          const oldRpcEndpoint = config.upstream.ethServer.rpcProviderEndpoints[this._currentEndpointIndex.rpcProviderEndpoint];
-          ++this._currentEndpointIndex.rpcProviderEndpoint;
-
-          if (this._currentEndpointIndex.rpcProviderEndpoint === config.upstream.ethServer.rpcProviderEndpoints.length) {
-            this._currentEndpointIndex.rpcProviderEndpoint = 0;
-          }
-
-          const { ethClient, ethProvider } = await initClients(config, this._currentEndpointIndex);
-          indexer.switchClients({ ethClient, ethProvider });
-          setActiveUpstreamEndpointMetric(config, this._currentEndpointIndex.rpcProviderEndpoint);
-
-          log(`RPC endpoint ${oldRpcEndpoint} is not working; failing over to new RPC endpoint ${ethProvider.connection.url}`);
-        }
-      });
+      jobQueue
+    );
 
     // Delete all active and pending (before completed) jobs to start job-runner without old queued jobs
     await jobRunner.jobQueue.deleteAllJobs('completed');
@@ -154,7 +125,7 @@ export class JobRunnerCmd {
     await startJobRunner(jobRunner);
     jobRunner.handleShutdown();
 
-    await startMetricsServer(config, jobQueue, indexer, this._currentEndpointIndex);
+    await startMetricsServer(config, jobQueue, indexer);
   }
 
   _getArgv (): any {

diff --git a/packages/cli/src/utils/index.ts b/packages/cli/src/utils/index.ts
@@ -9,7 +9,7 @@ import { providers } from 'ethers';
 
 // @ts-expect-error https://github.com/microsoft/TypeScript/issues/49721#issuecomment-1319854183
 import { PeerIdObj } from '@cerc-io/peer';
-import { Config, EthClient, getCustomProvider } from '@cerc-io/util';
+import { EthClient, UpstreamConfig, getCustomProvider } from '@cerc-io/util';
 import { getCache } from '@cerc-io/cache';
 import { EthClient as GqlEthClient } from '@cerc-io/ipld-eth-client';
 import { EthClient as RpcEthClient } from '@cerc-io/rpc-eth-client';
@@ -22,16 +22,10 @@ export function readPeerId (filePath: string): PeerIdObj {
   return JSON.parse(peerIdJson);
 }
 
-export const initClients = async (config: Config, endpointIndexes = { rpcProviderEndpoint: 0 }): Promise<{
+export const initClients = async (upstreamConfig: UpstreamConfig, endpointIndexes = { rpcProviderEndpoint: 0 }): Promise<{
   ethClient: EthClient,
   ethProvider: providers.JsonRpcProvider
 }> => {
-  const { database: dbConfig, upstream: upstreamConfig, server: serverConfig } = config;
-
-  assert(serverConfig, 'Missing server config');
-  assert(dbConfig, 'Missing database config');
-  assert(upstreamConfig, 'Missing upstream config');
-
   const { ethServer: { gqlApiEndpoint, rpcProviderEndpoints, rpcClient = false }, cache: cacheConfig } = upstreamConfig;
 
   assert(rpcProviderEndpoints, 'Missing upstream ethServer.rpcProviderEndpoints');

diff --git a/packages/codegen/package.json b/packages/codegen/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/codegen",
-  "version": "0.2.100",
+  "version": "0.2.101",
   "description": "Code generator",
   "private": true,
   "main": "index.js",
@@ -20,7 +20,7 @@
   },
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
-    "@cerc-io/util": "^0.2.100",
+    "@cerc-io/util": "^0.2.101",
     "@graphql-tools/load-files": "^6.5.2",
     "@npmcli/package-json": "^5.0.0",
     "@poanet/solidity-flattener": "https://github.com/vulcanize/solidity-flattener.git",

diff --git a/packages/codegen/src/templates/config-template.handlebars b/packages/codegen/src/templates/config-template.handlebars
@@ -83,6 +83,10 @@
     # Boolean flag to filter event logs by topics
     filterLogsByTopics = true
 
+    # Switch clients if eth_getLogs call takes more than threshold (in secs)
+    # Set to 0 for disabling switching
+    getLogsClientSwitchThresholdInSecs = 0
+
   [upstream.cache]
     name = "requests"
     enabled = false

diff --git a/packages/codegen/src/templates/indexer-template.handlebars b/packages/codegen/src/templates/indexer-template.handlebars
@@ -8,13 +8,12 @@ import debug from 'debug';
 {{#if queries}}
 import JSONbig from 'json-bigint';
 {{/if}}
-import { ethers, constants } from 'ethers';
+import { ethers, constants, providers } from 'ethers';
 {{#if (subgraphPath)}}
 import { GraphQLResolveInfo } from 'graphql';
 {{/if}}
 
 import { JsonFragment } from '@ethersproject/abi';
-import { BaseProvider } from '@ethersproject/providers';
 import { MappingKey, StorageLayout } from '@cerc-io/solidity-mapper';
 import {
   Indexer as BaseIndexer,
@@ -49,6 +48,7 @@ import {
   EthFullTransaction,
   ExtraEventData
 } from '@cerc-io/util';
+import { initClients } from '@cerc-io/cli';
 {{#if (subgraphPath)}}
 import { GraphWatcher } from '@cerc-io/graph-node';
 {{/if}}
@@ -91,7 +91,7 @@ const {{capitalize event}}_EVENT = '{{event}}';
 export class Indexer implements IndexerInterface {
   _db: Database;
   _ethClient: EthClient;
-  _ethProvider: BaseProvider;
+  _ethProvider: providers.JsonRpcProvider;
   _baseIndexer: BaseIndexer;
   _serverConfig: ServerConfig;
   _upstreamConfig: UpstreamConfig;
@@ -118,7 +118,7 @@ export class Indexer implements IndexerInterface {
     },
     db: DatabaseInterface,
     clients: Clients,
-    ethProvider: BaseProvider,
+    ethProvider: providers.JsonRpcProvider,
     jobQueue: JobQueue{{#if (subgraphPath)}},{{/if}}
     {{#if (subgraphPath)}}
     graphWatcher?: GraphWatcherInterface
@@ -199,15 +199,19 @@ export class Indexer implements IndexerInterface {
     await this._baseIndexer.fetchStateStatus();
   }
 
-  switchClients ({ ethClient, ethProvider }: { ethClient: EthClient, ethProvider: BaseProvider }): void {
+  async switchClients (): Promise<void> {
+    const { ethClient, ethProvider } = await this._baseIndexer.switchClients(initClients);
     this._ethClient = ethClient;
     this._ethProvider = ethProvider;
-    this._baseIndexer.switchClients({ ethClient, ethProvider });
     {{#if (subgraphPath)}}
     this._graphWatcher.switchClients({ ethClient, ethProvider });
     {{/if}}
   }
 
+  async isGetLogsRequestsSlow (): Promise<boolean> {
+    return this._baseIndexer.isGetLogsRequestsSlow();
+  }
+
   {{#if (subgraphPath)}}
   async getMetaData (block: BlockHeight): Promise<ResultMeta | null> {
     return this._baseIndexer.getMetaData(block);

diff --git a/packages/codegen/src/templates/package-template.handlebars b/packages/codegen/src/templates/package-template.handlebars
@@ -41,12 +41,12 @@
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
     "@apollo/client": "^3.3.19",
-    "@cerc-io/cli": "^0.2.100",
-    "@cerc-io/ipld-eth-client": "^0.2.100",
-    "@cerc-io/solidity-mapper": "^0.2.100",
-    "@cerc-io/util": "^0.2.100",
+    "@cerc-io/cli": "^0.2.101",
+    "@cerc-io/ipld-eth-client": "^0.2.101",
+    "@cerc-io/solidity-mapper": "^0.2.101",
+    "@cerc-io/util": "^0.2.101",
     {{#if (subgraphPath)}}
-    "@cerc-io/graph-node": "^0.2.100",
+    "@cerc-io/graph-node": "^0.2.101",
     {{/if}}
     "@ethersproject/providers": "^5.4.4",
     "debug": "^4.3.1",

diff --git a/packages/graph-node/package.json b/packages/graph-node/package.json
@@ -1,10 +1,10 @@
 {
   "name": "@cerc-io/graph-node",
-  "version": "0.2.100",
+  "version": "0.2.101",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "devDependencies": {
-    "@cerc-io/solidity-mapper": "^0.2.100",
+    "@cerc-io/solidity-mapper": "^0.2.101",
     "@ethersproject/providers": "^5.4.4",
     "@graphprotocol/graph-ts": "^0.22.0",
     "@nomiclabs/hardhat-ethers": "^2.0.2",
@@ -51,9 +51,9 @@
   "dependencies": {
     "@apollo/client": "^3.3.19",
     "@cerc-io/assemblyscript": "0.19.10-watcher-ts-0.1.2",
-    "@cerc-io/cache": "^0.2.100",
-    "@cerc-io/ipld-eth-client": "^0.2.100",
-    "@cerc-io/util": "^0.2.100",
+    "@cerc-io/cache": "^0.2.101",
+    "@cerc-io/ipld-eth-client": "^0.2.101",
+    "@cerc-io/util": "^0.2.101",
     "@types/json-diff": "^0.5.2",
     "@types/yargs": "^17.0.0",
     "bn.js": "^4.11.9",

diff --git a/packages/graph-node/test/utils/indexer.ts b/packages/graph-node/test/utils/indexer.ts
@@ -340,7 +340,11 @@ export class Indexer implements IndexerInterface {
     return [];
   }
 
-  switchClients ({ ethClient, ethProvider }: { ethClient: EthClient, ethProvider: providers.BaseProvider }): void {
+  async switchClients (): Promise<void> {
     return undefined;
   }
+
+  async isGetLogsRequestsSlow (): Promise<boolean> {
+    return false;
+  }
 }
diff --git a/packages/ipld-eth-client/package.json b/packages/ipld-eth-client/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/ipld-eth-client",
-  "version": "0.2.100",
+  "version": "0.2.101",
   "description": "IPLD ETH Client",
   "main": "dist/index.js",
   "scripts": {
@@ -20,8 +20,8 @@
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
     "@apollo/client": "^3.7.1",
-    "@cerc-io/cache": "^0.2.100",
-    "@cerc-io/util": "^0.2.100",
+    "@cerc-io/cache": "^0.2.101",
+    "@cerc-io/util": "^0.2.101",
     "cross-fetch": "^3.1.4",
     "debug": "^4.3.1",
     "ethers": "^5.4.4",

diff --git a/packages/peer/package.json b/packages/peer/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/peer",
-  "version": "0.2.100",
+  "version": "0.2.101",
   "description": "libp2p module",
   "main": "dist/index.js",
   "exports": "./dist/index.js",

diff --git a/packages/rpc-eth-client/package.json b/packages/rpc-eth-client/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/rpc-eth-client",
-  "version": "0.2.100",
+  "version": "0.2.101",
   "description": "RPC ETH Client",
   "main": "dist/index.js",
   "scripts": {
@@ -19,9 +19,9 @@
   },
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
-    "@cerc-io/cache": "^0.2.100",
-    "@cerc-io/ipld-eth-client": "^0.2.100",
-    "@cerc-io/util": "^0.2.100",
+    "@cerc-io/cache": "^0.2.101",
+    "@cerc-io/ipld-eth-client": "^0.2.101",
+    "@cerc-io/util": "^0.2.101",
     "chai": "^4.3.4",
     "ethers": "^5.4.4",
     "left-pad": "^1.3.0",

diff --git a/packages/solidity-mapper/package.json b/packages/solidity-mapper/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/solidity-mapper",
-  "version": "0.2.100",
+  "version": "0.2.101",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "devDependencies": {

diff --git a/packages/test/package.json b/packages/test/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/test",
-  "version": "0.2.100",
+  "version": "0.2.101",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "private": true,

diff --git a/packages/tracing-client/package.json b/packages/tracing-client/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/tracing-client",
-  "version": "0.2.100",
+  "version": "0.2.101",
   "description": "ETH VM tracing client",
   "main": "dist/index.js",
   "scripts": {