Merge pull request #1400 from edenia/fix/max-connections-reached

Fix/max connections reached

hapi-evm/src/services/block.service.ts

@@ -30,10 +30,6 @@ const syncFullBlock = async (blockNumber: number | bigint) => {
     throw new Error('Wrong block format')
   }
 
-  const blockExist = await blockModel.queries.exist(block.hash.toString())
-
-  if (blockExist) return
-
   const blockTimestamp = new Date(Number(block.timestamp) * 1000)
   const isBefore = moment(blockTimestamp).isBefore(
     moment().subtract(networkConfig.keepHistoryForYears, 'years')
@@ -119,20 +115,28 @@ const incrementTotalTransactions = async (transactionsCount: number) => {
   }
 }
 
-const getBlock = async () => {
-  let blockNumber: bigint
+const getLastBlockInDB = async () => {
   const lastBlockInDB = (await blockModel.queries.default.get(
     { timestamp: { _gt: moment().subtract(30, 'minutes') } },
     { number: 'desc' }
   )) as blockModel.interfaces.CappedBlock
 
+  return lastBlockInDB?.number || 0
+}
+
+const getBlock = async (lastInserted: number | null) => {
+  let blockNumber: bigint
+  const lastBlockInDB = lastInserted || (await getLastBlockInDB())
+
   if (!lastBlockInDB) {
     blockNumber = await web3.eth.getBlockNumber()
   } else {
-    blockNumber = BigInt(lastBlockInDB.number + 1)
+    blockNumber = BigInt(lastBlockInDB + 1)
   }
 
   await syncFullBlock(blockNumber)
+
+  return Number(blockNumber)
 }
 
 const syncOldBlocks = async (): Promise<void> => {
@@ -173,8 +177,9 @@ const syncOldBlocks = async (): Promise<void> => {
   )
 }
 
+let lastInserted: number | null = null
 const blockWorker = async () => {
-  getBlock()
+  lastInserted = await getBlock(lastInserted)
 }
 
 const cleanOldBlocks = async () => {

hapi-evm/src/services/partial-ath.service.ts

@@ -1,59 +1,71 @@
+import Web3 from 'web3'
+
 import { Stats } from '../models/stats/interfaces'
-import { sequelizeUtil } from '../utils'
+import { networkConfig } from '../config'
 
 export const getATHInRange = async (
   lowerBlockNumber: number,
   upperBlockNumber: number
 ) => {
-  const [rows] = await sequelizeUtil.sequelize.query(`
-  SELECT
-    ath_in_range.blocks AS ath_blocks,
-    COALESCE(
-        (ath_in_range.max_transaction_sum) :: numeric,
-        (0) :: numeric
-    ) AS ath_transactions_count,
-    COALESCE(ath_in_range.gas_used_sum, (0) :: numeric) AS ath_gas_used
-    FROM
-    (
-      WITH subquery AS (
-        SELECT
-          array_to_string(array_agg(block.number), ',' :: text) AS blocks,
-          sum(jsonb_array_length(block.transactions)) AS total_transaction_count,
-          sum(block.gas_used) AS gas_used_sum
-        FROM
-          evm.block
-        WHERE (block.number >= ${lowerBlockNumber} and block.number <= ${upperBlockNumber})
-        GROUP BY
-          block."timestamp"
+  const httpProvider = new Web3.providers.HttpProvider(
+    networkConfig.evmEndpoint
+  )
+  const web3 = new Web3(httpProvider)
+
+  const rangeOfBlocks = Array.from(
+    { length: upperBlockNumber - lowerBlockNumber },
+    (_, index) => index + lowerBlockNumber
+  )
+
+  const blocks = await Promise.allSettled(
+    rangeOfBlocks.map(async blockNumber => await web3.eth.getBlock(blockNumber))
+  )
+
+  const blocksPerSecond = blocks.reduce(
+    (status: { [timestamp: number]: Stats }, blockPromise) => {
+      if (blockPromise.status !== 'fulfilled') return status
+
+      const block = blockPromise.value
+      const timestamp = Number(block.timestamp)
+      const stats = {
+        ath_blocks: Number(block.number).toString(),
+        ath_transactions_count: block.transactions?.length || 0,
+        ath_gas_used: Number(block.gasUsed) || 0
+      }
+      if (!status[timestamp]) {
+        return { ...status, [timestamp]: stats }
+      } else {
+        return {
+          ...status,
+          [timestamp]: {
+            ath_blocks: status[timestamp].ath_blocks + ',' + stats.ath_blocks,
+            ath_transactions_count:
+              status[timestamp].ath_transactions_count +
+              stats.ath_transactions_count,
+            ath_gas_used: status[timestamp].ath_gas_used + stats.ath_gas_used
+          }
+        }
+      }
+    },
+    {}
+  )
+
+  const partialATH = Object.keys(blocksPerSecond || {}).reduce(
+    (max: Stats, current: string): Stats => {
+      if (
+        max.ath_transactions_count >
+        blocksPerSecond[Number(current)].ath_transactions_count
       )
-      SELECT
-        q2.blocks,
-        q1.max_transaction_sum,
-        q2.gas_used_sum
-      FROM
-        (
-          (
-            SELECT
-              max(subquery.total_transaction_count) AS max_transaction_sum
-            FROM
-              subquery
-          ) q1
-          JOIN subquery q2 ON (
-            (
-              q1.max_transaction_sum = q2.total_transaction_count
-            )
-          )
-        )
-      LIMIT
-        1
-    ) ath_in_range
-  `)
-
-  const row = rows[0] as Stats
-
-  return {
-    ath_blocks: row?.ath_blocks,
-    ath_transactions_count: Number(row?.ath_transactions_count) || 0,
-    ath_gas_used: Number(row?.ath_gas_used) || 0
-  } as Stats
+        return max
+
+      return blocksPerSecond[Number(current)]
+    },
+    {
+      ath_blocks: '',
+      ath_transactions_count: 0,
+      ath_gas_used: 0
+    }
+  )
+
+  return partialATH as Stats
 }

