Implement failover for RPC endpoints in watcher (#506)

* Handle RPC endpoint server errors and switch failover endpoints

* Add config maxNewBlockRetries for switching to failover endpoint

* Upgrade package versions

* Move unknown events removal after event processing for historical sync

* Rename doFailOverEndpoints to switchClients

packages/cli/package.json

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

packages/cli/src/job-runner.ts

@@ -7,6 +7,8 @@ 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 {
@@ -20,10 +22,14 @@ import {
   GraphWatcherInterface,
   startMetricsServer,
   Config,
-  UpstreamConfig
+  UpstreamConfig,
+  NEW_BLOCK_MAX_RETRIES_ERROR
 } from '@cerc-io/util';
 
 import { BaseCmd } from './base';
+import { initClients } from './utils/index';
+
+const log = debug('vulcanize:job-runner');
 
 interface Arguments {
   configFile: string;
@@ -33,6 +39,10 @@ export class JobRunnerCmd {
   _argv?: Arguments;
   _baseCmd: BaseCmd;
 
+  _currentEndpointIndex = {
+    rpcProviderEndpoint: 0
+  };
+
   constructor () {
     this._baseCmd = new BaseCmd();
   }
@@ -110,7 +120,27 @@ export class JobRunnerCmd {
       await indexer.addContracts();
     }
 
-    const jobRunner = new JobRunner(config.jobQueue, indexer, jobQueue);
+    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 });
+          log(`RPC endpoint ${oldRpcEndpoint} is not working; failing over to new RPC endpoint ${ethProvider.connection.url}`);
+        }
+      });
 
     // Delete all active and pending (before completed) jobs to start job-runner without old queued jobs
     await jobRunner.jobQueue.deleteAllJobs('completed');
@@ -121,7 +151,7 @@ export class JobRunnerCmd {
     await startJobRunner(jobRunner);
     jobRunner.handleShutdown();
 
-    await startMetricsServer(config, indexer);
+    await startMetricsServer(config, indexer, this._currentEndpointIndex);
   }
 
   _getArgv (): any {

packages/cli/src/utils/index.ts

@@ -22,7 +22,7 @@ export function readPeerId (filePath: string): PeerIdObj {
   return JSON.parse(peerIdJson);
 }
 
-export const initClients = async (config: Config): Promise<{
+export const initClients = async (config: Config, endpointIndexes = { rpcProviderEndpoint: 0 }): Promise<{
   ethClient: EthClient,
   ethProvider: providers.JsonRpcProvider
 }> => {
@@ -32,9 +32,10 @@ export const initClients = async (config: Config): Promise<{
   assert(dbConfig, 'Missing database config');
   assert(upstreamConfig, 'Missing upstream config');
 
-  const { ethServer: { gqlApiEndpoint, rpcProviderEndpoint, rpcClient = false }, cache: cacheConfig } = upstreamConfig;
+  const { ethServer: { gqlApiEndpoint, rpcProviderEndpoints, rpcClient = false }, cache: cacheConfig } = upstreamConfig;
 
-  assert(rpcProviderEndpoint, 'Missing upstream ethServer.rpcProviderEndpoint');
+  assert(rpcProviderEndpoints, 'Missing upstream ethServer.rpcProviderEndpoints');
+  assert(rpcProviderEndpoints.length, 'No endpoints configured in ethServer.rpcProviderEndpoints');
 
   const cache = await getCache(cacheConfig);
 
@@ -42,12 +43,13 @@ export const initClients = async (config: Config): Promise<{
 
   if (rpcClient) {
     ethClient = new RpcEthClient({
-      rpcEndpoint: rpcProviderEndpoint,
+      rpcEndpoint: rpcProviderEndpoints[endpointIndexes.rpcProviderEndpoint],
       cache
     });
   } else {
     assert(gqlApiEndpoint, 'Missing upstream ethServer.gqlApiEndpoint');
 
+    // TODO: Implement failover for GQL endpoint
     ethClient = new GqlEthClient({
       gqlEndpoint: gqlApiEndpoint,
       cache
@@ -55,7 +57,7 @@ export const initClients = async (config: Config): Promise<{
   }
 
   const ethProvider = getCustomProvider({
-    url: rpcProviderEndpoint,
+    url: rpcProviderEndpoints[endpointIndexes.rpcProviderEndpoint],
     allowGzip: true
   });