Merge branch 'main' into dependabot/npm_and_yarn/eslint-plugin-jsdoc-50.6.9

Author: pdp2121

.github/workflows/node.js.yml

@@ -12,7 +12,7 @@ jobs:
 
     strategy:
       matrix:
-        node-version: [14.x, 16.x, 18.x, 20.x]
+        node-version: [18.x, 20.x]
         # See supported Node.js release schedule at https://nodejs.org/en/about/releases/
 
     services:

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "validator-history-service",
-  "version": "0.8.0",
+  "version": "0.8.3-rc.2",
   "description": "Service for ingesting, aggregating, storing, and disbursing validation related data",
   "main": "./build/index.js",
   "scripts": {

package.json

@@ -17,7 +17,9 @@ import {
   ValidationRaw,
   ValidatorKeys,
   Ballot,
+  Chain,
 } from '../shared/types'
+import { getLists, overlaps } from '../shared/utils'
 import logger from '../shared/utils/logger'
 
 import chains from './chains'
@@ -120,6 +122,30 @@ function isPreceedingFlagLedger(ledger_index: string): boolean {
   return parseInt(ledger_index, 10) % 256 === 255
 }
 
+/**
+ * Finds network name from chain id.
+ *
+ * @param chain - A chain object.
+ * @returns String.
+ */
+async function getNetworkNameFromChainId(chain: Chain): Promise<string> {
+  let id = chain.id
+  const lists = await getLists().catch((err) => {
+    log.error('Error getting validator lists', err)
+    return undefined
+  })
+
+  if (lists != null) {
+    Object.entries(lists).forEach(([network, set]) => {
+      if (overlaps(chain.validators, set)) {
+        id = network
+      }
+    })
+  }
+
+  return id
+}
+
 /**
  *
  */
@@ -153,6 +179,14 @@ class Agreement {
     for (const chain of agreementChains) {
       const ledger_hashes = chain.ledgers
 
+      const networkName = await getNetworkNameFromChainId(chain)
+
+      log.info(
+        `Agreement: ${chain.id}:${networkName}:${Array.from(
+          chain.validators,
+        ).join(',')}`,
+      )
+
       for (const signing_key of chain.validators) {
         promises.push(
           this.calculateValidatorAgreement(

src/connection-manager/agreement.ts

@@ -27,44 +27,55 @@ const ports = [443, 80, 6005, 6006, 51233, 51234]
 const protocols = ['wss://', 'ws://']
 const connections: Map<string, WebSocket> = new Map()
 const networkFee: Map<string, FeeVote> = new Map()
+const validationNetworkDb: Map<string, string> = new Map()
 const CM_INTERVAL = 60 * 60 * 1000
 const WS_TIMEOUT = 10000
 const REPORTING_INTERVAL = 15 * 60 * 1000
 const BACKTRACK_INTERVAL = 30 * 60 * 1000
+const BASE_RETRY_DELAY = 1 * 1000
+const MAX_RETRY_DELAY = 30 * 1000
 
 // The frequent closing codes seen so far after connections established include:
 //  1008: Policy error: client is too slow. (Most frequent)
 //  1006: Abnormal Closure: The connection was closed abruptly without a proper handshake or a clean closure.
 //  1005: No Status Received: An empty or undefined status code is used to indicate no further details about the closure.
 // Reconnection should happen after seeing these codes for established connections.
 const CLOSING_CODES = [1005, 1006, 1008]
-let connectionsInitialized = false
 let cmStarted = false
 
 /**
  * Sets the handlers for each WebSocket object.
  *
- * @param ip - The ip address of the node we are trying to reach.
+ * @param ws_url - The WebSocket address of the node we are trying to reach.
  * @param ws - A WebSocket object.
  * @param networks - The networks of the node we are trying to reach where it retrieves validations.
  * @param isInitialNode - Whether source node is an entry/initial node for the network.
+ * @param retryCount - Retry count for exponential backoff.
  * @returns A Promise that resolves to void once a connection has been created or timeout has occured.
  */
+// eslint-disable-next-line max-params -- Required here
 async function setHandlers(
-  ip: string,
+  ws_url: string,
   ws: WebSocket,
   networks: string | undefined,
   isInitialNode = false,
+  retryCount = 0,
 ): Promise<void> {
   const ledger_hashes: string[] = []
   return new Promise(function setHandlersPromise(resolve, _reject) {
     ws.on('open', () => {
-      if (connections.has(ip)) {
+      log.info(
+        `Websocket connection opened for: ${ws.url} on ${
+          networks ?? 'unknown network'
+        }`,
+      )
+
+      if (connections.has(ws.url)) {
         resolve()
         return
       }
       void saveNodeWsUrl(ws.url, true)
-      connections.set(ip, ws)
+      connections.set(ws.url, ws)
       subscribe(ws)
 
       // Use LedgerEntry to look for amendments that has already been enabled on a network when connections
@@ -103,45 +114,63 @@ async function setHandlers(
           networks,
           networkFee,
           ws,
+          validationNetworkDb,
         )
       }
     })
     ws.on('close', async (code, reason) => {
-      const nodeNetworks = networks ?? 'unknown network'
-      if (connectionsInitialized) {
-        log.error(
-          `Websocket closed for ${
-            ws.url
-          } on ${nodeNetworks} with code ${code} and reason ${reason.toString(
-            'utf-8',
-          )}.`,
+      log.error(
+        `Websocket closed for ${ws.url} on ${
+          networks ?? 'unknown network'
+        } with code ${code} and reason ${reason.toString('utf-8')}.`,
+      )
+
+      const delay = BASE_RETRY_DELAY * 2 ** retryCount
+
+      if (CLOSING_CODES.includes(code) && delay <= MAX_RETRY_DELAY) {
+        log.info(
+          `Reconnecting to ${ws.url} on ${
+            networks ?? 'unknown network'
+          } after ${delay}ms...`,
         )
-        if (CLOSING_CODES.includes(code)) {
-          log.info(
-            `Reconnecting to ${ws.url} on ${networks ?? 'unknown network'}...`,
-          )
+        // Clean up the old Websocket connection
+        connections.delete(ws.url)
+        ws.terminate()
+        resolve()
+
+        setTimeout(async () => {
           // Open a new Websocket connection for the same url
           const newWS = new WebSocket(ws.url, { handshakeTimeout: WS_TIMEOUT })
-          // Clean up the old Websocket connection
-          connections.delete(ip)
-          ws.terminate()
-          resolve()
 
-          await setHandlers(ip, newWS, networks, isInitialNode)
-          // return since the old websocket connection has already been terminated
-          return
-        }
+          await setHandlers(
+            ws_url,
+            newWS,
+            networks,
+            isInitialNode,
+            retryCount + 1,
+          )
+        }, delay)
+
+        // return since the old websocket connection has already been terminated
+        return
       }
-      if (connections.get(ip)?.url === ws.url) {
-        connections.delete(ip)
+
+      if (connections.get(ws.url)?.url === ws.url) {
+        connections.delete(ws.url)
         void saveNodeWsUrl(ws.url, false)
       }
       ws.terminate()
       resolve()
     })
-    ws.on('error', () => {
-      if (connections.get(ip)?.url === ws.url) {
-        connections.delete(ip)
+    ws.on('error', (err) => {
+      log.error(
+        `Websocket connection error for ${ws.url} on ${
+          networks ?? 'unknown network'
+        } - ${err.message}`,
+      )
+
+      if (connections.get(ws.url)?.url === ws.url) {
+        connections.delete(ws.url)
       }
       ws.terminate()
       resolve()
@@ -169,13 +198,13 @@ async function findConnection(node: WsNode): Promise<void> {
     return Promise.resolve()
   }
 
-  if (connections.has(node.ip)) {
+  if (Array.from(connections.keys()).some((key) => key.includes(node.ip))) {
     return Promise.resolve()
   }
 
   if (node.ws_url) {
     const ws = new WebSocket(node.ws_url, { handshakeTimeout: WS_TIMEOUT })
-    return setHandlers(node.ip, ws, node.networks)
+    return setHandlers(node.ws_url, ws, node.networks)
   }
 
   const promises: Array<Promise<void>> = []
@@ -185,7 +214,7 @@ async function findConnection(node: WsNode): Promise<void> {
       const ws = new WebSocket(url, { handshakeTimeout: WS_TIMEOUT })
       promises.push(
         setHandlers(
-          node.ip,
+          url,
           ws,
           node.networks,
           networkInitialIps.includes(node.ip),
@@ -197,13 +226,23 @@ async function findConnection(node: WsNode): Promise<void> {
   return Promise.resolve()
 }
 
+async function getValidationNetworkDb(): Promise<void> {
+  const validatorNetwork: Array<{ signing_key: string; networks: string }> =
+    await query('validators').select('signing_key', 'networks')
+  for (const entry of validatorNetwork) {
+    validationNetworkDb.set(entry.signing_key, entry.networks)
+  }
+}
+
 /**
  * Creates connections to nodes found in the database.
  *
  * @returns A promise that resolves to void once all possible connections have been created.
  */
 async function createConnections(): Promise<void> {
   log.info('Finding Connections...')
+  validationNetworkDb.clear()
+  await getValidationNetworkDb()
   const tenMinutesAgo = new Date()
   tenMinutesAgo.setMinutes(tenMinutesAgo.getMinutes() - 10)
 
@@ -222,12 +261,12 @@ async function createConnections(): Promise<void> {
   })
 
   const promises: Array<Promise<void>> = []
-  connectionsInitialized = false
+
   nodes.forEach((node: WsNode) => {
     promises.push(findConnection(node))
   })
   await Promise.all(promises)
-  connectionsInitialized = true
+
   log.info(`${connections.size} connections created`)
 }