Add watchdog for watchers

.env.example

@@ -29,6 +29,7 @@ HOME_START_BLOCK=
 FOREIGN_START_BLOCK=
 
 LOG_LEVEL=debug
+MAX_PROCESSING_TIME=20000
 
 #testing accs
 USER_ADDRESS=0x59c4474184579b9c31b5e51445b6eef91cebf370

README.md

@@ -213,6 +213,7 @@ where the _watcher_ could be one of:
 | `REDIS_LOCK_TTL` | Threshold in milliseconds for locking a resource in the Redis DB. Until the threshold is exceeded, the resource is unlocked. Usually it is `1000`. | integer |
 | `ALLOW_HTTP` | **Only use in test environments - must be omitted in production environments.**. If this parameter is specified and set to `yes`, RPC URLs can be specified in form of HTTP links. A warning that the connection is insecure will be written to the logs. | `yes` / `no` |
 | `LOG_LEVEL` | Set the level of details in the logs. | `trace` / `debug` / `info` / `warn` / `error` / `fatal` |
+| `MAX_PROCESSING_TIME` | The workers processes will be killed if this amount of time (in milliseconds) is ellapsed before they finish processing. It is recommended to set this value to 4 times the value of the longest polling time (set with the `HOME_POLLING_INTERVAL` and `FOREIGN_POLLING_INTERVAL` variables). To disable this, set the time to 0. | integer |
 
 ### Useful Commands for Development
 

config/base.config.js

@@ -45,13 +45,24 @@ switch (process.env.BRIDGE_MODE) {
     }
 }
 
