Possibility to resend old pending transactions (#425)

oracle/config/affirmation-request-watcher.config.js

@@ -24,7 +24,7 @@ module.exports = {
   ...baseConfig.bridgeConfig,
   ...baseConfig.foreignConfig,
   event: 'UserRequestForAffirmation',
-  queue: 'home',
+  queue: 'home-prioritized',
   name: `watcher-${id}`,
   id
 }

oracle/config/base.config.js

@@ -73,6 +73,7 @@ const bridgeConfig = {
 }
 
 const homeConfig = {
+  chain: 'home',
   eventContractAddress: process.env.COMMON_HOME_BRIDGE_ADDRESS,
   eventAbi: homeAbi,
   bridgeContractAddress: process.env.COMMON_HOME_BRIDGE_ADDRESS,
@@ -83,6 +84,7 @@ const homeConfig = {
 }
 
 const foreignConfig = {
+  chain: 'foreign',
   eventContractAddress: process.env.COMMON_FOREIGN_BRIDGE_ADDRESS,
   eventAbi: foreignAbi,
   bridgeContractAddress: process.env.COMMON_FOREIGN_BRIDGE_ADDRESS,

oracle/config/collected-signatures-watcher.config.js

@@ -6,7 +6,7 @@ module.exports = {
   ...baseConfig.bridgeConfig,
   ...baseConfig.homeConfig,
   event: 'CollectedSignatures',
-  queue: 'foreign',
+  queue: 'foreign-prioritized',
   name: `watcher-${id}`,
   id
 }

oracle/config/convert-to-chai-worker.config.js

@@ -14,7 +14,7 @@ module.exports = {
   ...baseConfig.bridgeConfig,
   ...baseConfig.foreignConfig,
   workerQueue: 'convert-to-chai',
-  senderQueue: 'foreign',
+  senderQueue: 'foreign-prioritized',
   name: `worker-${id}`,
   id
 }

oracle/config/foreign-sender.config.js

@@ -4,7 +4,8 @@ const { web3Foreign } = require('../src/services/web3')
 
 module.exports = {
   ...baseConfig.bridgeConfig,
-  queue: 'foreign',
+  queue: 'foreign-prioritized',
+  oldQueue: 'foreign',
   id: 'foreign',
   name: 'sender-foreign',
   web3: web3Foreign

oracle/config/home-sender.config.js

@@ -4,7 +4,8 @@ const { web3Home } = require('../src/services/web3')
 
 module.exports = {
   ...baseConfig.bridgeConfig,
-  queue: 'home',
+  queue: 'home-prioritized',
+  oldQueue: 'home',
   id: 'home',
   name: 'sender-home',
   web3: web3Home

oracle/config/signature-request-watcher.config.js

@@ -6,7 +6,7 @@ module.exports = {
   ...baseConfig.bridgeConfig,
   ...baseConfig.homeConfig,
   event: 'UserRequestForSignature',
-  queue: 'home',
+  queue: 'home-prioritized',
   name: `watcher-${id}`,
   id
 }

oracle/config/transfer-watcher.config.js

@@ -41,7 +41,7 @@ module.exports = {
   eventContractAddress: initialChecks.bridgeableTokenAddress,
   eventAbi: ERC20_ABI,
   eventFilter: { to: process.env.COMMON_FOREIGN_BRIDGE_ADDRESS },
-  queue: 'home',
+  queue: 'home-prioritized',
   ...workerQueueConfig,
   name: `watcher-${id}`,
   id

oracle/src/confirmRelay.js

@@ -138,8 +138,8 @@ async function main({ sendJob, txHash }) {
 }
 
 async function sendJobTx(jobs) {
-  const gasPrice = await GasPrice.start(config.queue, true)
-  const chainId = await getChainId(config.queue)
+  const gasPrice = await GasPrice.start(config.chain, true)
+  const chainId = await getChainId(config.chain)
   let nonce = await getNonce(web3Instance, ORACLE_VALIDATOR_ADDRESS)
 
   await syncForEach(jobs, async job => {
@@ -153,7 +153,7 @@ async function sendJobTx(jobs) {
     try {
       logger.info(`Sending transaction with nonce ${nonce}`)
       const txHash = await sendTx({
-        chain: config.queue,
+        chain: config.chain,
         data: job.data,
         nonce,
         gasPrice: gasPrice.toString(10),

oracle/src/sender.js

@@ -1,5 +1,6 @@
 require('../env')
 const path = require('path')
+const { toBN } = require('web3-utils')
 const { connectSenderToQueue } = require('./services/amqpClient')
 const { redis } = require('./services/redisClient')
 const GasPrice = require('./services/gasPrice')
@@ -45,6 +46,7 @@ async function initialize() {
     chainId = await getChainId(config.id)
     connectSenderToQueue({
       queueName: config.queue,
+      oldQueueName: config.oldQueue,
       cb: options => {
         if (config.maxProcessingTime) {
           return watchdog(() => main(options), config.maxProcessingTime, () => {
@@ -88,7 +90,7 @@ function updateNonce(nonce) {
   return redis.set(nonceKey, nonce)
 }
 
-async function main({ msg, ackMsg, nackMsg, channel, scheduleForRetry }) {
+async function main({ msg, ackMsg, nackMsg, channel, scheduleForRetry, scheduleTransactionResend }) {
   try {
     if (redis.status !== 'ready') {
       nackMsg(msg)
@@ -103,8 +105,15 @@ async function main({ msg, ackMsg, nackMsg, channel, scheduleForRetry }) {
     let insufficientFunds = false
     let minimumBalance = null
     const failedTx = []
+    const sentTx = []
 
-    logger.debug(`Sending ${txArray.length} transactions`)
+    const isResend = txArray.length > 0 && !!txArray[0].txHash
+
+    if (isResend) {
+      logger.debug(`Checking status of ${txArray.length} transactions`)
+    } else {
+      logger.debug(`Sending ${txArray.length} transactions`)
+    }
     await syncForEach(txArray, async job => {
       let gasLimit
       if (typeof job.extraGas === 'number') {
@@ -114,11 +123,37 @@ async function main({ msg, ackMsg, nackMsg, channel, scheduleForRetry }) {
       }
 
       try {
-        logger.info(`Sending transaction with nonce ${nonce}`)
+        let txNonce
+        if (isResend) {
+          const tx = await web3Instance.eth.getTransaction(job.txHash)
+
+          if (tx === null) {
+            logger.info(`Transaction ${job.txHash} was not found, dropping it`)
+            return
+          }
+          if (tx.blockNumber !== null) {
+            logger.info(`Transaction ${job.txHash} was successfully mined`)
+            return
+          }
+
+          logger.info(
+            `Previously sent transaction is stuck, updating gasPrice: ${tx.gasPrice} -> ${gasPrice.toString(10)}`
+          )
+          if (toBN(tx.gasPrice).gte(toBN(gasPrice))) {
+            logger.info("Gas price returned from the oracle didn't increase, will reinspect this transaction later")
+            sentTx.push(job)
+            return
+          }
+
+          txNonce = tx.nonce
+        } else {
+          txNonce = nonce++
+        }
+        logger.info(`Sending transaction with nonce ${txNonce}`)
         const txHash = await sendTx({
           chain: config.id,
           data: job.data,
-          nonce,
+          nonce: txNonce,
           gasPrice: gasPrice.toString(10),
           amount: '0',
           gasLimit,
@@ -127,8 +162,11 @@ async function main({ msg, ackMsg, nackMsg, channel, scheduleForRetry }) {
           chainId,
           web3: web3Instance
         })
+        sentTx.push({
+          ...job,
+          txHash
+        })
 
-        nonce++
         logger.info(
           { eventTransactionHash: job.transactionReference, generatedTransactionHash: txHash },
           `Tx generated ${txHash} for event Tx ${job.transactionReference}`
@@ -163,6 +201,10 @@ async function main({ msg, ackMsg, nackMsg, channel, scheduleForRetry }) {
       logger.info(`Sending ${failedTx.length} Failed Tx to Queue`)
       await scheduleForRetry(failedTx, msg.properties.headers['x-retries'])
     }
+    if (sentTx.length) {
+      logger.info(`Sending ${sentTx.length} Tx Delayed Resend Requests to Queue`)
+      await scheduleTransactionResend(sentTx)
+    }
     ackMsg(msg)
     logger.debug(`Finished processing msg`)