Add watchdog for watchers

.env.example

@@ -29,6 +29,7 @@ HOME_START_BLOCK=
 FOREIGN_START_BLOCK=
 
 LOG_LEVEL=debug
+MAX_PROCESSING_TIME=600000
 
 #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` | If this parameter is specified, the workers processes will be killed if this amount of time (in milliseconds) is ellapsed before they finish processing. | integer |
 
 ### Useful Commands for Development
 

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,11 +10,12 @@ 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
 
@@ -32,6 +33,8 @@ const nonceLock = `lock:${config.id}:nonce`
 const nonceKey = `${config.id}:nonce`
 let chainId = 0
 
+const maxProcessingTime = process.env.MAX_PROCESSING_TIME
+
 async function initialize() {
   try {
     const checkHttps = checkHTTPS(process.env.ALLOW_HTTP, logger)
@@ -44,7 +47,16 @@ async function initialize() {
     chainId = await getChainId(web3Instance)
     connectSenderToQueue({
       queueName: config.queue,
-      cb: main
+      cb: options => {
+        if (maxProcessingTime) {
+          return watchdog(() => main(options), maxProcessingTime, () => {
+            logger.fatal('Max processing time reached')
+            process.exit(EXIT_CODES.MAX_TIME_REACHED)
+          })
+        }
+
+        return main(options)
+      }
     })
   } catch (e) {
     logger.error(e.message)

src/utils/constants.js

@@ -7,5 +7,9 @@ module.exports = {
     maxTimeout: 360000,
     randomize: true
   },
-  DEFAULT_UPDATE_INTERVAL: 600000
+  DEFAULT_UPDATE_INTERVAL: 600000,
+  EXIT_CODES: {
+    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,7 +7,8 @@ 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')
@@ -30,6 +31,8 @@ const eventContract = new web3Instance.eth.Contract(config.eventAbi, config.even
 const lastBlockRedisKey = `${config.id}:lastProcessedBlock`
 let lastProcessedBlock = BN.max(config.startBlock.sub(ONE), ZERO)
 
+const maxProcessingTime = process.env.MAX_PROCESSING_TIME
+
 async function initialize() {
   try {
     const checkHttps = checkHTTPS(process.env.ALLOW_HTTP, logger)
@@ -51,7 +54,14 @@ async function initialize() {
 async function runMain({ sendToQueue }) {
   try {
     if (connection.isConnected() && redis.status === 'ready') {
-      await main({ sendToQueue })
+      if (maxProcessingTime) {
+        await watchdog(() => main({ sendToQueue }), maxProcessingTime, () => {
+          logger.fatal('Max processing time reached')
+          process.exit(EXIT_CODES.MAX_TIME_REACHED)
+        })
+      } else {
+        await main({ sendToQueue })
+      }
     }
   } catch (e) {
     logger.error(e)