Auto relay exchange transactions (#227)

* auto relay exchange transactions

* docker + auto-relay-tx

* clippy

* jsonrpsee in Cargo.lock ???

* fix tests compilation

* Show sccache

* mute clippy

* move

* Update relays/ethereum/src/exchange.rs

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

* finish comment

* (bool, String) -> StringifiedMaybeConnectionError

* Update deployments/rialto/docker-compose.yml

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

Co-authored-by: Denis S. Soldatov aka General-Beck <general.beck@gmail.com>
Co-authored-by: Hernando Castano <HCastano@users.noreply.github.com>

bridges/relays/ethereum/Cargo.toml

@@ -29,6 +29,7 @@ rustc-hex = "2.0.1"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0.57"
 sysinfo = "0.15"
+sp-currency-exchange = { path = "../../primitives/currency-exchange" }
 sp-bridge-eth-poa = { path = "../../primitives/ethereum-poa" }
 time = "0.2"
 web3 = "0.13"

bridges/relays/ethereum/src/cli.yml

@@ -111,12 +111,18 @@ subcommands:
         args:
             - eth-host: *eth-host
             - eth-port: *eth-port
+            - eth-start-with-block:
+                long: eth-start-with-block
+                value_name: ETH_START_WITH_BLOCK
+                help: Auto-relay transactions starting with given block number. If not specified, starts with best finalized Ethereum block (known to Substrate node) transactions.
+                takes_value: true
+                conflicts_with:
+                    - eth-tx-hash
             - 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

bridges/relays/ethereum/src/ethereum_client.rs

@@ -129,6 +129,23 @@ impl EthereumRpc for EthereumRpcClient {
 		}
 	}
 
+	async fn header_by_number_with_transactions(&self, number: u64) -> Result<HeaderWithTransactions> {
+		let get_full_tx_objects = true;
+		let header = Ethereum::get_block_by_number_with_transactions(&self.client, number, 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 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?;

bridges/relays/ethereum/src/ethereum_exchange.rs

@@ -18,11 +18,16 @@
 
 use crate::ethereum_client::{EthereumConnectionParams, EthereumRpcClient};
 use crate::ethereum_types::{
-	EthereumHeaderId, Transaction as EthereumTransaction, TransactionHash as EthereumTransactionHash, H256,
+	EthereumHeaderId, HeaderWithTransactions as EthereumHeaderWithTransactions, Transaction as EthereumTransaction,
+	TransactionHash as EthereumTransactionHash, H256,
 };
-use crate::exchange::{relay_single_transaction_proof, SourceClient, TargetClient, TransactionProofPipeline};
+use crate::exchange::{
+	relay_single_transaction_proof, SourceBlock, SourceClient, SourceTransaction, TargetClient,
+	TransactionProofPipeline,
+};
+use crate::exchange_loop::{run as run_loop, InMemoryStorage};
 use crate::rpc::{EthereumRpc, SubstrateRpc};
-use crate::rpc_errors::{EthereumNodeError, RpcError};
+use crate::rpc_errors::RpcError;
 use crate::substrate_client::{
 	SubmitEthereumExchangeTransactionProof, SubstrateConnectionParams, SubstrateRpcClient, SubstrateSigningParams,
 };
@@ -30,6 +35,7 @@ use crate::sync_types::HeaderId;
 
 use async_trait::async_trait;
 use bridge_node_runtime::exchange::EthereumTransactionInclusionProof;
+use sp_currency_exchange::MaybeLockFundsTransaction;
 use std::time::Duration;
 
 /// Interval at which we ask Ethereum node for updates.
@@ -37,17 +43,26 @@ 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);
 
+/// Exchange relay mode.
+#[derive(Debug)]
+pub enum ExchangeRelayMode {
+	/// Relay single transaction and quit.
+	Single(EthereumTransactionHash),
+	/// Auto-relay transactions starting with given block.
+	Auto(Option<u64>),
+}
+
 /// PoA exchange transaction relay params.
-#[derive(Debug, Default)]
+#[derive(Debug)]
 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,
+	/// Relay working mode.
+	pub mode: ExchangeRelayMode,
 }
 
 /// Ethereum to Substrate exchange pipeline.
@@ -57,13 +72,46 @@ 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 Block = EthereumSourceBlock;
 	type TransactionProof = EthereumTransactionInclusionProof;
 }
 
+/// Ethereum source block.
+struct EthereumSourceBlock(EthereumHeaderWithTransactions);
+
+impl SourceBlock for EthereumSourceBlock {
+	type Hash = H256;
+	type Number = u64;
+	type Transaction = EthereumSourceTransaction;
+
+	fn id(&self) -> EthereumHeaderId {
+		HeaderId(
+			self.0.number.expect(crate::ethereum_types::HEADER_ID_PROOF).as_u64(),
+			self.0.hash.expect(crate::ethereum_types::HEADER_ID_PROOF),
+		)
+	}
+
+	fn transactions(&self) -> Vec<Self::Transaction> {
+		self.0
+			.transactions
+			.iter()
+			.cloned()
+			.map(EthereumSourceTransaction)
+			.collect()
+	}
+}
+
+/// Ethereum source transaction.
+struct EthereumSourceTransaction(EthereumTransaction);
+
+impl SourceTransaction for EthereumSourceTransaction {
+	type Hash = EthereumTransactionHash;
+
+	fn hash(&self) -> Self::Hash {
+		self.0.hash
+	}
+}
+
 /// Ethereum node as transactions proof source.
 struct EthereumTransactionsSource {
 	client: EthereumRpcClient,
@@ -77,63 +125,60 @@ impl SourceClient<EthereumToSubstrateExchange> for EthereumTransactionsSource {
 		async_std::task::sleep(ETHEREUM_TICK_INTERVAL).await;
 	}
 
-	async fn transaction(
+	async fn block_by_hash(&self, hash: H256) -> Result<EthereumSourceBlock, Self::Error> {
+		self.client
+			.header_by_hash_with_transactions(hash)
+			.await
+			.map(EthereumSourceBlock)
+	}
+
+	async fn block_by_number(&self, number: u64) -> Result<EthereumSourceBlock, Self::Error> {
+		self.client
+			.header_by_number_with_transactions(number)
+			.await
+			.map(EthereumSourceBlock)
+	}
+
+	async fn transaction_block(
 		&self,
 		hash: &EthereumTransactionHash,
-	) -> Result<Option<(EthereumHeaderId, EthereumTransaction)>, Self::Error> {
+	) -> Result<Option<(EthereumHeaderId, usize)>, 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),
+		let (eth_header_id, eth_tx_index) = match (eth_tx.block_number, eth_tx.block_hash, eth_tx.transaction_index) {
+			(Some(block_number), Some(block_hash), Some(transaction_index)) => (
+				HeaderId(block_number.as_u64(), block_hash),
+				transaction_index.as_u64() as _,
+			),
 			_ => return Ok(None),
 		};
 
-		Ok(Some((eth_header_id, eth_tx)))
+		Ok(Some((eth_header_id, eth_tx_index)))
 	}
 
 	async fn transaction_proof(
 		&self,
-		eth_header_id: &EthereumHeaderId,
-		eth_tx: EthereumTransaction,
+		block: &EthereumSourceBlock,
+		tx_index: usize,
 	) -> Result<EthereumTransactionInclusionProof, Self::Error> {
 		const TRANSACTION_HAS_RAW_FIELD_PROOF: &str = "RPC level checks that transactions from Ethereum\
 			node are having `raw` field; qed";
+		const BLOCK_HAS_HASH_FIELD_PROOF: &str = "RPC level checks that block has `hash` 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);
-			}
-		}
+		let transaction_proof = block
+			.0
+			.transactions
+			.iter()
+			.map(|tx| tx.raw.clone().expect(TRANSACTION_HAS_RAW_FIELD_PROOF).0)
+			.collect();
 
 		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,
