feat: reorgs outside of confirmations range handling

.circleci/config.yml

@@ -41,6 +41,8 @@ jobs:
         type: string
     docker:
       - image: circleci/node:<< parameters.version >>
+    environment:
+      ALLOW_CONFIG_MUTATIONS: true
     steps:
       - checkout
       - run:

.tasegir.js

@@ -4,7 +4,7 @@ module.exports = {
   },
   depCheck: {
     ignore: [
-      'tasegir', 'reflect-metadata', '@types/*', 'sqlite3', '@oclif/*',
+      'cross-env', 'tasegir', 'reflect-metadata', '@types/*', 'sqlite3', '@oclif/*',
     ]
   }
 }

config/default.json5

@@ -1,6 +1,9 @@
 // For full syntax see /src/types.ts::Config interface
 {
-  db: 'sqlite:db.sqlite',
+  db: 'db.sqlite',
+
+  // DB back up config
+  dbBackUp: { blocks: 10, path: 'db-backups' },
 
   // CORS setting, please consult https://expressjs.com/en/resources/middleware/cors.html for more details
   cors: {

config/test.json5

@@ -1,5 +1,5 @@
 {
-  db: 'sqlite://db_test.sqlite',
+  db: 'db_test.sqlite',
   blockchain: {
     provider: 'ws://localhost:8545',
     waitBlockCountBeforeConfirmationRemoved: 10

package.json

@@ -44,7 +44,7 @@
     "types-check": "tasegir types-check",
     "lint": "tasegir lint",
     "release": "tasegir release",
-    "test": "tasegir test --target node",
+    "test": "cross-env ALLOW_CONFIG_MUTATIONS=true tasegir test --target node",
     "start": "tasegir run --watch ./src/index.ts"
   },
   "dependencies": {
@@ -113,6 +113,7 @@
     "@types/validator": "^13.0.0",
     "chai": "^4.2.0",
     "chai-as-promised": "^7.1.1",
+    "cross-env": "^7.0.2",
     "dirty-chai": "^2.0.1",
     "sinon": "^9.0.1",
     "sinon-chai": "^3.5.0",

src/app.ts

@@ -7,7 +7,7 @@ import feathers from '@feathersjs/feathers'
 import express from '@feathersjs/express'
 import socketio from '@feathersjs/socketio'
 
-import { Application, SupportedServices } from './definitions'
+import { Application, ServiceAddresses, SupportedServices } from './definitions'
 import { loggingFactory } from './logger'
 import sequelize from './sequelize'
 import blockchain from './blockchain'
@@ -20,6 +20,7 @@ import authentication from './services/authentication'
 import storage from './services/storage'
 import rates from './services/rates'
 import rns from './services/rns'
+import { REORG_OUT_OF_RANGE_EVENT_NAME } from './blockchain/events'
 
 const logger = loggingFactory()
 
@@ -29,7 +30,9 @@ export const services = {
   [SupportedServices.RNS]: rns
 }
 
-export async function appFactory (): Promise<Application> {
+export type AppOptions = { appResetCallBack: (...args: any) => void }
+
+export async function startApp (options: AppOptions): Promise<{ stop: () => void }> {
   const app: Application = express(feathers())
 
   logger.verbose('Current configuration: ', config)
@@ -58,13 +61,13 @@ export async function appFactory (): Promise<Application> {
   /**********************************************************/
   // Configure each services
 
-  const servicePromises: Promise<void>[] = []
+  const servicePromises: Promise<{ stop: () => void }>[] = []
   for (const service of Object.values(services)) {
     app.configure((app) => servicePromises.push(errorHandler(service.initialize, logger)(app)))
   }
 
   // Wait for services to initialize
-  await Promise.all(servicePromises)
+  const servicesInstances = await Promise.all(servicePromises)
 
   // Log errors in hooks
   app.hooks({
@@ -77,5 +80,29 @@ export async function appFactory (): Promise<Application> {
   app.use(express.notFound())
   app.use(express.errorHandler({ logger }))
 
-  return app
+  // Start server
+  const port = config.get('port')
+  const server = app.listen(port)
+
+  server.on('listening', () =>
+    logger.info(`Server started on port ${port}`)
+  )
+
+  process.on('unhandledRejection', err =>
+    logger.error(`Unhandled Rejection at: ${err}`)
+  )
+
+  // Subscribe for reorg event
+  const reorgService = app.service(ServiceAddresses.REORG_EMITTER)
+  reorgService.on(REORG_OUT_OF_RANGE_EVENT_NAME, () => {
+    // wait 5 seconds to be sure that reorg event received by connected services
+    setTimeout(() => options.appResetCallBack(), 5000)
+  })
+
+  return {
+    stop: () => {
+      server.close()
+      servicesInstances.forEach(service => service.stop())
+    }
+  }
 }

src/blockchain/index.ts

@@ -8,6 +8,7 @@ import { loggingFactory } from '../logger'
 import { NEW_BLOCK_EVENT_NAME, NewBlockEmitterService } from './new-block-emitters'
 import { getNewBlockEmitter } from './utils'
 import { waitForReadyApp } from '../utils'
+import { ReorgEmitterService } from './reorg-emitter'
 
 const logger = loggingFactory('blockchain')
 
@@ -20,6 +21,7 @@ export function ethFactory (): Eth {
 
 const CONFIRMATION_CHANNEL = 'confirmations'
 const NEW_BLOCK_EMITTER_CHANNEL = 'new-block'
+const REORG_CHANNEL = 'reorg'
 
 function channelSetup (app: Application): void {
   if (typeof app.channel !== 'function') {
@@ -29,9 +31,11 @@ function channelSetup (app: Application): void {
   app.on('connection', (connection: any) => {
     app.channel(CONFIRMATION_CHANNEL).join(connection)
     app.channel(NEW_BLOCK_EMITTER_CHANNEL).join(connection)
+    app.channel(REORG_CHANNEL).join(connection)
   })
   app.service(ServiceAddresses.CONFIRMATIONS).publish(() => app.channel(CONFIRMATION_CHANNEL))
   app.service(ServiceAddresses.NEW_BLOCK_EMITTER).publish(() => app.channel(NEW_BLOCK_EMITTER_CHANNEL))
+  app.service(ServiceAddresses.REORG_EMITTER).publish(() => app.channel(REORG_CHANNEL))
 }
 
 function subscribeAndEmitNewBlocks (app: Application): void {
@@ -55,6 +59,7 @@ export default async function (app: Application): Promise<void> {
   app.set('eth', eth)
   app.use(ServiceAddresses.CONFIRMATIONS, new ConfirmatorService(eth))
   app.use(ServiceAddresses.NEW_BLOCK_EMITTER, new NewBlockEmitterService())
+  app.use(ServiceAddresses.REORG_EMITTER, new ReorgEmitterService())
 
   subscribeAndEmitNewBlocks(app)
   channelSetup(app)

src/blockchain/reorg-emitter.ts

@@ -0,0 +1,45 @@
+import { ServiceMethods } from '@feathersjs/feathers'
+
+import { loggingFactory } from '../logger'
+import { REORG_OUT_OF_RANGE_EVENT_NAME } from './events'
+
+const logger = loggingFactory('blockchain:reorg-service')
+const DEFAULT_DEBOUNCE_TIME = 5000
+
+export class ReorgEmitterService implements Partial<ServiceMethods<any>> {
+  private readonly debounceTime: number
+  private reorgContract: string[] = []
+  private lastProcessedBlockNumber = 0
+  private timeoutStarted = false
+  emit?: Function
+  events: string[]
+
+  constructor (debounceTime?: number) {
+    this.debounceTime = debounceTime || DEFAULT_DEBOUNCE_TIME
+    this.events = [REORG_OUT_OF_RANGE_EVENT_NAME]
+  }
+
+  // eslint-disable-next-line require-await
+  async get (): Promise<void> {
+    return Promise.resolve()
+  }
+
+  emitReorg (lastProcessedBlockNumber: number, contractName: string): void {
+    if (!this.emit) {
+      throw new Error('ReorgEmitterService invalid setup. Missing \'emit\' function')
+    }
+
+    if (!this.timeoutStarted) {
+      setTimeout(() => {
+        logger.warn(`Reorg outside of confirmation range happens on block number ${lastProcessedBlockNumber} for [${this.reorgContract}] contracts`)
+        this.emit!(REORG_OUT_OF_RANGE_EVENT_NAME, { contracts: this.reorgContract, lastProcessedBlockNumber: this.lastProcessedBlockNumber })
+        this.reorgContract = []
+        this.lastProcessedBlockNumber = 0
+      }, this.debounceTime)
+      this.timeoutStarted = true
+    }
+
+    this.reorgContract = [...this.reorgContract, contractName]
+    this.lastProcessedBlockNumber = lastProcessedBlockNumber
+  }
+}

src/blockchain/utils.ts

@@ -1,11 +1,10 @@
 import { AbiItem, keccak256 } from 'web3-utils'
 import Eth from 'web3-eth'
-import { EventEmitter } from 'events'
 import config from 'config'
 import { getObject } from 'sequelize-store'
 
 import { loggingFactory } from '../logger'
-import eventsEmitterFactory, { EventsEmitterOptions } from './events'
+import eventsEmitterFactory, { BaseEventsEmitter, EventsEmitterOptions } from './events'
 import { NewBlockEmitterOptions } from '../definitions'
 import { BlockTracker } from './block-tracker'
 import { AutoStartStopEventEmitter, ListeningNewBlockEmitter, PollingNewBlockEmitter } from './new-block-emitters'
@@ -29,7 +28,7 @@ export function isServiceInitialized (serviceName: string): boolean {
   return blockTracker.getLastFetchedBlock()[0] !== undefined
 }
 
-export function getEventsEmitterForService (serviceName: string, eth: Eth, contractAbi: AbiItem[]): EventEmitter {
+export function getEventsEmitterForService (serviceName: string, eth: Eth, contractAbi: AbiItem[]): BaseEventsEmitter {
   const contractAddresses = config.get<string>(`${serviceName}.contractAddress`)
   const contract = new eth.Contract(contractAbi, contractAddresses)
   const logger = loggingFactory(`${serviceName}:blockchain`)

src/cli/start.ts

@@ -1,12 +1,12 @@
 import config from 'config'
 import { flags } from '@oclif/command'
 
-import { appFactory, services } from '../app'
+import { startApp, services } from '../app'
 import { loggingFactory } from '../logger'
 import { Flags, Config, SupportedServices, isSupportedServices } from '../definitions'
 import { BaseCLICommand } from '../utils'
-import { sequelizeFactory } from '../sequelize'
-import Event from '../blockchain/event.model'
+import DbBackUpJob from '../db-backup'
+import { ethFactory } from '../blockchain'
 
 const logger = loggingFactory('cli:start')
 
@@ -26,10 +26,6 @@ ${formattedServices}`
     port: flags.integer({ char: 'p', description: 'port to attach the server to' }),
     db: flags.string({ description: 'database connection URI', env: 'RIFM_DB' }),
     provider: flags.string({ description: 'blockchain provider connection URI', env: 'RIFM_PROVIDER' }),
-    purge: flags.boolean({
-      char: 'u',
-      description: 'will purge services that should be lunched (eq. enable/disable is applied)'
-    }),
     enable: flags.string({ char: 'e', multiple: true, description: 'enable specific service' }),
     disable: flags.string({ char: 'd', multiple: true, description: 'disable specific service' })
   }
@@ -84,17 +80,12 @@ ${formattedServices}`
     return output
   }
 
-  private async purge (): Promise<void> {
-    const toBePurgedServices = (Object.keys(services) as Array<keyof typeof services>)
+  public async precache () {
+    const toBePrecache = (Object.keys(services) as Array<keyof typeof services>)
       .filter(service => config.get<boolean>(`${service}.enabled`))
-
-    logger.info(`Purging services: ${toBePurgedServices.join(', ')}`)
-
     await Promise.all(
-      toBePurgedServices.map((service) => services[service].purge())
+      toBePrecache.map((service) => services[service].precache(ethFactory()))
     )
-
-    await Event.destroy({ where: {}, truncate: true, cascade: true })
   }
 
   async run (): Promise<void> {
@@ -103,23 +94,42 @@ ${formattedServices}`
     const configOverrides = this.buildConfigObject(flags)
     config.util.extendDeep(config, configOverrides)
 
-    if (flags.purge) {
-      sequelizeFactory()
-      await this.purge()
-    }
+    const backUpJob = new DbBackUpJob(ethFactory())
+    // An infinite loop which you can exit only with SIGINT/SIGKILL
+    while (true) {
+      let stopCallback: () => void = () => {
+        throw new Error('No stop callback was assigned!')
+      }
 
-    const app = await appFactory()
-    const port = config.get('port')
-    const server = app.listen(port)
+      // Run backup job
+      backUpJob.run()
 
-    process.on('unhandledRejection', err =>
-      logger.error(`Unhandled Rejection at: ${err}`)
-    )
+      // Promise that resolves when reset callback is called
+      const resetPromise = new Promise(resolve => {
+        startApp({
+          appResetCallBack: () => resolve()
+        }).then(value => {
+          // Lets save the function that stops the app
+          stopCallback = value.stop
+        })
+      })
 
-    server.on('listening', () =>
-      this.log('Server started on port %d', port)
-    )
+      // Let see if we have to restart the app at some point most probably because
+      // reorgs outside of confirmation range.
+      await resetPromise
+
+      logger.warn('Reorg detected outside of confirmation range. Rebuilding the service\'s state!')
+      logger.info('Stopping service')
+      stopCallback()
+      backUpJob.stop()
+
+      // Restore DB from backup
+      await backUpJob.restoreDb()
 
-    return Promise.resolve()
+      // Run pre-cache
+      await this.precache()
+
+      logger.info('Restarting the app')
+    }
   }
 }