ApiHandler / ChainData refactoring

Removed a lot of retry/reconnect logic from ApiHandler and ChainData, as
it was spawning a lot of PJS instances, which also handling reconnects
on its own.
This change also properly adpots the idea of talking both to relay chain
and to parachains, depending on queried data.

With new logic, each ApiHandler instance governs an interface to a
single chain, where it can switch endpoints and do reconnects on its
own.
`ApiHandler.getApi()` returns a PJS instance that should work, and will
block in case we're reconnecting.

PeopleApi endpoints now live in configs, instead of being hardcoded.

Instead of passing around ApiHandlers, passing a ChainData instance now, as
it contains both ApiHandlers instance now, and as it is an intended
abstraction for it

Author: mutantcornholio

docs/docs/backend/config.md

@@ -42,7 +42,8 @@ An example config may look something like:
 
 - `dryRun`: Boolean (true/false). If set to true, the Nominator accounts that are added will calculate what a nomination would look like (how many and which validators), but not craft or submit any transactions. No nominations will be done when this flag is set. In the `nominate` function of the `Nominator` class, the `dryRun` flag is checked, and if it is set to true, the function will return after logging the validators it would nominate without doing anything. This flag is optional and set to `false` by default.
 - `networkPrefix`: Integer. Defines the network prefix. For Kusama, this is `2`, and for Polkadot, this is `0`. It can be set to `3` for running a local test network, although this isn't used much anymore. **This flag is required for `core` and `worker` services.**
-- `apiEndpoints`: Array of strings. Lists the RPC endpoints for the chain. When given a list of multiple, it will pick one at random to create a websocket connection to - this single connection is used throughout the entire service for any queries or submitting transactions. **This is required for `core` and `worker` services.
+- `apiEndpoints`: Array of strings. Lists the RPC endpoints for the chain. When given a list of multiple, it will pick one at random to create a websocket connection to - this single connection is used throughout the entire service for any queries or submitting transactions. **This is required for `core` and `worker` services.**
+- `apiPeopleEndpoints`: Optional array of strings. Lists the RPC endpoints for People parachain, if it's enabled on the network. 
 - `bootstrap`: Boolean. An **optional** flag that can be set to `true` to enable the bootstrap process. This can be used when running a instance of the backend and would query the main Kusama or Polkadot instances at the api endpoints specified below to populate the db with non-deterministic values like `rank` or `discoveredAt`. _This isn't currently used anywhere yet_
 - `kusamaBootstrapEndpoint`: String. URL for the Kusama bootstrap endpoint. **optional**. _This isn't currently used anywhere yet_. 
 - `polkadotBootstrapEndpoint`: String. URL for the Polkadot bootstrap endpoint. **optional**. _This isn't currently used anywhere yet_.
@@ -645,4 +646,4 @@ An example Worker config run as microservices may look something like:
   }
 }
 
-```
+```

helmfile.d/config/kusama/otv-backend-ci.yaml.gotmpl

@@ -27,6 +27,7 @@ config: |
         "wss://rpc.dotters.network/kusama",
         "wss://ksm-rpc.stakeworld.io"
       ],