-				))
-			})?,
+			block: block.0.hash.expect(BLOCK_HAS_HASH_FIELD_PROOF),
+			index: tx_index as _,
 			proof: transaction_proof,
 		})
 	}
@@ -170,15 +215,51 @@ impl TargetClient<EthereumToSubstrateExchange> for SubstrateTransactionsTarget {
 		Ok(id.0 <= best_finalized_ethereum_block.0)
 	}
 
+	async fn best_finalized_header_id(&self) -> Result<EthereumHeaderId, Self::Error> {
+		self.client.best_ethereum_finalized_block().await
+	}
+
+	async fn filter_transaction_proof(&self, proof: &EthereumTransactionInclusionProof) -> Result<bool, Self::Error> {
+		// let's try to parse transaction locally
+		let parse_result = bridge_node_runtime::exchange::EthTransaction::parse(&proof.proof[proof.index as usize]);
+		if parse_result.is_err() {
+			return Ok(false);
+		}
+
+		// seems that transaction is relayable - let's check if runtime is able to import it
+		// (we can't if e.g. header is pruned or there's some issue with tx data)
+		self.client.verify_exchange_transaction_proof(proof.clone()).await
+	}
+
 	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.
+impl Default for EthereumExchangeParams {
+	fn default() -> Self {
+		EthereumExchangeParams {
+			eth: Default::default(),
+			sub: Default::default(),
+			sub_sign: Default::default(),
+			mode: ExchangeRelayMode::Auto(None),
+		}
+	}
+}
+
+/// Relay exchange transaction proof(s) to Substrate node.
 pub fn run(params: EthereumExchangeParams) {
-	let eth_tx_hash = params.eth_tx_hash;
+	match params.mode {
+		ExchangeRelayMode::Single(eth_tx_hash) => run_single_transaction_relay(params, eth_tx_hash),
+		ExchangeRelayMode::Auto(eth_start_with_block_number) => {
+			run_auto_transactions_relay_loop(params, eth_start_with_block_number)
+		}
+	};
+}
+
+/// Run single transaction proof relay and stop.
+fn run_single_transaction_relay(params: EthereumExchangeParams, eth_tx_hash: H256) {
 	let mut local_pool = futures::executor::LocalPool::new();
 
 	let result = local_pool.run_until(async move {
@@ -212,3 +293,46 @@ pub fn run(params: EthereumExchangeParams) {
 		}
 	}
 }
+
+/// Run auto-relay loop.
+fn run_auto_transactions_relay_loop(params: EthereumExchangeParams, eth_start_with_block_number: Option<u64>) {
+	let do_run_loop = move || -> Result<(), String> {
+		let eth_client = EthereumRpcClient::new(params.eth);
+		let sub_client = async_std::task::block_on(SubstrateRpcClient::new(params.sub))
+			.map_err(|err| format!("Error starting Substrate client: {:?}", err))?;
+
+		let eth_start_with_block_number = match eth_start_with_block_number {
+			Some(eth_start_with_block_number) => eth_start_with_block_number,
+			None => {
+				async_std::task::block_on(sub_client.best_ethereum_finalized_block())
+					.map_err(|err| {
+						format!(
+							"Error retrieving best finalized Ethereum block from Substrate node: {:?}",
+							err
+						)
+					})?
+					.0
+			}
+		};
+
+		run_loop(
+			InMemoryStorage::new(eth_start_with_block_number),
+			EthereumTransactionsSource { client: eth_client },
+			SubstrateTransactionsTarget {
+				client: sub_client,
+				sign_params: params.sub_sign,
+			},
+			futures::future::pending(),
+		);
+
+		Ok(())
+	};
+
+	if let Err(err) = do_run_loop() {
+		log::error!(
+			target: "bridge",
+			"Error auto-relaying Ethereum transactions proofs to Substrate node: {}",
+			err,
+		);
+	}
+}

bridges/relays/ethereum/src/exchange.rs

@@ -14,10 +14,15 @@
 // 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.
+//! Relaying proofs of exchange transaction.
+
+use crate::sync_types::MaybeConnectionError;
 
 use async_trait::async_trait;
-use std::fmt::{Debug, Display};
+use std::{
+	fmt::{Debug, Display},
+	string::ToString,
+};
 
 /// Transaction proof pipeline.
 pub trait TransactionProofPipeline {
@@ -26,50 +31,76 @@ pub trait TransactionProofPipeline {
 	/// 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;
+	/// Block type.
+	type Block: SourceBlock;
 	/// Transaction inclusion proof type.
 	type TransactionProof;
 }
 
+/// Block that is participating in exchange.
+pub trait SourceBlock {
+	/// Block hash type.
+	type Hash: Clone + Debug + Display;
+	/// Block number type.
+	type Number: Debug + Display + Clone + Copy + std::cmp::Ord + std::ops::Add<Output = Self::Number> + num_traits::One;
+	/// Block transaction.
+	type Transaction: SourceTransaction;
+
+	/// Return hash of the block.
+	fn id(&self) -> crate::sync_types::HeaderId<Self::Hash, Self::Number>;
+	/// Return block transactions iterator.
+	fn transactions(&self) -> Vec<Self::Transaction>;
+}
+
+/// Transaction that is participating in exchange.
+pub trait SourceTransaction {
+	/// Transaction hash type.
+	type Hash: Debug + Display;
+
+	/// Return transaction hash.
+	fn hash(&self) -> Self::Hash;
+}
+
+/// Block hash for given pipeline.
+pub type BlockHashOf<P> = <<P as TransactionProofPipeline>::Block as SourceBlock>::Hash;
+
+/// Block number for given pipeline.
+pub type BlockNumberOf<P> = <<P as TransactionProofPipeline>::Block as SourceBlock>::Number;
+
+/// Transaction hash for given pipeline.
+pub type TransactionOf<P> = <<P as TransactionProofPipeline>::Block as SourceBlock>::Transaction;
+
+/// Transaction hash for given pipeline.
+pub type TransactionHashOf<P> = <TransactionOf<P> as SourceTransaction>::Hash;
+
 /// Header id.
-pub type HeaderId<P> = crate::sync_types::HeaderId<
-	<P as TransactionProofPipeline>::BlockHash,
-	<P as TransactionProofPipeline>::BlockNumber,
->;
+pub type HeaderId<P> = crate::sync_types::HeaderId<BlockHashOf<P>, BlockNumberOf<P>>;
 
 /// Source client API.
 #[async_trait]
 pub trait SourceClient<P: TransactionProofPipeline> {
 	/// Error type.
-	type Error: Debug;
+	type Error: Debug + MaybeConnectionError;
 
 	/// 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>;
+	/// Get block by hash.
+	async fn block_by_hash(&self, hash: BlockHashOf<P>) -> Result<P::Block, Self::Error>;
+	/// Get canonical block by number.
+	async fn block_by_number(&self, number: BlockNumberOf<P>) -> Result<P::Block, Self::Error>;
+	/// Return block + index where transaction has been **mined**. May return `Ok(None)` if transaction
+	/// is unknown to the source node.
+	async fn transaction_block(&self, hash: &TransactionHashOf<P>)
+		-> Result<Option<(HeaderId<P>, usize)>, Self::Error>;
 	/// Prepare transaction proof.
-	async fn transaction_proof(
-		&self,
-		header: &HeaderId<P>,
-		transaction: P::Transaction,
-	) -> Result<P::TransactionProof, Self::Error>;
+	async fn transaction_proof(&self, block: &P::Block, tx_index: usize) -> Result<P::TransactionProof, Self::Error>;
 }
 
 /// Target client API.
 #[async_trait]
 pub trait TargetClient<P: TransactionProofPipeline> {
 	/// Error type.
-	type Error: Debug;
+	type Error: Debug + MaybeConnectionError;
 
 	/// Sleep until exchange-related data is (probably) updated.
 	async fn tick(&self);
@@ -77,44 +108,209 @@ pub trait TargetClient<P: TransactionProofPipeline> {
 	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>;
+	/// Returns best finalized header id.
+	async fn best_finalized_header_id(&self) -> Result<HeaderId<P>, Self::Error>;
+	/// Returns `Ok(true)` if transaction proof is need to be relayed.
+	async fn filter_transaction_proof(&self, proof: &P::TransactionProof) -> Result<bool, Self::Error>;
 	/// Submits transaction proof to the target node.
 	async fn submit_transaction_proof(&self, proof: P::TransactionProof) -> Result<(), Self::Error>;
 }
 
+/// Block transaction statistics.
+#[derive(Debug, Default)]
+#[cfg_attr(test, derive(PartialEq))]
+pub struct RelayedBlockTransactions {
+	/// Total number of transactions processed (either relayed or ignored) so far.
+	pub processed: usize,
+	/// Total number of transactions successfully relayed so far.
+	pub relayed: usize,
+	/// Total number of transactions that we have failed to relay so far.
+	pub failed: usize,
+}
+
+/// Stringified error that may be either connection-related or not.
+enum StringifiedMaybeConnectionError {
+	/// The error is connection-related error.
+	Connection(String),
+	/// The error is connection-unrelated error.
+	NonConnection(String),
+}
+
+/// Relay all suitable transactions from single block.
+///
+/// If connection error occurs, returns Err with number of successfully processed transactions.
+/// If some other error occurs, it is ignored and other transactions are processed.
+///
+/// All transaction-level traces are written by this function. This function is not tracing
+/// any information about block.
+pub async fn relay_block_transactions<P: TransactionProofPipeline>(
+	source_client: &impl SourceClient<P>,
+	target_client: &impl TargetClient<P>,
+	source_block: &P::Block,
+	mut relayed_transactions: RelayedBlockTransactions,
+) -> Result<RelayedBlockTransactions, RelayedBlockTransactions> {
+	let transactions_to_process = source_block
+		.transactions()
+		.into_iter()
+		.enumerate()
+		.skip(relayed_transactions.processed);
+	for (source_tx_index, source_tx) in transactions_to_process {
+		let result = async {
+			let source_tx_id = format!("{}/{}", source_block.id().1, source_tx_index);
+			let source_tx_proof =
+				prepare_transaction_proof(source_client, &source_tx_id, source_block, source_tx_index).await?;
+
+			let needs_to_be_relayed =
+				target_client
+					.filter_transaction_proof(&source_tx_proof)
+					.await
+					.map_err(|err| {
+						StringifiedMaybeConnectionError::new(
+							err.is_connection_error(),
+							format!("Transaction filtering has failed with {:?}", err),
+						)
+					})?;
+
+			if !needs_to_be_relayed {
+				return Ok(false);
+			}
+
+			relay_ready_transaction_proof(target_client, &source_tx_id, source_tx_proof)
+				.await
+				.map(|_| true)
+		}
+		.await;
+
+		// We have two options here:
+		// 1) retry with the same transaction later;
+		// 2) report error and proceed with next transaction.
+		//
+		// Option#1 may seems better, but:
+		// 1) we do not track if transaction is mined (without an error) by the target node;
+		// 2) error could be irrecoverable (e.g. when block is already pruned by bridge module or tx
+		//    has invalid format) && we'll end up in infinite loop of retrying the same transaction proof.
+		//
+		// So we're going with option#2 here (the only exception are connection errors).
+		match result {
+			Ok(false) => {
+				relayed_transactions.processed += 1;
+			}
+			Ok(true) => {
+				log::info!(
+					target: "bridge",
+					"{} transaction {} proof has been successfully submitted to {} node",
+					P::SOURCE_NAME,
+					source_tx.hash(),
+					P::TARGET_NAME,
+				);
+
+				relayed_transactions.processed += 1;
+				relayed_transactions.relayed += 1;
+			}
+			Err(err) => {
+				log::error!(
+					target: "bridge",
+					"Error relaying {} transaction {} proof to {} node: {}. {}",
+					P::SOURCE_NAME,
+					source_tx.hash(),
+					P::TARGET_NAME,
+					err.to_string(),
+					if err.is_connection_error() {
+						"Going to retry after delay..."
+					} else {
+						"You may need to submit proof of this transaction manually"
+					},
+				);
+
+				if err.is_connection_error() {
+					return Err(relayed_transactions);
+				}
+
+				relayed_transactions.processed += 1;
+				relayed_transactions.failed += 1;
+			}
+		}
+	}
+
+	Ok(relayed_transactions)
+}
+
 /// 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,
+	source_tx_hash: TransactionHashOf<P>,
 ) -> 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,
-			)
-		})?;
+	let (source_header_id, source_tx_index) = wait_transaction_mined(source_client, &source_tx_hash).await?;
+	let source_block = source_client.block_by_hash(source_header_id.1.clone()).await;
+	let source_block = source_block.map_err(|err| {
+		format!(
+			"Error retrieving block {} from {} node: {:?}",
+			source_header_id.1,
+			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)
+	// and finally - prepare and submit transaction proof to target node
+	let source_tx_id = format!("{}", source_tx_hash);
+	relay_ready_transaction_proof(
+		target_client,
+		&source_tx_id,
+		prepare_transaction_proof(source_client, &source_tx_id, &source_block, source_tx_index)
+			.await
+			.map_err(|err| err.to_string())?,
+	)
+	.await
+	.map_err(|err| err.to_string())
+}
+
+/// Prepare transaction proof.
+async fn prepare_transaction_proof<P: TransactionProofPipeline>(
+	source_client: &impl SourceClient<P>,
+	source_tx_id: &str,
+	source_block: &P::Block,
+	source_tx_index: usize,
+) -> Result<P::TransactionProof, StringifiedMaybeConnectionError> {
+	source_client
+		.transaction_proof(source_block, source_tx_index)
 		.await
 		.map_err(|err| {
-			format!(
-				"Error submitting transaction {} proof to {} node: {:?}",
-				source_tx_hash,
-				P::TARGET_NAME,
-				err,
+			StringifiedMaybeConnectionError::new(
+				err.is_connection_error(),
+				format!(
+					"Error building transaction {} proof on {} node: {:?}",
+					source_tx_id,
+					P::SOURCE_NAME,
+					err,
+				),
+			)
+		})
+}
+
+/// Relay prepared proof of transaction.
+async fn relay_ready_transaction_proof<P: TransactionProofPipeline>(
+	target_client: &impl TargetClient<P>,
+	source_tx_id: &str,
+	source_tx_proof: P::TransactionProof,
+) -> Result<(), StringifiedMaybeConnectionError> {
+	target_client
+		.submit_transaction_proof(source_tx_proof)
+		.await
+		.map_err(|err| {
+			StringifiedMaybeConnectionError::new(
+				err.is_connection_error(),
+				format!(
+					"Error submitting transaction {} proof to {} node: {:?}",
+					source_tx_id,
+					P::TARGET_NAME,
+					err,
+				),
 			)
 		})
 }
@@ -122,10 +318,10 @@ pub async fn relay_single_transaction_proof<P: TransactionProofPipeline>(
 /// 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> {
+	source_tx_hash: &TransactionHashOf<P>,
+) -> Result<(HeaderId<P>, usize), String> {
 	loop {
-		let source_header_and_tx = source_client.transaction(&source_tx_hash).await.map_err(|err| {
+		let source_header_and_tx = source_client.transaction_block(&source_tx_hash).await.map_err(|err| {
 			format!(
 				"Error retrieving transaction {} from {} node: {:?}",
 				source_tx_hash,
@@ -245,68 +441,144 @@ async fn wait_header_finalized<P: TransactionProofPipeline>(
 	}
 }
 
+impl StringifiedMaybeConnectionError {
+	fn new(is_connection_error: bool, error: String) -> Self {
+		if is_connection_error {
+			StringifiedMaybeConnectionError::Connection(error)
+		} else {
+			StringifiedMaybeConnectionError::NonConnection(error)
+		}
+	}
+}
+
+impl MaybeConnectionError for StringifiedMaybeConnectionError {
+	fn is_connection_error(&self) -> bool {
+		match *self {
+			StringifiedMaybeConnectionError::Connection(_) => true,
+			StringifiedMaybeConnectionError::NonConnection(_) => false,
+		}
+	}
+}
+
+impl ToString for StringifiedMaybeConnectionError {
+	fn to_string(&self) -> String {
+		match *self {
+			StringifiedMaybeConnectionError::Connection(ref err) => err.clone(),
+			StringifiedMaybeConnectionError::NonConnection(ref err) => err.clone(),
+		}
+	}
+}
+
 #[cfg(test)]
-mod tests {
+pub(crate) mod tests {
 	use super::*;
 	use crate::sync_types::HeaderId;
 
 	use parking_lot::Mutex;
+	use std::{
+		collections::{HashMap, HashSet},
+		sync::Arc,
+	};
 
-	fn test_header_id() -> TestHeaderId {
-		HeaderId(100, 100)
+	pub fn test_block_id() -> TestHeaderId {
+		HeaderId(1, 1)
 	}
 
-	fn test_transaction_hash() -> TestTransactionHash {
-		200
+	pub fn test_next_block_id() -> TestHeaderId {
+		HeaderId(2, 2)
 	}
 
-	fn test_transaction() -> TestTransaction {
-		300
+	pub fn test_transaction_hash(tx_index: u64) -> TestTransactionHash {
+		200 + tx_index
 	}
 
-	fn test_transaction_proof() -> TestTransactionProof {
-		400
+	pub fn test_transaction(tx_index: u64) -> TestTransaction {
+		TestTransaction(test_transaction_hash(tx_index))
 	}
 
-	type TestError = u64;
-	type TestBlockNumber = u64;
-	type TestBlockHash = u64;
-	type TestTransactionHash = u64;
-	type TestTransaction = u64;
-	type TestTransactionProof = u64;
-	type TestHeaderId = HeaderId<TestBlockHash, TestBlockNumber>;
+	pub fn test_block() -> TestBlock {
+		TestBlock(test_block_id(), vec![test_transaction(0)])
+	}
 
-	struct TestTransactionProofPipeline;
+	pub fn test_next_block() -> TestBlock {
+		TestBlock(test_next_block_id(), vec![test_transaction(1)])
+	}
+
+	pub type TestBlockNumber = u64;
+	pub type TestBlockHash = u64;
+	pub type TestTransactionHash = u64;
+	pub type TestHeaderId = HeaderId<TestBlockHash, TestBlockNumber>;
+
+	#[derive(Debug, Clone, PartialEq)]
+	pub struct TestError(pub bool);
+
+	impl MaybeConnectionError for TestError {
+		fn is_connection_error(&self) -> bool {
+			self.0
+		}
+	}
+
+	pub 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 Block = TestBlock;
 		type TransactionProof = TestTransactionProof;
 	}
 
-	struct TestTransactionsSource {
-		on_tick: Box<dyn Fn(&mut TestTransactionsSourceData) + Send + Sync>,
-		data: Mutex<TestTransactionsSourceData>,
+	#[derive(Debug, Clone)]
+	pub struct TestBlock(pub TestHeaderId, pub Vec<TestTransaction>);
+
+	impl SourceBlock for TestBlock {
+		type Hash = TestBlockHash;
+		type Number = TestBlockNumber;
+		type Transaction = TestTransaction;
+
+		fn id(&self) -> TestHeaderId {
+			self.0
+		}
+
+		fn transactions(&self) -> Vec<TestTransaction> {
+			self.1.clone()
+		}
 	}
 
-	struct TestTransactionsSourceData {
-		transaction: Result<Option<(TestHeaderId, TestTransaction)>, TestError>,
-		transaction_proof: Result<TestTransactionProof, TestError>,
+	#[derive(Debug, Clone)]
+	pub struct TestTransaction(pub TestTransactionHash);
+
+	impl SourceTransaction for TestTransaction {
+		type Hash = TestTransactionHash;
+
+		fn hash(&self) -> Self::Hash {
+			self.0
+		}
+	}
+
+	#[derive(Debug, Clone, PartialEq)]
+	pub struct TestTransactionProof(pub TestTransactionHash);
+
+	pub struct TestTransactionsSource {
+		pub on_tick: Box<dyn Fn(&mut TestTransactionsSourceData) + Send + Sync>,
+		pub data: Arc<Mutex<TestTransactionsSourceData>>,
+	}
+
+	pub struct TestTransactionsSourceData {
+		pub block: Result<TestBlock, TestError>,
+		pub transaction_block: Result<Option<(TestHeaderId, usize)>, TestError>,
+		pub proofs_to_fail: HashMap<TestTransactionHash, TestError>,
 	}
 
 	impl TestTransactionsSource {
-		fn new(on_tick: Box<dyn Fn(&mut TestTransactionsSourceData) + Send + Sync>) -> Self {
+		pub 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()),
-				}),
+				data: Arc::new(Mutex::new(TestTransactionsSourceData {
+					block: Ok(test_block()),
+					transaction_block: Ok(Some((test_block_id(), 0))),
+					proofs_to_fail: HashMap::new(),
+				})),
 			}
 		}
 	}
@@ -319,42 +591,53 @@ mod tests {
 			(self.on_tick)(&mut *self.data.lock())
 		}
 
-		async fn transaction(
-			&self,
-			_: &TestTransactionHash,
-		) -> Result<Option<(TestHeaderId, TestTransaction)>, TestError> {
-			self.data.lock().transaction
+		async fn block_by_hash(&self, _: TestBlockHash) -> Result<TestBlock, TestError> {
+			self.data.lock().block.clone()
 		}
 
-		async fn transaction_proof(
-			&self,
-			_: &TestHeaderId,
-			_: TestTransaction,
-		) -> Result<TestTransactionProof, TestError> {
-			self.data.lock().transaction_proof
+		async fn block_by_number(&self, _: TestBlockNumber) -> Result<TestBlock, TestError> {
+			self.data.lock().block.clone()
+		}
+
+		async fn transaction_block(&self, _: &TestTransactionHash) -> Result<Option<(TestHeaderId, usize)>, TestError> {
+			self.data.lock().transaction_block.clone()
+		}
+
+		async fn transaction_proof(&self, block: &TestBlock, index: usize) -> Result<TestTransactionProof, TestError> {
+			let tx_hash = block.1[index].hash();
+			let proof_error = self.data.lock().proofs_to_fail.get(&tx_hash).cloned();
+			if let Some(err) = proof_error {
+				return Err(err);
+			}
+
+			Ok(TestTransactionProof(tx_hash))
 		}
 	}
 
-	struct TestTransactionsTarget {
-		on_tick: Box<dyn Fn(&mut TestTransactionsTargetData) + Send + Sync>,
-		data: Mutex<TestTransactionsTargetData>,
+	pub struct TestTransactionsTarget {
+		pub on_tick: Box<dyn Fn(&mut TestTransactionsTargetData) + Send + Sync>,
+		pub data: Arc<Mutex<TestTransactionsTargetData>>,
 	}
 
-	struct TestTransactionsTargetData {
-		is_header_known: Result<bool, TestError>,
-		is_header_finalized: Result<bool, TestError>,
-		submitted_proofs: Vec<TestTransactionProof>,
+	pub struct TestTransactionsTargetData {
+		pub is_header_known: Result<bool, TestError>,
+		pub is_header_finalized: Result<bool, TestError>,
+		pub best_finalized_header_id: Result<TestHeaderId, TestError>,
+		pub transactions_to_accept: HashSet<TestTransactionHash>,
+		pub submitted_proofs: Vec<TestTransactionProof>,
 	}
 
 	impl TestTransactionsTarget {
-		fn new(on_tick: Box<dyn Fn(&mut TestTransactionsTargetData) + Send + Sync>) -> Self {
+		pub fn new(on_tick: Box<dyn Fn(&mut TestTransactionsTargetData) + Send + Sync>) -> Self {
 			Self {
 				on_tick,
-				data: Mutex::new(TestTransactionsTargetData {
+				data: Arc::new(Mutex::new(TestTransactionsTargetData {
 					is_header_known: Ok(true),
 					is_header_finalized: Ok(true),
+					best_finalized_header_id: Ok(test_block_id()),
+					transactions_to_accept: vec![test_transaction_hash(0)].into_iter().collect(),
 					submitted_proofs: Vec::new(),
-				}),
+				})),
 			}
 		}
 	}
@@ -368,11 +651,19 @@ mod tests {
 		}
 
 		async fn is_header_known(&self, _: &TestHeaderId) -> Result<bool, TestError> {
-			self.data.lock().is_header_known
+			self.data.lock().is_header_known.clone()
 		}
 
 		async fn is_header_finalized(&self, _: &TestHeaderId) -> Result<bool, TestError> {
-			self.data.lock().is_header_finalized
+			self.data.lock().is_header_finalized.clone()
+		}
+
+		async fn best_finalized_header_id(&self) -> Result<TestHeaderId, TestError> {
+			self.data.lock().best_finalized_header_id.clone()
+		}
+
+		async fn filter_transaction_proof(&self, proof: &TestTransactionProof) -> Result<bool, TestError> {
+			Ok(self.data.lock().transactions_to_accept.contains(&proof.0))
 		}
 
 		async fn submit_transaction_proof(&self, proof: TestTransactionProof) -> Result<(), TestError> {
@@ -381,23 +672,22 @@ mod tests {
 		}
 	}
 
-	fn ensure_success(source: TestTransactionsSource, target: TestTransactionsTarget) {
+	fn ensure_relay_single_success(source: &TestTransactionsSource, target: &TestTransactionsTarget) {
 		assert_eq!(
-			async_std::task::block_on(relay_single_transaction_proof(
-				&source,
-				&target,
-				test_transaction_hash(),
-			)),
+			async_std::task::block_on(relay_single_transaction_proof(source, target, test_transaction_hash(0),)),
 			Ok(()),
 		);
-		assert_eq!(target.data.lock().submitted_proofs, vec![test_transaction_proof()],);
+		assert_eq!(
+			target.data.lock().submitted_proofs,
+			vec![TestTransactionProof(test_transaction_hash(0))],
+		);
 	}
 
-	fn ensure_failure(source: TestTransactionsSource, target: TestTransactionsTarget) {
+	fn ensure_relay_single_failure(source: TestTransactionsSource, target: TestTransactionsTarget) {
 		assert!(async_std::task::block_on(relay_single_transaction_proof(
 			&source,
 			&target,
-			test_transaction_hash(),
+			test_transaction_hash(0),
 		))
 		.is_err(),);
 		assert!(target.data.lock().submitted_proofs.is_empty());
@@ -407,21 +697,21 @@ mod tests {
 	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)
+		ensure_relay_single_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())));
+			assert_eq!(source_data.transaction_block, Ok(None));
+			source_data.transaction_block = Ok(Some((test_block_id(), 0)));
 		}));
 		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);
+		source.data.lock().transaction_block = Ok(None);
 
-		ensure_success(source, target)
+		ensure_relay_single_success(&source, &target)
 	}
 
 	#[test]
@@ -429,9 +719,9 @@ mod tests {
 		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);
+		source.data.lock().transaction_block = Err(TestError(false));
 
-		ensure_failure(source, target)
+		ensure_relay_single_failure(source, target)
 	}
 
 	#[test]
@@ -439,9 +729,13 @@ mod tests {
 		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);
+		source
+			.data
+			.lock()
+			.proofs_to_fail
+			.insert(test_transaction_hash(0), TestError(false));
 
-		ensure_failure(source, target)
+		ensure_relay_single_failure(source, target)
 	}
 
 	#[test]
@@ -455,7 +749,7 @@ mod tests {
 		// 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)
+		ensure_relay_single_success(&source, &target)
 	}
 
 	#[test]
@@ -463,9 +757,9 @@ mod tests {
 		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);
+		target.data.lock().is_header_known = Err(TestError(false));
 
-		ensure_failure(source, target)
+		ensure_relay_single_failure(source, target)
 	}
 
 	#[test]
@@ -479,7 +773,7 @@ mod tests {
 		// 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)
+		ensure_relay_single_success(&source, &target)
 	}
 
 	#[test]
@@ -487,8 +781,147 @@ mod tests {
 		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);
+		target.data.lock().is_header_finalized = Err(TestError(false));
 
-		ensure_failure(source, target)
+		ensure_relay_single_failure(source, target)
+	}
+
+	#[test]
+	fn relay_transaction_proof_fails_when_target_node_rejects_proof() {
+		let source = TestTransactionsSource::new(Box::new(|_| unreachable!("no ticks allowed")));
+		let target = TestTransactionsTarget::new(Box::new(|_| unreachable!("no ticks allowed")));
+
+		target
+			.data
+			.lock()
+			.transactions_to_accept
+			.remove(&test_transaction_hash(0));
+
+		ensure_relay_single_success(&source, &target)
+	}
+
+	fn test_relay_block_transactions(
+		source: &TestTransactionsSource,
+		target: &TestTransactionsTarget,
+		pre_relayed: RelayedBlockTransactions,
+	) -> Result<RelayedBlockTransactions, RelayedBlockTransactions> {
+		async_std::task::block_on(relay_block_transactions(
+			source,
+			target,
+			&TestBlock(
+				test_block_id(),
+				vec![test_transaction(0), test_transaction(1), test_transaction(2)],
+			),
+			pre_relayed,
+		))
+	}
+
+	#[test]
+	fn relay_block_transactions_process_all_transactions() {
+		let source = TestTransactionsSource::new(Box::new(|_| unreachable!("no ticks allowed")));
+		let target = TestTransactionsTarget::new(Box::new(|_| unreachable!("no ticks allowed")));
+
+		// let's only accept tx#1
+		target
+			.data
+			.lock()
+			.transactions_to_accept
+			.remove(&test_transaction_hash(0));
+		target
+			.data
+			.lock()
+			.transactions_to_accept
+			.insert(test_transaction_hash(1));
+
+		let relayed_transactions = test_relay_block_transactions(&source, &target, Default::default());
+		assert_eq!(
+			relayed_transactions,
+			Ok(RelayedBlockTransactions {
+				processed: 3,
+				relayed: 1,
+				failed: 0,
+			}),
+		);
+		assert_eq!(
+			target.data.lock().submitted_proofs,
+			vec![TestTransactionProof(test_transaction_hash(1))],
+		);
+	}
+
+	#[test]
+	fn relay_block_transactions_ignores_transaction_failure() {
+		let source = TestTransactionsSource::new(Box::new(|_| unreachable!("no ticks allowed")));
+		let target = TestTransactionsTarget::new(Box::new(|_| unreachable!("no ticks allowed")));
+
+		// let's reject proof for tx#0
+		source
+			.data
+			.lock()
+			.proofs_to_fail
+			.insert(test_transaction_hash(0), TestError(false));
+
+		let relayed_transactions = test_relay_block_transactions(&source, &target, Default::default());
+		assert_eq!(
+			relayed_transactions,
+			Ok(RelayedBlockTransactions {
+				processed: 3,
+				relayed: 0,
+				failed: 1,
+			}),
+		);
+		assert_eq!(target.data.lock().submitted_proofs, vec![],);
+	}
+
+	#[test]
+	fn relay_block_transactions_fails_on_connection_error() {
+		let source = TestTransactionsSource::new(Box::new(|_| unreachable!("no ticks allowed")));
+		let target = TestTransactionsTarget::new(Box::new(|_| unreachable!("no ticks allowed")));
+
+		// fail with connection error when preparing proof for tx#1
+		source
+			.data
+			.lock()
+			.proofs_to_fail
+			.insert(test_transaction_hash(1), TestError(true));
+
+		let relayed_transactions = test_relay_block_transactions(&source, &target, Default::default());
+		assert_eq!(
+			relayed_transactions,
+			Err(RelayedBlockTransactions {
+				processed: 1,
+				relayed: 1,
+				failed: 0,
+			}),
+		);
+		assert_eq!(
+			target.data.lock().submitted_proofs,
+			vec![TestTransactionProof(test_transaction_hash(0))],
+		);
+
+		// now do not fail on tx#2
+		source.data.lock().proofs_to_fail.clear();
+		// and also relay tx#3
+		target
+			.data
+			.lock()
+			.transactions_to_accept
+			.insert(test_transaction_hash(2));
+
+		let relayed_transactions = test_relay_block_transactions(&source, &target, relayed_transactions.unwrap_err());
+		assert_eq!(
+			relayed_transactions,
+			Ok(RelayedBlockTransactions {
+				processed: 3,
+				relayed: 2,
+				failed: 0,
+			}),
+		);
+		assert_eq!(
+			target.data.lock().submitted_proofs,
+			vec![
+				TestTransactionProof(test_transaction_hash(0)),
+				TestTransactionProof(test_transaction_hash(2))
+			],
+		);
 	}
 }

bridges/relays/ethereum/src/exchange_loop.rs

@@ -0,0 +1,267 @@
+// 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 crate::exchange::{
+	relay_block_transactions, BlockNumberOf, RelayedBlockTransactions, SourceClient, TargetClient,
+	TransactionProofPipeline,
+};
+use crate::utils::retry_backoff;
+
+use backoff::backoff::Backoff;
+use futures::{future::FutureExt, select};
+use num_traits::One;
+use std::{future::Future, time::Duration};
+
+/// Delay after connection-related error happened before we'll try
+/// reconnection again.
+const CONNECTION_ERROR_DELAY: Duration = Duration::from_secs(10);
+
+/// Transactions proofs relay state.
+#[derive(Debug)]
+pub struct TransactionProofsRelayState<BlockNumber> {
+	/// Number of last header we have processed so far.
+	pub best_processed_header_number: BlockNumber,
+}
+
+/// Transactions proofs relay storage.
+pub trait TransactionProofsRelayStorage {
+	/// Associated block number.
+	type BlockNumber;
+
+	/// Get relay state.
+	fn state(&self) -> TransactionProofsRelayState<Self::BlockNumber>;
+	/// Update relay state.
+	fn set_state(&mut self, state: &TransactionProofsRelayState<Self::BlockNumber>);
+}
+
+/// In-memory storage for auto-relay loop.
+#[derive(Debug)]
+pub struct InMemoryStorage<BlockNumber> {
+	best_processed_header_number: BlockNumber,
+}
+
+impl<BlockNumber> InMemoryStorage<BlockNumber> {
+	/// Created new in-memory storage with given best processed block number.
+	pub fn new(best_processed_header_number: BlockNumber) -> Self {
+		InMemoryStorage {
+			best_processed_header_number,
+		}
+	}
+}
+
+impl<BlockNumber: Clone + Copy> TransactionProofsRelayStorage for InMemoryStorage<BlockNumber> {
+	type BlockNumber = BlockNumber;
+
+	fn state(&self) -> TransactionProofsRelayState<BlockNumber> {
+		TransactionProofsRelayState {
+			best_processed_header_number: self.best_processed_header_number,
+		}
+	}
+
+	fn set_state(&mut self, state: &TransactionProofsRelayState<BlockNumber>) {
+		self.best_processed_header_number = state.best_processed_header_number;
+	}
+}
+
+/// Run proofs synchronization.
+pub fn run<P: TransactionProofPipeline>(
+	mut storage: impl TransactionProofsRelayStorage<BlockNumber = BlockNumberOf<P>>,
+	source_client: impl SourceClient<P>,
+	target_client: impl TargetClient<P>,
+	exit_signal: impl Future<Output = ()>,
+) {
+	let mut local_pool = futures::executor::LocalPool::new();
+
+	local_pool.run_until(async move {
+		let mut retry_backoff = retry_backoff();
+		let mut state = storage.state();
+		let mut current_finalized_block = None;
+
+		let exit_signal = exit_signal.fuse();
+
+		futures::pin_mut!(exit_signal);
+
+		loop {
+			let iteration_result = run_loop_iteration(
+				&mut storage,
+				&source_client,
+				&target_client,
+				&mut state,
+				&mut current_finalized_block,
+			)
+			.await;
+
+			match iteration_result {
+				Ok(_) => {
+					retry_backoff.reset();
+
+					select! {
+						_ = source_client.tick().fuse() => {},
+						_ = exit_signal => return,
+					}
+				}
+				Err(_) => {
+					let retry_timeout = retry_backoff.next_backoff().unwrap_or(CONNECTION_ERROR_DELAY);
+
+					select! {
+						_ = async_std::task::sleep(retry_timeout).fuse() => {},
+						_ = exit_signal => return,
+					}
+				}
+			}
+		}
+	});
+}
+
+/// Run exchange loop until we need to break.
+async fn run_loop_iteration<P: TransactionProofPipeline>(
+	storage: &mut impl TransactionProofsRelayStorage<BlockNumber = BlockNumberOf<P>>,
+	source_client: &impl SourceClient<P>,
+	target_client: &impl TargetClient<P>,
+	state: &mut TransactionProofsRelayState<BlockNumberOf<P>>,
+	current_finalized_block: &mut Option<(P::Block, RelayedBlockTransactions)>,
+) -> Result<(), ()> {
+	let best_finalized_header_id = match target_client.best_finalized_header_id().await {
+		Ok(best_finalized_header_id) => {
+			log::trace!(
+				target: "bridge",
+				"Got best finalized {} block from {} node: {:?}",
+				P::SOURCE_NAME,
+				P::TARGET_NAME,
+				best_finalized_header_id,
+			);
+
+			best_finalized_header_id
+		}
+		Err(err) => {
+			log::error!(
+				target: "bridge",
+				"Failed to retrieve best {} header id from {} node: {:?}. Going to retry...",
+				P::SOURCE_NAME,
+				P::TARGET_NAME,
+				err,
+			);
+
+			return Err(());
+		}
+	};
+
+	loop {
+		// if we already have some finalized block body, try to relay its transactions
+		if let Some((block, relayed_transactions)) = current_finalized_block.take() {
+			let result = relay_block_transactions(source_client, target_client, &block, relayed_transactions).await;
+
+			match result {
+				Ok(relayed_transactions) => {
+					log::info!(
+						target: "bridge",
+						"Relay has processed {} block #{}. Total/Relayed/Failed transactions: {}/{}/{}",
+						P::SOURCE_NAME,
+						state.best_processed_header_number,
+						relayed_transactions.processed,
+						relayed_transactions.relayed,
+						relayed_transactions.failed,
+					);
+
+					state.best_processed_header_number = state.best_processed_header_number + One::one();
+					storage.set_state(state);
+
+					// we have just updated state => proceed to next block retrieval
+				}
+				Err(relayed_transactions) => {
+					*current_finalized_block = Some((block, relayed_transactions));
+					return Err(());
+				}
+			}
+		}
+
+		// we may need to retrieve finalized block body from source node
+		if best_finalized_header_id.0 > state.best_processed_header_number {
+			let next_block_number = state.best_processed_header_number + One::one();
+			let result = source_client.block_by_number(next_block_number).await;
+
+			match result {
+				Ok(block) => {
+					*current_finalized_block = Some((block, RelayedBlockTransactions::default()));
+
+					// we have received new finalized block => go back to relay its transactions
+					continue;
+				}
+				Err(err) => {
+					log::error!(
+						target: "bridge",
+						"Failed to retrieve canonical block #{} from {} node: {:?}. Going to retry...",
+						next_block_number,
+						P::SOURCE_NAME,
+						err,
+					);
+
+					return Err(());
+				}
+			}
+		}
+
+		// there are no any transactions we need to relay => wait for new data
+		return Ok(());
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::exchange::tests::{
+		test_next_block, test_next_block_id, test_transaction_hash, TestTransactionProof, TestTransactionsSource,
+		TestTransactionsTarget,
+	};
+	use futures::{future::FutureExt, stream::StreamExt};
+
+	#[test]
+	fn exchange_loop_is_able_to_relay_proofs() {
+		let storage = InMemoryStorage {
+			best_processed_header_number: 0,
+		};
+		let target = TestTransactionsTarget::new(Box::new(|_| unreachable!("no target ticks allowed")));
+		let target_data = target.data.clone();
+		let (exit_sender, exit_receiver) = futures::channel::mpsc::unbounded();
+
+		let source = TestTransactionsSource::new(Box::new(move |data| {
+			let transaction1_relayed = target_data
+				.lock()
+				.submitted_proofs
+				.contains(&TestTransactionProof(test_transaction_hash(0)));
+			let transaction2_relayed = target_data
+				.lock()
+				.submitted_proofs
+				.contains(&TestTransactionProof(test_transaction_hash(1)));
+			match (transaction1_relayed, transaction2_relayed) {
+				(true, true) => exit_sender.unbounded_send(()).unwrap(),
+				(true, false) => {
+					data.block = Ok(test_next_block());
+					target_data.lock().best_finalized_header_id = Ok(test_next_block_id());
+					target_data
+						.lock()
+						.transactions_to_accept
+						.insert(test_transaction_hash(1));
+				}
+				_ => (),
+			}
+		}));
+
+		run(storage, source, target, exit_receiver.into_future().map(|(_, _)| ()));
+	}
+}

bridges/relays/ethereum/src/main.rs

@@ -22,6 +22,7 @@ mod ethereum_exchange;
 mod ethereum_sync_loop;
 mod ethereum_types;
 mod exchange;
+mod exchange_loop;
 mod headers;
 mod metrics;
 mod rpc;
@@ -260,11 +261,21 @@ fn ethereum_exchange_params(matches: &clap::ArgMatches) -> Result<ethereum_excha
 	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))?;
+	params.mode = match matches.value_of("eth-tx-hash") {
+		Some(eth_tx_hash) => ethereum_exchange::ExchangeRelayMode::Single(
+			eth_tx_hash
+				.parse()
+				.map_err(|e| format!("Failed to parse eth-tx-hash: {}", e))?,
+		),
+		None => ethereum_exchange::ExchangeRelayMode::Auto(match matches.value_of("eth-start-with-block") {
+			Some(eth_start_with_block) => Some(
+				eth_start_with_block
+					.parse()
+					.map_err(|e| format!("Failed to parse eth-start-with-block: {}", e))?,
+			),
+			None => None,
+		}),
+	};
 
 	Ok(params)
 }

bridges/relays/ethereum/src/rpc.rs

@@ -50,6 +50,8 @@ jsonrpsee::rpc_api! {
 		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, full_tx_objs: bool) -> EthereumHeader;
+		#[rpc(method = "eth_getBlockByNumber", positional_params)]
+		fn get_block_by_number_with_transactions(number: U64, full_tx_objs: bool) -> EthereumHeaderWithTransactions;
 		#[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)]
@@ -91,6 +93,8 @@ 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 number from Ethereum node.
+	async fn header_by_number_with_transactions(&self, block_number: u64) -> Result<EthereumHeaderWithTransactions>;
 	/// 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.

bridges/relays/ethereum/src/rpc_errors.rs

@@ -14,7 +14,6 @@
 // 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;
@@ -103,10 +102,6 @@ pub enum EthereumNodeError {
 	/// 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 {
@@ -122,14 +117,6 @@ impl ToString for EthereumNodeError {
 			}
 			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,
-			),
 		}
 	}
 }