+let maxProcessingTime = null
+if (String(process.env.MAX_PROCESSING_TIME) === '0') {
+  maxProcessingTime = 0
+} else if (!process.env.MAX_PROCESSING_TIME) {
+  maxProcessingTime =
+    4 * Math.max(process.env.HOME_POLLING_INTERVAL, process.env.FOREIGN_POLLING_INTERVAL)
+} else {
+  maxProcessingTime = Number(process.env.MAX_PROCESSING_TIME)
+}
+
 const bridgeConfig = {
   homeBridgeAddress: process.env.HOME_BRIDGE_ADDRESS,
   homeBridgeAbi: homeAbi,
   foreignBridgeAddress: process.env.FOREIGN_BRIDGE_ADDRESS,
   foreignBridgeAbi: foreignAbi,
   eventFilter: {},
-  validatorAddress: VALIDATOR_ADDRESS || privateKeyToAddress(VALIDATOR_ADDRESS_PRIVATE_KEY)
+  validatorAddress: VALIDATOR_ADDRESS || privateKeyToAddress(VALIDATOR_ADDRESS_PRIVATE_KEY),
+  maxProcessingTime
 }
 
 const homeConfig = {

config/foreign-sender.config.js

@@ -1,8 +1,10 @@
 require('dotenv').config()
+const baseConfig = require('./base.config')
 
 const { web3Foreign } = require('../src/services/web3')
 
 module.exports = {
+  ...baseConfig.bridgeConfig,
   queue: 'foreign',
   id: 'foreign',
   name: 'sender-foreign',

config/home-sender.config.js

@@ -1,8 +1,10 @@
 require('dotenv').config()
+const baseConfig = require('./base.config')
 
 const { web3Home } = require('../src/services/web3')
 
 module.exports = {
+  ...baseConfig.bridgeConfig,
   queue: 'home',
   id: 'home',
   name: 'sender-home',

scripts/privateKeyToAddress.js

@@ -3,12 +3,13 @@ require('dotenv').config({
   path: path.join(__dirname, '..', '.env')
 })
 const { privateKeyToAddress } = require('../src/utils/utils')
+const { EXIT_CODES } = require('../src/utils/constants')
 
 const privateKey = process.env.VALIDATOR_ADDRESS_PRIVATE_KEY
 
 if (!privateKey) {
   console.error('Environment variable VALIDATOR_ADDRESS_PRIVATE_KEY is not set')
-  process.exit(1)
+  process.exit(EXIT_CODES.GENERAL_ERROR)
 }
 
 const address = privateKeyToAddress(privateKey)

scripts/resetLastBlock.js

@@ -4,6 +4,7 @@ require('dotenv').config({
   path: path.join(__dirname, '../.env')
 })
 const { id } = require('../config/base.config')
+const { EXIT_CODES } = require('../src/utils/constants')
 
 const redis = new Redis(process.env.REDIS_URL)
 
@@ -22,7 +23,7 @@ if (process.argv.length < 4) {
 
 function logError(message) {
   console.log(message)
-  process.exit(1)
+  process.exit(EXIT_CODES.GENERAL_ERROR)
 }
 
 function getRedisKey(name) {

src/events/processAffirmationRequests/index.js

@@ -3,7 +3,7 @@ const rootLogger = require('../../services/logger')
 const { web3Home } = require('../../services/web3')
 const promiseLimit = require('promise-limit')
 const bridgeValidatorsABI = require('../../../abis/BridgeValidators.abi')
-const { MAX_CONCURRENT_EVENTS } = require('../../utils/constants')
+const { EXIT_CODES, MAX_CONCURRENT_EVENTS } = require('../../utils/constants')
 const estimateGas = require('./estimateGas')
 const {
   AlreadyProcessedError,
@@ -64,7 +64,7 @@ function processAffirmationRequestsBuilder(config) {
             )
           } else if (e instanceof InvalidValidatorError) {
             logger.fatal({ address: config.validatorAddress }, 'Invalid validator')
-            process.exit(10)
+            process.exit(EXIT_CODES.INCOMPATIBILITY)
           } else if (e instanceof AlreadySignedError) {
             logger.info(`Already signed affirmationRequest ${affirmationRequest.transactionHash}`)
             return

src/events/processSignatureRequests/index.js

@@ -11,7 +11,7 @@ const {
   AlreadySignedError,
   InvalidValidatorError
 } = require('../../utils/errors')
-const { MAX_CONCURRENT_EVENTS } = require('../../utils/constants')
+const { EXIT_CODES, MAX_CONCURRENT_EVENTS } = require('../../utils/constants')
 
 const { VALIDATOR_ADDRESS_PRIVATE_KEY } = process.env
 
@@ -81,7 +81,7 @@ function processSignatureRequestsBuilder(config) {
             )
           } else if (e instanceof InvalidValidatorError) {
             logger.fatal({ address: config.validatorAddress }, 'Invalid validator')
-            process.exit(10)
+            process.exit(EXIT_CODES.INCOMPATIBILITY)
           } else if (e instanceof AlreadySignedError) {
             logger.info(`Already signed signatureRequest ${signatureRequest.transactionHash}`)
             return

src/events/processTransfers/index.js

@@ -9,7 +9,7 @@ const {
   AlreadySignedError,
   InvalidValidatorError
 } = require('../../utils/errors')
-const { MAX_CONCURRENT_EVENTS } = require('../../utils/constants')
+const { EXIT_CODES, MAX_CONCURRENT_EVENTS } = require('../../utils/constants')
 const estimateGas = require('../processAffirmationRequests/estimateGas')
 
 const limit = promiseLimit(MAX_CONCURRENT_EVENTS)
@@ -61,7 +61,7 @@ function processTransfersBuilder(config) {
             )
           } else if (e instanceof InvalidValidatorError) {
             logger.fatal({ address: config.validatorAddress }, 'Invalid validator')
-            process.exit(10)
+            process.exit(EXIT_CODES.INCOMPATIBILITY)
           } else if (e instanceof AlreadySignedError) {
             logger.info(`Already signed transfer ${transfer.transactionHash}`)
             return

src/sender.js

@@ -10,19 +10,20 @@ const { getNonce, getChainId } = require('./tx/web3')
 const {
   addExtraGas,
   checkHTTPS,
+  privateKeyToAddress,
   syncForEach,
   waitForFunds,
-  privateKeyToAddress
+  watchdog
 } = require('./utils/utils')
-const { EXTRA_GAS_PERCENTAGE } = require('./utils/constants')
+const { EXIT_CODES, EXTRA_GAS_PERCENTAGE } = require('./utils/constants')
 
 const { VALIDATOR_ADDRESS_PRIVATE_KEY, REDIS_LOCK_TTL } = process.env
 
 const VALIDATOR_ADDRESS = privateKeyToAddress(VALIDATOR_ADDRESS_PRIVATE_KEY)
 
 if (process.argv.length < 3) {
   logger.error('Please check the number of arguments, config file was not provided')
-  process.exit(1)
+  process.exit(EXIT_CODES.GENERAL_ERROR)
 }
 
 const config = require(path.join('../config/', process.argv[2]))
@@ -44,11 +45,20 @@ async function initialize() {
     chainId = await getChainId(web3Instance)
     connectSenderToQueue({
       queueName: config.queue,
-      cb: main
+      cb: options => {
+        if (config.maxProcessingTime) {
+          return watchdog(() => main(options), config.maxProcessingTime, () => {
+            logger.fatal('Max processing time reached')
+            process.exit(EXIT_CODES.MAX_TIME_REACHED)
+          })
+        }
+
+        return main(options)
+      }
     })
   } catch (e) {
     logger.error(e.message)
-    process.exit(1)
+    process.exit(EXIT_CODES.GENERAL_ERROR)
   }
 }
 

src/utils/constants.js

@@ -7,5 +7,10 @@ module.exports = {
     maxTimeout: 360000,
     randomize: true
   },
-  DEFAULT_UPDATE_INTERVAL: 600000
+  DEFAULT_UPDATE_INTERVAL: 600000,
+  EXIT_CODES: {
+    GENERAL_ERROR: 1,
+    INCOMPATIBILITY: 10,
+    MAX_TIME_REACHED: 11
+  }
 }

src/utils/utils.js

@@ -65,6 +65,23 @@ function setIntervalAndRun(f, interval) {
   return handler
 }
 
+/**
+ * Run function `f` and return its result, unless `timeout` milliseconds pass before `f` ends. If that happens, run
+ * `kill` instead.
+ *
+ * @param {Function} f The function to run. It is assumed that it's an async function.
+ * @param {Number} timeout Max time in milliseconds to wait for `f` to finish.
+ * @param {Function} kill Function that will be called if `f` takes more than `timeout` milliseconds.
+ */
+async function watchdog(f, timeout, kill) {
+  const timeoutHandler = setTimeout(kill, timeout)
+
+  const result = await f()
+  clearTimeout(timeoutHandler)
+
+  return result
+}
+
 function add0xPrefix(s) {
   if (s.indexOf('0x') === 0) {
     return s
@@ -85,5 +102,6 @@ module.exports = {
   waitForFunds,
   addExtraGas,
   setIntervalAndRun,
+  watchdog,
   privateKeyToAddress
 }

src/watcher.js

@@ -7,11 +7,12 @@ const { redis } = require('./services/redisClient')
 const logger = require('./services/logger')
 const rpcUrlsManager = require('./services/getRpcUrlsManager')
 const { getRequiredBlockConfirmations, getEvents } = require('./tx/web3')
-const { checkHTTPS } = require('./utils/utils')
+const { checkHTTPS, watchdog } = require('./utils/utils')
+const { EXIT_CODES } = require('./utils/constants')
 
 if (process.argv.length < 3) {
   logger.error('Please check the number of arguments, config file was not provided')
-  process.exit(1)
+  process.exit(EXIT_CODES.GENERAL_ERROR)
 }
 
 const config = require(path.join('../config/', process.argv[2]))
@@ -44,14 +45,21 @@ async function initialize() {
     })
   } catch (e) {
     logger.error(e)
-    process.exit(1)
+    process.exit(EXIT_CODES.GENERAL_ERROR)
   }
 }
 
 async function runMain({ sendToQueue }) {
   try {
     if (connection.isConnected() && redis.status === 'ready') {
-      await main({ sendToQueue })
+      if (config.maxProcessingTime) {
+        await watchdog(() => main({ sendToQueue }), config.maxProcessingTime, () => {
+          logger.fatal('Max processing time reached')
+          process.exit(EXIT_CODES.MAX_TIME_REACHED)
+        })
+      } else {
+        await main({ sendToQueue })
+      }
     }
   } catch (e) {
     logger.error(e)