Updates to the VHS connection-manager

src/connection-manager/agreement.ts

@@ -18,6 +18,7 @@
   ValidatorKeys,
   Ballot,
   Chain,
+  LedgerHashIndex,
 } from '../shared/types'
 import { getLists, overlaps } from '../shared/utils'
 import logger from '../shared/utils/logger'
@@ -268,7 +269,7 @@
    */
   private async calculateValidatorAgreement(
     signing_key: string,
-    ledger_hashes: Set<string>,
+    ledger_hashes: Set<LedgerHashIndex>,
     incomplete: boolean,
   ): Promise<void> {
     const master_key = await signingToMaster(signing_key)
@@ -284,24 +285,38 @@
     await updateDailyAgreement(validator_keys)
   }
 
   /**
    * Calculate the agreement score for the last hour of validations.
    *
    * @param validator_keys - Signing keys of validations for one validator.
    * @param validations - Set of ledger_hashes validated by signing_key.
    * @param ledgers - Set of ledger_hashes validated by network.
    * @param incomplete - Is this agreement score incomplete.
    * @returns Void.
    */
   private async calculateHourlyAgreement(
     validator_keys: ValidatorKeys,
     validations: Map<string, number>,
-    ledgers: Set<string>,
+    ledgerHashIndexMap: Set<LedgerHashIndex>,
     incomplete: boolean,
   ): Promise<void> {
+    // obtain ledger_hashes validated by the network, strip out the ledger_index info for agreement calculation purposes
+    let ledgers = new Set<string>()
+    for(const value of ledgerHashIndexMap) {
+      ledgers.add(value.ledger_hash)
+    }
     const missed = setDifference(ledgers, validations)
     const validated = setIntersection(ledgers, validations)
 
+    if(validator_keys.master_key == 'nHU4bLE3EmSqNwfL4AP1UZeTNPrSPPP6FXLKXo2uqfHuvBQxDVKd' || validator_keys.signing_key == 'n9LbM9S5jeGopF5J1vBDoGxzV6rNS8K1T5DzhNynkFLqR9N2fywX') {
+      console.log('DEBUG: XRPL Mainnet received the following ledgers: ', Array.from(ledgerHashIndexMap))
+      console.log('DEBUG: Number of Validations received by the Ripple validator: ', validations.size)
+      console.log('DEBUG: Validations received by the Ripple validator: <ignore the second value, which is the timestamp>', validations)
+      console.log('DEBUG: Missed ledgers: ', missed)
+      console.log('DEBUG: Validated ledgers: ', validated)
+      console.log('DEBUG: Incomplete: ', incomplete)
+    }
+
     const agreement: AgreementScore = {
       validated: validated.size,
       missed: missed.size,

src/connection-manager/chains.ts

@@ -1,7 +1,7 @@
 import { Knex } from 'knex'
 
 import { query } from '../shared/database'
-import { Ledger, ValidationRaw, Chain } from '../shared/types'
+import { Ledger, ValidationRaw, Chain, LedgerHashIndex } from '../shared/types'
 import { getLists, overlaps } from '../shared/utils'
 import logger from '../shared/utils/logger'
 
@@ -17,20 +17,26 @@
   return chain2.current - chain2.first - (chain1.current - chain1.first)
 }
 
+let LAST_SEEN_MAINNET_LEDGER_INDEX = -1
+
 /**
  * Adds ledger information to chain.
  *
  * @param ledger - Ledger to add to chain.
  * @param chain - Chain to update.
  */
 function addLedgerToChain(ledger: Ledger, chain: Chain): void {
-  chain.ledgers.add(ledger.ledger_hash)
+  chain.ledgers.add({ ledger_hash: ledger.ledger_hash, ledger_index: ledger.ledger_index } as LedgerHashIndex)
   for (const validator of ledger.validations) {
     chain.validators.add(validator)
   }
 
-  chain.current = ledger.ledger_index
-  chain.updated = ledger.first_seen
+  // is the incoming ledger more recent than the current chain's tip (i.e. chain.current)? If so, update the chain's tip.
+  chain.current = ledger.ledger_index > chain.current ? ledger.ledger_index : chain.current
+  chain.updated = ledger.ledger_index > chain.current ? ledger.first_seen : chain.updated
+
+  log.info(`Ledgers in the chain ${chain.id}: ${Array.from(chain.ledgers)}}`)
+  log.info(`Validators belonging to the chain ${chain.id}: ${Array.from(chain.validators)}}`)
 }
 
 /**
@@ -102,7 +108,7 @@
    *
    * @returns List of chains being monitored by the system.
    */
   public calculateChainsFromLedgers(): Chain[] {
     const list = []
     const now = Date.now()
 
@@ -113,6 +119,15 @@
         list.push(ledger)
       }
 
+      if (ledger.validations.size == 1) {
+        console.log('DEBUG: Throwing away this ledger because of only one validation: ')
+        console.log('DEBUG: Ledger hash: ', ledger_hash)
+        console.log('DEBUG: Ledger index: ', ledger.ledger_index)
+        console.log('DEBUG: Ledger validations: ', ledger.validations)
+        console.log('DEBUG: Ledger first seen: ', ledger.first_seen)
+        console.log('DEBUG: Is the ledger older than 10 seconds? ', tenSecondsOld)
+      }
+
       if (tenSecondsOld) {
         this.ledgersByHash.delete(ledger_hash)
       }
@@ -124,6 +139,35 @@
       this.updateChains(ledger)
     }
 
