Merge pull request #244 from poanetwork/support-two-tokens-oracle

Support two tokens deposit requests in Oracle

commons/abis.js

@@ -13,6 +13,7 @@ const REWARDABLE_VALIDATORS_ABI = require('../contracts/build/contracts/Rewardab
 const HOME_AMB_ABI = require('../contracts/build/contracts/HomeAMB').abi
 const FOREIGN_AMB_ABI = require('../contracts/build/contracts/ForeignAMB').abi
 const BOX_ABI = require('../contracts/build/contracts/Box').abi
+const SAI_TOP = require('../contracts/build/contracts/SaiTopMock').abi
 
 const { HOME_V1_ABI, FOREIGN_V1_ABI } = require('./v1Abis')
 const { BRIDGE_MODES } = require('./constants')
@@ -92,5 +93,6 @@ module.exports = {
   ERC20_BYTES32_ABI,
   HOME_AMB_ABI,
   FOREIGN_AMB_ABI,
-  BOX_ABI
+  BOX_ABI,
+  SAI_TOP
 }

contracts

@@ -1 +1 @@
-Subproject commit 20d262702defd3aed39870f3379f1cf0c314305e
+Subproject commit 055bcb62a794ed913dfec10b6558e3b57a314151

deployment-e2e/molecule/ultimate-commons/oracle-add-docker-external-network.yml

@@ -0,0 +1,26 @@
+---
+- name: Slurp docker compose file
+  slurp:
+    src: "/home/poadocker/bridge/oracle/{{ item }}.yml"
+  register: docker_compose_slurp
+- name: Parse docker compose file
+  set_fact:
+    docker_compose_parsed: "{{ docker_compose_slurp['content'] | b64decode | from_yaml }}"
+
+- name: Add the external network used to connect to Parity nodes
+  set_fact:
+    docker_compose_parsed: "{{ docker_compose_parsed |combine({'networks': {'ultimate': {'external': 'true'}}}, recursive=True) }}"
+
+- name: Add all Oracle containers to the network
+  set_fact:
+    docker_compose_parsed: "{{ docker_compose_parsed | combine({'services': {item: {'networks': docker_compose_parsed.services[item].networks | union(['ultimate'])}}}, recursive=True) }}"
+  with_items: "{{ docker_compose_parsed.services }}"
+
+- name: Expose Redis port to allow connecting from redis-cli
+  set_fact:
+    docker_compose_parsed: "{{ docker_compose_parsed | combine({'services': {'redis': {'ports': ['6379:6379']}}}, recursive=True) }}"
+
+- name: Write updated docker file
+  copy:
+    content: "{{ docker_compose_parsed | to_yaml }}"
+    dest: "/home/poadocker/bridge/oracle/{{ item }}.yml"

deployment-e2e/molecule/ultimate-commons/oracle-docker-compose.yml

@@ -5,58 +5,12 @@
   tasks:
   - name: stop the service
     shell: service poabridge stop
-  
-  - name: Slurp docker compose file
-    slurp:
-      src: "/home/poadocker/bridge/oracle/docker-compose.yml"
-    register: docker_compose_slurp
-  - name: Parse docker compose file
-    set_fact:
-      docker_compose_parsed: "{{ docker_compose_slurp['content'] | b64decode | from_yaml }}"
 
-  - name: Add the external network used to connect to Parity nodes
-    set_fact:
-      docker_compose_parsed: "{{ docker_compose_parsed |combine({'networks': {'ultimate': {'external': 'true'}}}, recursive=True) }}"
-
-  - name: Add all Oracle containers to the network
-    set_fact:
-      docker_compose_parsed: "{{ docker_compose_parsed | combine({'services': {item: {'networks': docker_compose_parsed.services[item].networks | union(['ultimate'])}}}, recursive=True) }}"
-    with_items: "{{ docker_compose_parsed.services }}"
-  
-  - name: Expose Redis port to allow connecting from redis-cli
-    set_fact:
-      docker_compose_parsed: "{{ docker_compose_parsed | combine({'services': {'redis': {'ports': ['6379:6379']}}}, recursive=True) }}"
-
-  - name: Write new docker-compose file
-    copy:
-      content: "{{ docker_compose_parsed | to_yaml }}"
-      dest: "/home/poadocker/bridge/oracle/docker-compose.yml"
-
-  - name: Slurp docker compose extended file
-    slurp:
-      src: "/home/poadocker/bridge/oracle/docker-compose-transfer.yml"
-    register: docker_compose_extended_slurp
-  - name: Parse docker compose file
-    set_fact:
-      docker_compose_extended_parsed: "{{ docker_compose_extended_slurp['content'] | b64decode | from_yaml }}"
-
-  - name: Add the external network used to connect to Parity nodes in compose extended file
-    set_fact:
-      docker_compose_extended_parsed: "{{ docker_compose_extended_parsed |combine({'networks': {'ultimate': {'external': 'true'}}}, recursive=True) }}"
-
-  - name: Add all Oracle containers to the network in compose extended file
-    set_fact:
-      docker_compose_extended_parsed: "{{ docker_compose_extended_parsed | combine({'services': {item: {'networks': docker_compose_extended_parsed.services[item].networks | union(['ultimate'])}}}, recursive=True) }}"
-    with_items: "{{ docker_compose_extended_parsed.services }}"
-
-  - name: Expose Redis port to allow connecting from redis-cli in compose extended file
-    set_fact:
-      docker_compose_extended_parsed: "{{ docker_compose_extended_parsed | combine({'services': {'redis': {'ports': ['6379:6379']}}}, recursive=True) }}"
-
-  - name: Write new docker-compose extended file
-    copy:
-      content: "{{ docker_compose_extended_parsed | to_yaml }}"
-      dest: "/home/poadocker/bridge/oracle/docker-compose-transfer.yml"
+  - include_tasks: oracle-add-docker-external-network.yml
+    with_items:
+      - docker-compose
+      - docker-compose-transfer
+      - docker-compose-erc-native
 
   - name: start the service
     shell: service poabridge start

deployment/roles/oracle/tasks/logging.yml

@@ -1,41 +1,9 @@
 ---
-- name: Slurp docker compose file
-  slurp:
-    src: "{{ bridge_path }}/oracle/docker-compose.yml"
-  register: docker_compose_slurp
-
-- name: Parse docker compose file
-  set_fact:
-    docker_compose_parsed: "{{ docker_compose_slurp['content'] | b64decode | from_yaml }}"
-
-- name: Set logger to remote server
-  set_fact:
-    docker_compose_parsed: "{{ docker_compose_parsed |combine({'services': {item: {'logging': {'driver': 'syslog','options': {'tag': '{{.Name}}/{{.ID}}'}}}}}, recursive=True) }}"
-  with_items: "{{ docker_compose_parsed.services }}"
-
-- name: Write new docker-compose file
-  copy:
-    content: "{{ docker_compose_parsed | to_yaml }}"
-    dest: "{{ bridge_path }}/oracle/docker-compose.yml"
-
-- name: Slurp docker compose extended file
-  slurp:
-    src: "{{ bridge_path }}/oracle/docker-compose-transfer.yml"
-  register: docker_compose_extended_slurp
-
-- name: Parse docker compose extended file
-  set_fact:
-    docker_compose_extended_parsed: "{{ docker_compose_extended_slurp['content'] | b64decode | from_yaml }}"
-
-- name: Set logger to remote server for extended file
-  set_fact:
-    docker_compose_extended_parsed: "{{ docker_compose_extended_parsed |combine({'services': {item: {'logging': {'driver': 'syslog','options': {'tag': '{{.Name}}/{{.ID}}'}}}}}, recursive=True) }}"
-  with_items: "{{ docker_compose_extended_parsed.services }}"
-
-- name: Write new docker-compose-extended file
-  copy:
-    content: "{{ docker_compose_extended_parsed | to_yaml }}"
-    dest: "{{ bridge_path }}/oracle/docker-compose-transfer.yml"
+- include_tasks: logging_by_syslog.yml
+  with_items:
+    - docker-compose
+    - docker-compose-transfer
+    - docker-compose-erc-native
 
 - name: Set the local container logs configuration file
   template:

deployment/roles/oracle/tasks/logging_by_syslog.yml

@@ -0,0 +1,19 @@
+---
+- name: Slurp docker compose file
+  slurp:
+    src: "{{ bridge_path }}/oracle/{{ item }}.yml"
+  register: docker_compose_slurp
+
+- name: Parse docker compose file
+  set_fact:
+    docker_compose_parsed: "{{ docker_compose_slurp['content'] | b64decode | from_yaml }}"
+
+- name: Set logger to remote server
+  set_fact:
+    docker_compose_parsed: "{{ docker_compose_parsed |combine({'services': {item: {'logging': {'driver': 'syslog','options': {'tag': '{{.Name}}/{{.ID}}'}}}}}, recursive=True) }}"
+  with_items: "{{ docker_compose_parsed.services }}"
+
+- name: Write updated docker file
+  copy:
+    content: "{{ docker_compose_parsed | to_yaml }}"
+    dest: "{{ bridge_path }}/oracle/{{ item }}.yml"

deployment/roles/oracle/tasks/post_config.yml

@@ -40,7 +40,11 @@
 
 - name: Extend docker compose file
   set_fact: composefileoverride="-f docker-compose-transfer.yml"
-  when: ORACLE_BRIDGE_MODE == "ERC_TO_NATIVE" or ( ORACLE_BRIDGE_MODE == "ERC_TO_ERC" and FOREIGN_ERC_TYPE != "ERC677")
+  when: ORACLE_BRIDGE_MODE == "ERC_TO_ERC" and FOREIGN_ERC_TYPE != "ERC677"
+
+- name: Extend docker compose file for erc to native
+  set_fact: composefileoverride="-f docker-compose-erc-native.yml"
+  when: ORACLE_BRIDGE_MODE == "ERC_TO_NATIVE"
 
 - name: Install .key config
   template:

e2e-commons/constants.json

@@ -33,7 +33,9 @@
   "ercToNativeBridge": {
     "home": "0x488Af810997eD1730cB3a3918cD83b3216E6eAda",
     "foreign": "0x488Af810997eD1730cB3a3918cD83b3216E6eAda",
-    "foreignToken": "0x3C665A31199694Bf723fD08844AD290207B5797f",
+    "foreignToken": "0x7cc4b1851c35959d34e635a470f6b5c43ba3c9c9",
+    "halfDuplexToken": "0x89d24A6b4CcB1B6fAA2625fE562bDD9a23260359",
+    "saiTop": "0x9b0ccf7C8994E19F39b2B4CF708e0A7DF65fA8a3",
     "ui": "http://localhost:3002",
     "monitor": "http://monitor-erc20-native:3012"
   },

e2e-commons/contracts-envs/erc-to-native.env

@@ -32,7 +32,7 @@ FOREIGN_GAS_PRICE=10000000000
 FOREIGN_REWARDABLE=false
 
 BLOCK_REWARD_ADDRESS=0xF9698Eb93702dfdd0e2d802088d4c21822a8A977
-ERC20_TOKEN_ADDRESS=0x3C665A31199694Bf723fD08844AD290207B5797f
+ERC20_TOKEN_ADDRESS=0x89d24A6b4CcB1B6fAA2625fE562bDD9a23260359
 
 REQUIRED_NUMBER_OF_VALIDATORS=1
 VALIDATORS="0xaaB52d66283F7A1D5978bcFcB55721ACB467384b"

e2e-commons/up.sh

@@ -22,6 +22,8 @@ while [ "$1" != "" ]; do
     docker-compose run -d oracle-erc20-native yarn watcher:collected-signatures
     docker-compose run -d oracle-erc20-native yarn watcher:affirmation-request
     docker-compose run -d oracle-erc20-native yarn watcher:transfer
+    docker-compose run -d oracle-erc20-native yarn watcher:half-duplex-transfer
+    docker-compose run -d oracle-erc20-native yarn worker:swap-tokens
     docker-compose run -d oracle-amb yarn watcher:signature-request
     docker-compose run -d oracle-amb yarn watcher:collected-signatures
     docker-compose run -d oracle-amb yarn watcher:affirmation-request

monitor-e2e/test/ercToNative.js

@@ -25,7 +25,7 @@ describe('ERC TO NATIVE with changing state of contracts', () => {
   })
 
   it('should change balanceDiff', async () => {
-    await sendTokens(foreignRPC.URL, user, ercToNativeBridge.foreignToken, ercToNativeBridge.foreign)
+    await sendTokens(foreignRPC.URL, user, ercToNativeBridge.halfDuplexToken, ercToNativeBridge.foreign)
 
     await waitUntil(async () => {
       ;({ data } = await axios.get(`${baseUrl}`))

oracle-e2e/test/ercToNative.js

@@ -1,8 +1,15 @@
 const Web3 = require('web3')
 const assert = require('assert')
 const promiseRetry = require('promise-retry')
-const { user, secondUser, ercToNativeBridge, homeRPC, foreignRPC } = require('../../e2e-commons/constants.json')
-const { ERC677_BRIDGE_TOKEN_ABI, FOREIGN_ERC_TO_NATIVE_ABI } = require('../../commons')
+const {
+  user,
+  secondUser,
+  validator,
+  ercToNativeBridge,
+  homeRPC,
+  foreignRPC
+} = require('../../e2e-commons/constants.json')
+const { ERC677_BRIDGE_TOKEN_ABI, FOREIGN_ERC_TO_NATIVE_ABI, SAI_TOP } = require('../../commons')
 const { generateNewBlock } = require('../../e2e-commons/utils')
 
 const homeWeb3 = new Web3(new Web3.providers.HttpProvider(homeRPC.URL))
@@ -15,11 +22,134 @@ const { toBN } = foreignWeb3.utils
 
 homeWeb3.eth.accounts.wallet.add(user.privateKey)
 foreignWeb3.eth.accounts.wallet.add(user.privateKey)
+foreignWeb3.eth.accounts.wallet.add(validator.privateKey)
 
 const erc20Token = new foreignWeb3.eth.Contract(ERC677_BRIDGE_TOKEN_ABI, ercToNativeBridge.foreignToken)
 const foreignBridge = new foreignWeb3.eth.Contract(FOREIGN_ERC_TO_NATIVE_ABI, COMMON_FOREIGN_BRIDGE_ADDRESS)
 
 describe('erc to native', () => {
+  let halfDuplexTokenAddress
+  let halfDuplexToken
+  before(async () => {
+    halfDuplexTokenAddress = await foreignBridge.methods.halfDuplexErc20token().call()
+    halfDuplexToken = new foreignWeb3.eth.Contract(ERC677_BRIDGE_TOKEN_ABI, halfDuplexTokenAddress)
+  })
+  it('should continue working after migration', async () => {
+    const originalBalanceOnHome = await homeWeb3.eth.getBalance(user.address)
+
+    const transferValue = homeWeb3.utils.toWei('0.01')
+
+    // erc20 token address and half duplex address are the same before migration
+    const tokenAddress = await foreignBridge.methods.erc20token().call()
+
+    const erc20AndhalfDuplexToken = new foreignWeb3.eth.Contract(ERC677_BRIDGE_TOKEN_ABI, tokenAddress)
+
+    // send tokens to foreign bridge
+    await erc20AndhalfDuplexToken.methods
+      .transfer(COMMON_FOREIGN_BRIDGE_ADDRESS, transferValue)
+      .send({
+        from: user.address,
+        gas: '1000000'
+      })
+      .catch(e => {
+        console.error(e)
+      })
+
+    // Send a trivial transaction to generate a new block since the watcher
+    // is configured to wait 1 confirmation block
+    await generateNewBlock(foreignWeb3, user.address)
+
+    // check that balance increases
+    await promiseRetry(async (retry, number) => {
+      const balance = await homeWeb3.eth.getBalance(user.address)
+      await generateNewBlock(foreignWeb3, user.address)
+      // retry at least 4 times to check transfer is not double processed by the two watchers
+      if (toBN(balance).lte(toBN(originalBalanceOnHome)) || number < 4) {
+        retry()
+      } else {
+        assert(
+          toBN(balance).eq(toBN(originalBalanceOnHome).add(toBN(transferValue))),
+          'User balance should be increased only by second transfer'
+        )
+      }
+    })
+
+    // call migration
+    await foreignBridge.methods.migrateToMCD().send({
+      from: validator.address,
+      gas: '4000000'
+    })
+
+    // update min threshold for swap
+    await foreignBridge.methods.setMinHDTokenBalance(foreignWeb3.utils.toWei('1', 'ether')).send({
+      from: validator.address,
+      gas: '1000000'
+    })
+
+    const AfterMigrateBalance = await homeWeb3.eth.getBalance(user.address)
+
+    // send tokens to foreign bridge
+    await erc20Token.methods
+      .transfer(COMMON_FOREIGN_BRIDGE_ADDRESS, transferValue)
+      .send({
+        from: user.address,
+        gas: '1000000'
+      })
+      .catch(e => {
+        console.error(e)
+      })
+
+    // Send a trivial transaction to generate a new block since the watcher
+    // is configured to wait 1 confirmation block
+    await generateNewBlock(foreignWeb3, user.address)
+
+    // check that balance increases
+    await promiseRetry(async (retry, number) => {
+      const balance = await homeWeb3.eth.getBalance(user.address)
+      await generateNewBlock(foreignWeb3, user.address)
+      // retry at least 4 times to check transfer is not double processed by the two watchers
+      if (toBN(balance).lte(toBN(AfterMigrateBalance)) || number < 4) {
+        retry()
+      } else {
+        assert(
+          toBN(balance).eq(toBN(AfterMigrateBalance).add(toBN(transferValue))),
+          'User balance should be increased only by second transfer'
+        )
+      }
+    })
+
+    const afterMigrateAndTransferBalance = await homeWeb3.eth.getBalance(user.address)
+
+    // send tokens to foreign bridge
+    await halfDuplexToken.methods
+      .transfer(COMMON_FOREIGN_BRIDGE_ADDRESS, transferValue)
+      .send({
+        from: user.address,
+        gas: '1000000'
+      })
+      .catch(e => {
+        console.error(e)
+      })
+
+    // Send a trivial transaction to generate a new block since the watcher
+    // is configured to wait 1 confirmation block
+    await generateNewBlock(foreignWeb3, user.address)
+
+    // check that balance increases
+    await promiseRetry(async (retry, number) => {
+      const balance = await homeWeb3.eth.getBalance(user.address)
+      await generateNewBlock(foreignWeb3, user.address)
+      // retry at least 4 times to check transfer is not double processed by the two watchers
+      if (toBN(balance).lte(toBN(afterMigrateAndTransferBalance)) || number < 4) {
+        retry()
+      } else {
+        assert(
+          toBN(balance).eq(toBN(afterMigrateAndTransferBalance).add(toBN(transferValue))),
+          'User balance should be increased only by second transfer'
+        )
+      }
+    })
+  })
   it('should convert tokens in foreign to coins in home', async () => {
     const balance = await erc20Token.methods.balanceOf(user.address).call()
     const originalBalanceOnHome = await homeWeb3.eth.getBalance(user.address)
@@ -92,6 +222,238 @@ describe('erc to native', () => {
       }
     })
   })
+  it('should convert half duplex token in foreign to native token in home', async () => {
+    const originalBalanceOnHome = await homeWeb3.eth.getBalance(user.address)
+    const bridgeErc20TokenBalance = await erc20Token.methods.balanceOf(COMMON_FOREIGN_BRIDGE_ADDRESS).call()
+    const bridgeHalfDuplexBalance = await halfDuplexToken.methods.balanceOf(COMMON_FOREIGN_BRIDGE_ADDRESS).call()
+
+    const valueToTransfer = foreignWeb3.utils.toWei('1', 'ether')
+
+    // this transfer won't trigger a call to swap tokens
+    await halfDuplexToken.methods.transfer(COMMON_FOREIGN_BRIDGE_ADDRESS, valueToTransfer).send({
+      from: user.address,
+      gas: '1000000'
+    })
+
+    // Send a trivial transaction to generate a new block since the watcher
+    // is configured to wait 1 confirmation block
+    await generateNewBlock(foreignWeb3, user.address)
+
+    // check that balance increases
+    await promiseRetry(async retry => {
+      const balance = await homeWeb3.eth.getBalance(user.address)
+      if (toBN(balance).lte(toBN(originalBalanceOnHome))) {
+        retry()
+      } else {
+        assert(
+          toBN(balance).eq(toBN(originalBalanceOnHome).add(toBN(valueToTransfer))),
+          'User balance should be increased by the half duplex token transfer'
+        )
+      }
+    })
+
+    const updatedBalanceOnHome = await homeWeb3.eth.getBalance(user.address)
+    const updatedBridgeHalfDuplexBalance = await halfDuplexToken.methods.balanceOf(COMMON_FOREIGN_BRIDGE_ADDRESS).call()
+    assert(
+      toBN(updatedBridgeHalfDuplexBalance).eq(toBN(bridgeHalfDuplexBalance).add(toBN(valueToTransfer))),
+      'Bridge balance should reflect the transfer value'
+    )
+
+    // this transfer will trigger call to swap tokens
+    await halfDuplexToken.methods.transfer(COMMON_FOREIGN_BRIDGE_ADDRESS, valueToTransfer).send({
+      from: user.address,
+      gas: '1000000'
+    })
+
+    await generateNewBlock(foreignWeb3, user.address)
+
+    await promiseRetry(async retry => {
+      const userBalance = await homeWeb3.eth.getBalance(user.address)
+      const updatedBridgeErc20TokenBalance = await erc20Token.methods.balanceOf(COMMON_FOREIGN_BRIDGE_ADDRESS).call()
+
+      if (
+        toBN(userBalance).lte(toBN(updatedBalanceOnHome)) ||
+        toBN(updatedBridgeErc20TokenBalance).lte(toBN(bridgeErc20TokenBalance))
+      ) {
+        retry()
+      } else {
+        assert(
+          toBN(userBalance).eq(toBN(updatedBalanceOnHome).add(toBN(valueToTransfer))),
+          'User balance should be increased by the half duplex token transfer'
+        )
+        const updatedBalance = await halfDuplexToken.methods.balanceOf(COMMON_FOREIGN_BRIDGE_ADDRESS).call()
+        assert(toBN(updatedBalance).isZero(), 'Half duplex bridge balance should be zero')
+        assert(
+          toBN(updatedBridgeErc20TokenBalance).eq(
+            toBN(bridgeErc20TokenBalance)
+              .add(toBN(bridgeHalfDuplexBalance))
+              .add(toBN(foreignWeb3.utils.toWei('2', 'ether')))
+          ),
+          'Erc20 token balance should be correctly increased by the token swap'
+        )
+      }
+    })
+  })
+  it('should convert half duplex token in foreign to native token in home for alternative receiver ', async () => {
+    const originalBalanceOnHome = await homeWeb3.eth.getBalance(user.address)
+    const initialBalanceSecondUser = await homeWeb3.eth.getBalance(secondUser.address)
+    const bridgeErc20TokenBalance = await erc20Token.methods.balanceOf(COMMON_FOREIGN_BRIDGE_ADDRESS).call()
+
+    const valueToTransfer = foreignWeb3.utils.toWei('1', 'ether')
+
+    // approve tokens to foreign bridge
+    await halfDuplexToken.methods
+      .approve(COMMON_FOREIGN_BRIDGE_ADDRESS, valueToTransfer)
+      .send({
+        from: user.address,
+        gas: '1000000'
+      })
+      .catch(e => {
+        console.error(e)
+      })
+
+    // call bridge method to transfer tokens to a different recipient
+    await foreignBridge.methods['relayTokens(address,uint256,address)'](
+      secondUser.address,
+      valueToTransfer,
+      halfDuplexTokenAddress
+    )
+      .send({
+        from: user.address,
+        gas: '1000000'
+      })
+      .catch(e => {
+        console.error(e)
+      })
+
+    // Send a trivial transaction to generate a new block since the watcher
+    // is configured to wait 1 confirmation block
+    await generateNewBlock(foreignWeb3, user.address)
+
+    // check that balance increases
+    await promiseRetry(async retry => {
+      const secondUserbalance = await homeWeb3.eth.getBalance(secondUser.address)
+      const updatedBridgeErc20TokenBalance = await erc20Token.methods.balanceOf(COMMON_FOREIGN_BRIDGE_ADDRESS).call()
+      const userbalance = await homeWeb3.eth.getBalance(user.address)
+      assert(toBN(userbalance).lte(toBN(originalBalanceOnHome)), 'User balance should be the same')
+      if (
+        toBN(secondUserbalance).lte(toBN(initialBalanceSecondUser)) ||
+        toBN(updatedBridgeErc20TokenBalance).lte(toBN(bridgeErc20TokenBalance))
+      ) {
+        retry()
+      } else {
+        assert(
+          toBN(secondUserbalance).eq(toBN(initialBalanceSecondUser).add(toBN(valueToTransfer))),
+          'User balance should be increased by the half duplex token transfer'
+        )
+        const updatedHDBalance = await halfDuplexToken.methods.balanceOf(COMMON_FOREIGN_BRIDGE_ADDRESS).call()
+        assert(toBN(updatedHDBalance).isZero(), 'Half duplex bridge balance should be zero')
+        assert(
+          toBN(updatedBridgeErc20TokenBalance).eq(toBN(bridgeErc20TokenBalance).add(toBN(valueToTransfer))),
+          'Erc20 token balance should be correctly increased by the token swap'
+        )
+      }
+    })
+  })
+  it('should not relay half duplex token transfer after Emergency Shutdown', async () => {
+    const originalBalanceOnHome = await homeWeb3.eth.getBalance(user.address)
+    const bridgeErc20TokenBalance = await erc20Token.methods.balanceOf(COMMON_FOREIGN_BRIDGE_ADDRESS).call()
+    const bridgeHalfDuplexBalance = await halfDuplexToken.methods.balanceOf(COMMON_FOREIGN_BRIDGE_ADDRESS).call()
+
+    const block = await foreignWeb3.eth.getBlock('latest')
+    const saiTop = new foreignWeb3.eth.Contract(SAI_TOP, ercToNativeBridge.saiTop)
+
+    // Trigger Emergency Shutdown
+    await saiTop.methods
+      .setCaged(block.timestamp)
+      .send({
+        from: user.address,
+        gas: '1000000'
+      })
+      .catch(e => {
+        console.error(e)
+      })
+
+    const valueToTransfer = foreignWeb3.utils.toWei('1', 'ether')
+
+    await generateNewBlock(foreignWeb3, user.address)
+
+    // this transfer won't trigger a call to swap tokens
+    await halfDuplexToken.methods.transfer(COMMON_FOREIGN_BRIDGE_ADDRESS, valueToTransfer).send({
+      from: user.address,
+      gas: '1000000'
+    })
+
+    // Send a trivial transaction to generate a new block since the watcher
+    // is configured to wait 1 confirmation block
+    await generateNewBlock(foreignWeb3, user.address)
+
+    // check that transfer and swap are not processed in the next blocks.
+    await promiseRetry(async (retry, number) => {
+      const balanceOnHome = await homeWeb3.eth.getBalance(user.address)
+      const currentBridgeErc20TokenBalance = await erc20Token.methods.balanceOf(COMMON_FOREIGN_BRIDGE_ADDRESS).call()
+      const currentBridgeHalfDuplexBalance = await halfDuplexToken.methods
+        .balanceOf(COMMON_FOREIGN_BRIDGE_ADDRESS)
+        .call()
+
+      assert(toBN(balanceOnHome).eq(toBN(originalBalanceOnHome)), 'User balance should be the same')
+      assert(
+        toBN(currentBridgeHalfDuplexBalance).eq(toBN(bridgeHalfDuplexBalance).add(toBN(valueToTransfer))),
+        'Half duplex balance should be the value of transfer'
+      )
+      assert(toBN(currentBridgeErc20TokenBalance).eq(toBN(bridgeErc20TokenBalance)), 'erc20 balance should not change')
+
+      // generate new blocks
+      await generateNewBlock(foreignWeb3, user.address)
+
+      // after several retries, the state is corrects
+      if (number < 4) {
+        retry()
+      }
+    })
+
+    // let's undo the Emergency Shutdown to check that the oracle is still working
+    await saiTop.methods.setCaged('0').send({
+      from: user.address,
+      gas: '1000000'
+    })
+
+    const newValueToTransfer = foreignWeb3.utils.toWei('2', 'ether')
+
+    await halfDuplexToken.methods.transfer(COMMON_FOREIGN_BRIDGE_ADDRESS, newValueToTransfer).send({
+      from: user.address,
+      gas: '1000000'
+    })
+
+    await generateNewBlock(foreignWeb3, user.address)
+
+    await promiseRetry(async retry => {
+      const userBalance = await homeWeb3.eth.getBalance(user.address)
+      const updatedBridgeErc20TokenBalance = await erc20Token.methods.balanceOf(COMMON_FOREIGN_BRIDGE_ADDRESS).call()
+
+      if (
+        toBN(userBalance).lte(toBN(originalBalanceOnHome)) ||
+        toBN(updatedBridgeErc20TokenBalance).lte(toBN(bridgeErc20TokenBalance))
+      ) {
+        retry()
+      } else {
+        assert(
+          toBN(userBalance).eq(toBN(originalBalanceOnHome).add(toBN(newValueToTransfer))),
+          'User balance should be increased by the half duplex token transfer'
+        )
+        const updatedHDBalance = await halfDuplexToken.methods.balanceOf(COMMON_FOREIGN_BRIDGE_ADDRESS).call()
+        assert(toBN(updatedHDBalance).isZero(), 'Half duplex bridge balance should be zero')
+        assert(
+          toBN(updatedBridgeErc20TokenBalance).eq(
+            toBN(bridgeErc20TokenBalance)
+              .add(toBN(valueToTransfer))
+              .add(toBN(newValueToTransfer))
+          ),
+          'Erc20 token balance should be correctly increased by the token swap'
+        )
+      }
+    })
+  })
   it('should convert coins in home to tokens in foreign', async () => {
     const originalBalance = await erc20Token.methods.balanceOf(user.address).call()
 

oracle/config/half-duplex-transfer-watcher.config.js

@@ -0,0 +1,39 @@
+const baseConfig = require('./base.config')
+const { ERC20_ABI } = require('../../commons')
+const { EXIT_CODES } = require('../src/utils/constants')
+
+const initialChecksJson = process.argv[3]
+
+if (!initialChecksJson) {
+  throw new Error('initial check parameter was not provided.')
+}
+
+let initialChecks
+try {
+  initialChecks = JSON.parse(initialChecksJson)
+} catch (e) {
+  throw new Error('Error on decoding values from initial checks.')
+}
+
+const id = `${baseConfig.id}-half-duplex-transfer`
+
+const transferWatcherRequired = baseConfig.id === 'erc-native'
+
+if (!transferWatcherRequired) {
+  console.error(`Transfer watcher not required for bridge mode ${process.env.ORACLE_BRIDGE_MODE}`)
+  process.exit(EXIT_CODES.WATCHER_NOT_REQUIRED)
+}
+
+module.exports = {
+  ...baseConfig.bridgeConfig,
+  ...baseConfig.foreignConfig,
+  event: 'Transfer',
+  eventContractAddress: initialChecks.halfDuplexTokenAddress,
+  eventAbi: ERC20_ABI,
+  eventFilter: { to: process.env.COMMON_FOREIGN_BRIDGE_ADDRESS },
+  queue: 'home',
+  workerQueue: 'swap-tokens',
+  name: `watcher-${id}`,
+  id,
+  idle: initialChecks.idle
+}

oracle/config/swap-tokens-worker.config.js

@@ -0,0 +1,20 @@
+const baseConfig = require('./base.config')
+const { EXIT_CODES } = require('../src/utils/constants')
+
+const id = `${baseConfig.id}-swap-tokens`
+
+const workerRequired = baseConfig.id === 'erc-native'
+
+if (!workerRequired) {
+  console.error(`Swap tokens worker not required for bridge mode ${process.env.ORACLE_BRIDGE_MODE}`)
+  process.exit(EXIT_CODES.WATCHER_NOT_REQUIRED)
+}
+
+module.exports = {
+  ...baseConfig.bridgeConfig,
+  ...baseConfig.foreignConfig,
+  workerQueue: 'swap-tokens',
+  senderQueue: 'foreign',
+  name: `worker-${id}`,
+  id
+}

oracle/docker-compose-erc-native.yml

@@ -0,0 +1,126 @@
+---
+version: '2.4'
+services:
+  rabbit:
+    extends:
+      file: docker-compose.yml
+      service: rabbit
+    networks:
+      - net_rabbit_bridge_transfer
+  redis:
+    extends:
+      file: docker-compose.yml
+      service: redis
+    networks:
+      - net_db_bridge_transfer
+  bridge_request:
+    extends:
+      file: docker-compose.yml
+      service: bridge_request
+    networks:
+      - net_db_bridge_request
+      - net_rabbit_bridge_request
+  bridge_collected:
+    extends:
+      file: docker-compose.yml
+      service: bridge_collected
+    networks:
+      - net_db_bridge_request
+      - net_rabbit_bridge_request
+  bridge_affirmation:
+    extends:
+      file: docker-compose.yml
+      service: bridge_affirmation
+    networks:
+      - net_db_bridge_request
+      - net_rabbit_bridge_request
+  bridge_transfer:
+    cpus: 0.1
+    mem_limit: 500m
+    build:
+      context: ..
+      dockerfile: oracle/Dockerfile
+    env_file: ./.env
+    environment:
+      - NODE_ENV=production
+      - ORACLE_VALIDATOR_ADDRESS=${ORACLE_VALIDATOR_ADDRESS}
+    restart: unless-stopped
+    entrypoint: yarn watcher:transfer
+    networks:
+      - net_db_bridge_transfer
+      - net_rabbit_bridge_transfer
+  bridge_half_duplex_transfer:
+    cpus: 0.1
+    mem_limit: 500m
+    build:
+      context: ..
+      dockerfile: oracle/Dockerfile
+    env_file: ./.env
+    environment:
+      - NODE_ENV=production
+      - ORACLE_VALIDATOR_ADDRESS=${ORACLE_VALIDATOR_ADDRESS}
+    restart: unless-stopped
+    entrypoint: yarn watcher:half-duplex-transfer
+    networks:
+      - net_db_bridge_half_duplex_transfer
+      - net_rabbit_bridge_half_duplex_transfer
+  bridge_swap_tokens_worker:
+    cpus: 0.1
+    mem_limit: 500m
+    build:
+      context: ..
+      dockerfile: oracle/Dockerfile
+    env_file: ./.env
+    environment:
+      - NODE_ENV=production
+      - ORACLE_VALIDATOR_ADDRESS=${ORACLE_VALIDATOR_ADDRESS}
+    restart: unless-stopped
+    entrypoint: yarn worker:swap-tokens
+    networks:
+      - net_rabbit_bridge_swap_tokens_worker
+  bridge_senderhome:
+    extends:
+      file: docker-compose.yml
+      service: bridge_senderhome
+    networks:
+      - net_db_bridge_request
+      - net_rabbit_bridge_request
+  bridge_senderforeign:
+    extends:
+      file: docker-compose.yml
+      service: bridge_senderforeign
+    networks:
+      - net_db_bridge_request
+      - net_rabbit_bridge_request
+
+networks:
+  net_db_bridge_request:
+    driver: bridge
+  net_db_bridge_collected:
+    driver: bridge
+  net_db_bridge_affirmation:
+    driver: bridge
+  net_db_bridge_transfer:
+    driver: bridge
+  net_db_bridge_half_duplex_transfer:
+    driver: bridge
+  net_db_bridge_senderhome:
+    driver: bridge
+  net_db_bridge_senderforeign:
+    driver: bridge
+  net_rabbit_bridge_request:
+    driver: bridge
+  net_rabbit_bridge_collected:
+    driver: bridge
+  net_rabbit_bridge_affirmation:
+    driver: bridge
+  net_rabbit_bridge_transfer:
+    driver: bridge
+  net_rabbit_bridge_half_duplex_transfer:
+    driver: bridge
+  net_rabbit_bridge_swap_tokens_worker:
+    driver: bridge
+  net_rabbit_bridge_senderhome:
+    driver: bridge
+  net_rabbit_bridge_senderforeign:
+    driver: bridge

oracle/package.json

@@ -9,9 +9,11 @@
     "watcher:collected-signatures": "./scripts/start-worker.sh watcher collected-signatures-watcher",
     "watcher:affirmation-request": "./scripts/start-worker.sh watcher affirmation-request-watcher",
     "watcher:transfer": "./scripts/start-worker.sh watcher transfer-watcher",
+    "watcher:half-duplex-transfer": "./scripts/start-worker.sh watcher half-duplex-transfer-watcher",
+    "worker:swap-tokens": "./scripts/start-worker.sh worker swap-tokens-worker",
     "sender:home": "./scripts/start-worker.sh sender home-sender",
     "sender:foreign": "./scripts/start-worker.sh sender foreign-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'",
+    "dev": "concurrently -n 'watcher:signature-request,watcher:collected-signatures,watcher:affirmation-request,watcher:transfer,watcher:half-duplex-transfer, worker:swap-tokens, sender:home,sender:foreign' -c 'red,green,yellow,blue,white,gray,magenta,cyan' 'yarn watcher:signature-request' 'yarn watcher:collected-signatures' 'yarn watcher:affirmation-request' 'yarn watcher:transfer' 'yarn watcher:half-duplex-transfer' 'yarn worker:swap-tokens' 'yarn sender:home' 'yarn sender:foreign'",
     "test": "NODE_ENV=test mocha",
     "test:watch": "NODE_ENV=test mocha --watch --reporter=min",
     "coverage": "NODE_ENV=test nyc --reporter=text --reporter=html mocha",

oracle/scripts/initialChecks.js

@@ -1,5 +1,6 @@
 require('../env')
 const Web3 = require('web3')
+const { getTokensState } = require('../src/utils/tokenState')
 const {
   ERC677_BRIDGE_TOKEN_ABI,
   FOREIGN_ERC_TO_ERC_ABI,
@@ -9,7 +10,7 @@ const {
 
 async function initialChecks() {
   const { ORACLE_BRIDGE_MODE, COMMON_FOREIGN_RPC_URL, COMMON_FOREIGN_BRIDGE_ADDRESS } = process.env
-  const result = {}
+  let result = {}
   const foreignWeb3 = new Web3(new Web3.providers.HttpProvider(COMMON_FOREIGN_RPC_URL))
 
   if (ORACLE_BRIDGE_MODE === 'ERC_TO_ERC') {
@@ -17,7 +18,7 @@ async function initialChecks() {
     result.bridgeableTokenAddress = await bridge.methods.erc20token().call()
   } else if (ORACLE_BRIDGE_MODE === 'ERC_TO_NATIVE') {
     const bridge = new foreignWeb3.eth.Contract(FOREIGN_ERC_TO_NATIVE_ABI, COMMON_FOREIGN_BRIDGE_ADDRESS)
-    result.bridgeableTokenAddress = await bridge.methods.erc20token().call()
+    result = await getTokensState(bridge)
   }
 
   if (ORACLE_BRIDGE_MODE === 'ERC_TO_ERC') {

oracle/src/events/processHalfDuplexTransfers/index.js

@@ -0,0 +1,134 @@
+require('../../../env')
+const promiseLimit = require('promise-limit')
+const { HttpListProviderError } = require('http-list-provider')
+const { BRIDGE_VALIDATORS_ABI, ZERO_ADDRESS } = require('../../../../commons')
+const rootLogger = require('../../services/logger')
+const { web3Home, web3Foreign } = require('../../services/web3')
+const { AlreadyProcessedError, AlreadySignedError, InvalidValidatorError } = require('../../utils/errors')
+const { EXIT_CODES, MAX_CONCURRENT_EVENTS } = require('../../utils/constants')
+const estimateGas = require('../processAffirmationRequests/estimateGas')
+
+const limit = promiseLimit(MAX_CONCURRENT_EVENTS)
+
+let validatorContract = null
+
+function processTransfersBuilder(config) {
+  const homeBridge = new web3Home.eth.Contract(config.homeBridgeAbi, config.homeBridgeAddress)
+  const foreignBridge = new web3Foreign.eth.Contract(config.foreignBridgeAbi, config.foreignBridgeAddress)
+  const userRequestForAffirmationAbi = config.foreignBridgeAbi.filter(
+    e => e.type === 'event' && e.name === 'UserRequestForAffirmation'
+  )[0]
+  const tokensSwappedAbi = config.foreignBridgeAbi.filter(e => e.type === 'event' && e.name === 'TokensSwapped')[0]
+  const userRequestForAffirmationHash = web3Home.eth.abi.encodeEventSignature(userRequestForAffirmationAbi)
+  const tokensSwappedHash = tokensSwappedAbi ? web3Home.eth.abi.encodeEventSignature(tokensSwappedAbi) : '0x'
+
+  return async function processTransfers(transfers, blockNumber) {
+    const txToSend = []
+
+    if (validatorContract === null) {
+      rootLogger.debug('Getting validator contract address')
+      const validatorContractAddress = await homeBridge.methods.validatorContract().call()
+      rootLogger.debug({ validatorContractAddress }, 'Validator contract address obtained')
+
+      validatorContract = new web3Home.eth.Contract(BRIDGE_VALIDATORS_ABI, validatorContractAddress)
+    }
+
+    rootLogger.debug(`Processing ${transfers.length} Transfer events`)
+    const callbacks = transfers
+      .map(transfer => async () => {
+        const { from, value } = transfer.returnValues
+
+        const logger = rootLogger.child({
+          eventTransactionHash: transfer.transactionHash
+        })
+
+        logger.info({ from, value }, `Processing transfer ${transfer.transactionHash}`)
+
+        const block = await web3Foreign.eth.getBlock(blockNumber)
+        logger.debug({ blockNumber, timestamp: block.timestamp }, `Block obtained`)
+
+        const tokenSwapAllowed = await foreignBridge.methods.isTokenSwapAllowed(block.timestamp).call()
+
+        if (!tokenSwapAllowed) {
+          logger.info(
+            `Transfer event discarded because SCD Emergency Shutdown has happened ${transfer.transactionHash}`
+          )
+          return
+        }
+
+        const receipt = await web3Foreign.eth.getTransactionReceipt(transfer.transactionHash)
+
+        const existsAffirmationEvent = receipt.logs.some(
+          e => e.address === config.foreignBridgeAddress && e.topics[0] === userRequestForAffirmationHash
+        )
+
+        if (existsAffirmationEvent) {
+          logger.info(
+            `Transfer event discarded because a transaction with alternative receiver detected in transaction ${
+              transfer.transactionHash
+            }`
+          )
+          return
+        }
+
+        const existsTokensSwappedEvent = tokensSwappedAbi
+          ? receipt.logs.some(e => e.address === config.foreignBridgeAddress && e.topics[0] === tokensSwappedHash)
+          : false
+
+        if (from === ZERO_ADDRESS && existsTokensSwappedEvent) {
+          logger.info(
+            `Transfer event discarded because token swap is detected in transaction ${transfer.transactionHash}`
+          )
+          return
+        }
+
+        let gasEstimate
+        try {
+          logger.debug('Estimate gas')
+          gasEstimate = await estimateGas({
+            web3: web3Home,
+            homeBridge,
+            validatorContract,
+            recipient: from,
+            value,
+            txHash: transfer.transactionHash,
+            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 InvalidValidatorError) {
+            logger.fatal({ address: config.validatorAddress }, 'Invalid validator')
+            process.exit(EXIT_CODES.INCOMPATIBILITY)
+          } else if (e instanceof AlreadySignedError) {
+            logger.info(`Already signed transfer ${transfer.transactionHash}`)
+            return
+          } else if (e instanceof AlreadyProcessedError) {
+            logger.info(`transfer ${transfer.transactionHash} was already processed by other validators`)
+            return
+          } else {
+            logger.error(e, 'Unknown error while processing transaction')
+            throw e
+          }
+        }
+
+        const data = await homeBridge.methods
+          .executeAffirmation(from, value, transfer.transactionHash)
+          .encodeABI({ from: config.validatorAddress })
+
+        txToSend.push({
+          data,
+          gasEstimate,
+          transactionReference: transfer.transactionHash,
+          to: config.homeBridgeAddress
+        })
+      })
+      .map(promise => limit(promise))
+
+    await Promise.all(callbacks)
+    return txToSend
+  }
+}
+
+module.exports = processTransfersBuilder

oracle/src/services/amqpClient.js

@@ -11,17 +11,23 @@ connection.on('disconnect', () => {
   logger.error('Disconnected from amqp Broker')
 })
 
-function connectWatcherToQueue({ queueName, cb }) {
+function connectWatcherToQueue({ queueName, workerQueue, cb }) {
+  const queueList = workerQueue ? [queueName, workerQueue] : [queueName]
+
   const channelWrapper = connection.createChannel({
     json: true,
     setup(channel) {
-      return Promise.all([channel.assertQueue(queueName, { durable: true })])
+      return Promise.all(queueList.map(queue => channel.assertQueue(queue, { durable: true })))
     }
   })
 
   const sendToQueue = data => channelWrapper.sendToQueue(queueName, data, { persistent: true })
+  let sendToWorker
+  if (workerQueue) {
+    sendToWorker = data => channelWrapper.sendToQueue(workerQueue, data, { persistent: true })
+  }
 
-  cb({ sendToQueue, channel: channelWrapper })
+  cb({ sendToQueue, sendToWorker, channel: channelWrapper })
 }
 
 function connectSenderToQueue({ queueName, cb }) {
@@ -59,6 +65,43 @@ function connectSenderToQueue({ queueName, cb }) {
   })
 }
 
+function connectWorkerToQueue({ queueName, senderQueue, cb }) {
+  const deadLetterExchange = `${queueName}-retry`
+
+  const channelWrapper = connection.createChannel({
+    json: true
+  })
+
+  channelWrapper.addSetup(channel => {
+    return Promise.all([
+      channel.assertExchange(deadLetterExchange, 'fanout', { durable: true }),
+      channel.assertQueue(queueName, { durable: true }),
+      channel.assertQueue(senderQueue, { durable: true }),
+      channel.bindQueue(queueName, deadLetterExchange),
+      channel.prefetch(1),
+      channel.consume(queueName, msg =>
+        cb({
+          msg,
+          channel: channelWrapper,
+          ackMsg: job => channelWrapper.ack(job),
+          nackMsg: job => channelWrapper.nack(job, false, true),
+          sendToSenderQueue: data => channelWrapper.sendToQueue(senderQueue, data, { persistent: true }),
+          scheduleForRetry: async (data, msgRetries = 0) => {
+            await generateRetry({
+              data,
+              msgRetries,
+              channelWrapper,
+              channel,
+              queueName,
+              deadLetterExchange
+            })
+          }
+        })
+      )
+    ])
+  })
+}
+
 async function generateRetry({ data, msgRetries, channelWrapper, channel, queueName, deadLetterExchange }) {
   const retries = msgRetries + 1
   const delay = getRetrySequence(retries) * 1000
@@ -78,6 +121,7 @@ async function generateRetry({ data, msgRetries, channelWrapper, channel, queueN
 module.exports = {
   connectWatcherToQueue,
   connectSenderToQueue,
+  connectWorkerToQueue,
   connection,
   generateRetry
 }

oracle/src/utils/tokenState.js

@@ -0,0 +1,20 @@
+async function getTokensState(bridgeContract) {
+  const context = {}
+  context.bridgeableTokenAddress = await bridgeContract.methods.erc20token().call()
+  try {
+    const halfDuplexErc20tokenAddress = await bridgeContract.methods.halfDuplexErc20token().call()
+    if (halfDuplexErc20tokenAddress !== context.bridgeableTokenAddress) {
+      context.halfDuplexTokenAddress = halfDuplexErc20tokenAddress
+    } else {
+      context.idle = true
+    }
+  } catch (e) {
+    context.idle = true
+  }
+
+  return context
+}
+
+module.exports = {
+  getTokensState
+}

oracle/src/watcher.js

@@ -21,16 +21,21 @@ const processSignatureRequests = require('./events/processSignatureRequests')(co
 const processCollectedSignatures = require('./events/processCollectedSignatures')(config)
 const processAffirmationRequests = require('./events/processAffirmationRequests')(config)
 const processTransfers = require('./events/processTransfers')(config)
+const processHalfDuplexTransfers = require('./events/processHalfDuplexTransfers')(config)
 const processAMBSignatureRequests = require('./events/processAMBSignatureRequests')(config)
 const processAMBCollectedSignatures = require('./events/processAMBCollectedSignatures')(config)
 const processAMBAffirmationRequests = require('./events/processAMBAffirmationRequests')(config)
 
+const { getTokensState } = require('./utils/tokenState')
+
 const ZERO = toBN(0)
 const ONE = toBN(1)
 
 const web3Instance = config.web3
 const bridgeContract = new web3Instance.eth.Contract(config.bridgeAbi, config.bridgeContractAddress)
-const eventContract = new web3Instance.eth.Contract(config.eventAbi, config.eventContractAddress)
+let { eventContractAddress } = config
+let eventContract = new web3Instance.eth.Contract(config.eventAbi, eventContractAddress)
+let skipEvents = config.idle
 const lastBlockRedisKey = `${config.id}:lastProcessedBlock`
 let lastProcessedBlock = BN.max(config.startBlock.sub(ONE), ZERO)
 
@@ -44,6 +49,7 @@ async function initialize() {
     await getLastProcessedBlock()
     connectWatcherToQueue({
       queueName: config.queue,
+      workerQueue: config.workerQueue,
       cb: runMain
     })
   } catch (e) {
@@ -52,16 +58,16 @@ async function initialize() {
   }
 }
 
-async function runMain({ sendToQueue }) {
+async function runMain({ sendToQueue, sendToWorker }) {
   try {
     if (connection.isConnected() && redis.status === 'ready') {
       if (config.maxProcessingTime) {
-        await watchdog(() => main({ sendToQueue }), config.maxProcessingTime, () => {
+        await watchdog(() => main({ sendToQueue, sendToWorker }), config.maxProcessingTime, () => {
           logger.fatal('Max processing time reached')
           process.exit(EXIT_CODES.MAX_TIME_REACHED)
         })
       } else {
-        await main({ sendToQueue })
+        await main({ sendToQueue, sendToWorker })
       }
     }
   } catch (e) {
@@ -69,7 +75,7 @@ async function runMain({ sendToQueue }) {
   }
 
   setTimeout(() => {
-    runMain({ sendToQueue })
+    runMain({ sendToQueue, sendToWorker })
   }, config.pollingInterval)
 }
 
@@ -84,7 +90,7 @@ function updateLastProcessedBlock(lastBlockNumber) {
   return redis.set(lastBlockRedisKey, lastProcessedBlock.toString())
 }
 
-function processEvents(events) {
+function processEvents(events, blockNumber) {
   switch (config.id) {
     case 'native-erc-signature-request':
     case 'erc-erc-signature-request':
@@ -102,6 +108,8 @@ function processEvents(events) {
     case 'erc-erc-transfer':
     case 'erc-native-transfer':
       return processTransfers(events)
+    case 'erc-native-half-duplex-transfer':
+      return processHalfDuplexTransfers(events, blockNumber)
     case 'amb-signature-request':
       return processAMBSignatureRequests(events)
     case 'amb-collected-signatures':
@@ -113,6 +121,29 @@ function processEvents(events) {
   }
 }
 
+async function checkConditions() {
+  let state
+  switch (config.id) {
+    case 'erc-native-transfer':
+      state = await getTokensState(bridgeContract)
+      updateEventContract(state.bridgeableTokenAddress)
+      break
+    case 'erc-native-half-duplex-transfer':
+      state = await getTokensState(bridgeContract)
+      skipEvents = state.idle
+      updateEventContract(state.halfDuplexTokenAddress)
+      break
+    default:
+  }
+}
+
+function updateEventContract(address) {
+  if (eventContractAddress !== address) {
+    eventContractAddress = address
+    eventContract = new web3Instance.eth.Contract(config.eventAbi, eventContractAddress)
+  }
+}
+
 async function getLastBlockToProcess() {
   const lastBlockNumberPromise = getBlockNumber(web3Instance).then(toBN)
   const requiredBlockConfirmationsPromise = getRequiredBlockConfirmations(bridgeContract).then(toBN)
@@ -124,8 +155,15 @@ async function getLastBlockToProcess() {
   return lastBlockNumber.sub(requiredBlockConfirmations)
 }
 
-async function main({ sendToQueue }) {
+async function main({ sendToQueue, sendToWorker }) {
   try {
+    await checkConditions()
+
+    if (skipEvents) {
+      logger.debug('Watcher in idle mode, skipping getting events')
+      return
+    }
+
     const lastBlockToProcess = await getLastBlockToProcess()
 
     if (lastBlockToProcess.lte(lastProcessedBlock)) {
@@ -146,7 +184,11 @@ async function main({ sendToQueue }) {
     logger.info(`Found ${events.length} ${config.event} events`)
 
     if (events.length) {
-      const job = await processEvents(events)
+      if (sendToWorker) {
+        await sendToWorker({ blockNumber: toBlock.toString() })
+      }
+
+      const job = await processEvents(events, toBlock.toString())
       logger.info('Transactions to send:', job.length)
 
       if (job.length) {

oracle/src/worker.js

@@ -0,0 +1,73 @@
+const path = require('path')
+const logger = require('./services/logger')
+const rpcUrlsManager = require('./services/getRpcUrlsManager')
+const { checkHTTPS, watchdog } = require('./utils/utils')
+const { EXIT_CODES } = require('./utils/constants')
+const { connectWorkerToQueue } = require('./services/amqpClient')
+
+const config = require(path.join('../config/', process.argv[2]))
+
+const swapTokens = require('./workers/swapTokens')(config)
+
+async function initialize() {
+  try {
+    const checkHttps = checkHTTPS(process.env.ORACLE_ALLOW_HTTP_FOR_RPC, logger)
+
+    rpcUrlsManager.homeUrls.forEach(checkHttps('home'))
+    rpcUrlsManager.foreignUrls.forEach(checkHttps('foreign'))
+
+    connectWorkerToQueue({
+      queueName: config.workerQueue,
+      senderQueue: config.senderQueue,
+      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(EXIT_CODES.GENERAL_ERROR)
+  }
+}
+
+async function run(blockNumber) {
+  if (config.id === 'erc-native-swap-tokens') {
+    return swapTokens(blockNumber)
+  } else {
+    return []
+  }
+}
+
+async function main({ msg, ackMsg, nackMsg, sendToSenderQueue, scheduleForRetry }) {
+  try {
+    const { blockNumber } = JSON.parse(msg.content)
+    logger.info(`Msg received with block number ${blockNumber}`)
+
+    try {
+      const job = await run(blockNumber)
+
+      logger.info('Transactions to send:', job.length)
+
+      if (job.length) {
+        await sendToSenderQueue(job)
+      }
+    } catch (e) {
+      logger.info(`Sending failed msg to retry`)
+      await scheduleForRetry({ blockNumber }, msg.properties.headers['x-retries'])
+    }
+
+    ackMsg(msg)
+  } catch (e) {
+    logger.error(e)
+    nackMsg(msg)
+  }
+  logger.debug(`Finished worker operation`)
+}
+
+initialize()

oracle/src/workers/swapTokens.js

@@ -0,0 +1,111 @@
+require('../../env')
+const { HttpListProviderError } = require('http-list-provider')
+const rootLogger = require('../services/logger')
+const { web3Foreign } = require('../services/web3')
+
+const { BRIDGE_VALIDATORS_ABI, ERC20_ABI } = require('../../../commons')
+
+let validatorContract = null
+let halfDuplexTokenContract = null
+
+function swapTokensBuilder(config) {
+  const foreignBridge = new web3Foreign.eth.Contract(config.foreignBridgeAbi, config.foreignBridgeAddress)
+
+  return async function swapTokens(blockNumber) {
+    const txToSend = []
+
+    const logger = rootLogger.child({
+      blockNumber: blockNumber.toString()
+    })
+
+    logger.debug(`Starting swap tokens operation`)
+
+    if (validatorContract === null) {
+      logger.debug('Getting validator contract address')
+      const validatorContractAddress = await foreignBridge.methods.validatorContract().call()
+      logger.debug({ validatorContractAddress }, 'Validator contract address obtained')
+
+      validatorContract = new web3Foreign.eth.Contract(BRIDGE_VALIDATORS_ABI, validatorContractAddress)
+    }
+
+    logger.debug(`Checking if is validator duty`)
+    const validatorDuty = await validatorContract.methods.isValidatorDuty(config.validatorAddress).call()
+
+    if (!validatorDuty) {
+      logger.info(`Token swap discarded because is not validator duty`)
+      return txToSend
+    }
+
+    logger.debug(`Checking if half duplex token balance is above the threshold`)
+    const hdTokenBalanceAboveMinBalance = await foreignBridge.methods.isHDTokenBalanceAboveMinBalance().call()
+
+    if (!hdTokenBalanceAboveMinBalance) {
+      logger.info(`Token swap discarded because half duplex balance is below the threshold`)
+      return txToSend
+    }
+
+    const block = await web3Foreign.eth.getBlock(blockNumber)
+    logger.debug({ timestamp: block.timestamp }, `Block obtained`)
+
+    logger.debug(`Checking if SCD Emergency Shutdown has happened`)
+    const tokenSwapAllowed = await foreignBridge.methods.isTokenSwapAllowed(block.timestamp).call()
+
+    if (!tokenSwapAllowed) {
+      logger.info(`Token swap discarded because SCD Emergency Shutdown has happened`)
+      return txToSend
+    }
+
+    let gasEstimate
+
+    try {
+      logger.debug(`Estimate gas`)
+      gasEstimate = await foreignBridge.methods.swapTokens().estimateGas({
+        from: config.validatorAddress
+      })
+
+      logger.debug({ gasEstimate }, 'Gas estimated')
+    } catch (e) {
+      if (e instanceof HttpListProviderError) {
+        const errorMsg = 'RPC Connection Error: swapTokens Gas Estimate cannot be obtained.'
+        logger.error(e, errorMsg)
+        throw new Error(errorMsg)
+      } else {
+        if (halfDuplexTokenContract === null) {
+          logger.debug('Getting half duplex token contract address')
+          const halfDuplexErc20Token = await foreignBridge.methods.halfDuplexErc20token().call()
+          logger.debug({ halfDuplexErc20Token }, 'Half duplex token contract address obtained')
+
+          halfDuplexTokenContract = new web3Foreign.eth.Contract(ERC20_ABI, halfDuplexErc20Token)
+        }
+
+        const balance = web3Foreign.utils.toBN(
+          await halfDuplexTokenContract.methods.balanceOf(config.foreignBridgeAddress).call()
+        )
+        logger.debug({ balance: balance.toString() }, 'Half duplex token bridge balance obtained')
+
+        if (balance.isZero()) {
+          logger.info(`Gas estimate failed because half duplex token balance is zero. Tokens swap is discarded.`)
+          return txToSend
+        }
+
+        logger.error(e, 'Unknown error while processing transaction')
+        throw e
+      }
+    }
+
+    // generate data
+    const data = await foreignBridge.methods.swapTokens().encodeABI()
+
+    // push to job
+    txToSend.push({
+      data,
+      gasEstimate,
+      transactionReference: `swap tokens operation for block number ${blockNumber.toString()}`,
+      to: config.foreignBridgeAddress
+    })
+
+    return txToSend
+  }
+}
+
+module.exports = swapTokensBuilder