+      "apiPeopleEndpoints": ["wss://kusama-people-rpc.polkadot.io"],
       "candidatesUrl": "https://raw.githubusercontent.com/w3f/1k-validators-be/master/candidates/kusama.json"
     },
     "constraints": {

packages/common/src/ApiHandler/ApiHandler.ts

@@ -2,7 +2,6 @@ import { ApiPromise, WsProvider } from "@polkadot/api";
 import EventEmitter from "eventemitter3";
 
 import logger from "../logger";
-import { sleep } from "../utils/util";
 import { API_PROVIDER_TIMEOUT, POLKADOT_API_TIMEOUT } from "../constants";
 
 export const apiLabel = { label: "ApiHandler" };
@@ -12,167 +11,147 @@ export const apiLabel = { label: "ApiHandler" };
  * to a different provider if one proves troublesome.
  */
 class ApiHandler extends EventEmitter {
-  private _wsProvider?: WsProvider;
-  private _api: ApiPromise | null = null;
-  private readonly _endpoints: string[] = [];
-  static isConnected = false;
-  private healthCheckInProgress = false;
-  private _currentEndpoint?: string;
-  public upSince: number = Date.now();
+  private wsProvider?: WsProvider;
+  private api: ApiPromise | null = null;
+  private readonly endpoints: string[] = [];
+
+  // If we're reconnecting right now, awaiting on this promise will block until connection succedes
+  private connectionAttempt: Promise<void> | null = null;
+
+  public upSince = -1;
+  public isConnected = false;
+
   constructor(endpoints: string[]) {
     super();
-    this._endpoints = endpoints.sort(() => Math.random() - 0.5);
-    this.upSince = Date.now();
+    this.endpoints = endpoints.sort(() => Math.random() - 0.5);
   }
 
-  async healthCheck(retries = 0): Promise<boolean> {
-    if (retries < 10) {
+  /**
+   * This copies connectWithRetry() logic from WsProvider
+   * The issue with original logic is that `autoConnectMs` is set to 0 when disconnect() is called, but we
+   * want to call it from nextEndpoint()
+   *
+   * This function can be called multiple times, and it'll wait on the same promise, without spamming reconnects.
+   * @see https://github.com/polkadot-js/api/blob/2ef84c5dcdbbff8aec9ba01e4f13a50130d1a6f3/packages/rpc-provider/src/ws/index.ts#L239-L271
+   */
+  private async connectWithRetry(): Promise<void> {
+    if (!this.wsProvider) {
+      throw new Error(
+        "connectWithRetry() is called before initializing WsProvider",
+      );
+    }
+
+    if (this.connectionAttempt instanceof Promise) {
+      await this.connectionAttempt;
+      return;
+    }
+
+    this.isConnected = false;
+    this.connectionAttempt = new Promise(async (resolve) => {
       try {
-        this.healthCheckInProgress = true;
-        let chain;
-
-        const isConnected = this._wsProvider?.isConnected;
-        if (isConnected && !this._api?.isConnected) {
-          try {
-            chain = await this._api?.rpc.system.chain();
-          } catch (e) {
-            await sleep(API_PROVIDER_TIMEOUT);
-          }
-        }
-        chain = await this._api?.rpc.system.chain();
-
-        if (isConnected && chain) {
-          this.healthCheckInProgress = false;
-          return true;
-        } else {
-          await sleep(API_PROVIDER_TIMEOUT);
-          logger.info(`api still disconnected, disconnecting.`, apiLabel);
-          await this._wsProvider?.disconnect();
-          await this.getProvider(this._endpoints);
-          await this.getAPI();
-          return false;
-        }
-      } catch (e: unknown) {
-        const errorMessage =
-          e instanceof Error ? e.message : "An unknown error occurred";
-        logger.error(
-          `Error in health check for WS Provider for rpc. ${errorMessage}`,
+        await this.wsProvider.connect();
+
+        await new Promise<void>((resolve, reject) => {
+          const unsubConnect = this.wsProvider.on("connected", resolve);
+          const unsubDisconnect = this.wsProvider.on("disconnected", reject);
+          this.connectionAttempt.finally(() => {
+            unsubConnect();
+            unsubDisconnect();
+          });
+        });
+
+        this.connectionAttempt = null;
+        this.upSince = Date.now();
+        this.isConnected = true;
+        logger.info(`Connected to ${this.currentEndpoint()}`, apiLabel);
+        resolve();
+      } catch (err) {
+        logger.warn(
+          `Connection attempt to ${this.currentEndpoint()} failed: ${JSON.stringify(err)}, trying next endpoint`,
           apiLabel,
         );
-        this.healthCheckInProgress = false;
-        return false;
+        setTimeout(() => {
+          this.connectionAttempt = null;
+          this.connectWithRetry().then(resolve);
+        }, API_PROVIDER_TIMEOUT);
       }
-    }
-    return false;
-  }
+    });
 
-  public currentEndpoint() {
-    return this._currentEndpoint;
+    await this.connectionAttempt;
   }
 
-  async getProvider(endpoints: string[]): Promise<WsProvider> {
-    return await new Promise<WsProvider>((resolve, reject) => {
-      const wsProvider = new WsProvider(
-        endpoints,
-        5000,
-        undefined,
-        POLKADOT_API_TIMEOUT,
-      );
+  /**
+   * In case of errors like RPC rate limit, we might want to force endpoint change
+   * PJS handles endpoint rotation internally, changing the endpoint on every next connection attempt.
+   * We only disconnect here; reconnect happens inside `"disconnected"` event handler
+   */
+  async nextEndpoint() {
+    logger.info("Rotating API endpoint", apiLabel);
+    await this.wsProvider.disconnect();
+    await this.connectWithRetry();
+  }
 
-      wsProvider.on("disconnected", async () => {
-        try {
-          const isHealthy = await this.healthCheck();
-          logger.info(
-            `[Disconnection] ${this._currentEndpoint}} Health check result: ${isHealthy}`,
-            apiLabel,
-          );
-          resolve(wsProvider);
-        } catch (error: any) {
-          logger.warn(
-            `WS provider for rpc ${endpoints[0]} disconnected!`,
-            apiLabel,
-          );
-          reject(error);
-        }
-      });
-      wsProvider.on("connected", () => {
-        logger.info(`WS provider for rpc ${endpoints[0]} connected`, apiLabel);
-        this._currentEndpoint = endpoints[0];
-        resolve(wsProvider);
-      });
-      wsProvider.on("error", async () => {
-        try {
-          const isHealthy = await this.healthCheck();
-          logger.info(
-            `[Error] ${this._currentEndpoint} Health check result: ${isHealthy}`,
-            apiLabel,
-          );
-          resolve(wsProvider);
-        } catch (error: any) {
-          logger.error(`Error thrown for rpc ${this._endpoints[0]}`, apiLabel);
-          reject(error);
-        }
-      });
-    });
+  currentEndpoint(): string | undefined {
+    return this.wsProvider?.endpoint;
   }
 
-  async getAPI(retries = 0): Promise<ApiPromise> {
-    if (this._wsProvider && this._api && this._api?.isConnected) {
-      return this._api;
+  private async healthCheck(): Promise<void> {
+    if (this.connectionAttempt instanceof Promise) {
+      return;
     }
-    const endpoints = this._endpoints.sort(() => Math.random() - 0.5);
-
     try {
-      logger.info(
-        `[getAPI]: try ${retries} creating provider with endpoint ${endpoints[0]}`,
+      const api = await this.getApi();
+      await api.rpc.system.chain();
+    } catch (err) {
+      logger.warn(
+        `Healthcheck on ${this.currentEndpoint()} failed: ${JSON.stringify(err)}, trying next endpoint`,
         apiLabel,
       );
-      const provider = await this.getProvider(endpoints);
-      this._wsProvider = provider;
-      logger.info(
-        `[getAPI]: provider created with endpoint: ${endpoints[0]}`,
-        apiLabel,
-      );
-      const api = await ApiPromise.create({
-        provider: provider,
-        noInitWarn: true,
-      });
-      await api.isReadyOrError;
-      logger.info(`[getApi] Api is ready`, apiLabel);
-      return api;
-    } catch (e) {
-      if (retries < 15) {
-        return await this.getAPI(retries + 1);
-      } else {
-        const provider = await this.getProvider(endpoints);
-        return await ApiPromise.create({
-          provider: provider,
-          noInitWarn: true,
-        });
-      }
+      await this.nextEndpoint();
     }
   }
 
-  async setAPI() {
-    const api = await this.getAPI(0);
-    this._api = api;
-    this._registerEventHandlers(this._api);
-    return api;
-  }
+  /**
+   * This function provides access to PJS api. While the ApiPromise instance never changes,
+   * the function will block if we're reconnecting.
+   * It's intended to be called every time instead of saving ApiPromise instance long-term.
+   */
+  async getApi(): Promise<ApiPromise> {
+    if (!this.wsProvider) {
+      this.wsProvider = new WsProvider(
+        this.endpoints,
+        false, // Do not autoconnect
+        undefined,
+        POLKADOT_API_TIMEOUT,
+      );
+      await this.connectWithRetry();
+      this.wsProvider.on("disconnected", () => {
+        logger.warn(`WsProvider disconnected`, apiLabel);
+        this.connectWithRetry();
+      });
+    }
+    if (!this.api) {
+      this.api = await ApiPromise.create({
+        provider: this.wsProvider,
+        noInitWarn: true,
+      });
+      await this.api.isReady;
+      this.registerEventHandlers(this.api);
 
-  isConnected(): boolean {
-    return this._wsProvider?.isConnected || false;
-  }
+      // healthcheck queries RPC, thus its interval can't be shorter than RPC timout
+      setInterval(() => {
+        void this.healthCheck();
+      }, POLKADOT_API_TIMEOUT);
+    }
 
-  getApi(): ApiPromise | null {
-    if (!this._api) {
-      return null;
-    } else {
-      return this._api;
+    if (this.connectionAttempt instanceof Promise) {
+      await this.connectionAttempt;
     }
+
+    return this.api;
   }
 
-  _registerEventHandlers(api: ApiPromise): void {
+  private registerEventHandlers(api: ApiPromise): void {
     if (!api) {
       logger.warn(`API is null, cannot register event handlers.`, apiLabel);
       return;

packages/common/src/ApiHandler/__mocks__/ApiHandler.ts

@@ -1,37 +1,30 @@
 import EventEmitter from "eventemitter3";
 import { ApiPromise } from "@polkadot/api";
+import { vi } from "vitest";
 
 const createMockApiPromise = (): any => ({
-  isConnected: jest.fn().mockReturnValue(true),
+  isConnected: vi.fn().mockReturnValue(true),
   query: {
     system: {
-      events: jest.fn().mockImplementation((callback) => callback([])), // Simplified; adjust as needed
+      events: vi.fn().mockImplementation((callback) => callback([])), // Simplified; adjust as needed
     },
   },
   isReadyOrError: Promise.resolve(),
   // Use the function itself to provide a new instance for the 'create' method
-  create: jest.fn().mockImplementation(() => createMockApiPromise()),
+  create: vi.fn().mockImplementation(() => createMockApiPromise()),
   // Add more mocked methods and properties as needed
 });
 
 // A mock class for ApiHandler
 class ApiHandlerMock extends EventEmitter {
-  private _endpoints: string[];
-  private _api: ApiPromise = createMockApiPromise as unknown as ApiPromise;
-  private healthCheckInProgress = false;
+  private endpoints: string[];
+  private api: ApiPromise = createMockApiPromise as unknown as ApiPromise;
 
   constructor(endpoints: string[]) {
     super();
     // Initialize with mock data or behavior as needed
-    this._endpoints = endpoints.sort(() => Math.random() - 0.5);
-  }
-
-  async healthCheck(): Promise<boolean> {
-    this.healthCheckInProgress = true;
-    // Simulate the health check logic; adjust the logic as needed for your tests
-    const isConnected = this._api.isConnected;
-    this.healthCheckInProgress = false;
-    return isConnected;
+    this.endpoints = endpoints.sort(() => Math.random() - 0.5);
+    this.api = createMockApiPromise();
   }
 
   async getProvider(endpoints: string[]): Promise<ApiPromise | undefined> {
@@ -42,25 +35,15 @@ class ApiHandlerMock extends EventEmitter {
     return undefined;
   }
 
-  async getAPI(retries = 0): Promise<ApiPromise> {
-    // Use the mockApiPromise directly for simplicity
-    return this._api;
-  }
-
-  async setAPI(): Promise<void> {
-    // Directly set the mock _api; in a real scenario, you might want to simulate more complex logic
-    this._api = await this.getAPI();
-  }
-
-  isConnected(): boolean {
-    return this._api.isConnected;
+  get isConnected(): boolean {
+    return this.api.isConnected;
   }
 
-  getApi(): ApiPromise {
-    return this._api;
+  async getApi(): Promise<ApiPromise> {
+    return this.api;
   }
 
-  _registerEventHandlers(api: ApiPromise): void {
+  private registerEventHandlers(api: ApiPromise): void {
     // Simplify the event handler registration for testing purposes
     // In a real scenario, you might want to simulate more complex event handling
     api.query.system.events((events) => {