Bump node-fetch from 2.6.1 to 2.6.7

Bumps [node-fetch](https://github.com/node-fetch/node-fetch) from 2.6.1 to 2.6.7.
- [Release notes](https://github.com/node-fetch/node-fetch/releases)
- [Commits](https://github.com/node-fetch/node-fetch/compare/v2.6.1...v2.6.7)

---
updated-dependencies:
- dependency-name: node-fetch
  dependency-type: direct:production
...

Signed-off-by: dependabot[bot] <support@github.com>

alm/package.json

@@ -59,6 +59,6 @@
   },
   "devDependencies": {
     "eslint-plugin-prettier": "^3.1.3",
-    "node-fetch": "^2.6.1"
+    "node-fetch": "^2.6.7"
   }
 }

commons/package.json

@@ -11,7 +11,7 @@
     "@mycrypto/gas-estimation": "^1.1.0",
     "gas-price-oracle": "^0.1.5",
     "web3-utils": "^1.3.0",
-    "node-fetch": "^2.1.2"
+    "node-fetch": "^2.6.7"
   },
   "devDependencies": {
     "bn-chai": "^1.0.1",

monitor/package.json

@@ -18,7 +18,7 @@
     "cors": "^2.8.5",
     "dotenv": "^5.0.1",
     "express": "^4.16.3",
-    "node-fetch": "^2.1.2",
+    "node-fetch": "^2.6.7",
     "web3": "^1.3.0"
   },
   "engines": {

oracle/config/foreign-mev-sender.config.js

@@ -0,0 +1,37 @@
+const baseConfig = require('./base.config')
+
+const { DEFAULT_TRANSACTION_RESEND_INTERVAL } = require('../src/utils/constants')
+const { MEV_HELPER_ABI } = require('../src/utils/mev')
+const { web3Foreign, getFlashbotsProvider } = require('../src/services/web3')
+
+const {
+  ORACLE_FOREIGN_TX_RESEND_INTERVAL,
+  ORACLE_MEV_FOREIGN_HELPER_CONTRACT_ADDRESS,
+  ORACLE_MEV_FOREIGN_MIN_GAS_PRICE,
+  ORACLE_MEV_FOREIGN_FLAT_MINER_FEE,
+  ORACLE_MEV_FOREIGN_MAX_PRIORITY_FEE_PER_GAS,
+  ORACLE_MEV_FOREIGN_MAX_FEE_PER_GAS,
+  ORACLE_MEV_FOREIGN_BUNDLES_BLOCK_RANGE
+} = process.env
+
+const contract = new baseConfig.foreign.web3.eth.Contract(MEV_HELPER_ABI, ORACLE_MEV_FOREIGN_HELPER_CONTRACT_ADDRESS)
+
+module.exports = {
+  ...baseConfig,
+  pollingInterval: baseConfig.foreign.pollingInterval,
+  mevForeign: {
+    contractAddress: ORACLE_MEV_FOREIGN_HELPER_CONTRACT_ADDRESS,
+    contract,
+    minGasPrice: ORACLE_MEV_FOREIGN_MIN_GAS_PRICE,
+    flatMinerFee: ORACLE_MEV_FOREIGN_FLAT_MINER_FEE,
+    maxPriorityFeePerGas: ORACLE_MEV_FOREIGN_MAX_PRIORITY_FEE_PER_GAS,
+    maxFeePerGas: ORACLE_MEV_FOREIGN_MAX_FEE_PER_GAS,
+    bundlesPerIteration: Math.max(parseInt(ORACLE_MEV_FOREIGN_BUNDLES_BLOCK_RANGE, 10) || 5, 1),
+    getFlashbotsProvider
+  },
+  mevJobsRedisKey: `${baseConfig.id}-collected-signatures-mev:mevJobs`,
+  id: 'mev-sender-foreign',
+  name: 'mev-sender-foreign',
+  web3: web3Foreign,
+  resendInterval: parseInt(ORACLE_FOREIGN_TX_RESEND_INTERVAL, 10) || DEFAULT_TRANSACTION_RESEND_INTERVAL
+}

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

@@ -0,0 +1,30 @@
+const baseConfig = require('./base.config')
+const { MEV_HELPER_ABI } = require('../src/utils/mev')
+
+const {
+  ORACLE_MEV_FOREIGN_HELPER_CONTRACT_ADDRESS,
+  ORACLE_MEV_FOREIGN_MIN_GAS_PRICE,
+  ORACLE_MEV_FOREIGN_FLAT_MINER_FEE,
+  ORACLE_MEV_FOREIGN_MAX_PRIORITY_FEE_PER_GAS,
+  ORACLE_MEV_FOREIGN_MAX_FEE_PER_GAS
+} = process.env
+
+const id = `${baseConfig.id}-collected-signatures-mev`
+
+const contract = new baseConfig.foreign.web3.eth.Contract(MEV_HELPER_ABI, ORACLE_MEV_FOREIGN_HELPER_CONTRACT_ADDRESS)
+
+module.exports = {
+  ...baseConfig,
+  mevForeign: {
+    contractAddress: ORACLE_MEV_FOREIGN_HELPER_CONTRACT_ADDRESS,
+    contract,
+    minGasPrice: ORACLE_MEV_FOREIGN_MIN_GAS_PRICE,
+    flatMinerFee: ORACLE_MEV_FOREIGN_FLAT_MINER_FEE,
+    maxPriorityFeePerGas: ORACLE_MEV_FOREIGN_MAX_PRIORITY_FEE_PER_GAS,
+    maxFeePerGas: ORACLE_MEV_FOREIGN_MAX_FEE_PER_GAS
+  },
+  main: baseConfig.home,
+  event: 'CollectedSignatures',
+  name: `watcher-${id}`,
+  id
+}

oracle/docker-compose-helpers.yml

@@ -1,6 +1,14 @@
 ---
 version: '2.4'
 services:
+  redis:
+    cpus: 0.1
+    mem_limit: 500m
+    command: [ redis-server, --appendonly, 'yes' ]
+    hostname: redis
+    image: redis:4
+    restart: unless-stopped
+    volumes: [ '~/bridge_data/helpers/redis:/data' ]
   interestFetcher:
     cpus: 0.1
     mem_limit: 500m
@@ -13,3 +21,41 @@ services:
       INTERVAL: 300000
     restart: unless-stopped
     entrypoint: yarn helper:interestFether
+  mevWatcher:
+    cpus: 0.1
+    mem_limit: 500m
+    image: poanetwork/tokenbridge-oracle:latest
+    env_file: ./.env
+    environment:
+      NODE_ENV: production
+      ORACLE_VALIDATOR_ADDRESS: ${ORACLE_VALIDATOR_ADDRESS}
+      ORACLE_MEV_FOREIGN_HELPER_CONTRACT_ADDRESS: 'TBD'
+      ORACLE_MEV_FOREIGN_MIN_GAS_PRICE: '50000000000' # 50 gwei
+      ORACLE_MEV_FOREIGN_FLAT_MINER_FEE: '1500000000000000' # 0.0015 eth = 300k gas * 5 gwei
+      ORACLE_MEV_FOREIGN_MAX_PRIORITY_FEE_PER_GAS: '0' # 0 gwei
+      ORACLE_MEV_FOREIGN_MAX_FEE_PER_GAS: '1000000000000' # 1000 gwei
+      ORACLE_FOREIGN_RPC_POLLING_INTERVAL: '15000' # CollectedSignatures event polling interval
+      ORACLE_HOME_START_BLOCK: 'TBD'
+      ORACLE_HOME_SKIP_MANUAL_LANE: 'true'
+    restart: unless-stopped
+    entrypoint: yarn mev:watcher:collected-signatures
+  mevSender:
+    cpus: 0.1
+    mem_limit: 500m
+    image: poanetwork/tokenbridge-oracle:latest
+    env_file: ./.env
+    environment:
+      NODE_ENV: production
+      ORACLE_VALIDATOR_ADDRESS: ${ORACLE_VALIDATOR_ADDRESS}
+      ORACLE_VALIDATOR_ADDRESS_PRIVATE_KEY: ${ORACLE_VALIDATOR_ADDRESS_PRIVATE_KEY}
+      ORACLE_MEV_FOREIGN_HELPER_CONTRACT_ADDRESS: 'TBD'
+      ORACLE_MEV_FOREIGN_MIN_GAS_PRICE: '50000000000' # 50 gwei
+      ORACLE_MEV_FOREIGN_FLAT_MINER_FEE: '1500000000000000' # 0.0015 eth = 300k gas * 5 gwei
+      ORACLE_MEV_FOREIGN_MAX_PRIORITY_FEE_PER_GAS: '0' # 0 gwei
+      ORACLE_MEV_FOREIGN_MAX_FEE_PER_GAS: '1000000000000' # 1000 gwei
+      ORACLE_MEV_FOREIGN_FLASHBOTS_RPC_URL: 'https://relay-goerli.flashbots.net'
+      ORACLE_MEV_FOREIGN_FLASHBOTS_AUTH_SIGNING_KEY: 82db7175932f4e6c8e45283b78b54fd5f195149378ec90d95b8fd0ec8bdadf1d
+      ORACLE_MEV_FOREIGN_BUNDLES_BLOCK_RANGE: '5'
+      ORACLE_FOREIGN_RPC_POLLING_INTERVAL: '70000' # time between sending different batches of MEV bundles (~= 5 blocks * 14 seconds)
+    restart: unless-stopped
+    entrypoint: yarn mev:sender:foreign

oracle/package.json

@@ -19,6 +19,8 @@
     "confirm:information-request": "./scripts/start-worker.sh confirmRelay information-request-watcher",
     "manager:shutdown": "./scripts/start-worker.sh shutdownManager shutdown-manager",
     "helper:interestFether": "node ./scripts/interestFetcher.js",
+    "mev:watcher:collected-signatures": "./scripts/start-worker.sh mevWatcher mev-collected-signatures-watcher",
+    "mev:sender:foreign": "./scripts/start-worker.sh mevSender foreign-mev-sender",
     "dev": "concurrently -n 'watcher:signature-request,watcher:collected-signatures,watcher:affirmation-request,watcher:transfer, sender:home,sender:foreign' -c 'red,green,yellow,blue,magenta,cyan' 'yarn watcher:signature-request' 'yarn watcher:collected-signatures' 'yarn watcher:affirmation-request' 'yarn watcher:transfer' 'yarn sender:home' 'yarn sender:foreign'",
     "test": "NODE_ENV=test mocha",
     "test:watch": "NODE_ENV=test mocha --watch --reporter=min",
@@ -28,12 +30,14 @@
   "author": "",
   "license": "ISC",
   "dependencies": {
+    "@flashbots/ethers-provider-bundle": "^0.4.3",
     "amqp-connection-manager": "^2.0.0",
     "amqplib": "^0.5.2",
     "bignumber.js": "^7.2.1",
     "dotenv": "^5.0.1",
+    "ethers": "^5.5.3",
     "ioredis": "^3.2.2",
-    "node-fetch": "^2.1.2",
+    "node-fetch": "^2.6.7",
     "pino": "^4.17.3",
     "pino-pretty": "^2.0.1",
     "promise-limit": "^2.7.0",

oracle/scripts/erc20_to_native/sendForeign.js

@@ -36,7 +36,7 @@ async function main() {
         data,
         nonce,
         gasPrice: FOREIGN_TEST_TX_GAS_PRICE,
-        amount: '0',
+        value: '0',
         gasLimit,
         to: bridgeableTokenAddress,
         web3: web3Foreign,

oracle/scripts/erc20_to_native/sendHome.js

@@ -29,7 +29,7 @@ async function main() {
         data: '0x',
         nonce,
         gasPrice: HOME_TEST_TX_GAS_PRICE,
-        amount: HOME_MIN_AMOUNT_PER_TX,
+        value: web3Home.utils.toWei(HOME_MIN_AMOUNT_PER_TX),
         gasLimit: 100000,
         to: COMMON_HOME_BRIDGE_ADDRESS,
         web3: web3Home,

oracle/scripts/interestFetcher.js

@@ -54,7 +54,7 @@ async function main() {
       nonce,
       gasPrice,
       gasLimit: Math.round(gasLimit * 1.5),
-      amount: '0',
+      value: '0',
       chainId,
       web3: web3Home
     })

oracle/src/confirmRelay.js

@@ -174,7 +174,7 @@ async function sendJobTx(jobs) {
       const txHash = await sendTx({
         data: job.data,
         nonce,
-        amount: '0',
+        value: '0',
         gasLimit,
         privateKey: config.validatorPrivateKey,
         to: job.to,

oracle/src/events/processAMBCollectedSignatures/estimateGas.js

@@ -4,7 +4,6 @@ const { AlreadyProcessedError, IncompatibleContractError, InvalidValidatorError
 const logger = require('../../services/logger').child({
   module: 'processCollectedSignatures:estimateGas'
 })
-const { parseAMBHeader } = require('../../utils/message')
 
 const web3 = new Web3()
 const { toBN } = Web3.utils
@@ -22,15 +21,9 @@ async function estimateGas({
   address
 }) {
   try {
-    const gasEstimate = await foreignBridge.methods.executeSignatures(message, signatures).estimateGas({
+    return await foreignBridge.methods.executeSignatures(message, signatures).estimateGas({
       from: address
     })
-    const msgGasLimit = parseAMBHeader(message).gasLimit
-
-    // + estimateExtraGas(len)
-    // is not needed here, since estimateGas will already take into account gas
-    // needed for memory expansion, message processing, etc.
-    return gasEstimate + msgGasLimit
   } catch (e) {
     if (e instanceof HttpListProviderError) {
       throw e

oracle/src/events/processAMBCollectedSignaturesMEV/index.js

@@ -0,0 +1,183 @@
+require('dotenv').config()
+const promiseLimit = require('promise-limit')
+const { HttpListProviderError } = require('../../services/HttpListProvider')
+const { getValidatorContract } = require('../../tx/web3')
+const rootLogger = require('../../services/logger')
+const { signatureToVRS, packSignatures } = require('../../utils/message')
+const { readAccessListFile, isRevertError } = require('../../utils/utils')
+const { parseAMBMessage } = require('../../../../commons')
+const estimateGas = require('../processAMBCollectedSignatures/estimateGas')
+const { AlreadyProcessedError, IncompatibleContractError, InvalidValidatorError } = require('../../utils/errors')
+const { MAX_CONCURRENT_EVENTS, EXTRA_GAS_ABSOLUTE } = require('../../utils/constants')
+
+const limit = promiseLimit(MAX_CONCURRENT_EVENTS)
+
+const { ORACLE_HOME_TO_FOREIGN_ALLOWANCE_LIST, ORACLE_HOME_TO_FOREIGN_BLOCK_LIST } = process.env
+const ORACLE_HOME_SKIP_MANUAL_LANE = process.env.ORACLE_HOME_SKIP_MANUAL_LANE === 'true'
+
+function processCollectedSignaturesBuilder(config) {
+  const { home, foreign, mevForeign } = config
+
+  let validatorContract = null
+
+  return async function processCollectedSignatures(signatures) {
+    const txToSend = []
+
+    if (validatorContract === null) {
+      validatorContract = await getValidatorContract(foreign.bridgeContract, foreign.web3)
+    }
+
+    rootLogger.debug(`Processing ${signatures.length} CollectedSignatures events`)
+    const callbacks = signatures
+      .map(colSignature => async () => {
+        const { messageHash, NumberOfCollectedSignatures } = colSignature.returnValues
+
+        const logger = rootLogger.child({
+          eventTransactionHash: colSignature.transactionHash
+        })
+
+        logger.info(`Processing CollectedSignatures ${colSignature.transactionHash}`)
+        const message = await home.bridgeContract.methods.message(messageHash).call()
+        const parsedMessage = parseAMBMessage(message)
+
+        if (ORACLE_HOME_TO_FOREIGN_ALLOWANCE_LIST || ORACLE_HOME_TO_FOREIGN_BLOCK_LIST) {
+          const sender = parsedMessage.sender.toLowerCase()
+          const executor = parsedMessage.executor.toLowerCase()
+
+          if (ORACLE_HOME_TO_FOREIGN_ALLOWANCE_LIST) {
+            const allowanceList = await readAccessListFile(ORACLE_HOME_TO_FOREIGN_ALLOWANCE_LIST, logger)
+            if (!allowanceList.includes(executor) && !allowanceList.includes(sender)) {
+              logger.info(
+                { sender, executor },
+                'Validator skips a message. Neither sender nor executor addresses are in the allowance list.'
+              )
+              return
+            }
+          } else if (ORACLE_HOME_TO_FOREIGN_BLOCK_LIST) {
+            const blockList = await readAccessListFile(ORACLE_HOME_TO_FOREIGN_BLOCK_LIST, logger)
+            if (blockList.includes(executor)) {
+              logger.info({ executor }, 'Validator skips a message. Executor address is in the block list.')
+              return
+            }
+            if (blockList.includes(sender)) {
+              logger.info({ sender }, 'Validator skips a message. Sender address is in the block list.')
+              return
+            }
+          }
+        }
+
+        if (ORACLE_HOME_SKIP_MANUAL_LANE && parsedMessage.decodedDataType.manualLane) {
+          logger.info(
+            { dataType: parsedMessage.dataType },
+            'Validator skips a message. Message was forwarded to the manual lane by the extension'
+          )
+          return
+        }
+
+        logger.debug({ NumberOfCollectedSignatures }, 'Number of signatures to get')
+
+        const requiredSignatures = []
+        requiredSignatures.length = NumberOfCollectedSignatures
+        requiredSignatures.fill(0)
+
+        const signaturesArray = []
+        const [v, r, s] = [[], [], []]
+        logger.debug('Getting message signatures')
+        const signaturePromises = requiredSignatures.map(async (el, index) => {
+          logger.debug({ index }, 'Getting message signature')
+          const signature = await home.bridgeContract.methods.signature(messageHash, index).call()
+          const vrs = signatureToVRS(signature)
+          v.push(vrs.v)
+          r.push(vrs.r)
+          s.push(vrs.s)
+          signaturesArray.push(vrs)
+        })
+
+        await Promise.all(signaturePromises)
+        const signatures = packSignatures(signaturesArray)
+        logger.info(`Processing messageId: ${parsedMessage.messageId}`)
+
+        let gasEstimate
+        try {
+          logger.debug('Estimate gas')
+          gasEstimate = await estimateGas({
+            foreignBridge: foreign.bridgeContract,
+            validatorContract,
+            v,
+            r,
+            s,
+            signatures,
+            message,
+            numberOfCollectedSignatures: NumberOfCollectedSignatures,
+            messageId: parsedMessage.messageId,
+            address: config.validatorAddress
+          })
+          logger.debug({ gasEstimate }, 'Gas estimated')
+        } catch (e) {
+          if (e instanceof HttpListProviderError) {
+            throw new Error('RPC Connection Error: submitSignature Gas Estimate cannot be obtained.')
+          } else if (e instanceof AlreadyProcessedError) {
+            logger.info(`Already processed CollectedSignatures ${colSignature.transactionHash}`)
+            return
+          } else if (e instanceof IncompatibleContractError || e instanceof InvalidValidatorError) {
+            logger.error(`The message couldn't be processed; skipping: ${e.message}`)
+            return
+          } else {
+            logger.error(e, 'Unknown error while processing transaction')
+            throw e
+          }
+        }
+
+        const executeData = foreign.bridgeContract.methods.executeSignatures(message, signatures).encodeABI()
+        const profit = await estimateProfit(
+          mevForeign.contract,
+          mevForeign.minGasPrice,
+          executeData,
+          mevForeign.flatMinerFee
+        )
+        if (profit === '0') {
+          logger.error('No MEV opportunity found when testing with min gas price, skipping job')
+          return
+        }
+        logger.info(`Estimated profit of ${profit} when simulating with ${mevForeign.minGasPrice} gas price`)
+
+        txToSend.push({
+          profit,
+          executeData,
+          data: mevForeign.contract.methods.execute(executeData).encodeABI(),
+          gasEstimate,
+          extraGas: EXTRA_GAS_ABSOLUTE,
+          maxFeePerGas: mevForeign.maxFeePerGas,
+          maxPriorityFeePerGas: mevForeign.maxPriorityFeePerGas,
+          transactionReference: colSignature.transactionHash,
+          to: mevForeign.contractAddress,
+          value: mevForeign.flatMinerFee
+        })
+      })
+      .map(promise => limit(promise))
+
+    await Promise.all(callbacks)
+
+    return txToSend
+  }
+}
+
+async function estimateProfit(contract, gasPrice, data, minerFee) {
+  return contract.methods
+    .estimateProfit(gasPrice, data)
+    .call({ value: minerFee })
+    .then(
+      res => res.toString(),
+      e => {
+        if (isRevertError(e)) {
+          return '0'
+        }
+        throw e
+      }
+    )
+}
+
+module.exports = {
+  processCollectedSignaturesBuilder,
+  estimateProfit
+}

oracle/src/events/processAffirmationRequests/estimateGas.js

@@ -6,11 +6,9 @@ const logger = require('../../services/logger').child({
 
 async function estimateGas({ web3, homeBridge, validatorContract, recipient, value, txHash, address }) {
   try {
-    const gasEstimate = await homeBridge.methods.executeAffirmation(recipient, value, txHash).estimateGas({
+    return await homeBridge.methods.executeAffirmation(recipient, value, txHash).estimateGas({
       from: address
     })
-
-    return gasEstimate
   } catch (e) {
     if (e instanceof HttpListProviderError) {
       throw e

oracle/src/events/processCollectedSignatures/estimateGas.js

@@ -20,8 +20,7 @@ async function estimateGas({
   signatures
 }) {
   try {
-    const gasEstimate = await foreignBridge.methods.executeSignatures(message, signatures).estimateGas()
-    return gasEstimate
+    return await foreignBridge.methods.executeSignatures(message, signatures).estimateGas()
   } catch (e) {
     if (e instanceof HttpListProviderError) {
       throw e

oracle/src/events/processSignatureRequests/estimateGas.js

@@ -6,10 +6,9 @@ const logger = require('../../services/logger').child({
 
 async function estimateGas({ web3, homeBridge, validatorContract, signature, message, address }) {
   try {
-    const gasEstimate = await homeBridge.methods.submitSignature(signature, message).estimateGas({
+    return await homeBridge.methods.submitSignature(signature, message).estimateGas({
       from: address
     })
-    return gasEstimate
   } catch (e) {
     if (e instanceof HttpListProviderError) {
       throw e

oracle/src/mevSender.js

@@ -0,0 +1,159 @@
+require('../env')
+const path = require('path')
+const BigNumber = require('bignumber.js')
+const { redis } = require('./services/redisClient')
+const logger = require('./services/logger')
+const { sendTx } = require('./tx/sendTx')
+const { getNonce, getChainId, getBlock } = require('./tx/web3')
+const { addExtraGas, checkHTTPS, watchdog } = require('./utils/utils')
+const { EXIT_CODES, EXTRA_GAS_PERCENTAGE, MAX_GAS_LIMIT } = require('./utils/constants')
+const { estimateProfit } = require('./events/processAMBCollectedSignaturesMEV')
+
+if (process.argv.length < 3) {
+  logger.error('Please check the number of arguments, config file was not provided')
+  process.exit(EXIT_CODES.GENERAL_ERROR)
+}
+
+const config = require(path.join('../config/', process.argv[2]))
+
+const { web3, mevForeign, validatorAddress } = config
+
+let chainId = 0
+let flashbotsProvider
+
+async function initialize() {
+  try {
+    const checkHttps = checkHTTPS(process.env.ORACLE_ALLOW_HTTP_FOR_RPC, logger)
+
+    web3.currentProvider.urls.forEach(checkHttps(config.id))
+
+    chainId = await getChainId(web3)
+    flashbotsProvider = await mevForeign.getFlashbotsProvider(chainId)
+    return runMain()
+  } catch (e) {
+    logger.error(e.message)
+    process.exit(EXIT_CODES.GENERAL_ERROR)
+  }
+}
+
+async function runMain() {
+  try {
+    if (redis.status === 'ready') {
+      if (config.maxProcessingTime) {
+        await watchdog(main, config.maxProcessingTime, () => {
+          logger.fatal('Max processing time reached')
+          process.exit(EXIT_CODES.MAX_TIME_REACHED)
+        })
+      } else {
+        await main()
+      }
+    }
+  } catch (e) {
+    logger.error(e)
+  }
+
+  setTimeout(runMain, config.pollingInterval)
+}
+
+async function main() {
+  try {
+    const jobs = Object.values(await redis.hgetall(config.mevJobsRedisKey)).map(JSON.parse)
+    const totalJobs = jobs.length
+
+    if (totalJobs === 0) {
+      logger.debug('Nothing to process')
+      return
+    }
+
+    const { baseFeePerGas: pendingBaseFee, number: pendingBlockNumber } = await getBlock(web3, 'pending')
+    const bestJob = pickBestJob(jobs, pendingBaseFee)
+
+    if (!bestJob) {
+      logger.info({ totalJobs, pendingBaseFee }, 'No suitable job was found, waiting for a lower gas price')
+      return
+    }
+
+    const jobLogger = logger.child({ eventTransactionHash: bestJob.transactionReference })
+
+    const maxProfit = await estimateProfit(
+      mevForeign.contract,
+      mevForeign.minGasPrice,
+      bestJob.executeData,
+      bestJob.value
+    )
+
+    if (maxProfit === '0') {
+      jobLogger.info(`No MEV opportunity found when testing with min gas price ${mevForeign.minGasPrice}, removing job`)
+      await redis.hdel(config.mevJobsRedisKey, bestJob.transactionReference)
+      return
+    }
+    jobLogger.info(`Estimated profit of ${maxProfit} when simulating with ${mevForeign.minGasPrice} gas price`)
+    bestJob.profit = maxProfit
+
+    if (new BigNumber(pendingBaseFee).gt(mevForeign.minGasPrice)) {
+      const profit = await estimateProfit(mevForeign.contract, pendingBaseFee, bestJob.executeData, bestJob.value)
+      if (profit === '0') {
+        jobLogger.info(
+          `No MEV opportunity found when testing with current gas price ${pendingBaseFee}, waiting for lower gas price`
+        )
+        bestJob.maxFeePerGas = pendingBaseFee
+        await redis.hset(config.mevJobsRedisKey, bestJob.transactionReference, JSON.stringify(bestJob))
+        return
+      }
+      jobLogger.info(`Estimated profit of ${profit} when simulating with ${pendingBaseFee} gas price`)
+    }
+
+    let gasLimit
+    if (typeof bestJob.extraGas === 'number') {
+      gasLimit = addExtraGas(bestJob.gasEstimate + bestJob.extraGas, 0, MAX_GAS_LIMIT)
+    } else {
+      gasLimit = addExtraGas(bestJob.gasEstimate, EXTRA_GAS_PERCENTAGE, MAX_GAS_LIMIT)
+    }
+
+    const nonce = await getNonce(web3, validatorAddress)
+    jobLogger.info(
+      { nonce, fromBlock: pendingBlockNumber, toBlock: pendingBlockNumber + mevForeign.bundlesPerIteration - 1 },
+      'Sending MEV bundles'
+    )
+    const txHash = await sendTx({
+      data: bestJob.data,
+      nonce,
+      value: bestJob.value,
+      gasLimit,
+      privateKey: config.validatorPrivateKey,
+      to: bestJob.to,
+      chainId,
+      web3,
+      gasPriceOptions: {
+        maxFeePerGas: bestJob.maxFeePerGas,
+        maxPriorityFeePerGas: bestJob.maxPriorityFeePerGas
+      },
+      mevOptions: {
+        provider: flashbotsProvider,
+        fromBlock: pendingBlockNumber,
+        toBlock: pendingBlockNumber + mevForeign.bundlesPerIteration - 1,
+        logger
+      }
+    })
+
+    jobLogger.info({ txHash }, `Tx generated ${txHash} for event Tx ${bestJob.transactionReference}`)
+
+    await redis.hset(config.mevJobsRedisKey, bestJob.transactionReference, JSON.stringify(bestJob))
+    jobLogger.debug(`Finished processing msg`)
+  } catch (e) {
+    logger.error(e)
+  }
+}
+
+function pickBestJob(jobs, feePerGas) {
+  const feePerGasBN = new BigNumber(feePerGas)
+  let best = null
+  jobs.forEach(job => {
+    if (feePerGasBN.lt(job.maxFeePerGas) && (!best || new BigNumber(best.profit).lt(job.profit))) {
+      best = job
+    }
+  })
+  return best
+}
+
+initialize()

oracle/src/mevWatcher.js

@@ -0,0 +1,251 @@
+require('../env')
+const path = require('path')
+const { redis } = require('./services/redisClient')
+const logger = require('./services/logger')
+const { getBlockNumber, getRequiredBlockConfirmations, getEvents } = require('./tx/web3')
+const { checkHTTPS, watchdog, syncForEach } = require('./utils/utils')
+const { processCollectedSignaturesBuilder } = require('./events/processAMBCollectedSignaturesMEV')
+const {
+  EXIT_CODES,
+  BLOCK_NUMBER_PROGRESS_ITERATIONS_LIMIT,
+  MAX_HISTORY_BLOCK_TO_REPROCESS
+} = require('./utils/constants')
+
+if (process.argv.length < 3) {
+  logger.error('Please check the number of arguments, config file was not provided')
+  process.exit(EXIT_CODES.GENERAL_ERROR)
+}
+
+const config = require(path.join('../config/', process.argv[2]))
+
+const processAMBCollectedSignaturesMEV = processCollectedSignaturesBuilder(config)
+
+const {
+  web3,
+  bridgeContract,
+  eventContract,
+  startBlock,
+  pollingInterval,
+  chain,
+  reprocessingOptions,
+  blockPollingLimit
+} = config.main
+const lastBlockRedisKey = `${config.id}:lastProcessedBlock`
+const lastReprocessedBlockRedisKey = `${config.id}:lastReprocessedBlock`
+const seenEventsRedisKey = `${config.id}:seenEvents`
+const mevJobsRedisKey = `${config.id}:mevJobs`
+let lastProcessedBlock = Math.max(startBlock - 1, 0)
+let lastReprocessedBlock
+let lastSeenBlockNumber = 0
+let sameBlockNumberCounter = 0
+
+async function initialize() {
+  try {
+    const checkHttps = checkHTTPS(process.env.ORACLE_ALLOW_HTTP_FOR_RPC, logger)
+
+    web3.currentProvider.urls.forEach(checkHttps(chain))
+
+    await getLastProcessedBlock()
+    await getLastReprocessedBlock()
+    runMain({ sendToQueue: saveJobsToRedis })
+  } catch (e) {
+    logger.error(e)
+    process.exit(EXIT_CODES.GENERAL_ERROR)
+  }
+}
+
+async function runMain({ sendToQueue }) {
+  try {
+    if (redis.status === 'ready') {
+      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)
+  }
+
+  setTimeout(() => {
+    runMain({ sendToQueue })
+  }, pollingInterval)
+}
+
+async function saveJobsToRedis(jobs) {
+  return syncForEach(jobs, job => redis.hset(mevJobsRedisKey, job.transactionReference, JSON.stringify(job)))
+}
+
+async function getLastProcessedBlock() {
+  const result = await redis.get(lastBlockRedisKey)
+  logger.debug({ fromRedis: result, fromConfig: lastProcessedBlock }, 'Last Processed block obtained')
+  lastProcessedBlock = result ? parseInt(result, 10) : lastProcessedBlock
+}
+
+async function getLastReprocessedBlock() {
+  if (reprocessingOptions.enabled) {
+    const result = await redis.get(lastReprocessedBlockRedisKey)
+    if (result) {
+      lastReprocessedBlock = Math.max(parseInt(result, 10), lastProcessedBlock - MAX_HISTORY_BLOCK_TO_REPROCESS)
+    } else {
+      lastReprocessedBlock = lastProcessedBlock
+    }
+    logger.debug({ block: lastReprocessedBlock }, 'Last reprocessed block obtained')
+  } else {
+    // when reprocessing is being enabled not for the first time,
+    // we do not want to process blocks for which we didn't recorded seen events,
+    // instead, we want to start from the current block.
+    // Thus we should delete this reprocessing pointer once it is disabled.
+    await redis.del(lastReprocessedBlockRedisKey)
+  }
+}
+
+function updateLastProcessedBlock(lastBlockNumber) {
+  lastProcessedBlock = lastBlockNumber
+  return redis.set(lastBlockRedisKey, lastProcessedBlock)
+}
+
+function updateLastReprocessedBlock(lastBlockNumber) {
+  lastReprocessedBlock = lastBlockNumber
+  return redis.set(lastReprocessedBlockRedisKey, lastReprocessedBlock)
+}
+
+function processEvents(events) {
+  switch (config.id) {
+    case 'amb-collected-signatures-mev':
+      return processAMBCollectedSignaturesMEV(events)
+    default:
+      return []
+  }
+}
+
+const eventKey = e => `${e.transactionHash}-${e.logIndex}`
+
+async function reprocessOldLogs(sendToQueue) {
+  const fromBlock = lastReprocessedBlock + 1
+  const toBlock = lastReprocessedBlock + reprocessingOptions.batchSize
+  const events = await getEvents({
+    contract: eventContract,
+    event: config.event,
+    fromBlock,
+    toBlock,
+    filter: config.eventFilter
+  })
+  const alreadySeenEvents = await getSeenEvents(fromBlock, toBlock)
+  const missingEvents = events.filter(e => !alreadySeenEvents[eventKey(e)])
+  if (missingEvents.length === 0) {
+    logger.debug('No missed events were found')
+  } else {
+    logger.info(`Found ${missingEvents.length} ${config.event} missed events`)
+    const job = await processEvents(missingEvents)
+    logger.info('Missed events transactions to send:', job.length)
+    if (job.length) {
+      await sendToQueue(job)
+    }
+  }
+
+  await updateLastReprocessedBlock(toBlock)
+  await deleteSeenEvents(0, toBlock)
+}
+
+async function getSeenEvents(fromBlock, toBlock) {
+  const keys = await redis.zrangebyscore(seenEventsRedisKey, fromBlock, toBlock)
+  const res = {}
+  keys.forEach(k => {
+    res[k] = true
+  })
+  return res
+}
+
+function deleteSeenEvents(fromBlock, toBlock) {
+  return redis.zremrangebyscore(seenEventsRedisKey, fromBlock, toBlock)
+}
+
+function addSeenEvents(events) {
+  return redis.zadd(seenEventsRedisKey, ...events.flatMap(e => [e.blockNumber, eventKey(e)]))
+}
+
+async function getLastBlockToProcess(web3, bridgeContract) {
+  const [lastBlockNumber, requiredBlockConfirmations] = await Promise.all([
+    getBlockNumber(web3),
+    getRequiredBlockConfirmations(bridgeContract)
+  ])
+
+  if (lastBlockNumber < lastSeenBlockNumber) {
+    sameBlockNumberCounter = 0
+    logger.warn({ lastBlockNumber, lastSeenBlockNumber }, 'Received block number less than already seen block')
+    web3.currentProvider.switchToFallbackRPC()
+  } else if (lastBlockNumber === lastSeenBlockNumber) {
+    sameBlockNumberCounter++
+    if (sameBlockNumberCounter > 1) {
+      logger.info({ lastBlockNumber, sameBlockNumberCounter }, 'Received the same block number more than twice')
+      if (sameBlockNumberCounter >= BLOCK_NUMBER_PROGRESS_ITERATIONS_LIMIT) {
+        sameBlockNumberCounter = 0
+        logger.warn(
+          { lastBlockNumber, n: BLOCK_NUMBER_PROGRESS_ITERATIONS_LIMIT },
+          'Received the same block number for too many times. Probably node is not synced anymore'
+        )
+        web3.currentProvider.switchToFallbackRPC()
+      }
+    }
+  } else {
+    sameBlockNumberCounter = 0
+    lastSeenBlockNumber = lastBlockNumber
+  }
+  return lastBlockNumber - requiredBlockConfirmations
+}
+
+async function main({ sendToQueue }) {
+  try {
+    const lastBlockToProcess = await getLastBlockToProcess(web3, bridgeContract)
+
+    if (reprocessingOptions.enabled) {
+      if (lastReprocessedBlock + reprocessingOptions.batchSize + reprocessingOptions.blockDelay < lastBlockToProcess) {
+        await reprocessOldLogs(sendToQueue)
+        return
+      }
+    }
+
+    if (lastBlockToProcess <= lastProcessedBlock) {
+      logger.debug('All blocks already processed')
+      return
+    }
+
+    const fromBlock = lastProcessedBlock + 1
+    const rangeEndBlock = blockPollingLimit ? fromBlock + blockPollingLimit : lastBlockToProcess
+    const toBlock = Math.min(lastBlockToProcess, rangeEndBlock)
+
+    const events = await getEvents({
+      contract: eventContract,
+      event: config.event,
+      fromBlock,
+      toBlock,
+      filter: config.eventFilter
+    })
+    logger.info(`Found ${events.length} ${config.event} events`)
+
+    if (events.length) {
+      const job = await processEvents(events)
+      logger.info('Transactions to send:', job.length)
+
+      if (job.length) {
+        await sendToQueue(job)
+      }
+      if (reprocessingOptions.enabled) {
+        await addSeenEvents(events)
+      }
+    }
+
+    logger.debug({ lastProcessedBlock: toBlock.toString() }, 'Updating last processed block')
+    await updateLastProcessedBlock(toBlock)
+  } catch (e) {
+    logger.error(e)
+  }
+
+  logger.debug('Finished')
+}
+
+initialize()

oracle/src/sender.js

@@ -169,7 +169,7 @@ async function main({ msg, ackMsg, nackMsg, channel, scheduleForRetry, scheduleT
         const txHash = await sendTx({
           data: job.data,
           nonce,
-          amount: '0',
+          value: '0',
           gasLimit,
           privateKey: config.validatorPrivateKey,
           to: job.to,

oracle/src/services/web3.js

@@ -1,4 +1,6 @@
 const Web3 = require('web3')
+const ethers = require('ethers')
+const flashbots = require('@flashbots/ethers-provider-bundle')
 const { HttpListProvider } = require('./HttpListProvider')
 const { SafeEthLogsProvider } = require('./SafeEthLogsProvider')
 const { RedundantHttpListProvider } = require('./RedundantHttpListProvider')
@@ -9,6 +11,8 @@ const {
   COMMON_FOREIGN_RPC_URL,
   ORACLE_SIDE_RPC_URL,
   ORACLE_FOREIGN_ARCHIVE_RPC_URL,
+  ORACLE_MEV_FOREIGN_FLASHBOTS_RPC_URL,
+  ORACLE_MEV_FOREIGN_FLASHBOTS_AUTH_SIGNING_KEY,
   ORACLE_RPC_REQUEST_TIMEOUT,
   ORACLE_HOME_RPC_POLLING_INTERVAL,
   ORACLE_FOREIGN_RPC_POLLING_INTERVAL
@@ -94,6 +98,15 @@ if (foreignUrls.length > 1) {
   web3ForeignRedundant = new Web3(redundantProvider)
 }
 
+let getFlashbotsProvider
+if (ORACLE_MEV_FOREIGN_FLASHBOTS_RPC_URL) {
+  const provider = new ethers.providers.JsonRpcProvider(foreignUrls[0])
+  const authSigner = new ethers.Wallet(ORACLE_MEV_FOREIGN_FLASHBOTS_AUTH_SIGNING_KEY, provider)
+
+  getFlashbotsProvider = chainId =>
+    flashbots.FlashbotsBundleProvider.create(provider, authSigner, ORACLE_MEV_FOREIGN_FLASHBOTS_RPC_URL, chainId)
+}
+
 module.exports = {
   web3Home,
   web3Foreign,
@@ -102,5 +115,6 @@ module.exports = {
   web3HomeRedundant,
   web3ForeignRedundant,
   web3HomeFallback,
-  web3ForeignFallback
+  web3ForeignFallback,
+  getFlashbotsProvider
 }

oracle/src/tx/sendTx.js

@@ -1,6 +1,5 @@
-const { toWei } = require('web3').utils
-
-async function sendTx({ privateKey, data, nonce, gasPrice, gasPriceOptions, amount, gasLimit, to, chainId, web3 }) {
+async function sendTx(opts) {
+  const { privateKey, data, nonce, gasPrice, gasPriceOptions, value, gasLimit, to, chainId, web3, mevOptions } = opts
   const gasOpts = gasPriceOptions || { gasPrice }
   const serializedTx = await web3.eth.accounts.signTransaction(
     {
@@ -8,19 +7,32 @@ async function sendTx({ privateKey, data, nonce, gasPrice, gasPriceOptions, amou
       chainId,
       to,
       data,
-      value: toWei(amount),
+      value,
       gas: gasLimit,
       ...gasOpts
     },
     privateKey
   )
 
-  return new Promise((res, rej) =>
-    web3.eth
-      .sendSignedTransaction(serializedTx.rawTransaction)
-      .once('transactionHash', res)
-      .once('error', rej)
+  if (!mevOptions) {
+    return new Promise((res, rej) =>
+      web3.eth
+        .sendSignedTransaction(serializedTx.rawTransaction)
+        .once('transactionHash', res)
+        .once('error', rej)
+    )
+  }
+
+  mevOptions.logger.debug(
+    { rawTx: serializedTx.rawTransaction, txHash: serializedTx.transactionHash },
+    'Signed MEV helper transaction'
   )
+
+  for (let blockNumber = mevOptions.fromBlock; blockNumber <= mevOptions.toBlock; blockNumber++) {
+    mevOptions.logger.debug({ txHash: serializedTx.transactionHash, blockNumber }, 'Sending MEV bundle transaction')
+    await mevOptions.provider.sendRawBundle([serializedTx.rawTransaction], blockNumber)
+  }
+  return Promise.resolve(serializedTx.transactionHash)
 }
 
 module.exports = {

oracle/src/utils/mev.js

@@ -0,0 +1,43 @@
+const MEV_HELPER_ABI = [
+  {
+    constant: false,
+    inputs: [
+      {
+        name: '_data',
+        type: 'bytes'
+      }
+    ],
+    name: 'execute',
+    outputs: [],
+    payable: false,
+    stateMutability: 'nonpayable',
+    type: 'function'
+  },
+  {
+    constant: false,
+    inputs: [
+      {
+        name: '_gasPrice',
+        type: 'uint256'
+      },
+      {
+        name: '_data',
+        type: 'bytes'
+      }
+    ],
+    name: 'estimateProfit',
+    outputs: [
+      {
+        name: '',
+        type: 'uint256'
+      }
+    ],
+    payable: true,
+    stateMutability: 'nonpayable',
+    type: 'function'
+  }
+]
+
+module.exports = {
+  MEV_HELPER_ABI
+}

oracle/src/watcher.js

@@ -30,7 +30,16 @@ const processAMBInformationRequests = require('./events/processAMBInformationReq
 
 const { getTokensState } = require('./utils/tokenState')
 
-const { web3, bridgeContract, eventContract, startBlock, pollingInterval, chain, reprocessingOptions } = config.main
+const {
+  web3,
+  bridgeContract,
+  eventContract,
+  startBlock,
+  pollingInterval,
+  chain,
+  reprocessingOptions,
+  blockPollingLimit
+} = config.main
 const lastBlockRedisKey = `${config.id}:lastProcessedBlock`
 const lastReprocessedBlockRedisKey = `${config.id}:lastReprocessedBlock`
 const seenEventsRedisKey = `${config.id}:seenEvents`
@@ -268,7 +277,7 @@ async function main({ sendToQueue }) {
     }
 
     const fromBlock = lastProcessedBlock + 1
-    const rangeEndBlock = config.blockPollingLimit ? fromBlock + config.blockPollingLimit : lastBlockToProcess
+    const rangeEndBlock = blockPollingLimit ? fromBlock + blockPollingLimit : lastBlockToProcess
     let toBlock = Math.min(lastBlockToProcess, rangeEndBlock)
 
     let events = await getEvents({