BLUE-209 Db perf

archiver-config.json

@@ -57,6 +57,5 @@
     }
   ],
   "ARCHIVER_MODE": "release",
-  "DevPublicKey": "",
-  "EXISTING_ARCHIVER_DB_PATH": ""
+  "DevPublicKey": ""
 }

debug_mode.patch

@@ -8,8 +8,7 @@ index a892baf..ffac25a 100644
    ],
 -  "ARCHIVER_MODE": "release",
 +  "ARCHIVER_MODE": "debug",
-   "DevPublicKey": "",
-   "EXISTING_ARCHIVER_DB_PATH": ""
+   "DevPublicKey": ""
  }
 \ No newline at end of file
 diff --git a/src/Config.ts b/src/Config.ts

src/Config.ts

@@ -20,7 +20,6 @@ export interface Config {
     receiptDB: string
     originalTxDataDB: string
   }
-  EXISTING_ARCHIVER_DB_PATH: string
   DATASENDER_TIMEOUT: number
   RATE_LIMIT: number // number of allowed request per second,
   N_NODE_REJECT_PERCENT: number
@@ -104,7 +103,6 @@ let config: Config = {
     receiptDB: 'receipts.sqlite3',
     originalTxDataDB: 'originalTxsData.sqlite3',
   },
-  EXISTING_ARCHIVER_DB_PATH: '',
   DATASENDER_TIMEOUT: 1000 * 60 * 5,
   RATE_LIMIT: 100, // 100 req per second,
   N_NODE_REJECT_PERCENT: 5, // Percentage of old nodes to remove from nodelist

src/Data/Collector.ts

