Relay PoA lock-funds transactions proofs (#177)

* relay exchange transaction(s)

* fixed Ethereum::get_block_by_hash

* added exchange trace

* fixed method name

* update for new web3

* svyatonik/rust-web3 -> tomusdrw/rust-web3

* if let Some() -> .expect()

* extracted loops in separate functions

* use yaml references (TIL)

* get eth header with transactions

* cargo fmt --all

* Update primitives/ethereum-poa/src/lib.rs

Co-authored-by: Hernando Castano <HCastano@users.noreply.github.com>

* Update relays/ethereum/src/ethereum_exchange.rs

Co-authored-by: Hernando Castano <HCastano@users.noreply.github.com>

* Update relays/ethereum/src/rpc_errors.rs

Co-authored-by: Hernando Castano <HCastano@users.noreply.github.com>

* Update relays/ethereum/src/exchange.rs

Co-authored-by: Hernando Castano <HCastano@users.noreply.github.com>

* removed comment

* Update relays/ethereum/src/ethereum_exchange.rs

Co-authored-by: Hernando Castano <HCastano@users.noreply.github.com>

* Update relays/ethereum/src/ethereum_exchange.rs

Co-authored-by: Hernando Castano <HCastano@users.noreply.github.com>

* module-level comments

* updated readme

* use web3 from crates.io

* added missing fields info in error description

* cargo fmt --all

Co-authored-by: Hernando Castano <HCastano@users.noreply.github.com>

bridges/relays/ethereum/Cargo.toml

@@ -30,7 +30,7 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0.56"
 sp-bridge-eth-poa = { path = "../../primitives/ethereum-poa" }
 time = "0.2"
-web3 = { version = "0.12.0", default-features = false }
+web3 = "0.13"
 
 [dependencies.jsonrpsee]
 git = "https://github.com/svyatonik/jsonrpsee.git"

bridges/relays/ethereum/src/cli.yml

@@ -6,22 +6,22 @@ subcommands:
     - eth-to-sub:
         about: Synchronize headers from Ethereum node to Substrate node.
         args:
-            - eth-host:
+            - eth-host: &eth-host
                 long: eth-host
                 value_name: ETH_HOST
                 help: Connect to Ethereum node at given host.
                 takes_value: true
-            - eth-port:
+            - eth-port: &eth-port
                 long: eth-port
                 value_name: ETH_PORT
                 help: Connect to Ethereum node at given port.
                 takes_value: true
-            - sub-host:
+            - sub-host: &sub-host
                 long: sub-host
                 value_name: SUB_HOST
                 help: Connect to Substrate node at given host.
                 takes_value: true
-            - sub-port:
+            - sub-port: &sub-port
                 long: sub-port
                 value_name: SUB_PORT
                 help: Connect to Substrate node at given port.
@@ -35,78 +35,43 @@ subcommands:
                     - signed
                     - unsigned
                     - backup
-            - sub-signer:
+            - sub-signer: &sub-signer
                 long: sub-signer
                 value_name: SUB_SIGNER
                 help: The SURI of secret key to use when transactions are submitted to the Substrate node.
-            - sub-signer-password:
+            - sub-signer-password: &sub-signer-password
                 long: sub-signer-password
                 value_name: SUB_SIGNER_PASSWORD
                 help: The password for the SURI of secret key to use when transactions are submitted to the Substrate node.
     - sub-to-eth:
         about: Synchronize headers from Substrate node to Ethereum node.
         args:
-            - eth-host:
-                long: eth-host
-                value_name: ETH_HOST
-                help: Connect to Ethereum node at given host.
-                takes_value: true
-            - eth-port:
-                long: eth-port
-                value_name: ETH_PORT
-                help: Connect to Ethereum node at given port.
-                takes_value: true
+            - eth-host: *eth-host
+            - eth-port: *eth-port
             - eth-contract:
                 long: eth-contract
                 value_name: ETH_CONTRACT
                 help: Address of deployed bridge contract.
                 takes_value: true
-            - eth-signer:
+            - eth-signer: &eth-signer
                 long: eth-signer
                 value_name: ETH_SIGNER
                 help: Hex-encoded secret to use when transactions are submitted to the Ethereum node.
-            - sub-host:
-                long: sub-host
-                value_name: SUB_HOST
-                help: Connect to Substrate node at given host.
-                takes_value: true
-            - sub-port:
-                long: sub-port
-                value_name: SUB_PORT
-                help: Connect to Substrate node at given port.
-                takes_value: true
+            - sub-host: *sub-host
+            - sub-port: *sub-port
     - eth-deploy-contract:
         about: Deploy Bridge contract on Ethereum node.
         args:
-            - eth-host:
-                long: eth-host
-                value_name: ETH_HOST
-                help: Connect to Ethereum node at given host.
-                takes_value: true
-            - eth-port:
-                long: eth-port
-                value_name: ETH_PORT
-                help: Connect to Ethereum node at given port.
-                takes_value: true
-            - eth-signer:
-                long: eth-signer
-                value_name: ETH_SIGNER
-                help: Hex-encoded secret to use when transactions are submitted to the Ethereum node.
+            - eth-host: *eth-host
+            - eth-port: *eth-port
+            - eth-signer: *eth-signer
             - eth-contract-code:
                 long: eth-contract-code
                 value_name: ETH_CONTRACT_CODE
                 help: Bytecode of bridge contract.
                 takes_value: true
-            - sub-host:
-                long: sub-host
-                value_name: SUB_HOST
-                help: Connect to Substrate node at given host.
-                takes_value: true
-            - sub-port:
-                long: sub-port
-                value_name: SUB_PORT
-                help: Connect to Substrate node at given port.
-                takes_value: true
+            - sub-host: *sub-host
+            - sub-port: *sub-port
             - sub-authorities-set-id:
                 long: sub-authorities-set-id
                 value_name: SUB_AUTHORITIES_SET_ID
@@ -122,3 +87,18 @@ subcommands:
                 value_name: SUB_INITIAL_HEADER
                 help: Encoded initial Substrate header.
                 takes_value: true
+    - eth-exchange-sub:
+        about: Submit proof of PoA lock funds transaction to Substrate node.
+        args:
+            - eth-host: *eth-host
+            - eth-port: *eth-port
+            - eth-tx-hash:
+                long: eth-tx-hash
+                value_name: ETH_TX_HASH
+                help: Hash of the lock funds transaction.
+                takes_value: true
+                required: true
+            - sub-host: *sub-host
+            - sub-port: *sub-port
+            - sub-signer: *sub-signer
+            - sub-signer-password: *sub-signer-password

bridges/relays/ethereum/src/ethereum_client.rs

@@ -15,7 +15,8 @@
 // along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
 
 use crate::ethereum_types::{
-	Address, Bytes, CallRequest, EthereumHeaderId, Header, Receipt, SignedRawTx, TransactionHash, H256, U256,
+	Address, Bytes, CallRequest, EthereumHeaderId, Header, HeaderWithTransactions, Receipt, SignedRawTx, Transaction,
+	TransactionHash, H256, U256,
 };
 use crate::rpc::{Ethereum, EthereumRpc};
 use crate::rpc_errors::{EthereumNodeError, RpcError};
@@ -120,13 +121,35 @@ impl EthereumRpc for EthereumRpcClient {
 	}
 
 	async fn header_by_hash(&self, hash: H256) -> Result<Header> {
-		let header = Ethereum::get_block_by_hash(&self.client, hash).await?;
+		let get_full_tx_objects = false;
+		let header = Ethereum::get_block_by_hash(&self.client, hash, get_full_tx_objects).await?;
 		match header.number.is_some() && header.hash.is_some() && header.logs_bloom.is_some() {
 			true => Ok(header),
 			false => Err(RpcError::Ethereum(EthereumNodeError::IncompleteHeader)),
 		}
 	}
 
+	async fn header_by_hash_with_transactions(&self, hash: H256) -> Result<HeaderWithTransactions> {
+		let get_full_tx_objects = true;
+		let header = Ethereum::get_block_by_hash_with_transactions(&self.client, hash, get_full_tx_objects).await?;
+
+		let is_complete_header = header.number.is_some() && header.hash.is_some() && header.logs_bloom.is_some();
+		if !is_complete_header {
+			return Err(RpcError::Ethereum(EthereumNodeError::IncompleteHeader));
+		}
+
+		let is_complete_transactions = header.transactions.iter().all(|tx| tx.raw.is_some());
+		if !is_complete_transactions {
+			return Err(RpcError::Ethereum(EthereumNodeError::IncompleteTransaction));
+		}
+
+		Ok(header)
+	}
+
+	async fn transaction_by_hash(&self, hash: H256) -> Result<Option<Transaction>> {
+		Ok(Ethereum::transaction_by_hash(&self.client, hash).await?)
+	}
+
 	async fn transaction_receipt(&self, transaction_hash: H256) -> Result<Receipt> {
 		let receipt = Ethereum::get_transaction_receipt(&self.client, transaction_hash).await?;
 

bridges/relays/ethereum/src/ethereum_exchange.rs

@@ -0,0 +1,214 @@
+// Copyright 2019-2020 Parity Technologies (UK) Ltd.
+// This file is part of Parity Bridges Common.
+
+// Parity Bridges Common is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity Bridges Common is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Relaying proofs of PoA -> Substrate exchange transactions.
+
+use crate::ethereum_client::{EthereumConnectionParams, EthereumRpcClient};
+use crate::ethereum_types::{
+	EthereumHeaderId, Transaction as EthereumTransaction, TransactionHash as EthereumTransactionHash, H256,
+};
+use crate::exchange::{relay_single_transaction_proof, SourceClient, TargetClient, TransactionProofPipeline};
+use crate::rpc::{EthereumRpc, SubstrateRpc};
+use crate::rpc_errors::{EthereumNodeError, RpcError};
+use crate::substrate_client::{
+	SubmitEthereumExchangeTransactionProof, SubstrateConnectionParams, SubstrateRpcClient, SubstrateSigningParams,
+};
+use crate::sync_types::HeaderId;
+
+use async_trait::async_trait;
+use bridge_node_runtime::exchange::EthereumTransactionInclusionProof;
+use std::time::Duration;
+
+/// Interval at which we ask Ethereum node for updates.
+const ETHEREUM_TICK_INTERVAL: Duration = Duration::from_secs(10);
+/// Interval at which we ask Substrate node for updates.
+const SUBSTRATE_TICK_INTERVAL: Duration = Duration::from_secs(5);
+
+/// PoA exchange transaction relay params.
+#[derive(Debug, Default)]
+pub struct EthereumExchangeParams {
+	/// Ethereum connection params.
+	pub eth: EthereumConnectionParams,
+	/// Hash of the Ethereum transaction to relay.
+	pub eth_tx_hash: EthereumTransactionHash,
+	/// Substrate connection params.
+	pub sub: SubstrateConnectionParams,
+	/// Substrate signing params.
+	pub sub_sign: SubstrateSigningParams,
+}
+
+/// Ethereum to Substrate exchange pipeline.
+struct EthereumToSubstrateExchange;
+
+impl TransactionProofPipeline for EthereumToSubstrateExchange {
+	const SOURCE_NAME: &'static str = "Ethereum";
+	const TARGET_NAME: &'static str = "Substrate";
+
+	type BlockHash = H256;
+	type BlockNumber = u64;
+	type TransactionHash = EthereumTransactionHash;
+	type Transaction = EthereumTransaction;
+	type TransactionProof = EthereumTransactionInclusionProof;
+}
+
+/// Ethereum node as transactions proof source.
+struct EthereumTransactionsSource {
+	client: EthereumRpcClient,
+}
+
+#[async_trait]
+impl SourceClient<EthereumToSubstrateExchange> for EthereumTransactionsSource {
+	type Error = RpcError;
+
+	async fn tick(&self) {
+		async_std::task::sleep(ETHEREUM_TICK_INTERVAL).await;
+	}
+
+	async fn transaction(
+		&self,
+		hash: &EthereumTransactionHash,
+	) -> Result<Option<(EthereumHeaderId, EthereumTransaction)>, Self::Error> {
+		let eth_tx = match self.client.transaction_by_hash(*hash).await? {
+			Some(eth_tx) => eth_tx,
+			None => return Ok(None),
+		};
+
+		// we need transaction to be mined => check if it is included in the block
+		let eth_header_id = match (eth_tx.block_number, eth_tx.block_hash) {
+			(Some(block_number), Some(block_hash)) => HeaderId(block_number.as_u64(), block_hash),
+			_ => return Ok(None),
+		};
+
+		Ok(Some((eth_header_id, eth_tx)))
+	}
+
+	async fn transaction_proof(
+		&self,
+		eth_header_id: &EthereumHeaderId,
+		eth_tx: EthereumTransaction,
+	) -> Result<EthereumTransactionInclusionProof, Self::Error> {
+		const TRANSACTION_HAS_RAW_FIELD_PROOF: &'static str = "RPC level checks that transactions from Ethereum\
+			node are having `raw` field; qed";
+
+		let eth_header = self.client.header_by_hash_with_transactions(eth_header_id.1).await?;
+		let eth_relay_tx_hash = eth_tx.hash;
+		let mut eth_relay_tx = Some(eth_tx);
+		let mut eth_relay_tx_index = None;
+		let mut transaction_proof = Vec::with_capacity(eth_header.transactions.len());
+		for (index, eth_tx) in eth_header.transactions.into_iter().enumerate() {
+			if eth_tx.hash != eth_relay_tx_hash {
+				let eth_raw_tx = eth_tx.raw.expect(TRANSACTION_HAS_RAW_FIELD_PROOF);
+				transaction_proof.push(eth_raw_tx.0);
+			} else {
+				let eth_raw_relay_tx = match eth_relay_tx.take() {
+					Some(eth_relay_tx) => eth_relay_tx.raw.expect(TRANSACTION_HAS_RAW_FIELD_PROOF),
+					None => {
+						return Err(
+							EthereumNodeError::DuplicateBlockTransaction(*eth_header_id, eth_relay_tx_hash).into(),
+						)
+					}
+				};
+				eth_relay_tx_index = Some(index as u64);
+				transaction_proof.push(eth_raw_relay_tx.0);
+			}
+		}
+
+		Ok(EthereumTransactionInclusionProof {
+			block: eth_header_id.1,
+			index: eth_relay_tx_index.ok_or_else(|| {
+				RpcError::from(EthereumNodeError::BlockMissingTransaction(
+					*eth_header_id,
+					eth_relay_tx_hash,
+				))
+			})?,
+			proof: transaction_proof,
+		})
+	}
+}
+
+/// Substrate node as transactions proof target.
+struct SubstrateTransactionsTarget {
+	client: SubstrateRpcClient,
+	sign_params: SubstrateSigningParams,
+}
+
+#[async_trait]
+impl TargetClient<EthereumToSubstrateExchange> for SubstrateTransactionsTarget {
+	type Error = RpcError;
+
+	async fn tick(&self) {
+		async_std::task::sleep(SUBSTRATE_TICK_INTERVAL).await;
+	}
+
+	async fn is_header_known(&self, id: &EthereumHeaderId) -> Result<bool, Self::Error> {
+		self.client.ethereum_header_known(*id).await
+	}
+
+	async fn is_header_finalized(&self, id: &EthereumHeaderId) -> Result<bool, Self::Error> {
+		// we check if header is finalized by simple comparison of the header number and
+		// number of best finalized PoA header known to Substrate node.
+		//
+		// this may lead to failure in tx proof import if PoA reorganization has happened
+		// after we have checked that our tx has been included into given block
+		//
+		// the fix is easy, but since this code is mostly developed for demonstration purposes,
+		// I'm leaving this KISS-based design here
+		let best_finalized_ethereum_block = self.client.best_ethereum_finalized_block().await?;
+		Ok(id.0 <= best_finalized_ethereum_block.0)
+	}
+
+	async fn submit_transaction_proof(&self, proof: EthereumTransactionInclusionProof) -> Result<(), Self::Error> {
+		let sign_params = self.sign_params.clone();
+		self.client.submit_exchange_transaction_proof(sign_params, proof).await
+	}
+}
+
+/// Relay exchange transaction proof to Substrate node.
+pub fn run(params: EthereumExchangeParams) {
+	let eth_tx_hash = params.eth_tx_hash;
+	let mut local_pool = futures::executor::LocalPool::new();
+
+	let result = local_pool.run_until(async move {
+		let eth_client = EthereumRpcClient::new(params.eth);
+		let sub_client = SubstrateRpcClient::new(params.sub).await?;
+
+		let source = EthereumTransactionsSource { client: eth_client };
+		let target = SubstrateTransactionsTarget {
+			client: sub_client,
+			sign_params: params.sub_sign,
+		};
+
+		relay_single_transaction_proof(&source, &target, eth_tx_hash).await
+	});
+
+	match result {
+		Ok(_) => {
+			log::info!(
+				target: "bridge",
+				"Ethereum transaction {} proof has been successfully submitted to Substrate node",
+				eth_tx_hash,
+			);
+		}
+		Err(err) => {
+			log::error!(
+				target: "bridge",
+				"Error submitting Ethereum transaction {} proof to Substrate node: {}",
+				eth_tx_hash,
+				err,
+			);
+		}
+	}
+}

bridges/relays/ethereum/src/ethereum_types.rs

@@ -31,9 +31,15 @@ pub const RECEIPT_GAS_USED_PROOF: &'static str = "checked on retrieval; qed";
 /// Ethereum transaction hash type.
 pub type TransactionHash = H256;
 
+/// Ethereum transaction type.
+pub type Transaction = web3::types::Transaction;
+
 /// Ethereum header type.
 pub type Header = web3::types::Block<H256>;
 
+/// Ethereum header with transactions type.
+pub type HeaderWithTransactions = web3::types::Block<Transaction>;
+
 /// Ethereum transaction receipt type.
 pub type Receipt = web3::types::TransactionReceipt;
 

bridges/relays/ethereum/src/exchange.rs

@@ -0,0 +1,494 @@
+// Copyright 2019-2020 Parity Technologies (UK) Ltd.
+// This file is part of Parity Bridges Common.
+
+// Parity Bridges Common is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity Bridges Common is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Relaying proofs of exchange transactions.
+
+use async_trait::async_trait;
+use std::fmt::{Debug, Display};
+
+/// Transaction proof pipeline.
+pub trait TransactionProofPipeline {
+	/// Name of the transaction proof source.
+	const SOURCE_NAME: &'static str;
+	/// Name of the transaction proof target.
+	const TARGET_NAME: &'static str;
+
+	/// Block hash type.
+	type BlockHash: Display;
+	/// Block number type.
+	type BlockNumber: Display;
+	/// Transaction hash type.
+	type TransactionHash: Display;
+	/// Transaction type.
+	type Transaction;
+	/// Transaction inclusion proof type.
+	type TransactionProof;
+}
+
+/// Header id.
+pub type HeaderId<P> = crate::sync_types::HeaderId<
+	<P as TransactionProofPipeline>::BlockHash,
+	<P as TransactionProofPipeline>::BlockNumber,
+>;
+
+/// Source client API.
+#[async_trait]
+pub trait SourceClient<P: TransactionProofPipeline> {
+	/// Error type.
+	type Error: Debug;
+
+	/// Sleep until exchange-related data is (probably) updated.
+	async fn tick(&self);
+	/// Return **mined** transaction by its hash. May return `Ok(None)` if transaction is unknown to the source node.
+	async fn transaction(
+		&self,
+		hash: &P::TransactionHash,
+	) -> Result<Option<(HeaderId<P>, P::Transaction)>, Self::Error>;
+	/// Prepare transaction proof.
+	async fn transaction_proof(
+		&self,
+		header: &HeaderId<P>,
+		transaction: P::Transaction,
+	) -> Result<P::TransactionProof, Self::Error>;
+}
+
+/// Target client API.
+#[async_trait]
+pub trait TargetClient<P: TransactionProofPipeline> {
+	/// Error type.
+	type Error: Debug;
+
+	/// Sleep until exchange-related data is (probably) updated.
+	async fn tick(&self);
+	/// Returns `Ok(true)` if header is known to the target node.
+	async fn is_header_known(&self, id: &HeaderId<P>) -> Result<bool, Self::Error>;
+	/// Returns `Ok(true)` if header is finalized by the target node.
+	async fn is_header_finalized(&self, id: &HeaderId<P>) -> Result<bool, Self::Error>;
+	/// Submits transaction proof to the target node.
+	async fn submit_transaction_proof(&self, proof: P::TransactionProof) -> Result<(), Self::Error>;
+}
+
+/// Relay single transaction proof.
+pub async fn relay_single_transaction_proof<P: TransactionProofPipeline>(
+	source_client: &impl SourceClient<P>,
+	target_client: &impl TargetClient<P>,
+	source_tx_hash: P::TransactionHash,
+) -> Result<(), String> {
+	// wait for transaction and header on source node
+	let (source_header_id, source_tx) = wait_transaction_mined(source_client, &source_tx_hash).await?;
+	let transaction_proof = source_client
+		.transaction_proof(&source_header_id, source_tx)
+		.await
+		.map_err(|err| {
+			format!(
+				"Error building transaction {} proof on {} node: {:?}",
+				source_tx_hash,
+				P::SOURCE_NAME,
+				err,
+			)
+		})?;
+
+	// wait for transaction and header on target node
+	wait_header_imported(target_client, &source_header_id).await?;
+	wait_header_finalized(target_client, &source_header_id).await?;
+
+	// and finally - submit transaction proof to target node
+	target_client
+		.submit_transaction_proof(transaction_proof)
+		.await
+		.map_err(|err| {
+			format!(
+				"Error submitting transaction {} proof to {} node: {:?}",
+				source_tx_hash,
+				P::TARGET_NAME,
+				err,
+			)
+		})
+}
+
+/// Wait until transaction is mined by source node.
+async fn wait_transaction_mined<P: TransactionProofPipeline>(
+	source_client: &impl SourceClient<P>,
+	source_tx_hash: &P::TransactionHash,
+) -> Result<(HeaderId<P>, P::Transaction), String> {
+	loop {
+		let source_header_and_tx = source_client.transaction(&source_tx_hash).await.map_err(|err| {
+			format!(
+				"Error retrieving transaction {} from {} node: {:?}",
+				source_tx_hash,
+				P::SOURCE_NAME,
+				err,
+			)
+		})?;
+		match source_header_and_tx {
+			Some((source_header_id, source_tx)) => {
+				log::info!(
+					target: "bridge",
+					"Transaction {} is retrieved from {} node. Continuing...",
+					source_tx_hash,
+					P::SOURCE_NAME,
+				);
+
+				return Ok((source_header_id, source_tx));
+			}
+			None => {
+				log::info!(
+					target: "bridge",
+					"Waiting for transaction {} to be mined by {} node...",
+					source_tx_hash,
+					P::SOURCE_NAME,
+				);
+
+				source_client.tick().await;
+			}
+		}
+	}
+}
+
+/// Wait until target node imports required header.
+async fn wait_header_imported<P: TransactionProofPipeline>(
+	target_client: &impl TargetClient<P>,
+	source_header_id: &HeaderId<P>,
+) -> Result<(), String> {
+	loop {
+		let is_header_known = target_client.is_header_known(&source_header_id).await.map_err(|err| {
+			format!(
+				"Failed to check existence of header {}/{} on {} node: {:?}",
+				source_header_id.0,
+				source_header_id.1,
+				P::TARGET_NAME,
+				err,
+			)
+		})?;
+		match is_header_known {
+			true => {
+				log::info!(
+					target: "bridge",
+					"Header {}/{} is known to {} node. Continuing.",
+					source_header_id.0,
+					source_header_id.1,
+					P::TARGET_NAME,
+				);
+
+				return Ok(());
+			}
+			false => {
+				log::info!(
+					target: "bridge",
+					"Waiting for header {}/{} to be imported by {} node...",
+					source_header_id.0,
+					source_header_id.1,
+					P::TARGET_NAME,
+				);
+
+				target_client.tick().await;
+			}
+		}
+	}
+}
+
+/// Wait until target node finalizes required header.
+async fn wait_header_finalized<P: TransactionProofPipeline>(
+	target_client: &impl TargetClient<P>,
+	source_header_id: &HeaderId<P>,
+) -> Result<(), String> {
+	loop {
+		let is_header_finalized = target_client
+			.is_header_finalized(&source_header_id)
+			.await
+			.map_err(|err| {
+				format!(
+					"Failed to check finality of header {}/{} on {} node: {:?}",
+					source_header_id.0,
+					source_header_id.1,
+					P::TARGET_NAME,
+					err,
+				)
+			})?;
+		match is_header_finalized {
+			true => {
+				log::info!(
+					target: "bridge",
+					"Header {}/{} is finalizd by {} node. Continuing.",
+					source_header_id.0,
+					source_header_id.1,
+					P::TARGET_NAME,
+				);
+
+				return Ok(());
+			}
+			false => {
+				log::info!(
+					target: "bridge",
+					"Waiting for header {}/{} to be finalized by {} node...",
+					source_header_id.0,
+					source_header_id.1,
+					P::TARGET_NAME,
+				);
+
+				target_client.tick().await;
+			}
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::sync_types::HeaderId;
+
+	use parking_lot::Mutex;
+
+	fn test_header_id() -> TestHeaderId {
+		HeaderId(100, 100)
+	}
+
+	fn test_transaction_hash() -> TestTransactionHash {
+		200
+	}
+
+	fn test_transaction() -> TestTransaction {
+		300
+	}
+
+	fn test_transaction_proof() -> TestTransactionProof {
+		400
+	}
+
+	type TestError = u64;
+	type TestBlockNumber = u64;
+	type TestBlockHash = u64;
+	type TestTransactionHash = u64;
+	type TestTransaction = u64;
+	type TestTransactionProof = u64;
+	type TestHeaderId = HeaderId<TestBlockHash, TestBlockNumber>;
+
+	struct TestTransactionProofPipeline;
+
+	impl TransactionProofPipeline for TestTransactionProofPipeline {
+		const SOURCE_NAME: &'static str = "TestSource";
+		const TARGET_NAME: &'static str = "TestTarget";
+
+		type BlockHash = TestBlockHash;
+		type BlockNumber = TestBlockNumber;
+		type TransactionHash = TestTransactionHash;
+		type Transaction = TestTransaction;
+		type TransactionProof = TestTransactionProof;
+	}
+
+	struct TestTransactionsSource {
+		on_tick: Box<dyn Fn(&mut TestTransactionsSourceData) + Send + Sync>,
+		data: Mutex<TestTransactionsSourceData>,
+	}
+
+	struct TestTransactionsSourceData {
+		transaction: Result<Option<(TestHeaderId, TestTransaction)>, TestError>,
+		transaction_proof: Result<TestTransactionProof, TestError>,
+	}
+
+	impl TestTransactionsSource {
+		fn new(on_tick: Box<dyn Fn(&mut TestTransactionsSourceData) + Send + Sync>) -> Self {
+			Self {
+				on_tick,
+				data: Mutex::new(TestTransactionsSourceData {
+					transaction: Ok(Some((test_header_id(), test_transaction()))),
+					transaction_proof: Ok(test_transaction_proof()),
+				}),
+			}
+		}
+	}
+
+	#[async_trait]
+	impl SourceClient<TestTransactionProofPipeline> for TestTransactionsSource {
+		type Error = TestError;
+
+		async fn tick(&self) {
+			(self.on_tick)(&mut *self.data.lock())
+		}
+
+		async fn transaction(
+			&self,
+			_: &TestTransactionHash,
+		) -> Result<Option<(TestHeaderId, TestTransaction)>, TestError> {
+			self.data.lock().transaction.clone()
+		}
+
+		async fn transaction_proof(
+			&self,
+			_: &TestHeaderId,
+			_: TestTransaction,
+		) -> Result<TestTransactionProof, TestError> {
+			self.data.lock().transaction_proof.clone()
+		}
+	}
+
+	struct TestTransactionsTarget {
+		on_tick: Box<dyn Fn(&mut TestTransactionsTargetData) + Send + Sync>,
+		data: Mutex<TestTransactionsTargetData>,
+	}
+
+	struct TestTransactionsTargetData {
+		is_header_known: Result<bool, TestError>,
+		is_header_finalized: Result<bool, TestError>,
+		submitted_proofs: Vec<TestTransactionProof>,
+	}
+
+	impl TestTransactionsTarget {
+		fn new(on_tick: Box<dyn Fn(&mut TestTransactionsTargetData) + Send + Sync>) -> Self {
+			Self {
+				on_tick,
+				data: Mutex::new(TestTransactionsTargetData {
+					is_header_known: Ok(true),
+					is_header_finalized: Ok(true),
+					submitted_proofs: Vec::new(),
+				}),
+			}
+		}
+	}
+
+	#[async_trait]
+	impl TargetClient<TestTransactionProofPipeline> for TestTransactionsTarget {
+		type Error = TestError;
+
+		async fn tick(&self) {
+			(self.on_tick)(&mut *self.data.lock())
+		}
+
+		async fn is_header_known(&self, _: &TestHeaderId) -> Result<bool, TestError> {
+			self.data.lock().is_header_known.clone()
+		}
+
+		async fn is_header_finalized(&self, _: &TestHeaderId) -> Result<bool, TestError> {
+			self.data.lock().is_header_finalized.clone()
+		}
+
+		async fn submit_transaction_proof(&self, proof: TestTransactionProof) -> Result<(), TestError> {
+			self.data.lock().submitted_proofs.push(proof);
+			Ok(())
+		}
+	}
+
+	fn ensure_success(source: TestTransactionsSource, target: TestTransactionsTarget) {
+		assert_eq!(
+			async_std::task::block_on(relay_single_transaction_proof(
+				&source,
+				&target,
+				test_transaction_hash(),
+			)),
+			Ok(()),
+		);
+		assert_eq!(target.data.lock().submitted_proofs, vec![test_transaction_proof()],);
+	}
+
+	fn ensure_failure(source: TestTransactionsSource, target: TestTransactionsTarget) {
+		assert!(async_std::task::block_on(relay_single_transaction_proof(
+			&source,
+			&target,
+			test_transaction_hash(),
+		))
+		.is_err(),);
+		assert!(target.data.lock().submitted_proofs.is_empty());
+	}
+
+	#[test]
+	fn ready_transaction_proof_relayed_immediately() {
+		let source = TestTransactionsSource::new(Box::new(|_| unreachable!("no ticks allowed")));
+		let target = TestTransactionsTarget::new(Box::new(|_| unreachable!("no ticks allowed")));
+		ensure_success(source, target)
+	}
+
+	#[test]
+	fn relay_transaction_proof_waits_for_transaction_to_be_mined() {
+		let source = TestTransactionsSource::new(Box::new(|source_data| {
+			assert_eq!(source_data.transaction, Ok(None));
+			source_data.transaction = Ok(Some((test_header_id(), test_transaction())));
+		}));
+		let target = TestTransactionsTarget::new(Box::new(|_| unreachable!("no ticks allowed")));
+
+		// transaction is not yet mined, but will be available after first wait (tick)
+		source.data.lock().transaction = Ok(None);
+
+		ensure_success(source, target)
+	}
+
+	#[test]
+	fn relay_transaction_fails_when_transaction_retrieval_fails() {
+		let source = TestTransactionsSource::new(Box::new(|_| unreachable!("no ticks allowed")));
+		let target = TestTransactionsTarget::new(Box::new(|_| unreachable!("no ticks allowed")));
+
+		source.data.lock().transaction = Err(0);
+
+		ensure_failure(source, target)
+	}
+
+	#[test]
+	fn relay_transaction_fails_when_proof_retrieval_fails() {
+		let source = TestTransactionsSource::new(Box::new(|_| unreachable!("no ticks allowed")));
+		let target = TestTransactionsTarget::new(Box::new(|_| unreachable!("no ticks allowed")));
+
+		source.data.lock().transaction_proof = Err(0);
+
+		ensure_failure(source, target)
+	}
+
+	#[test]
+	fn relay_transaction_proof_waits_for_header_to_be_imported() {
+		let source = TestTransactionsSource::new(Box::new(|_| unreachable!("no ticks allowed")));
+		let target = TestTransactionsTarget::new(Box::new(|target_data| {
+			assert_eq!(target_data.is_header_known, Ok(false));
+			target_data.is_header_known = Ok(true);
+		}));
+
+		// header is not yet imported, but will be available after first wait (tick)
+		target.data.lock().is_header_known = Ok(false);
+
+		ensure_success(source, target)
+	}
+
+	#[test]
+	fn relay_transaction_proof_fails_when_is_header_known_fails() {
+		let source = TestTransactionsSource::new(Box::new(|_| unreachable!("no ticks allowed")));
+		let target = TestTransactionsTarget::new(Box::new(|_| unreachable!("no ticks allowed")));
+
+		target.data.lock().is_header_known = Err(0);
+
+		ensure_failure(source, target)
+	}
+
+	#[test]
+	fn relay_transaction_proof_waits_for_header_to_be_finalized() {
+		let source = TestTransactionsSource::new(Box::new(|_| unreachable!("no ticks allowed")));
+		let target = TestTransactionsTarget::new(Box::new(|target_data| {
+			assert_eq!(target_data.is_header_finalized, Ok(false));
+			target_data.is_header_finalized = Ok(true);
+		}));
+
+		// header is not yet finalized, but will be available after first wait (tick)
+		target.data.lock().is_header_finalized = Ok(false);
+
+		ensure_success(source, target)
+	}
+
+	#[test]
+	fn relay_transaction_proof_fails_when_is_header_finalized_fails() {
+		let source = TestTransactionsSource::new(Box::new(|_| unreachable!("no ticks allowed")));
+		let target = TestTransactionsTarget::new(Box::new(|_| unreachable!("no ticks allowed")));
+
+		target.data.lock().is_header_finalized = Err(0);
+
+		ensure_failure(source, target)
+	}
+}

bridges/relays/ethereum/src/main.rs

@@ -18,8 +18,10 @@
 
 mod ethereum_client;
 mod ethereum_deploy_contract;
+mod ethereum_exchange;
 mod ethereum_sync_loop;
 mod ethereum_types;
+mod exchange;
 mod headers;
 mod rpc;
 mod rpc_errors;
@@ -83,6 +85,15 @@ fn main() {
 				}
 			});
 		}
+		("eth-exchange-sub", Some(eth_exchange_matches)) => {
+			ethereum_exchange::run(match ethereum_exchange_params(&eth_exchange_matches) {
+				Ok(eth_exchange_params) => eth_exchange_params,
+				Err(err) => {
+					log::error!(target: "bridge", "Error relaying Ethereum transactions proofs: {}", err);
+					return;
+				}
+			});
+		}
 		("", _) => {
 			log::error!(target: "bridge", "No subcommand specified");
 			return;
@@ -225,3 +236,18 @@ fn ethereum_deploy_contract_params(
 
 	Ok(eth_deploy_params)
 }
+
+fn ethereum_exchange_params(matches: &clap::ArgMatches) -> Result<ethereum_exchange::EthereumExchangeParams, String> {
+	let mut params = ethereum_exchange::EthereumExchangeParams::default();
+	params.eth = ethereum_connection_params(matches)?;
+	params.sub = substrate_connection_params(matches)?;
+	params.sub_sign = substrate_signing_params(matches)?;
+
+	params.eth_tx_hash = matches
+		.value_of("eth-tx-hash")
+		.expect("eth-tx-hash is a required parameter; clap verifies that required parameters have matches; qed")
+		.parse()
+		.map_err(|e| format!("Failed to parse eth-tx-hash: {}", e))?;
+
+	Ok(params)
+}

bridges/relays/ethereum/src/rpc.rs

@@ -25,7 +25,8 @@
 use std::result;
 
 use crate::ethereum_types::{
-	Address as EthAddress, Bytes, CallRequest, EthereumHeaderId, Header as EthereumHeader, Receipt, SignedRawTx,
+	Address as EthAddress, Bytes, CallRequest, EthereumHeaderId, Header as EthereumHeader,
+	HeaderWithTransactions as EthereumHeaderWithTransactions, Receipt, SignedRawTx, Transaction as EthereumTransaction,
 	TransactionHash as EthereumTxHash, H256, U256, U64,
 };
 use crate::rpc_errors::RpcError;
@@ -48,7 +49,11 @@ jsonrpsee::rpc_api! {
 		#[rpc(method = "eth_getBlockByNumber", positional_params)]
 		fn get_block_by_number(block_number: U64, full_tx_objs: bool) -> EthereumHeader;
 		#[rpc(method = "eth_getBlockByHash", positional_params)]
-		fn get_block_by_hash(hash: H256) -> EthereumHeader;
+		fn get_block_by_hash(hash: H256, full_tx_objs: bool) -> EthereumHeader;
+		#[rpc(method = "eth_getBlockByHash", positional_params)]
+		fn get_block_by_hash_with_transactions(hash: H256, full_tx_objs: bool) -> EthereumHeaderWithTransactions;
+		#[rpc(method = "eth_getTransactionByHash", positional_params)]
+		fn transaction_by_hash(hash: H256) -> Option<EthereumTransaction>;
 		#[rpc(method = "eth_getTransactionReceipt", positional_params)]
 		fn get_transaction_receipt(transaction_hash: H256) -> Receipt;
 		#[rpc(method = "eth_getTransactionCount", positional_params)]
@@ -86,6 +91,10 @@ pub trait EthereumRpc {
 	async fn header_by_number(&self, block_number: u64) -> Result<EthereumHeader>;
 	/// Retrieve block header by its hash from Ethereum node.
 	async fn header_by_hash(&self, hash: H256) -> Result<EthereumHeader>;
+	/// Retrieve block header and its transactions by its hash from Ethereum node.
+	async fn header_by_hash_with_transactions(&self, hash: H256) -> Result<EthereumHeaderWithTransactions>;
+	/// Retrieve transaction by its hash from Ethereum node.
+	async fn transaction_by_hash(&self, hash: H256) -> Result<Option<EthereumTransaction>>;
 	/// Retrieve transaction receipt by transaction hash.
 	async fn transaction_receipt(&self, transaction_hash: H256) -> Result<Receipt>;
 	/// Get the nonce of the given account.
@@ -117,6 +126,8 @@ pub trait SubstrateRpc {
 	async fn next_account_index(&self, account: node_primitives::AccountId) -> Result<node_primitives::Index>;
 	/// Returns best Ethereum block that Substrate runtime knows of.
 	async fn best_ethereum_block(&self) -> Result<EthereumHeaderId>;
+	/// Returns best finalized Ethereum block that Substrate runtime knows of.
+	async fn best_ethereum_finalized_block(&self) -> Result<EthereumHeaderId>;
 	/// Returns whether or not transactions receipts are required for Ethereum header submission.
 	async fn ethereum_receipts_required(&self, header: SubstrateEthereumHeader) -> Result<bool>;
 	/// Returns whether or not the given Ethereum header is known to the Substrate runtime.

bridges/relays/ethereum/src/rpc_errors.rs

@@ -14,6 +14,7 @@
 // You should have received a copy of the GNU General Public License
 // along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
 
+use crate::ethereum_types::{EthereumHeaderId, TransactionHash as EthereumTransactionHash};
 use crate::sync_types::MaybeConnectionError;
 
 use jsonrpsee::client::RequestError;
@@ -101,20 +102,39 @@ pub enum EthereumNodeError {
 	IncompleteHeader,
 	/// We have received a receipt missing a `gas_used` field.
 	IncompleteReceipt,
+	/// We have received a transaction missing a `raw` field.
+	IncompleteTransaction,
 	/// An invalid Substrate block number was received from
 	/// an Ethereum node.
 	InvalidSubstrateBlockNumber,
+	/// Block includes the same transaction more than once.
+	DuplicateBlockTransaction(EthereumHeaderId, EthereumTransactionHash),
+	/// Block is missing transaction we believe is a part of this block.
+	BlockMissingTransaction(EthereumHeaderId, EthereumTransactionHash),
 }
 
 impl ToString for EthereumNodeError {
 	fn to_string(&self) -> String {
 		match self {
-			Self::ResponseParseFailed(e) => e,
-			Self::IncompleteHeader => "Incomplete Ethereum Header Received",
-			Self::IncompleteReceipt => "Incomplete Ethereum Receipt Recieved",
-			Self::InvalidSubstrateBlockNumber => "Received an invalid Substrate block from Ethereum Node",
+			Self::ResponseParseFailed(e) => e.to_string(),
+			Self::IncompleteHeader => {
+				"Incomplete Ethereum Header Received (missing some of required fields - hash, number, logs_bloom)"
+					.to_string()
+			}
+			Self::IncompleteReceipt => {
+				"Incomplete Ethereum Receipt Recieved (missing required field - gas_used)".to_string()
+			}
+			Self::IncompleteTransaction => "Incomplete Ethereum Transaction (missing required field - raw)".to_string(),
+			Self::InvalidSubstrateBlockNumber => "Received an invalid Substrate block from Ethereum Node".to_string(),
+			Self::DuplicateBlockTransaction(header_id, tx_hash) => format!(
+				"Ethereum block {}/{} includes Ethereum transaction {} more than once",
+				header_id.0, header_id.1, tx_hash,
+			),
+			Self::BlockMissingTransaction(header_id, tx_hash) => format!(
+				"Ethereum block {}/{} is missing Ethereum transaction {} which we believe is a part of this block",
+				header_id.0, header_id.1, tx_hash,
+			),
 		}
-		.to_string()
 	}
 }
 

bridges/relays/ethereum/src/substrate_client.rs

@@ -37,6 +37,7 @@ use std::collections::VecDeque;
 const ETH_API_IMPORT_REQUIRES_RECEIPTS: &str = "EthereumHeadersApi_is_import_requires_receipts";
 const ETH_API_IS_KNOWN_BLOCK: &str = "EthereumHeadersApi_is_known_block";
 const ETH_API_BEST_BLOCK: &str = "EthereumHeadersApi_best_block";
+const ETH_API_BEST_FINALIZED_BLOCK: &str = "EthereumHeadersApi_finalized_block";
 const SUB_API_GRANDPA_AUTHORITIES: &str = "GrandpaApi_grandpa_authorities";
 
 type Result<T> = std::result::Result<T, RpcError>;
@@ -142,6 +143,17 @@ impl SubstrateRpc for SubstrateRpcClient {
 		Ok(best_header_id)
 	}
 
+	async fn best_ethereum_finalized_block(&self) -> Result<EthereumHeaderId> {
+		let call = ETH_API_BEST_FINALIZED_BLOCK.to_string();
+		let data = Bytes("0x".into());
+
+		let encoded_response = Substrate::state_call(&self.client, call, data, None).await?;
+		let decoded_response: (u64, sp_bridge_eth_poa::H256) = Decode::decode(&mut &encoded_response.0[..])?;
+
+		let best_header_id = HeaderId(decoded_response.0, decoded_response.1);
+		Ok(best_header_id)
+	}
+
 	async fn ethereum_receipts_required(&self, header: SubstrateEthereumHeader) -> Result<bool> {
 		let call = ETH_API_IMPORT_REQUIRES_RECEIPTS.to_string();
 		let data = Bytes(header.encode());
@@ -281,6 +293,42 @@ impl SubmitEthereumHeaders for SubstrateRpcClient {
 	}
 }
 
+/// A trait for RPC calls which are used to submit proof of Ethereum exchange transaction to a
+/// Substrate runtime. These are typically calls which use a combination of other low-level RPC
+/// calls.
+#[async_trait]
+pub trait SubmitEthereumExchangeTransactionProof: SubstrateRpc {
+	/// Submits Ethereum exchange transaction proof to Substrate runtime.
+	async fn submit_exchange_transaction_proof(
+		&self,
+		params: SubstrateSigningParams,
+		proof: bridge_node_runtime::exchange::EthereumTransactionInclusionProof,
+	) -> Result<()>;
+}
+
+#[async_trait]
+impl SubmitEthereumExchangeTransactionProof for SubstrateRpcClient {
+	async fn submit_exchange_transaction_proof(
+		&self,
+		params: SubstrateSigningParams,
+		proof: bridge_node_runtime::exchange::EthereumTransactionInclusionProof,
+	) -> Result<()> {
+		let account_id = params.signer.public().as_array_ref().clone().into();
+		let nonce = self.next_account_index(account_id).await?;
+
+		let transaction = create_signed_transaction(
+			bridge_node_runtime::Call::BridgeCurrencyExchange(
+				bridge_node_runtime::BridgeCurrencyExchangeCall::import_peer_transaction(proof),
+			),
+			&params.signer,
+			nonce,
+			self.genesis_hash,
+		);
+		let _ = self.submit_extrinsic(Bytes(transaction.encode())).await?;
+		Ok(())
+	}
+}
+
 /// Create signed Substrate transaction for submitting Ethereum headers.
 fn create_signed_submit_transaction(
 	headers: Vec<QueuedEthereumHeader>,
@@ -288,7 +336,7 @@ fn create_signed_submit_transaction(
 	index: node_primitives::Index,
 	genesis_hash: H256,
 ) -> bridge_node_runtime::UncheckedExtrinsic {
-	let function =
+	create_signed_transaction(
 		bridge_node_runtime::Call::BridgeEthPoA(bridge_node_runtime::BridgeEthPoACall::import_signed_headers(
 			headers
 				.into_iter()
@@ -299,8 +347,31 @@ fn create_signed_submit_transaction(
 					)
 				})
 				.collect(),
+		)),
+		signer,
+		index,
+		genesis_hash,
+	)
+}
+
+/// Create unsigned Substrate transaction for submitting Ethereum header.
+fn create_unsigned_submit_transaction(header: QueuedEthereumHeader) -> bridge_node_runtime::UncheckedExtrinsic {
+	let function =
+		bridge_node_runtime::Call::BridgeEthPoA(bridge_node_runtime::BridgeEthPoACall::import_unsigned_header(
+			into_substrate_ethereum_header(header.header()),
+			into_substrate_ethereum_receipts(header.extra()),
 		));
 
+	bridge_node_runtime::UncheckedExtrinsic::new_unsigned(function)
+}
+
+/// Create signed Substrate transaction.
+fn create_signed_transaction(
+	function: bridge_node_runtime::Call,
+	signer: &sp_core::sr25519::Pair,
+	index: node_primitives::Index,
+	genesis_hash: H256,
+) -> bridge_node_runtime::UncheckedExtrinsic {
 	let extra = |i: node_primitives::Index, f: node_primitives::Balance| {
 		(
 			frame_system::CheckSpecVersion::<bridge_node_runtime::Runtime>::new(),
@@ -331,14 +402,3 @@ fn create_signed_submit_transaction(
 
 	bridge_node_runtime::UncheckedExtrinsic::new_signed(function, signer.into_account().into(), signature.into(), extra)
 }
-
-/// Create unsigned Substrate transaction for submitting Ethereum header.
-fn create_unsigned_submit_transaction(header: QueuedEthereumHeader) -> bridge_node_runtime::UncheckedExtrinsic {
-	let function =
-		bridge_node_runtime::Call::BridgeEthPoA(bridge_node_runtime::BridgeEthPoACall::import_unsigned_header(
-			into_substrate_ethereum_header(header.header()),
-			into_substrate_ethereum_receipts(header.extra()),
-		));
-
-	bridge_node_runtime::UncheckedExtrinsic::new_unsigned(function)
-}