Merge pull request #533 from hop-protocol/feat/latency-optimization

Feat/latency optimization

Author: shanefontaine

packages/hop-node/src/cli/bondWithdrawal.ts

@@ -64,7 +64,6 @@ async function main (source: any) {
 
       const txParams: SendBondWithdrawalTxParams = {
         transferId: dbTransfer.transferId,
-        sender: dbTransfer.sender!,
         recipient: dbTransfer.recipient!,
         amount: dbTransfer.amount!,
         transferNonce: dbTransfer.transferNonce!,

packages/hop-node/src/cli/dbDump.ts

@@ -79,7 +79,7 @@ async function main (source: any) {
         output.push(item || { transferRootId })
       }
       const filtered = output.map((x: any) => {
-        const { transferRootId, transferRootHash, totalAmount, bonded, comitted, committedAt, confirmed, rootSetTimestamp, allSettled } = x
+        const { transferRootId, transferRootHash, totalAmount, bonded, comitted, committedAt, confirmed, rootSetTimestamp } = x
         return {
           transferRootId,
           transferRootHash,
@@ -88,8 +88,7 @@ async function main (source: any) {
           comitted,
           committedAt,
           confirmed,
-          rootSetTimestamp,
-          allSettled
+          rootSetTimestamp
         }
       })
       items = filtered

packages/hop-node/src/cli/settle.ts

@@ -12,7 +12,6 @@ root
   .option('--source-chain <slug>', 'Source chain', parseString)
   .option('--token <slug>', 'Token', parseString)
   .option('--transfer-root-hash <id>', 'Transfer root hash', parseString)
-  .option('--transfer-id <id>', 'Transfer ID', parseString)
   .option('--bonder <address>', 'Bonder address', parseString)
   .option('--use-db [boolean]', 'Use the DB to construct the roots', parseBool)
   .option(
@@ -23,15 +22,15 @@ root
   .action(actionHandler(main))
 
 async function main (source: any) {
-  let { sourceChain: chain, token, transferRootHash, transferId, bonder, useDb, dry: dryMode } = source
+  let { sourceChain: chain, token, transferRootHash, bonder, useDb, dry: dryMode } = source
   if (!chain) {
     throw new Error('source chain is required')
   }
   if (!token) {
     throw new Error('token is required')
   }
-  if (!(transferRootHash || transferId)) {
-    throw new Error('transferRootHash or transferId is required')
+  if (!transferRootHash) {
+    throw new Error('transferRootHash is required')
   }
 
   if (typeof useDb === 'undefined') {
@@ -44,11 +43,7 @@ async function main (source: any) {
   }
 
   if (useDb) {
-    if (transferRootHash) {
-      await watcher.checkTransferRootHash(transferRootHash, bonder)
-    } else {
-      await watcher.checkTransferId(transferId)
-    }
+    await watcher.checkTransferRootHash(transferRootHash, bonder)
   } else {
     const transferRoot: any = await getTransferRoot(chain, token, transferRootHash)
     const destinationChainId: number = transferRoot.destinationChainId

packages/hop-node/src/db/BaseDb.ts

@@ -9,6 +9,7 @@ import spread from 'lodash/spread'
 import sub from 'subleveldown'
 import wait from 'src/utils/wait'
 import { EventEmitter } from 'events'
+import { Migration } from 'src/db/migrations'
 import { Mutex } from 'async-mutex'
 import { TenSecondsMs } from 'src/constants'
 import { config as globalConfig } from 'src/config'
@@ -36,6 +37,10 @@ type QueueItem = {
   cb: any
 }
 
+type MigrationIndexEntry = {
+  index: number
+}
+
 // this are options that leveldb createReadStream accepts
 export type KeyFilter = {
   gt?: string
@@ -59,8 +64,9 @@ class BaseDb extends EventEmitter {
   batchSize: number = 10
   batchTimeLimit: number = 2 * 1000
   batchQueue: QueueItem[] = []
+  private readonly dbMigrationKey: string = '_dbMigrationIndex'
 
-  constructor (prefix: string, _namespace?: string) {
+  constructor (prefix: string, _namespace?: string, migrations?: Migration[]) {
     super()
     if (!prefix) {
       throw new Error('db prefix is required')
@@ -121,7 +127,7 @@ class BaseDb extends EventEmitter {
       })
 
     this.pollBatchQueue()
-    this._migration()
+    this._migrate(migrations)
       .then(() => {
         this.ready = true
         this.logger.debug('db ready')
@@ -132,15 +138,36 @@ class BaseDb extends EventEmitter {
       })
   }
 
-  // To add a migration, implement the shouldMigrate and migration functions in the child class.
   // Migrations are memory intensive. Ensure there is no unintentional memory overflow.
   // * Use stream instead of storing all entries at once
   // * Bypass the mutex
-  async _migration (): Promise<void> {
-    // Check if migration is needed
-    if (!this.shouldMigrate()) return
+  // This may take minutes or hours to complete
+  private async _migrate (migrations?: Migration[]): Promise<void> {
+    if (!migrations?.length) {
+      this.logger.debug('no migrations to process')
+      return
+    }
+
+    const currentMigrationIndex = await this.getMigrationIndex() ?? 0
+    const lastMigrationIndex = migrations.length - 1
+    if (currentMigrationIndex > lastMigrationIndex) {
+      this.logger.debug(`no migration required, currentMigrationIndex: ${currentMigrationIndex}`)
+      return
+    }
+
+    this.logger.debug(`processing migrations from ${currentMigrationIndex} to ${lastMigrationIndex}`)
+    for (let i = currentMigrationIndex; i <= lastMigrationIndex; i++) {
+      const migration: Migration = migrations[i]
+      this.logger.debug(`processing migration ${i}`)
+      await this._processMigration(migration)
+      await this.putMigrationIndex(i + 1)
+      this.logger.debug(`completed migration ${i}`)
+    }
+    this.logger.debug('migrations complete')
+  }
 
-    // Perform migration
+  private async _processMigration (migration: Migration): Promise<void> {
+    let migrationCount = 0
     return await new Promise((resolve, reject) => {
       const s = this.db.createReadStream({})
       s.on('data', async (key: any, value: any) => {
@@ -154,15 +181,15 @@ class BaseDb extends EventEmitter {
           return
         }
 
-        // Call the child class migration function
         try {
-          await this.migration(key, value)
+          await this._migrateEntry(migration, key, value)
+          migrationCount++
         } catch (err) {
           s.emit('error', err)
         }
       })
         .on('end', () => {
-          this.logger.debug('DB migration complete')
+          this.logger.debug(`DB migration complete. migrated ${migrationCount} entries`)
           s.destroy()
           resolve()
         })
@@ -174,15 +201,29 @@ class BaseDb extends EventEmitter {
     })
   }
 
-  shouldMigrate (): boolean {
-    // Optional
-    // Must return true in child class to perform migration
-    return false
+  // Ensure that these DB reads and writes do not use the mutex since a migration is memory-intensive
+  private async _migrateEntry (migration: Migration, key: string, value: any): Promise<void> {
+    const { key: migrationKey, value: migrationValue, migratedValue } = migration
+    if (
+      value?.[migrationKey] === undefined ||
+      value?.[migrationKey] === migrationValue
+    ) {
+      const { value: updatedValue } = await this._getUpdateData(key, value)
+      updatedValue[migrationKey] = migratedValue
+      return this.db.put(key, updatedValue)
+    }
+  }
+
+  private async getMigrationIndex (): Promise<number | undefined> {
+    const migrationEntry: MigrationIndexEntry = await this.getById(this.dbMigrationKey)
+    return migrationEntry?.index
   }
 
-  async migration (key: string, value: any): Promise<void> {
-    // Optional
-    // Must be implemented in child class to perform migration
+  private async putMigrationIndex (updatedMigrationIndex: number): Promise<void> {
+    const data: MigrationIndexEntry = {
+      index: updatedMigrationIndex
+    }
+    return this._updateSingle(this.dbMigrationKey, data)
   }
 
   protected async tilReady (): Promise<boolean> {