+    for(const chain of this.chains) {
+      if(chain.validators.has('nHU4bLE3EmSqNwfL4AP1UZeTNPrSPPP6FXLKXo2uqfHuvBQxDVKd') || chain.validators.has('n9LbM9S5jeGopF5J1vBDoGxzV6rNS8K1T5DzhNynkFLqR9N2fywX')) {
+        log.info('DEBUG: Validatig the continuity of XRPL Mainnet validated ledgers: \n')
+        log.info('DEBUG: Chain ID: {chain.id}')
+        log.info(`DEBUG: Chain updated at: ${chain.updated}`)
+        log.info(`DEBUG: Chain first: ${chain.first}`)
+        log.info(`DEBUG: Chain current: ${chain.current}`)
+        log.info(`DEBUG: Chain validators: ${Array.from(chain.validators)}`)
+        log.info(`DEBUG: Chain ledgers: ${Array.from(chain.ledgers)}`)
+        log.info(`DEBUG: Chain incomplete: ${chain.incomplete}`)
+
+        // check if the obtained ledgers are consecutive
+        for(const ledger of chain.ledgers) {
+
+          // initialization of this variable occurs exactly once, at the start of the program
+          if(LAST_SEEN_MAINNET_LEDGER_INDEX === -1) {
+            LAST_SEEN_MAINNET_LEDGER_INDEX = ledger.ledger_index
+            continue
+          }
+
+
+          if(ledger.ledger_index !== LAST_SEEN_MAINNET_LEDGER_INDEX + 1) {
+            log.error('ERROR: Ledgers are not consecutive. Void between indices: ' + LAST_SEEN_MAINNET_LEDGER_INDEX + ' and ' + ledger.ledger_index)
+          }
+          LAST_SEEN_MAINNET_LEDGER_INDEX = ledger.ledger_index
+        }
+      }
+    }
+
     return this.chains
   }
 