bridges/relays/ethereum/src/substrate_client.rs

@@ -38,6 +38,7 @@ const ETH_API_IMPORT_REQUIRES_RECEIPTS: &str = "EthereumHeadersApi_is_import_req
 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 EXCH_API_FILTER_TRANSACTION_PROOF: &str = "CurrencyExchangeApi_filter_transaction_proof";
 const SUB_API_GRANDPA_AUTHORITIES: &str = "GrandpaApi_grandpa_authorities";
 
 type Result<T> = std::result::Result<T, RpcError>;
@@ -145,7 +146,7 @@ impl SubstrateRpc for SubstrateRpcClient {
 
 	async fn best_ethereum_finalized_block(&self) -> Result<EthereumHeaderId> {
 		let call = ETH_API_BEST_FINALIZED_BLOCK.to_string();
-		let data = Bytes("0x".into());
+		let data = Bytes(Vec::new());
 
 		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[..])?;
@@ -298,6 +299,11 @@ impl SubmitEthereumHeaders for SubstrateRpcClient {
 /// calls.
 #[async_trait]
 pub trait SubmitEthereumExchangeTransactionProof: SubstrateRpc {
+	/// Pre-verify Ethereum exchange transaction proof.
+	async fn verify_exchange_transaction_proof(
+		&self,
+		proof: bridge_node_runtime::exchange::EthereumTransactionInclusionProof,
+	) -> Result<bool>;
 	/// Submits Ethereum exchange transaction proof to Substrate runtime.
 	async fn submit_exchange_transaction_proof(
 		&self,
@@ -308,6 +314,19 @@ pub trait SubmitEthereumExchangeTransactionProof: SubstrateRpc {
 
 #[async_trait]
 impl SubmitEthereumExchangeTransactionProof for SubstrateRpcClient {
+	async fn verify_exchange_transaction_proof(
+		&self,
+		proof: bridge_node_runtime::exchange::EthereumTransactionInclusionProof,
+	) -> Result<bool> {
+		let call = EXCH_API_FILTER_TRANSACTION_PROOF.to_string();
+		let data = Bytes(proof.encode());
+
+		let encoded_response = Substrate::state_call(&self.client, call, data, None).await?;
+		let is_allowed: bool = Decode::decode(&mut &encoded_response.0[..])?;
+
+		Ok(is_allowed)
+	}
+
 	async fn submit_exchange_transaction_proof(
 		&self,
 		params: SubstrateSigningParams,

bridges/relays/ethereum/src/sync_loop.rs

@@ -20,6 +20,7 @@ use crate::sync_loop_metrics::SyncLoopMetrics;
 use crate::sync_types::{
 	HeaderIdOf, HeaderStatus, HeadersSyncPipeline, MaybeConnectionError, QueuedHeader, SubmittedHeaders,
 };
+use crate::utils::retry_backoff;
 
 use async_trait::async_trait;
 use backoff::{backoff::Backoff, ExponentialBackoff};
@@ -48,9 +49,6 @@ const BACKUP_STALL_SYNC_TIMEOUT: Duration = Duration::from_secs(10 * 60);
 /// Delay after connection-related error happened before we'll try
 /// reconnection again.
 const CONNECTION_ERROR_DELAY: Duration = Duration::from_secs(10);
-/// Max delay after connection-unrelated error happened before we'll try the
-/// same request again.
-const MAX_BACKOFF_INTERVAL: Duration = Duration::from_secs(60);
 
 /// Source client trait.
 #[async_trait]
@@ -569,15 +567,6 @@ fn interval(timeout: Duration) -> impl futures::Stream<Item = ()> {
 	})
 }
 
-/// Exponential backoff for connection-unrelated errors retries.
-pub(crate) fn retry_backoff() -> ExponentialBackoff {
-	let mut backoff = ExponentialBackoff::default();
-	// we do not want relayer to stop
-	backoff.max_elapsed_time = None;
-	backoff.max_interval = MAX_BACKOFF_INTERVAL;
-	backoff
-}
-
 /// Process result of the future from a client.
 ///
 /// Returns whether or not the client we're interacting with is online. In this context

bridges/relays/ethereum/src/sync_loop_tests.rs

@@ -16,10 +16,11 @@
 
 #![cfg(test)]
 
-use crate::sync_loop::{process_future_result, retry_backoff, run, SourceClient, TargetClient};
+use crate::sync_loop::{process_future_result, run, SourceClient, TargetClient};
 use crate::sync_types::{
 	HeaderId, HeadersSyncPipeline, MaybeConnectionError, QueuedHeader, SourceHeader, SubmittedHeaders,
 };
+use crate::utils::retry_backoff;
 
 use async_trait::async_trait;
 use backoff::backoff::Backoff;

bridges/relays/ethereum/src/utils.rs

@@ -14,6 +14,13 @@
 // 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 backoff::ExponentialBackoff;
+use std::time::Duration;
+
+/// Max delay after connection-unrelated error happened before we'll try the
+/// same request again.
+const MAX_BACKOFF_INTERVAL: Duration = Duration::from_secs(60);
+
 /// Macro that returns (client, Err(error)) tuple from function if result is Err(error).
 #[macro_export]
 macro_rules! bail_on_error {
@@ -35,3 +42,12 @@ macro_rules! bail_on_arg_error {
 			}
 	};
 }
+
+/// Exponential backoff for connection-unrelated errors retries.
+pub fn retry_backoff() -> ExponentialBackoff {
+	let mut backoff = ExponentialBackoff::default();
+	// we do not want relayer to stop
+	backoff.max_elapsed_time = None;
+	backoff.max_interval = MAX_BACKOFF_INTERVAL;
+	backoff
+}