Possibility to send a tx through all provided RPC endpoints (#394)

CONFIGURATION.md

@@ -36,6 +36,7 @@ ORACLE_LOG_LEVEL | Set the level of details in the logs. | `trace` / `debug` / `
 ORACLE_MAX_PROCESSING_TIME | The workers processes will be killed if this amount of time (in milliseconds) is elapsed 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
 ORACLE_VALIDATOR_ADDRESS_PRIVATE_KEY | The private key of the bridge validator used to sign confirmations before sending transactions to the bridge contracts. The validator account is calculated automatically from the private key. Every bridge instance (set of watchers and senders) must have its own unique private key. The specified private key is used to sign transactions on both sides of the bridge. | hexidecimal without "0x"
 ORACLE_VALIDATOR_ADDRESS | The public address of the bridge validator | hexidecimal with "0x"
+ORACLE_TX_REDUNDANCY | If set to `true`, instructs oracle to send `eth_sendRawTransaction` requests through all available RPC urls defined in `COMMON_HOME_RPC_URL` and `COMMON_FOREIGN_RPC_URL` variables instead of using first available one
 
 
 ## UI configuration

deployment/roles/oracle/templates/.env.j2

@@ -47,6 +47,9 @@ COMMON_FOREIGN_GAS_PRICE_FACTOR={{ COMMON_FOREIGN_GAS_PRICE_FACTOR }}
 ORACLE_ALLOW_HTTP_FOR_RPC={{ "yes" if ORACLE_ALLOW_HTTP_FOR_RPC else "no" }}
 ORACLE_QUEUE_URL={{ ORACLE_QUEUE_URL }}
 ORACLE_REDIS_URL={{ ORACLE_REDIS_URL }}
+{% if ORACLE_TX_REDUNDANCY | default('') != '' %}
+ORACLE_TX_REDUNDANCY={{ ORACLE_TX_REDUNDANCY }}
+{% endif %}
 
 {% if ORACLE_HOME_START_BLOCK | default('') != '' %}
 ORACLE_HOME_START_BLOCK={{ ORACLE_HOME_START_BLOCK }}

oracle/package.json

@@ -30,7 +30,6 @@
     "dotenv": "^5.0.1",
     "http-list-provider": "0.0.5",
     "ioredis": "^3.2.2",
-    "lodash": "^4.17.10",
     "node-fetch": "^2.1.2",
     "pino": "^4.17.3",
     "pino-pretty": "^2.0.1",

oracle/src/services/RpcUrlsManager.js

@@ -1,7 +1,7 @@
-const _ = require('lodash')
 const promiseRetry = require('promise-retry')
 const tryEach = require('../utils/tryEach')
 const { RETRY_CONFIG } = require('../utils/constants')
+const { promiseAny } = require('../utils/utils')
 
 function RpcUrlsManager(homeUrls, foreignUrls) {
   if (!homeUrls) {
@@ -15,19 +15,22 @@ function RpcUrlsManager(homeUrls, foreignUrls) {
   this.foreignUrls = foreignUrls.split(',')
 }
 
-RpcUrlsManager.prototype.tryEach = async function(chain, f) {
+RpcUrlsManager.prototype.tryEach = async function(chain, f, redundant = false) {
   if (chain !== 'home' && chain !== 'foreign') {
     throw new Error(`Invalid argument chain: '${chain}'`)
   }
 
-  // save homeUrls to avoid race condition
-  const urls = chain === 'home' ? _.cloneDeep(this.homeUrls) : _.cloneDeep(this.foreignUrls)
+  // save urls to avoid race condition
+  const urls = chain === 'home' ? [...this.homeUrls] : [...this.foreignUrls]
 
-  const [result, index] = await promiseRetry(retry =>
-    tryEach(urls, f).catch(() => {
-      retry()
-    }, RETRY_CONFIG)
-  )
+  if (redundant) {
+    // result from first responded node will be returned immediately
+    // remaining nodes will continue to retry queries in separate promises
+    // promiseAny will throw only if all urls reached max retry number
+    return promiseAny(urls.map(url => promiseRetry(retry => f(url).catch(retry), RETRY_CONFIG)))
+  }
+
+  const [result, index] = await promiseRetry(retry => tryEach(urls, f).catch(retry), RETRY_CONFIG)
 
   if (index > 0) {
     // rotate urls

oracle/src/tx/sendTx.js

@@ -2,6 +2,8 @@ const Web3Utils = require('web3-utils')
 const fetch = require('node-fetch')
 const rpcUrlsManager = require('../services/getRpcUrlsManager')
 
+const { ORACLE_TX_REDUNDANCY } = process.env
+
 // eslint-disable-next-line consistent-return
 async function sendTx({ chain, privateKey, data, nonce, gasPrice, amount, gasLimit, to, chainId, web3 }) {
   const serializedTx = await web3.eth.accounts.signTransaction(
@@ -26,7 +28,9 @@ async function sendTx({ chain, privateKey, data, nonce, gasPrice, amount, gasLim
 
 // eslint-disable-next-line consistent-return
 async function sendRawTx({ chain, params, method }) {
-  const result = await rpcUrlsManager.tryEach(chain, async url => {
+  const result = await rpcUrlsManager.tryEach(
+    chain,
+    async url => {
       // curl -X POST --data '{"jsonrpc":"2.0","method":"eth_sendRawTransaction","params":[{see above}],"id":1}'
       const response = await fetch(url, {
         headers: {
@@ -46,7 +50,9 @@ async function sendRawTx({ chain, params, method }) {
       }
 
       return response
-  })
+    },
+    ORACLE_TX_REDUNDANCY === 'true' && method === 'eth_sendRawTransaction'
+  )
 
   const json = await result.json()
   if (json.error) {

oracle/src/utils/utils.js

@@ -100,6 +100,11 @@ function nonceError(e) {
   )
 }
 
+// Promise.all rejects on the first rejected Promise or fulfills with the list of results
+// inverted Promise.all fulfills with the first obtained result or rejects with the list of errors
+const invert = p => new Promise((res, rej) => p.then(rej, res))
+const promiseAny = ps => invert(Promise.all(ps.map(invert)))
+
 module.exports = {
   syncForEach,
   checkHTTPS,
@@ -109,5 +114,6 @@ module.exports = {
   watchdog,
   privateKeyToAddress,
   nonceError,
-  getRetrySequence
+  getRetrySequence,
+  promiseAny
 }

oracle/test/utils.test.js

@@ -3,9 +3,10 @@ const chai = require('chai')
 const chaiAsPromised = require('chai-as-promised')
 const BigNumber = require('bignumber.js')
 const proxyquire = require('proxyquire')
-const { addExtraGas, syncForEach } = require('../src/utils/utils')
+const { addExtraGas, syncForEach, promiseAny } = require('../src/utils/utils')
 
 chai.use(chaiAsPromised)
+chai.should()
 const { expect } = chai
 
 describe('utils', () => {
@@ -134,4 +135,34 @@ describe('utils', () => {
       })
     })
   })
+
+  describe('promiseAny', () => {
+    const f = x => new Promise((res, rej) => setTimeout(() => (x > 0 ? res : rej)(x), 10 * x))
+
+    it('should return first obtained result', async () => {
+      const array = [2, 1, 3]
+      const result = await promiseAny(array.map(f))
+
+      expect(result).to.equal(1)
+    })
+
+    it('should return first obtained result with one reject', async () => {
+      const array = [2, -1, 3]
+      const result = await promiseAny(array.map(f))
+
+      expect(result).to.equal(2)
+    })
+
+    it('should return first obtained result with several rejects', async () => {
+      const array = [2, -1, -3]
+      const result = await promiseAny(array.map(f))
+
+      expect(result).to.equal(2)
+    })
+
+    it('should reject if all functions failed', async () => {
+      const array = [-2, -1, -3]
+      await promiseAny(array.map(f)).should.be.rejected
+    })
+  })
 })