Multiple endpoints

packages/cli/src/controller/migrate-controller.ts

@@ -10,7 +10,7 @@ import {
   ChainTypes,
   loadSubstrateProjectManifest,
 } from '@subql/common-substrate';
-import {loadTerraProjectManifest, TerraProjectManifestVersioned} from '@subql/common-terra';
+import {loadTerraProjectManifest, TerraProjectManifestVersioned, TerraProjectNetworkV0_3_0} from '@subql/common-terra';
 import {classToPlain} from 'class-transformer';
 import {cli} from 'cli-ux';
 import inquirer from 'inquirer';
@@ -63,7 +63,9 @@ export async function prepare(
   if (project.runner.node.name === SUBSTRATE_NODE_NAME) {
     cli.action.start('Getting network genesis hash from endpoint for Chain ID');
     try {
-      genesisHash = await getGenesisHash(projectNetwork.endpoint);
+      genesisHash = await getGenesisHash(
+        typeof projectNetwork.endpoint === 'string' ? projectNetwork.endpoint : projectNetwork.endpoint[0]
+      );
     } catch (e) {
       genesisHash = null;
     }

packages/common-substrate/src/project/versioned/v0_0_1/model.ts

@@ -25,8 +25,8 @@ import {SubstrateProjectNetworkConfig} from '../../types';
 import {ManifestV0_0_1Mapping, ProjectManifestV0_0_1, RuntimeDataSourceV0_0_1} from './types';
 
 export class ProjectNetworkV0_0_1 extends ChainTypes implements SubstrateProjectNetworkConfig {
-  @IsString()
-  endpoint: string;
+  @IsString({each: true})
+  endpoint: string[];
   @IsString()
   @IsOptional()
   dictionary?: string;

packages/common-substrate/src/project/versioned/v0_2_0/model.ts

@@ -34,9 +34,9 @@ export class ProjectNetworkDeploymentV0_2_0 {
 }
 
 export class ProjectNetworkV0_2_0 extends ProjectNetworkDeploymentV0_2_0 {
-  @IsString()
+  @IsString({each: true})
   @IsOptional()
-  endpoint?: string;
+  endpoint?: string | string[];
   @IsString()
   @IsOptional()
   dictionary?: string;

packages/common-substrate/src/project/versioned/v0_3_0/types.ts

@@ -18,7 +18,7 @@ export interface SubstrateProjectManifestV0_3_0 extends ISubstrateProjectManifes
 
   network: {
     genesisHash: string;
-    endpoint?: string;
+    endpoint?: string | string[];
     dictionary?: string;
     chaintypes?: {
       file: string;

packages/common-substrate/src/project/versioned/v1_0_0/model.ts

@@ -74,9 +74,9 @@ export class ProjectNetworkDeploymentV1_0_0 {
 }
 
 export class ProjectNetworkV1_0_0 extends ProjectNetworkDeploymentV1_0_0 {
-  @IsString()
+  @IsString({each: true})
   @IsOptional()
-  endpoint?: string;
+  endpoint?: string | string[];
   @IsString()
   @IsOptional()
   dictionary?: string;

packages/common/src/project/types.ts

@@ -11,7 +11,7 @@ export interface IProjectManifest<D> {
 }
 
 export interface ProjectNetworkConfig {
-  endpoint: string;
+  endpoint: string | string[];
   dictionary?: string;
   bypassBlocks?: (number | string)[];
   //genesisHash?: string;

packages/common/src/project/utils.ts

@@ -59,7 +59,7 @@ export function validateSemver(current: string, required: string): boolean {
 @ValidatorConstraint({name: 'semver', async: false})
 export class SemverVersionValidator implements ValidatorConstraintInterface {
   validate(value: string | null | undefined): boolean {
-    if (valid(value, {includePrerelease: false}) === null) {
+    if (valid(value) === null) {
       return validRange(value, {includePrerelease: false}) !== null;
     } else {
       return prerelease(value) === null;

packages/common/src/project/versioned/v0_2_0/types.ts

@@ -11,7 +11,7 @@ export interface ProjectManifestV0_2_0<D extends object = BaseDataSource> {
   };
   network: {
     genesisHash: string;
-    endpoint?: string;
+    endpoint?: string | string[];
     dictionary?: string;
     chaintypes?: {
       file: string;

packages/common/src/project/versioned/v1_0_0/types.ts

@@ -26,7 +26,7 @@ export interface ProjectManifestV1_0_0<T extends object = TemplateBase, D extend
   templates?: T[];
   network: {
     chainId: string;
-    endpoint?: string;
+    endpoint?: string | string[];
     dictionary?: string;
     bypassBlocks?: (number | string)[];
     chaintypes?: {

packages/node-core/src/configure/NodeConfig.ts

@@ -22,7 +22,7 @@ export interface IConfig {
   readonly blockTime: number;
   readonly debug: boolean;
   readonly preferRange: boolean;
-  readonly networkEndpoint?: string;
+  readonly networkEndpoints?: string[];
   readonly networkDictionary?: string;
   readonly dictionaryResolver?: string;
   readonly outputFmt?: 'json';
@@ -109,8 +109,8 @@ export class NodeConfig implements IConfig {
     return this._config.batchSize;
   }
 
-  get networkEndpoint(): string | undefined {
-    return this._config.networkEndpoint;
+  get networkEndpoints(): string[] | undefined {
+    return this._config.networkEndpoints;
   }
 
   get networkDictionary(): string | undefined {

packages/node-core/src/indexer/connectionPool.service.ts

@@ -0,0 +1,96 @@
+// Copyright 2020-2022 OnFinality Limited authors & contributors
+// SPDX-License-Identifier: Apache-2.0
+
+import {Injectable, OnApplicationShutdown} from '@nestjs/common';
+import {toNumber} from 'lodash';
+import {getLogger} from '../logger';
+
+const logger = getLogger('api');
+
+export abstract class ApiConnection {
+  // eslint-disable-next-line @typescript-eslint/require-await
+  async create(endpoint: string, args: any): Promise<void> {
+    throw new Error('create() is not supported');
+  }
+
+  // eslint-disable-next-line @typescript-eslint/require-await
+  async apiConnect(): Promise<void> {
+    throw new Error('apiConnect() is not supported');
+  }
+
+  // eslint-disable-next-line @typescript-eslint/require-await
+  async apiDisconnect(): Promise<void> {
+    throw new Error('apiDisconnect() is not supported');
+  }
+}
+
+@Injectable()
+export class ConnectionPoolService<T extends ApiConnection> implements OnApplicationShutdown {
+  private allApi: T[] = [];
+  private connectionPool: Record<number, T> = {};
+  private taskCounter = 0;
+
+  async onApplicationShutdown(): Promise<void> {
+    await Promise.all(
+      Object.keys(this.connectionPool)?.map((key) => this.connectionPool[toNumber(key)].apiDisconnect())
+    );
+  }
+
+  addToConnections(api: T): void {
+    this.allApi.push(api);
+    this.connectionPool[this.allApi.length - 1] = api;
+  }
+
+  async connectToApi(apiIndex: number): Promise<void> {
+    await this.allApi[apiIndex].apiConnect();
+  }
+
+  get api(): T {
+    const index = this.getNextConnectedApiIndex();
+    if (index === -1) {
+      throw new Error('No connected api');
+    }
+    return this.connectionPool[index];
+  }
+
+  get firstConnectedApi(): T {
+    const index = this.getFirstConnectedApiIndex();
+    if (index === -1) {
+      throw new Error('No connected api');
+    }
+    return this.connectionPool[index];
+  }
+
+  getNextConnectedApiIndex(): number {
+    // get the next connected api index
+    if (Object.keys(this.connectionPool).length === 0) {
+      return -1;
+    }
+    const nextIndex = this.taskCounter % Object.keys(this.connectionPool).length;
+    this.taskCounter++;
+    return toNumber(Object.keys(this.connectionPool)[nextIndex]);
+  }
+
+  private getFirstConnectedApiIndex(): number {
+    // get the first connected api index
+    if (Object.keys(this.connectionPool).length === 0) {
+      return -1;
+    }
+    return toNumber(Object.keys(this.connectionPool)[0]);
+  }
+
+  get numConnections(): number {
+    return Object.keys(this.connectionPool).length;
+  }
+
+  async handleApiDisconnects(apiIndex: number, endpoint: string): Promise<void> {
+    logger.warn(`disconnected from ${endpoint}`);
+    delete this.connectionPool[apiIndex];
+
+    logger.debug(`reconnecting to ${endpoint}...`);
+    await this.connectToApi(apiIndex);
+
+    logger.info(`reconnected to ${endpoint}!`);
+    this.connectionPool[apiIndex] = this.allApi[apiIndex];
+  }
+}

packages/node-core/src/indexer/dictionary.service.test.ts

@@ -105,7 +105,7 @@ const HAPPY_PATH_CONDITIONS: DictionaryQueryEntry[] = [
 const nodeConfig = new NodeConfig({
   subquery: 'asdf',
   subqueryName: 'asdf',
-  networkEndpoint: 'wss://polkadot.api.onfinality.io/public-ws',
+  networkEndpoints: ['wss://polkadot.api.onfinality.io/public-ws'],
   dictionaryTimeout: 10,
 });
 

packages/node-core/src/indexer/index.ts

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 export * from './benchmark.service';
+export * from './connectionPool.service';
 export * from './entities';
 export * from './PoiBlock';
 export * from './types';
@@ -12,4 +13,4 @@ export * from './mmr.service';
 export * from './worker';
 export * from './dictionary.service';
 export * from './sandbox';
-export * from './smartBatch.service';
+export * from './smartBatch.service';

packages/node-core/src/utils/object.ts

@@ -20,3 +20,14 @@ export function camelCaseObjectKey(object: Record<string, any>): object {
     {}
   );
 }
+
+export function splitArrayByRatio(arr: number[], weights: number[]): number[][] {
+  const result: number[][] = [];
+  let start = 0;
+  for (let i = 0; i < weights.length; i++) {
+    const end = Math.floor(arr.length * weights[i]) + start;
+    result.push(arr.slice(start, end));
+    start = end;
+  }
+  return result;
+}

packages/node/src/configure/SubqueryProject.spec.ts

@@ -36,7 +36,7 @@ describe('SubqueryProject', () => {
     it('convert local 0.2.0 manifest to project object', async () => {
       //manually pass the endpoint
       const project = await SubqueryProject.create(projectDirV0_2_0, {
-        endpoint: 'wss://rpc.polkadot.io/public-ws',
+        endpoint: ['wss://rpc.polkadot.io/public-ws'],
       });
 
       expect((project.dataSources[1] as any).processor.file).toMatch(
@@ -47,7 +47,7 @@ describe('SubqueryProject', () => {
     it('convert local 0.3.0 manifest to project object', async () => {
       //manually pass the endpoint
       const project = await SubqueryProject.create(projectDirV0_3_0, {
-        endpoint: 'wss://rpc.polkadot.io/public-ws',
+        endpoint: ['wss://rpc.polkadot.io/public-ws'],
       });
 
       expect((project.dataSources[1] as any).processor.file).toMatch(
@@ -60,7 +60,7 @@ describe('SubqueryProject', () => {
       //manually pass the endpoint
       const project = await SubqueryProject.create(
         deployment,
-        { endpoint: 'wss://rpc.polkadot.io/public-ws' },
+        { endpoint: ['wss://rpc.polkadot.io/public-ws'] },
         { ipfs: 'http://127.0.0.1:8080' },
       );
     }, 5000000);
@@ -77,15 +77,15 @@ describe('SubqueryProject', () => {
         },
       };
       const project = await SubqueryProject.create(projectDirV1_0_0, {
-        endpoint: 'wss://rpc.polkadot.io/public-ws',
+        endpoint: ['wss://rpc.polkadot.io/public-ws'],
       });
 
       expect(project.runner).toMatchObject(expectedRunner);
     }, 500000);
 
     it('check processChainId', async () => {
       const project = await SubqueryProject.create(projectDirV1_0_0, {
-        endpoint: 'wss://rpc.polkadot.io/public-ws',
+        endpoint: ['wss://rpc.polkadot.io/public-ws'],
       });
       expect(project.network.chainId).toMatch(
         '0x401a1f9dca3da46f5c4091016c8a2f26dcea05865116b286f60f668207d1474b',
@@ -94,10 +94,10 @@ describe('SubqueryProject', () => {
 
     it('check loadProjectTemplates', async () => {
       const project = await SubqueryProject.create(templateProject, {
-        endpoint: 'wss://moonbeam-alpha.api.onfinality.io/public-ws',
+        endpoint: ['wss://moonbeam-alpha.api.onfinality.io/public-ws'],
       });
       const project_v1 = await SubqueryProject.create(projectDirV1_0_0, {
-        endpoint: 'wss://rpc.polkadot.io/public-ws',
+        endpoint: ['wss://rpc.polkadot.io/public-ws'],
       });
       expect(project_v1).not.toContain('template');
       expect(project.templates.length).toBe(1);

packages/node/src/configure/SubqueryProject.ts

@@ -108,7 +108,7 @@ export class SubqueryProject {
 
 export interface SubqueryProjectNetwork {
   chainId: string;
-  endpoint?: string;
+  endpoint?: string[];
   dictionary?: string;
   chaintypes?: FileType;
 }
@@ -137,6 +137,10 @@ async function loadProjectFromManifestBase(
 ): Promise<SubqueryProject> {
   const root = await getProjectRoot(reader);
 
+  if (typeof projectManifest.network.endpoint === 'string') {
+    projectManifest.network.endpoint = [projectManifest.network.endpoint];
+  }
+
   const network = processChainId({
     ...projectManifest.network,
     ...networkOverrides,

packages/node/src/configure/configure.module.ts

@@ -108,7 +108,7 @@ export class ConfigureModule {
         config.subquery,
         omitBy<SubstrateProjectNetworkConfig>(
           {
-            endpoint: config.networkEndpoint,
+            endpoint: config.networkEndpoints,
             dictionary: config.networkDictionary,
           },
           isNil,
@@ -170,7 +170,7 @@ export class ConfigureModule {
         argv.subquery,
         omitBy<SubstrateProjectNetworkConfig>(
           {
-            endpoint: config.networkEndpoint,
+            endpoint: config.networkEndpoints,
             dictionary: config.networkDictionary,
           },
           isNil,