@@ -169,7 +213,7 @@
   private addNewChain(ledger: Ledger): void {
     const current = ledger.ledger_index
     const validators = ledger.validations
-    const ledgerSet = new Set([ledger.ledger_hash])
+    const ledgerSet = new Set([{ ledger_hash: ledger.ledger_hash, ledger_index: ledger.ledger_index } as LedgerHashIndex])
 
     const chain: Chain = {
       id: this.getNextChainID(),
@@ -191,6 +235,7 @@
    * @param ledger - The Ledger being handled in order to update the chains.
    */
   private updateChains(ledger: Ledger): void {
+    log.info(`Updating chains for ledger ${JSON.stringify(ledger)}`)
     const next = ledger.ledger_index
     const validators = ledger.validations
 
@@ -203,6 +248,7 @@
       .shift()
 
     if (chainAtNextIndex !== undefined) {
+      log.info(`Adding ledger ${ledger.ledger_hash} to chain ${chainAtNextIndex.id} (chain at next index)`)
       addLedgerToChain(ledger, chainAtNextIndex)
       return
     }
@@ -216,6 +262,7 @@
       .shift()
 
     if (chainAtThisIndex !== undefined) {
+      log.info(`Found this ledger ${ledger.ledger_hash} in chain ${chainAtThisIndex.id}. No action taken.`)
       return
     }
 
@@ -224,24 +271,37 @@
       .shift()
 
     const chainWithLedger: Chain | undefined = this.chains.find(
-      (chain: Chain) => chain.ledgers.has(ledger.ledger_hash),
+      (chain: Chain) => chain.ledgers.has({ ledger_hash: ledger.ledger_hash, ledger_index: ledger.ledger_index } as LedgerHashIndex),
     )
 
     if (chainWithThisValidator !== undefined) {
       const skipped = ledger.ledger_index - chainWithThisValidator.current
-      log.warn(`Possibly skipped ${skipped} ledgers`)
-      if (skipped > 1 && skipped < 20) {
+      log.warn(`Processing ledger ${ledger.ledger_hash}: Discovered chain with an overlap of validators ${chainWithThisValidator.id}. Possibly skipped ${skipped} ledgers`)
+      if (skipped !== 0) {
+        log.info(`Adding ledger ${ledger.ledger_hash} to chain ${chainWithThisValidator.id} (chain with an overlap of validators); Existing chain id: ${JSON.stringify(chainWithThisValidator)}`)
         chainWithThisValidator.incomplete = true
         addLedgerToChain(ledger, chainWithThisValidator)
       }
     }
 
     if (chainWithThisValidator !== undefined || chainWithLedger !== undefined) {
+      log.info(`No action taken for ledger ${ledger.ledger_hash}; \nINFO: Chain with an overlap of validators: ${chainWithThisValidator?.id};\n  Chain that contains this ledger: ${chainWithLedger?.id}`)
       return
     }
 
+    log.info(`Adding new chain for ledger ${ledger.ledger_hash}`)
     this.addNewChain(ledger)
   }
+
+  /**
+   *
+   * VHS tests use this method to reset the chains singleton value. This ensures that tests are not contaminating each other.
+   */
+  public __resetChainsSingletonForTests(): void {
+    this.ledgersByHash.clear()
+    this.chains = []
+    this.index = 0
+  }
 }
 
 let chains: Chains | undefined

src/connection-manager/connections.ts

@@ -37,6 +37,7 @@ const REPORTING_INTERVAL = 15 * 60 * 1000
 const BACKTRACK_INTERVAL = 30 * 60 * 1000
 const BASE_RETRY_DELAY = 1 * 1000
 const MAX_RETRY_DELAY = 30 * 1000
+const WS_HEARTBEAT_INTERVAL = 10 * 60 * 1000
 
 // The frequent closing codes seen so far after connections established include:
 //  1008: Policy error: client is too slow. (Most frequent)
@@ -46,6 +47,8 @@ const MAX_RETRY_DELAY = 30 * 1000
 const CLOSING_CODES = [1005, 1006, 1008]
 let cmStarted = false
 
+const webSocketHeartbeat: Map<string, Date> = new Map()
+
 /**
  * Sets the handlers for each WebSocket object.
  *
@@ -79,9 +82,11 @@ async function setHandlers(
       })
       subscribe(ws)
 
+      webSocketHeartbeat.set(ws.url, new Date())
       resolve()
     })
     ws.on('message', function handleMessage(message: string) {
+      webSocketHeartbeat.set(ws.url, new Date())
       let data
       try {
         data = JSON.parse(message)
@@ -101,6 +106,7 @@ async function setHandlers(
       )
     })
     ws.on('close', async (code) => {
+      webSocketHeartbeat.delete(ws.url)
       void updateConnectionHealthStatus(ws.url, false)
       ws.terminate()
 
@@ -120,13 +126,31 @@ async function setHandlers(
       resolve()
     })
     ws.on('error', () => {
+      webSocketHeartbeat.delete(ws.url)
       void updateConnectionHealthStatus(ws.url, false)
       ws.terminate()
       resolve()
     })
   })
 }
 
+setInterval(() => {
+  log.info('WS INFO: Total number of live WS connections: ' + webSocketHeartbeat.size)
+  for (const [url, heartbeat] of webSocketHeartbeat.entries()) {
+    log.info('WS INFO: heartbeat for: ' + url + ' last received at: ' + (Date.now() - heartbeat.getTime()) + 'ms ago')
+  }
+
+  let staleConnections = 0
+  // report the stale connections whose heartbeat is older than 60 seconds (missing 15 validated ledgers)
+  for (const [url, heartbeat] of webSocketHeartbeat.entries()) {
+    if (Date.now() - heartbeat.getTime() > 60 * 1000) {
+      log.error('WS ERROR: stale connection for: ' + url + ' last received at: ' + heartbeat.toISOString() + '. (' + (Date.now() - heartbeat.getTime())/1000 + 'seconds ago)')
+      staleConnections++
+    }
+  }
+  log.error('WS ERROR: Total number of stale connections: ' + staleConnections)
+}, WS_HEARTBEAT_INTERVAL)
+
 /**
  * Tries to find a valid WebSockets endpoint for a node.
  *
@@ -198,6 +222,8 @@ async function createConnections(): Promise<void> {
 
   const promises: Array<Promise<void>> = []
 
+  log.info(`Checking/Initiating connections to the following nodes: ${nodes.map((node) => node.ip + ' | ' + node.ws_url + ' | ' + node.networks + ' | ' + node.public_key).join(', ')}`)
+
   nodes.forEach((node: WsNode) => {
     promises.push(findConnection(node))
   })

src/connection-manager/manifests.ts

@@ -32,7 +32,10 @@ let jobsStarted = false
  * @param networkName - The name of the network.
  * @returns The first UNL in the list of UNLs for the network.
  */
-async function getFirstUNL(networkName: string): Promise<string> {
+export async function getFirstUNL(networkName: string): Promise<string> {
+  if (!networkName) {
+    return Promise.reject(new Error('Network name is empty'))
+  }
   const networks = await getNetworks()
   const network = networks.filter((ntwk) => ntwk.id === networkName)[0]
   return network.unls[0]
@@ -96,9 +99,13 @@ export async function updateUNLManifests(): Promise<void> {
  * @param network - The network to update.
  * @returns A promise that resolves to void once all UNL validators are saved.
  */
-async function updateUNLManifestNetwork(network: string): Promise<void> {
+export async function updateUNLManifestNetwork(network: string): Promise<void> {
+  // TODO: In an ideal scenario, the networks table should not contain undefined network.id rows
+  if (!network) {
+    return Promise.reject(new Error('Network string should not be empty'))
+  }
   try {
-    log.info('Fetching UNL...')
+    log.info('Fetching Manifests for network: ' + network + '. Identified UNL: ' + await getFirstUNL(network))
     const unl: UNLBlob = await fetchValidatorList(await getFirstUNL(network))
     const promises: Array<Promise<void>> = []
 

src/shared/database/index.ts

@@ -91,26 +91,43 @@ export async function getNodes(sinceStartDate: Date): Promise<WsNode[]> {
  * @param manifest -- Incoming manifest.
  * @returns The original manifest with the revoked column updated.
  */
-async function handleRevocations(
+export async function handleRevocations(
   manifest: DatabaseManifest,
 ): Promise<DatabaseManifest> {
   // Mark all older manifests as revoked
-  const revokedSigningKeys = (await query('manifests')
-    .where({ master_key: manifest.master_key })
-    .andWhere('seq', '<', manifest.seq)
-    .update({ revoked: true }, ['manifests.signing_key'])
-    .catch((err: Error) =>
-      log.error('Error revoking older manifests', err),
-    )) as DatabaseManifest[]
+  let revokedSigningKeys = undefined
+  let numberOfAttempts = 0
+  while (numberOfAttempts < 3) {
+    try {
+      revokedSigningKeys = (await query('manifests')
+      .where({ master_key: manifest.master_key })
+      .andWhere('seq', '<', manifest.seq)
+      .update({ revoked: true }, ['manifests.signing_key'])
+      ) as DatabaseManifest[]
+      break
+
+    } catch (err: any) {
+
+      if (err.code === '40P01') {
+        log.error('Error revoking older manifests: Deadlock detected, retrying with Exponential Backoff')
+        await new Promise(resolve => setTimeout(resolve, Math.pow(2, numberOfAttempts) * 1000)); // Exponential backoff
+        numberOfAttempts++
+        continue
+      } else {
+        log.error('Error revoking older manifests', err)
+        break
+      }
+    }
+  }
 
   const revokedSigningKeysArray =
-    revokedSigningKeys.length > 0
-      ? await Promise.all(
-          revokedSigningKeys.map(async (obj) => {
-            return obj.signing_key
-          }),
-        )
-      : []
+    revokedSigningKeys && revokedSigningKeys.length > 0
+        ? await Promise.all(
+            revokedSigningKeys.map(async (obj) => {
+              return obj.signing_key
+            }),
+          )
+        : []
 
   // If there exists a newer manifest, mark this manifest as revoked
   const newer = (await query('manifests')
@@ -122,7 +139,7 @@ async function handleRevocations(
 
   const updated = { revoked: false, ...manifest }
 
-  if (newer.length !== 0) {
+  if (newer && newer.length !== 0) {
     updated.revoked = true
     revokedSigningKeysArray.push(manifest.signing_key)
   }
@@ -134,7 +151,7 @@ async function handleRevocations(
   // updates revocations in validators table
   await query('validators')
     .whereIn('signing_key', revokedSigningKeysCleaned)
-    .update({ revoked: true })
+    .update({ revoked: true }).catch((err) => log.error('Error updating revoked manifest in validators table', err))
 
   return updated
 }