@@ -99,7 +99,7 @@
     0,
     5 * config.RECEIPT_CONFIRMATIONS
   )
   const isReceiptEqual = (receipt1: any, receipt2: any): boolean => {
     if (!receipt1 || !receipt2) return false
     const r1 = Utils.deepCopy(receipt1)
     const r2 = Utils.deepCopy(receipt2)
@@ -124,7 +124,7 @@
     true
   )
   if (config.VERBOSE) Logger.mainLogger.debug('robustQuery', receipt.tx.txId, robustQuery)
   if (!robustQuery || !robustQuery.value || !(robustQuery.value as any).receipt) {
     Logger.mainLogger.error(
       `❌ 'null' response from all nodes in receipt-validation for txId: ${receipt.tx.txId} , ${receipt.cycle}, ${receipt.tx.timestamp}
       }`
@@ -134,7 +134,7 @@
     return result
   }
 
   const robustQueryReceipt = (robustQuery.value as any).receipt as Receipt.SignedReceipt
 
   if (robustQuery.count < minConfirmations) {
     // Wait for 500ms and try fetching the receipt from the nodes that did not respond in the robustQuery
@@ -157,7 +157,7 @@
       }
       const node = nodesToQuery[0]
       nodesToQuery.splice(0, 1)
       const receiptResult: any = await queryReceipt(node)
       if (!receiptResult || !receiptResult.receipt) continue
       if (isReceiptEqual(robustQueryReceipt, receiptResult.receipt)) {
         requiredConfirmations--
@@ -790,7 +790,7 @@
 
     if (verifyData) {
       if (config.usePOQo === false) {
         const existingReceipt = await Receipt.queryReceiptByReceiptId(txId)
         // if (
         //   existingReceipt &&
         //   receipt.appliedReceipt &&
@@ -1380,6 +1380,8 @@
     `Collecting missing receipt for txId ${txId} with timestamp ${txTimestamp} from archivers`,
     senderArchivers.map((a) => a.ip + ':' + a.port)
   )
+  if (nestedCountersInstance) nestedCountersInstance.countEvent('receipt', 'Collect_missing_receipt')
+  if (profilerInstance) profilerInstance.profileSectionStart('Collect_missing_receipt')
   for (const senderArchiver of senderArchivers) {
     if (
       (processedReceiptsMap.has(txId) && processedReceiptsMap.get(txId) === txTimestamp) ||
@@ -1405,11 +1407,14 @@
     if (foundTxData) break
   }
   if (!foundTxData) {
+    if (nestedCountersInstance)
+      nestedCountersInstance.countEvent('receipt', 'Failed to collect missing receipt from archivers')
     Logger.mainLogger.error(
       `Failed to collect receipt for txId ${txId} with timestamp ${txTimestamp} from archivers ${senders}`
     )
   }
-  collectingMissingOriginalTxsMap.delete(txId)
+  collectingMissingReceiptsMap.delete(txId)
+  if (profilerInstance) profilerInstance.profileSectionEnd('Collect_missing_receipt')
 }
 
 const collectMissingOriginalTxsData = async (
@@ -1424,6 +1429,9 @@
     `Collecting missing originalTxData for txId ${txId} with timestamp ${txTimestamp} from archivers`,
     senderArchivers.map((a) => a.ip + ':' + a.port)
   )
+  if (nestedCountersInstance)
+    nestedCountersInstance.countEvent('originalTxData', 'Collect_missing_originalTxData')
+  if (profilerInstance) profilerInstance.profileSectionStart('Collect_missing_originalTxData')
   for (const senderArchiver of senderArchivers) {
     if (
       (processedOriginalTxsMap.has(txId) && processedOriginalTxsMap.get(txId) === txTimestamp) ||
@@ -1447,11 +1455,14 @@
     if (foundTxData) break
   }
   if (!foundTxData) {
+    if (nestedCountersInstance)
+      nestedCountersInstance.countEvent('originalTxData', 'Failed to collect_missing_originalTxData')
     Logger.mainLogger.error(
       `Failed to collect originalTxData for txId ${txId} with timestamp ${txTimestamp} from archivers ${senders}`
     )
   }
-  collectingMissingReceiptsMap.delete(txId)
+  collectingMissingOriginalTxsMap.delete(txId)
+  if (profilerInstance) profilerInstance.profileSectionEnd('Collect_missing_originalTxData')
 }
 
 type TxDataFromArchiversResponse = {

src/Data/Data.ts

@@ -5,7 +5,7 @@
 import * as Cycles from './Cycles'
 import {
   getCurrentCycleCounter,
   currentCycleDuration,
   processCycles,
   validateCycle,
   validateCycleData,
@@ -525,7 +525,7 @@
   dataSenders.set(sender.nodeInfo.publicKey, sender)
 }
 
 async function syncFromNetworkConfig(): Promise<any> {
   try {
     // Define the query function to get the network config from a node
     const queryFn = async (node): Promise<object> => {
@@ -576,7 +576,7 @@
       //   typeof devPublicKey === typeof config.DevPublicKey &&
       //   devPublicKey !== config.DevPublicKey
       // )
-      //   updateConfig({ DevPublicKey: devPublicKey })
+      //   updateConfig({ DevPublicKey: devPublicKey })count query)
       if (
         !Utils.isUndefined(newPOQReceipt) &&
         typeof newPOQReceipt === typeof config.newPOQReceipt &&
@@ -1505,7 +1505,7 @@
       }
     } else {
       if (totalSavedReceiptsCount >= totalReceipts) {
         const res: any = await getTotalDataFromArchivers()
         if (res && res.totalReceipts > 0) {
           if (res.totalReceipts > totalReceipts) totalReceipts = res.totalReceipts
           if (res.totalCycles > totalCycles) totalCycles = res.totalCycles

src/Data/GossipData.ts

@@ -6,6 +6,7 @@ import { Signature } from '@shardus/crypto-utils'
 import { P2P as P2PTypes } from '@shardus/types'
 import * as Utils from '../Utils'
 import { config } from '../Config'
+import { nestedCountersInstance } from '../profiler/nestedCounters'
 
 // adjacentArchivers are one archiver from left and one archiver from right of the current archiver
 export let adjacentArchivers: State.ArchiverNodeInfo[] = []
@@ -102,12 +103,23 @@ export async function sendDataToAdjacentArchivers(
       }
     }
     try {
-      await Promise.allSettled(promises)
+      await Promise.allSettled(promises).then((results) => {
+        results.forEach((result) => {
+          if (nestedCountersInstance) {
+            if (result.status === 'fulfilled') {
+              if (result.value !== null) nestedCountersInstance.countEvent('gossip-data', 'success')
+              else nestedCountersInstance.countEvent('gossip-data', 'failure')
+            } else nestedCountersInstance.countEvent('gossip-data', 'failure')
+          }
+        })
+      })
     } catch (err) {
       Logger.mainLogger.error('Gossip Error: ' + err)
+      if (nestedCountersInstance) nestedCountersInstance.countEvent('gossip-data', 'error 1', err)
     }
   } catch (ex) {
     Logger.mainLogger.debug(ex)
     Logger.mainLogger.debug('Fail to gossip')
+    if (nestedCountersInstance) nestedCountersInstance.countEvent('gossip-data', 'error 2', ex)
   }
 }

src/dbstore/accounts.ts

@@ -1,5 +1,5 @@
 import * as db from './sqlite3storage'
-import { accountDatabase, extractValues, extractValuesFromArray } from './sqlite3storage'
+import { accountDatabase } from '.'
 import * as Logger from '../Logger'
 import { config } from '../Config'
 import { DeSerializeFromJsonString, SerializeToJsonString } from '../utils/serialization'
@@ -22,7 +22,7 @@ export async function insertAccount(account: AccountsCopy): Promise<void> {
   try {
     const fields = Object.keys(account).join(', ')
     const placeholders = Object.keys(account).fill('?').join(', ')
-    const values = extractValues(account)
+    const values = db.extractValues(account)
     const sql = 'INSERT OR REPLACE INTO accounts (' + fields + ') VALUES (' + placeholders + ')'
     await db.run(accountDatabase, sql, values)
     if (config.VERBOSE) {
@@ -41,7 +41,7 @@ export async function bulkInsertAccounts(accounts: AccountsCopy[]): Promise<void
   try {
     const fields = Object.keys(accounts[0]).join(', ')
     const placeholders = Object.keys(accounts[0]).fill('?').join(', ')
-    const values = extractValuesFromArray(accounts)
+    const values = db.extractValuesFromArray(accounts)
     let sql = 'INSERT OR REPLACE INTO accounts (' + fields + ') VALUES (' + placeholders + ')'
     for (let i = 1; i < accounts.length; i++) {
       sql = sql + ', (' + placeholders + ')'

src/dbstore/cycles.ts

@@ -1,5 +1,5 @@
 import * as db from './sqlite3storage'
-import { cycleDatabase, extractValues, extractValuesFromArray } from './sqlite3storage'
+import { cycleDatabase } from '.'
 import { P2P } from '@shardus/types'
 import * as Logger from '../Logger'
 import { config } from '../Config'
@@ -10,7 +10,7 @@ export async function insertCycle(cycle: Cycle): Promise<void> {
   try {
     const fields = Object.keys(cycle).join(', ')
     const placeholders = Object.keys(cycle).fill('?').join(', ')
-    const values = extractValues(cycle)
+    const values = db.extractValues(cycle)
     const sql = 'INSERT OR REPLACE INTO cycles (' + fields + ') VALUES (' + placeholders + ')'
     await db.run(cycleDatabase, sql, values)
     Logger.mainLogger.debug('Successfully inserted Cycle', cycle.cycleRecord.counter, cycle.cycleMarker)
@@ -28,7 +28,7 @@ export async function bulkInsertCycles(cycles: Cycle[]): Promise<void> {
   try {
     const fields = Object.keys(cycles[0]).join(', ')
     const placeholders = Object.keys(cycles[0]).fill('?').join(', ')
-    const values = extractValuesFromArray(cycles)
+    const values = db.extractValuesFromArray(cycles)
     let sql = 'INSERT OR REPLACE INTO cycles (' + fields + ') VALUES (' + placeholders + ')'
     for (let i = 1; i < cycles.length; i++) {
       sql = sql + ', (' + placeholders + ')'

src/dbstore/index.ts

@@ -1,24 +1,39 @@
+import * as fs from 'fs'
+import * as path from 'path'
+import { Database } from 'sqlite3'
 import { Config } from '../Config'
-import {
-  init,
-  runCreate,
-  close,
-  accountDatabase,
-  transactionDatabase,
-  cycleDatabase,
-  receiptDatabase,
-  originalTxDataDatabase,
-} from './sqlite3storage'
+import { createDB, runCreate, close } from './sqlite3storage'
+
+export let cycleDatabase: Database
+export let accountDatabase: Database
+export let transactionDatabase: Database
+export let receiptDatabase: Database
+export let originalTxDataDatabase: Database
 
 export const initializeDB = async (config: Config): Promise<void> => {
-  await init(config)
+  createDirectories(config.ARCHIVER_DB)
+  accountDatabase = await createDB(`${config.ARCHIVER_DB}/${config.ARCHIVER_DATA.accountDB}`, 'Account')
+  cycleDatabase = await createDB(`${config.ARCHIVER_DB}/${config.ARCHIVER_DATA.cycleDB}`, 'Cycle')
+  transactionDatabase = await createDB(
+    `${config.ARCHIVER_DB}/${config.ARCHIVER_DATA.transactionDB}`,
+    'Transaction'
+  )
+  receiptDatabase = await createDB(`${config.ARCHIVER_DB}/${config.ARCHIVER_DATA.receiptDB}`, 'Receipt')
+  originalTxDataDatabase = await createDB(
+    `${config.ARCHIVER_DB}/${config.ARCHIVER_DATA.originalTxDataDB}`,
+    'OriginalTxData'
+  )
   await runCreate(
     transactionDatabase,
     'CREATE TABLE if not exists `transactions` (`txId` TEXT NOT NULL UNIQUE PRIMARY KEY, `appReceiptId` TEXT, `timestamp` BIGINT NOT NULL, `cycleNumber` NUMBER NOT NULL, `data` JSON NOT NULL, `originalTxData` JSON NOT NULL)'
   )
   await runCreate(
     transactionDatabase,
-    'CREATE INDEX if not exists `transactions_idx` ON `transactions` (`cycleNumber` DESC, `timestamp` DESC)'
+    'CREATE INDEX if not exists `transactions_timestamp` ON `transactions` (`timestamp` ASC)'
+  )
+  await runCreate(
+    transactionDatabase,
+    'CREATE INDEX if not exists `transactions_cycleNumber_timestamp` ON `transactions` (`cycleNumber` ASC, `timestamp` ASC)'
   )
   await runCreate(
     transactionDatabase,
@@ -35,27 +50,48 @@
   )
   await runCreate(
     accountDatabase,
-    'CREATE INDEX if not exists `accounts_idx` ON `accounts` (`cycleNumber` DESC, `timestamp` DESC)'
+    'CREATE INDEX if not exists `accounts_cycleNumber` ON `accounts` (`cycleNumber` ASC)'
+  )
+  await runCreate(
+    accountDatabase,
+    'CREATE INDEX if not exists `accounts_timestamp` ON `accounts` (`timestamp` ASC)'
+  )
+  await runCreate(
+    accountDatabase,
+    'CREATE INDEX if not exists `accounts_cycleNumber_timestamp` ON `accounts` (`cycleNumber` ASC, `timestamp` ASC)'
   )
   await runCreate(
     receiptDatabase,
     'CREATE TABLE if not exists `receipts` (`receiptId` TEXT NOT NULL UNIQUE PRIMARY KEY, `tx` JSON NOT NULL, `cycle` NUMBER NOT NULL, `applyTimestamp` BIGINT NOT NULL, `timestamp` BIGINT NOT NULL, `signedReceipt` JSON NOT NULL, `afterStates` JSON, `beforeStates` JSON, `appReceiptData` JSON, `executionShardKey` TEXT NOT NULL, `globalModification` BOOLEAN NOT NULL)'
   )
+  await runCreate(receiptDatabase, 'CREATE INDEX if not exists `receipts_cycle` ON `receipts` (`cycle` ASC)')
   await runCreate(
     receiptDatabase,
-    'CREATE INDEX if not exists `receipts_idx` ON `receipts` (`cycle` ASC, `timestamp` ASC)'
+    'CREATE INDEX if not exists `receipts_timestamp` ON `receipts` (`timestamp` ASC)'
+  )
+  await runCreate(
+    receiptDatabase,
+    'CREATE INDEX if not exists `receipts_cycle_timestamp` ON `receipts` (`cycle` ASC, `timestamp` ASC)'
   )
   await runCreate(
     originalTxDataDatabase,
     'CREATE TABLE if not exists `originalTxsData` (`txId` TEXT NOT NULL, `timestamp` BIGINT NOT NULL, `cycle` NUMBER NOT NULL, `originalTxData` JSON NOT NULL, PRIMARY KEY (`txId`, `timestamp`))'
   )
   await runCreate(
     originalTxDataDatabase,
-    'CREATE INDEX if not exists `originalTxsData_idx` ON `originalTxsData` (`cycle` ASC, `timestamp` ASC)'
+    'CREATE INDEX if not exists `originalTxsData_cycle` ON `originalTxsData` (`cycle` ASC)'
+  )
+  await runCreate(
+    originalTxDataDatabase,
+    'CREATE INDEX if not exists `originalTxsData_timestamp` ON `originalTxsData` (`timestamp` ASC)'
   )
   await runCreate(
     originalTxDataDatabase,
-    'CREATE INDEX if not exists `originalTxsData_txId_idx` ON `originalTxsData` (`txId`)'
+    'CREATE INDEX if not exists `originalTxsData_cycle_timestamp` ON `originalTxsData` (`cycle` ASC, `timestamp` ASC)'
+  )
+  await runCreate(
+    originalTxDataDatabase,
+    'CREATE INDEX if not exists `originalTxsData_txId` ON `originalTxsData` (`txId`)'
   )
 }
 
@@ -66,3 +102,9 @@
   await close(receiptDatabase, 'Receipt')
   await close(originalTxDataDatabase, 'OriginalTxData')
 }
+
+function createDirectories(pathname: string): void {
+  const __dirname = path.resolve()
+  pathname = pathname.replace(/^\.*\/|\/?[^/]+\.[a-z]+|\/$/g, '') // Remove leading directory markers, and remove ending /file-name.extension
+  fs.mkdirSync(path.resolve(__dirname, pathname), { recursive: true }) // eslint-disable-line security/detect-non-literal-fs-filename
+}

src/dbstore/originalTxsData.ts

@@ -1,6 +1,6 @@
 // import { Signature } from 'shardus-crypto-types'
 import * as db from './sqlite3storage'
-import { originalTxDataDatabase, extractValues, extractValuesFromArray } from './sqlite3storage'
+import { originalTxDataDatabase } from '.'
 import * as Logger from '../Logger'
 import { config } from '../Config'
 import { DeSerializeFromJsonString } from '../utils/serialization'
@@ -31,7 +31,7 @@ export async function insertOriginalTxData(OriginalTxData: OriginalTxData): Prom
   try {
     const fields = Object.keys(OriginalTxData).join(', ')
     const placeholders = Object.keys(OriginalTxData).fill('?').join(', ')
-    const values = extractValues(OriginalTxData)
+    const values = db.extractValues(OriginalTxData)
     const sql = 'INSERT OR REPLACE INTO originalTxsData (' + fields + ') VALUES (' + placeholders + ')'
     await db.run(originalTxDataDatabase, sql, values)
     if (config.VERBOSE) {
@@ -50,7 +50,7 @@ export async function bulkInsertOriginalTxsData(originalTxsData: OriginalTxData[
   try {
     const fields = Object.keys(originalTxsData[0]).join(', ')
     const placeholders = Object.keys(originalTxsData[0]).fill('?').join(', ')
-    const values = extractValuesFromArray(originalTxsData)
+    const values = db.extractValuesFromArray(originalTxsData)
     let sql = 'INSERT OR REPLACE INTO originalTxsData (' + fields + ') VALUES (' + placeholders + ')'
     for (let i = 1; i < originalTxsData.length; i++) {
       sql = sql + ', (' + placeholders + ')'

src/dbstore/receipts.ts

@@ -1,6 +1,6 @@
 import { Signature } from '@shardus/crypto-utils'
 import * as db from './sqlite3storage'
-import { receiptDatabase, extractValues, extractValuesFromArray } from './sqlite3storage'
+import { receiptDatabase } from '.'
 import * as Logger from '../Logger'
 import { config } from '../Config'
 import { DeSerializeFromJsonString } from '../utils/serialization'
@@ -109,7 +109,7 @@ export async function insertReceipt(receipt: Receipt): Promise<void> {
   try {
     const fields = Object.keys(receipt).join(', ')
     const placeholders = Object.keys(receipt).fill('?').join(', ')
-    const values = extractValues(receipt)
+    const values = db.extractValues(receipt)
     const sql = 'INSERT OR REPLACE INTO receipts (' + fields + ') VALUES (' + placeholders + ')'
     await db.run(receiptDatabase, sql, values)
     if (config.VERBOSE) {
@@ -128,7 +128,7 @@ export async function bulkInsertReceipts(receipts: Receipt[]): Promise<void> {
   try {
     const fields = Object.keys(receipts[0]).join(', ')
     const placeholders = Object.keys(receipts[0]).fill('?').join(', ')
-    const values = extractValuesFromArray(receipts)
+    const values = db.extractValuesFromArray(receipts)
     let sql = 'INSERT OR REPLACE INTO receipts (' + fields + ') VALUES (' + placeholders + ')'
     for (let i = 1; i < receipts.length; i++) {
       sql = sql + ', (' + placeholders + ')'

src/dbstore/sqlite3storage.ts

@@ -1,32 +1,7 @@
-import * as fs from 'fs'
-import * as path from 'path'
-import { Config } from '../Config'
 import { SerializeToJsonString } from '../utils/serialization'
 import { Database } from 'sqlite3'
-// eslint-disable-next-line @typescript-eslint/no-var-requires
 
-export let cycleDatabase: Database
-export let accountDatabase: Database
-export let transactionDatabase: Database
-export let receiptDatabase: Database
-export let originalTxDataDatabase: Database
-
-export async function init(config: Config): Promise<void> {
-  createDirectories(config.ARCHIVER_DB)
-  accountDatabase = await createDB(`${config.ARCHIVER_DB}/${config.ARCHIVER_DATA.accountDB}`, 'Account')
-  cycleDatabase = await createDB(`${config.ARCHIVER_DB}/${config.ARCHIVER_DATA.cycleDB}`, 'Cycle')
-  transactionDatabase = await createDB(
-    `${config.ARCHIVER_DB}/${config.ARCHIVER_DATA.transactionDB}`,
-    'Transaction'
-  )
-  receiptDatabase = await createDB(`${config.ARCHIVER_DB}/${config.ARCHIVER_DATA.receiptDB}`, 'Receipt')
-  originalTxDataDatabase = await createDB(
-    `${config.ARCHIVER_DB}/${config.ARCHIVER_DATA.originalTxDataDB}`,
-    'OriginalTxData'
-  )
-}
-
-const createDB = async (dbPath: string, dbName: string): Promise<Database> => {
+export const createDB = async (dbPath: string, dbName: string): Promise<Database> => {
   console.log('dbName', dbName, 'dbPath', dbPath)
   const db = new Database(dbPath, (err) => {
     if (err) {
@@ -146,9 +121,3 @@ export function extractValuesFromArray(arr: object[]): unknown[] {
     return null
   }
 }
-
-function createDirectories(pathname: string): void {
-  const __dirname = path.resolve()
-  pathname = pathname.replace(/^\.*\/|\/?[^/]+\.[a-z]+|\/$/g, '') // Remove leading directory markers, and remove ending /file-name.extension
-  fs.mkdirSync(path.resolve(__dirname, pathname), { recursive: true }) // eslint-disable-line security/detect-non-literal-fs-filename
-}