hapi-evm/src/services/worker/index.ts

@@ -8,7 +8,7 @@ const run = async (worker: defaultModel.Worker) => {
   try {
     await worker.action()
   } catch (error: any) {
-    console.log(`${worker.name} ERROR =>`, error.message)
+    console.log(`${worker.name} ERROR =>`, error?.message || error)
   }
 
   if (!worker.intervalSec) {

hapi-evm/src/utils/sequelize.util.ts

@@ -6,5 +6,11 @@ export const sequelize = new Sequelize(hasuraConfig.databaseURL, {
   dialectOptions: {
     connectTimeout: 60000000
   },
+  pool: {
+    max: 2,
+    min: 0,
+    acquire: 30000,
+    idle: 10000
+  },
   logging: false
 })

hapi/src/services/producer.service.js

@@ -112,18 +112,41 @@ const saveEstimateNextUpdate = async lastUpdateAt => {
 }
 
 const getProducersSummary = async () => {
-  const [rows] = await sequelizeUtil.query(`
-    SELECT 
-        bp_json->>'type' as type, 
-        count(*)::integer as entities_count,
-        STRING_AGG (owner, ',') as entities
-    FROM producer
-    GROUP BY 
-        bp_json->>'type'
-    ;
-`)
-
-  return rows
+  if (eosConfig.networkName === eosConfig.knownNetworks.lacchain) {
+    const [rows] = await sequelizeUtil.query(`
+          SELECT 
+              bp_json->>'type' as type, 
+              count(*)::integer as entities_count,
+              STRING_AGG (owner, ',') as entities
+          FROM producer
+          GROUP BY 
+              bp_json->>'type'
+          ;
+      `)
+
+    return rows
+  } 
+
+  const query = `
+    {
+      producer_aggregate {
+        aggregate {
+          count
+        }
+      }
+    }
+  `
+
+  const { 
+    producer_aggregate: { aggregate: { count } } 
+  } = await hasuraUtil.request(query)
+
+  return [
+    {
+      type: null,
+      entities_count: count
+    }
+  ]
 }
 
 const syncNodes = async producers => {

hapi/src/utils/sequelize.util.js

@@ -6,6 +6,12 @@ const sequelize = new Sequelize(hasuraConfig.databaseUrl, {
   dialectOptions: {
     connectTimeout: 60000000
   },
+  pool: {
+    max: 5,
+    min: 0,
+    acquire: 30000,
+    idle: 10000
+  },
   logging: false
 })