Add async_trait to SourceClient and TargetClient (#123)

* Use async_trait for SourceClient

* Use aync_trait for TargetClient

* Revert async_trait usage for Source/Target client

This reverts commit f636ffaffd60197e90e887362b4a0c35a0dc5a6c.
This reverts commit 2c15755e8c93318f8e0a605852efe87d72edb769.

I'm having a very hard time finding out what is causing compilation
issues, and I think it's best to start over again.

* Use async_trait for TargetClient

* Use async_trait for SourceClient

* Move where non-async methods are

* RustFmt

* QueuedHeader holds Arc to actual data

* Clean up async return type

Co-authored-by: Svyatoslav Nikolsky <svyatonik@gmail.com>

* Clean up async return type

Co-authored-by: Svyatoslav Nikolsky <svyatonik@gmail.com>

* Clean up async return type

Co-authored-by: Svyatoslav Nikolsky <svyatonik@gmail.com>

* Remove unused import

Co-authored-by: Svyatoslav Nikolsky <svyatonik@gmail.com>

bridges/relays/ethereum/src/ethereum_client.rs

@@ -284,7 +284,7 @@ pub async fn submit_substrate_headers(
 				Some(contract_address),
 				Some(nonce),
 				false,
-				bridge_contract::functions::import_header::encode_input(header.extract().0.encode(),),
+				bridge_contract::functions::import_header::encode_input(header.header().encode(),),
 			)
 			.await
 		)

bridges/relays/ethereum/src/ethereum_sync_loop.rs

@@ -21,8 +21,10 @@ use crate::ethereum_types::{EthereumHeaderId, EthereumHeadersSyncPipeline, Heade
 use crate::substrate_client::{self, SubstrateConnectionParams, SubstrateSigningParams};
 use crate::sync::{HeadersSyncParams, TargetTransactionMode};
 use crate::sync_loop::{OwnedSourceFutureOutput, OwnedTargetFutureOutput, SourceClient, TargetClient};
-use futures::future::{ready, FutureExt, Ready};
-use std::{collections::HashSet, future::Future, pin::Pin, time::Duration};
+
+use async_trait::async_trait;
+use futures::future::FutureExt;
+use std::{collections::HashSet, time::Duration};
 use web3::types::H256;
 
 /// Interval at which we check new Ethereum headers when we are synced/almost synced.
@@ -80,40 +82,40 @@ struct EthereumHeadersSource {
 
 type EthereumFutureOutput<T> = OwnedSourceFutureOutput<EthereumHeadersSource, EthereumHeadersSyncPipeline, T>;
 
+#[async_trait]
 impl SourceClient<EthereumHeadersSyncPipeline> for EthereumHeadersSource {
 	type Error = ethereum_client::Error;
-	type BestBlockNumberFuture = Pin<Box<dyn Future<Output = EthereumFutureOutput<u64>>>>;
-	type HeaderByHashFuture = Pin<Box<dyn Future<Output = EthereumFutureOutput<Header>>>>;
-	type HeaderByNumberFuture = Pin<Box<dyn Future<Output = EthereumFutureOutput<Header>>>>;
-	type HeaderExtraFuture = Pin<Box<dyn Future<Output = EthereumFutureOutput<(EthereumHeaderId, Vec<Receipt>)>>>>;
-	type HeaderCompletionFuture = Ready<EthereumFutureOutput<(EthereumHeaderId, Option<()>)>>;
 
-	fn best_block_number(self) -> Self::BestBlockNumberFuture {
+	async fn best_block_number(self) -> EthereumFutureOutput<u64> {
 		ethereum_client::best_block_number(self.client)
 			.map(|(client, result)| (EthereumHeadersSource { client }, result))
-			.boxed()
+			.await
 	}
 
-	fn header_by_hash(self, hash: H256) -> Self::HeaderByHashFuture {
+	async fn header_by_hash(self, hash: H256) -> EthereumFutureOutput<Header> {
 		ethereum_client::header_by_hash(self.client, hash)
 			.map(|(client, result)| (EthereumHeadersSource { client }, result))
-			.boxed()
+			.await
 	}
 
-	fn header_by_number(self, number: u64) -> Self::HeaderByNumberFuture {
+	async fn header_by_number(self, number: u64) -> EthereumFutureOutput<Header> {
 		ethereum_client::header_by_number(self.client, number)
 			.map(|(client, result)| (EthereumHeadersSource { client }, result))
-			.boxed()
+			.await
 	}
 
-	fn header_extra(self, id: EthereumHeaderId, header: &Header) -> Self::HeaderExtraFuture {
-		ethereum_client::transactions_receipts(self.client, id, header.transactions.clone())
+	async fn header_completion(self, id: EthereumHeaderId) -> EthereumFutureOutput<(EthereumHeaderId, Option<()>)> {
+		(self, Ok((id, None)))
+	}
+
+	async fn header_extra(
+		self,
+		id: EthereumHeaderId,
+		header: QueuedEthereumHeader,
+	) -> EthereumFutureOutput<(EthereumHeaderId, Vec<Receipt>)> {
+		ethereum_client::transactions_receipts(self.client, id, header.header().transactions.clone())
 			.map(|(client, result)| (EthereumHeadersSource { client }, result))
-			.boxed()
-	}
-
-	fn header_completion(self, id: EthereumHeaderId) -> Self::HeaderCompletionFuture {
-		ready((self, Ok((id, None))))
+			.await
 	}
 }
 
@@ -129,16 +131,11 @@ struct SubstrateHeadersTarget {
 
 type SubstrateFutureOutput<T> = OwnedTargetFutureOutput<SubstrateHeadersTarget, EthereumHeadersSyncPipeline, T>;
 
+#[async_trait]
 impl TargetClient<EthereumHeadersSyncPipeline> for SubstrateHeadersTarget {
 	type Error = substrate_client::Error;
-	type BestHeaderIdFuture = Pin<Box<dyn Future<Output = SubstrateFutureOutput<EthereumHeaderId>>>>;
-	type IsKnownHeaderFuture = Pin<Box<dyn Future<Output = SubstrateFutureOutput<(EthereumHeaderId, bool)>>>>;
-	type RequiresExtraFuture = Pin<Box<dyn Future<Output = SubstrateFutureOutput<(EthereumHeaderId, bool)>>>>;
-	type SubmitHeadersFuture = Pin<Box<dyn Future<Output = SubstrateFutureOutput<Vec<EthereumHeaderId>>>>>;
-	type IncompleteHeadersFuture = Ready<SubstrateFutureOutput<HashSet<EthereumHeaderId>>>;
-	type CompleteHeadersFuture = Ready<SubstrateFutureOutput<EthereumHeaderId>>;
 
-	fn best_header_id(self) -> Self::BestHeaderIdFuture {
+	async fn best_header_id(self) -> SubstrateFutureOutput<EthereumHeaderId> {
 		let (sign_transactions, sign_params) = (self.sign_transactions, self.sign_params);
 		substrate_client::best_ethereum_block(self.client)
 			.map(move |(client, result)| {
@@ -151,10 +148,10 @@ impl TargetClient<EthereumHeadersSyncPipeline> for SubstrateHeadersTarget {
 					result,
 				)
 			})
-			.boxed()
+			.await
 	}
 
-	fn is_known_header(self, id: EthereumHeaderId) -> Self::IsKnownHeaderFuture {
+	async fn is_known_header(self, id: EthereumHeaderId) -> SubstrateFutureOutput<(EthereumHeaderId, bool)> {
 		let (sign_transactions, sign_params) = (self.sign_transactions, self.sign_params);
 		substrate_client::ethereum_header_known(self.client, id)
 			.map(move |(client, result)| {
@@ -167,29 +164,10 @@ impl TargetClient<EthereumHeadersSyncPipeline> for SubstrateHeadersTarget {
 					result,
 				)
 			})
-			.boxed()
+			.await
 	}
 
-	fn requires_extra(self, header: &QueuedEthereumHeader) -> Self::RequiresExtraFuture {
-		// we can minimize number of receipts_check calls by checking header
-		// logs bloom here, but it may give us false positives (when authorities
-		// source is contract, we never need any logs)
-		let (sign_transactions, sign_params) = (self.sign_transactions, self.sign_params);
-		substrate_client::ethereum_receipts_required(self.client, header.clone())
-			.map(move |(client, result)| {
-				(
-					SubstrateHeadersTarget {
-						client,
-						sign_transactions,
-						sign_params,
-					},
-					result,
-				)
-			})
-			.boxed()
-	}
-
-	fn submit_headers(self, headers: Vec<QueuedEthereumHeader>) -> Self::SubmitHeadersFuture {
+	async fn submit_headers(self, headers: Vec<QueuedEthereumHeader>) -> SubstrateFutureOutput<Vec<EthereumHeaderId>> {
 		let (sign_transactions, sign_params) = (self.sign_transactions, self.sign_params);
 		substrate_client::submit_ethereum_headers(self.client, sign_params.clone(), headers, sign_transactions)
 			.map(move |(client, result)| {
@@ -202,15 +180,34 @@ impl TargetClient<EthereumHeadersSyncPipeline> for SubstrateHeadersTarget {
 					result,
 				)
 			})
-			.boxed()
+			.await
 	}
 
-	fn incomplete_headers_ids(self) -> Self::IncompleteHeadersFuture {
-		ready((self, Ok(HashSet::new())))
+	async fn incomplete_headers_ids(self) -> SubstrateFutureOutput<HashSet<EthereumHeaderId>> {
+		(self, Ok(HashSet::new()))
 	}
 
-	fn complete_header(self, id: EthereumHeaderId, _completion: ()) -> Self::CompleteHeadersFuture {
-		ready((self, Ok(id)))
+	async fn complete_header(self, id: EthereumHeaderId, _completion: ()) -> SubstrateFutureOutput<EthereumHeaderId> {
+		(self, Ok(id))
+	}
+
+	async fn requires_extra(self, header: QueuedEthereumHeader) -> SubstrateFutureOutput<(EthereumHeaderId, bool)> {
+		// we can minimize number of receipts_check calls by checking header
+		// logs bloom here, but it may give us false positives (when authorities
+		// source is contract, we never need any logs)
+		let (sign_transactions, sign_params) = (self.sign_transactions, self.sign_params);
+		substrate_client::ethereum_receipts_required(self.client, header)
+			.map(move |(client, result)| {
+				(
+					SubstrateHeadersTarget {
+						client,
+						sign_transactions,
+						sign_params,
+					},
+					result,
+				)
+			})
+			.await
 	}
 }
 

bridges/relays/ethereum/src/substrate_client.rs

@@ -380,10 +380,9 @@ fn create_signed_submit_transaction(
 			headers
 				.into_iter()
 				.map(|header| {
-					let (header, receipts) = header.extract();
 					(
-						into_substrate_ethereum_header(&header),
-						into_substrate_ethereum_receipts(&receipts),
+						into_substrate_ethereum_header(header.header()),
+						into_substrate_ethereum_receipts(header.extra()),
 					)
 				})
 				.collect(),
@@ -422,11 +421,10 @@ fn create_signed_submit_transaction(
 
 /// Create unsigned Substrate transaction for submitting Ethereum header.
 fn create_unsigned_submit_transaction(header: QueuedEthereumHeader) -> bridge_node_runtime::UncheckedExtrinsic {
-	let (header, receipts) = header.extract();
 	let function =
 		bridge_node_runtime::Call::BridgeEthPoA(bridge_node_runtime::BridgeEthPoACall::import_unsigned_header(
-			into_substrate_ethereum_header(&header),
-			into_substrate_ethereum_receipts(&receipts),
+			into_substrate_ethereum_header(header.header()),
+			into_substrate_ethereum_receipts(header.extra()),
 		));
 
 	bridge_node_runtime::UncheckedExtrinsic::new_unsigned(function)

bridges/relays/ethereum/src/substrate_sync_loop.rs

@@ -25,8 +25,10 @@ use crate::substrate_types::{
 use crate::sync::{HeadersSyncParams, TargetTransactionMode};
 use crate::sync_loop::{OwnedSourceFutureOutput, OwnedTargetFutureOutput, SourceClient, TargetClient};
 use crate::sync_types::SourceHeader;
-use futures::future::{ready, FutureExt, Ready};
-use std::{collections::HashSet, future::Future, pin::Pin, time::Duration};
+
+use async_trait::async_trait;
+use futures::future::FutureExt;
+use std::{collections::HashSet, time::Duration};
 
 /// Interval at which we check new Substrate headers when we are synced/almost synced.
 const SUBSTRATE_TICK_INTERVAL: Duration = Duration::from_secs(10);
@@ -87,41 +89,43 @@ struct SubstrateHeadersSource {
 
 type SubstrateFutureOutput<T> = OwnedSourceFutureOutput<SubstrateHeadersSource, SubstrateHeadersSyncPipeline, T>;
 
+#[async_trait]
 impl SourceClient<SubstrateHeadersSyncPipeline> for SubstrateHeadersSource {
 	type Error = substrate_client::Error;
-	type BestBlockNumberFuture = Pin<Box<dyn Future<Output = SubstrateFutureOutput<Number>>>>;
-	type HeaderByHashFuture = Pin<Box<dyn Future<Output = SubstrateFutureOutput<Header>>>>;
-	type HeaderByNumberFuture = Pin<Box<dyn Future<Output = SubstrateFutureOutput<Header>>>>;
-	type HeaderExtraFuture = Ready<SubstrateFutureOutput<(SubstrateHeaderId, ())>>;
-	type HeaderCompletionFuture =
-		Pin<Box<dyn Future<Output = SubstrateFutureOutput<(SubstrateHeaderId, Option<GrandpaJustification>)>>>>;
 
-	fn best_block_number(self) -> Self::BestBlockNumberFuture {
+	async fn best_block_number(self) -> SubstrateFutureOutput<Number> {
 		substrate_client::best_header(self.client)
 			.map(|(client, result)| (SubstrateHeadersSource { client }, result.map(|header| header.number)))
-			.boxed()
+			.await
 	}
 
-	fn header_by_hash(self, hash: Hash) -> Self::HeaderByHashFuture {
+	async fn header_by_hash(self, hash: Hash) -> SubstrateFutureOutput<Header> {
 		substrate_client::header_by_hash(self.client, hash)
 			.map(|(client, result)| (SubstrateHeadersSource { client }, result))
-			.boxed()
+			.await
 	}
 
-	fn header_by_number(self, number: Number) -> Self::HeaderByNumberFuture {
+	async fn header_by_number(self, number: Number) -> SubstrateFutureOutput<Header> {
 		substrate_client::header_by_number(self.client, number)
 			.map(|(client, result)| (SubstrateHeadersSource { client }, result))
-			.boxed()
+			.await
 	}
 
-	fn header_extra(self, id: SubstrateHeaderId, _header: &Header) -> Self::HeaderExtraFuture {
-		ready((self, Ok((id, ()))))
-	}
-
-	fn header_completion(self, id: SubstrateHeaderId) -> Self::HeaderCompletionFuture {
+	async fn header_completion(
+		self,
+		id: SubstrateHeaderId,
+	) -> SubstrateFutureOutput<(SubstrateHeaderId, Option<GrandpaJustification>)> {
 		substrate_client::grandpa_justification(self.client, id)
 			.map(|(client, result)| (SubstrateHeadersSource { client }, result))
-			.boxed()
+			.await
+	}
+
+	async fn header_extra(
+		self,
+		id: SubstrateHeaderId,
+		_header: QueuedSubstrateHeader,
+	) -> SubstrateFutureOutput<(SubstrateHeaderId, ())> {
+		(self, Ok((id, ())))
 	}
 }
 
@@ -137,16 +141,11 @@ struct EthereumHeadersTarget {
 
 type EthereumFutureOutput<T> = OwnedTargetFutureOutput<EthereumHeadersTarget, SubstrateHeadersSyncPipeline, T>;
 
+#[async_trait]
 impl TargetClient<SubstrateHeadersSyncPipeline> for EthereumHeadersTarget {
 	type Error = ethereum_client::Error;
-	type BestHeaderIdFuture = Pin<Box<dyn Future<Output = EthereumFutureOutput<SubstrateHeaderId>>>>;
-	type IsKnownHeaderFuture = Pin<Box<dyn Future<Output = EthereumFutureOutput<(SubstrateHeaderId, bool)>>>>;
-	type RequiresExtraFuture = Ready<EthereumFutureOutput<(SubstrateHeaderId, bool)>>;
-	type SubmitHeadersFuture = Pin<Box<dyn Future<Output = EthereumFutureOutput<Vec<SubstrateHeaderId>>>>>;
-	type IncompleteHeadersFuture = Pin<Box<dyn Future<Output = EthereumFutureOutput<HashSet<SubstrateHeaderId>>>>>;
-	type CompleteHeadersFuture = Pin<Box<dyn Future<Output = EthereumFutureOutput<SubstrateHeaderId>>>>;
 
-	fn best_header_id(self) -> Self::BestHeaderIdFuture {
+	async fn best_header_id(self) -> EthereumFutureOutput<SubstrateHeaderId> {
 		let (contract, sign_params) = (self.contract, self.sign_params);
 		ethereum_client::best_substrate_block(self.client, contract)
 			.map(move |(client, result)| {
@@ -159,10 +158,10 @@ impl TargetClient<SubstrateHeadersSyncPipeline> for EthereumHeadersTarget {
 					result,
 				)
 			})
-			.boxed()
+			.await
 	}
 
-	fn is_known_header(self, id: SubstrateHeaderId) -> Self::IsKnownHeaderFuture {
+	async fn is_known_header(self, id: SubstrateHeaderId) -> EthereumFutureOutput<(SubstrateHeaderId, bool)> {
 		let (contract, sign_params) = (self.contract, self.sign_params);
 		ethereum_client::substrate_header_known(self.client, contract, id)
 			.map(move |(client, result)| {
@@ -175,14 +174,10 @@ impl TargetClient<SubstrateHeadersSyncPipeline> for EthereumHeadersTarget {
 					result,
 				)
 			})
-			.boxed()
+			.await
 	}
 
-	fn requires_extra(self, header: &QueuedSubstrateHeader) -> Self::RequiresExtraFuture {
-		ready((self, Ok((header.header().id(), false))))
-	}
-
-	fn submit_headers(self, headers: Vec<QueuedSubstrateHeader>) -> Self::SubmitHeadersFuture {
+	async fn submit_headers(self, headers: Vec<QueuedSubstrateHeader>) -> EthereumFutureOutput<Vec<SubstrateHeaderId>> {
 		let (contract, sign_params) = (self.contract, self.sign_params);
 		ethereum_client::submit_substrate_headers(self.client, sign_params.clone(), contract, headers)
 			.map(move |(client, result)| {
@@ -195,10 +190,10 @@ impl TargetClient<SubstrateHeadersSyncPipeline> for EthereumHeadersTarget {
 					result,
 				)
 			})
-			.boxed()
+			.await
 	}
 
-	fn incomplete_headers_ids(self) -> Self::IncompleteHeadersFuture {
+	async fn incomplete_headers_ids(self) -> EthereumFutureOutput<HashSet<SubstrateHeaderId>> {
 		let (contract, sign_params) = (self.contract, self.sign_params);
 		ethereum_client::incomplete_substrate_headers(self.client, contract)
 			.map(move |(client, result)| {
@@ -211,10 +206,14 @@ impl TargetClient<SubstrateHeadersSyncPipeline> for EthereumHeadersTarget {
 					result,
 				)
 			})
-			.boxed()
+			.await
 	}
 
-	fn complete_header(self, id: SubstrateHeaderId, completion: GrandpaJustification) -> Self::CompleteHeadersFuture {
+	async fn complete_header(
+		self,
+		id: SubstrateHeaderId,
+		completion: GrandpaJustification,
+	) -> EthereumFutureOutput<SubstrateHeaderId> {
 		let (contract, sign_params) = (self.contract, self.sign_params);
 		ethereum_client::complete_substrate_header(self.client, sign_params.clone(), contract, id, completion)
 			.map(move |(client, result)| {
@@ -227,7 +226,11 @@ impl TargetClient<SubstrateHeadersSyncPipeline> for EthereumHeadersTarget {
 					result,
 				)
 			})
-			.boxed()
+			.await
+	}
+
+	async fn requires_extra(self, header: QueuedSubstrateHeader) -> EthereumFutureOutput<(SubstrateHeaderId, bool)> {
+		(self, Ok((header.header().id(), false)))
 	}
 }
 

bridges/relays/ethereum/src/sync_loop.rs

@@ -16,11 +16,12 @@
 
 use crate::sync::HeadersSyncParams;
 use crate::sync_types::{HeaderId, HeaderStatus, HeadersSyncPipeline, MaybeConnectionError, QueuedHeader};
+
+use async_trait::async_trait;
 use futures::{future::FutureExt, stream::StreamExt};
 use num_traits::{Saturating, Zero};
 use std::{
 	collections::HashSet,
-	future::Future,
 	time::{Duration, Instant},
 };
 
@@ -48,69 +49,70 @@ pub type OwnedSourceFutureOutput<Client, P, T> = (Client, Result<T, <Client as S
 pub type OwnedTargetFutureOutput<Client, P, T> = (Client, Result<T, <Client as TargetClient<P>>::Error>);
 
 /// Source client trait.
+#[async_trait]
 pub trait SourceClient<P: HeadersSyncPipeline>: Sized {
 	/// Type of error this clients returns.
 	type Error: std::fmt::Debug + MaybeConnectionError;
-	/// Future that returns best block number.
-	type BestBlockNumberFuture: Future<Output = OwnedSourceFutureOutput<Self, P, P::Number>>;
-	/// Future that returns header by hash.
-	type HeaderByHashFuture: Future<Output = OwnedSourceFutureOutput<Self, P, P::Header>>;
-	/// Future that returns header by number.
-	type HeaderByNumberFuture: Future<Output = OwnedSourceFutureOutput<Self, P, P::Header>>;
-	/// Future that returns extra data associated with header.
-	type HeaderExtraFuture: Future<Output = OwnedSourceFutureOutput<Self, P, (HeaderId<P::Hash, P::Number>, P::Extra)>>;
-	/// Future that returns data required to 'complete' header.
-	type HeaderCompletionFuture: Future<
-		Output = OwnedSourceFutureOutput<Self, P, (HeaderId<P::Hash, P::Number>, Option<P::Completion>)>,
-	>;
 
 	/// Get best block number.
-	fn best_block_number(self) -> Self::BestBlockNumberFuture;
+	async fn best_block_number(self) -> OwnedSourceFutureOutput<Self, P, P::Number>;
+
 	/// Get header by hash.
-	fn header_by_hash(self, hash: P::Hash) -> Self::HeaderByHashFuture;
+	async fn header_by_hash(self, hash: P::Hash) -> OwnedSourceFutureOutput<Self, P, P::Header>;
+
 	/// Get canonical header by number.
-	fn header_by_number(self, number: P::Number) -> Self::HeaderByNumberFuture;
-	/// Get extra data by header hash.
-	fn header_extra(self, id: HeaderId<P::Hash, P::Number>, header: &P::Header) -> Self::HeaderExtraFuture;
+	async fn header_by_number(self, number: P::Number) -> OwnedSourceFutureOutput<Self, P, P::Header>;
+
 	/// Get completion data by header hash.
-	fn header_completion(self, id: HeaderId<P::Hash, P::Number>) -> Self::HeaderCompletionFuture;
+	async fn header_completion(
+		self,
+		id: HeaderId<P::Hash, P::Number>,
+	) -> OwnedSourceFutureOutput<Self, P, (HeaderId<P::Hash, P::Number>, Option<P::Completion>)>;
+
+	/// Get extra data by header hash.
+	async fn header_extra(
+		self,
+		id: HeaderId<P::Hash, P::Number>,
+		header: QueuedHeader<P>,
+	) -> OwnedSourceFutureOutput<Self, P, (HeaderId<P::Hash, P::Number>, P::Extra)>;
 }
 
 /// Target client trait.
+#[async_trait]
 pub trait TargetClient<P: HeadersSyncPipeline>: Sized {
 	/// Type of error this clients returns.
 	type Error: std::fmt::Debug + MaybeConnectionError;
-	/// Future that returns best header id.
-	type BestHeaderIdFuture: Future<Output = OwnedTargetFutureOutput<Self, P, HeaderId<P::Hash, P::Number>>>;
-	/// Future that returns known header check result.
-	type IsKnownHeaderFuture: Future<Output = OwnedTargetFutureOutput<Self, P, (HeaderId<P::Hash, P::Number>, bool)>>;
-	/// Future that returns extra check result.
-	type RequiresExtraFuture: Future<Output = OwnedTargetFutureOutput<Self, P, (HeaderId<P::Hash, P::Number>, bool)>>;
-	/// Future that returns header submission result.
-	type SubmitHeadersFuture: Future<Output = OwnedTargetFutureOutput<Self, P, Vec<HeaderId<P::Hash, P::Number>>>>;
-	/// Future that returns incomplete headers ids.
-	type IncompleteHeadersFuture: Future<
-		Output = OwnedTargetFutureOutput<Self, P, HashSet<HeaderId<P::Hash, P::Number>>>,
-	>;
-	/// Future that returns header completion result.
-	type CompleteHeadersFuture: Future<Output = OwnedTargetFutureOutput<Self, P, HeaderId<P::Hash, P::Number>>>;
 
 	/// Returns ID of best header known to the target node.
-	fn best_header_id(self) -> Self::BestHeaderIdFuture;
+	async fn best_header_id(self) -> OwnedTargetFutureOutput<Self, P, HeaderId<P::Hash, P::Number>>;
+
 	/// Returns true if header is known to the target node.
-	fn is_known_header(self, id: HeaderId<P::Hash, P::Number>) -> Self::IsKnownHeaderFuture;
-	/// Returns true if header requires extra data to be submitted.
-	fn requires_extra(self, header: &QueuedHeader<P>) -> Self::RequiresExtraFuture;
+	async fn is_known_header(
+		self,
+		id: HeaderId<P::Hash, P::Number>,
+	) -> OwnedTargetFutureOutput<Self, P, (HeaderId<P::Hash, P::Number>, bool)>;
+
 	/// Submit headers.
-	fn submit_headers(self, headers: Vec<QueuedHeader<P>>) -> Self::SubmitHeadersFuture;
+	async fn submit_headers(
+		self,
+		headers: Vec<QueuedHeader<P>>,
+	) -> OwnedTargetFutureOutput<Self, P, Vec<HeaderId<P::Hash, P::Number>>>;
+
 	/// Returns ID of headers that require to be 'completed' before children can be submitted.
-	fn incomplete_headers_ids(self) -> Self::IncompleteHeadersFuture;
+	async fn incomplete_headers_ids(self) -> OwnedTargetFutureOutput<Self, P, HashSet<HeaderId<P::Hash, P::Number>>>;
+
 	/// Submit completion data for header.
-	fn complete_header(
+	async fn complete_header(
 		self,
 		id: HeaderId<P::Hash, P::Number>,
 		completion: P::Completion,
-	) -> Self::CompleteHeadersFuture;
+	) -> OwnedTargetFutureOutput<Self, P, HeaderId<P::Hash, P::Number>>;
+
+	/// Returns true if header requires extra data to be submitted.
+	async fn requires_extra(
+		self,
+		header: QueuedHeader<P>,
+	) -> OwnedTargetFutureOutput<Self, P, (HeaderId<P::Hash, P::Number>, bool)>;
 }
 
 /// Run headers synchronization.
@@ -385,7 +387,7 @@ pub fn run<P: HeadersSyncPipeline>(
 						header.id(),
 					);
 
-					target_extra_check_future.set(target_client.requires_extra(header).fuse());
+					target_extra_check_future.set(target_client.requires_extra(header.clone()).fuse());
 				} else if let Some(header) = sync.headers().header(HeaderStatus::MaybeOrphan) {
 					// for MaybeOrphan we actually ask for parent' header existence
 					let parent_id = header.parent_id();
@@ -452,7 +454,7 @@ pub fn run<P: HeadersSyncPipeline>(
 						"Retrieving extra data for header: {:?}",
 						id,
 					);
-					source_extra_future.set(source_client.header_extra(id, header.header()).fuse());
+					source_extra_future.set(source_client.header_extra(id, header.clone()).fuse());
 				} else if let Some(header) = sync.headers().header(HeaderStatus::Orphan) {
 					// for Orphan we actually ask for parent' header
 					let parent_id = header.parent_id();

bridges/relays/ethereum/src/sync_types.rs

@@ -14,6 +14,8 @@
 // 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 std::{ops::Deref, sync::Arc};
+
 /// Ethereum header Id.
 #[derive(Debug, Clone, Copy, Eq, Hash, PartialEq)]
 pub struct HeaderId<Hash, Number>(pub Number, pub Hash);
@@ -70,7 +72,7 @@ pub trait HeadersSyncPipeline: Clone + Copy {
 		+ num_traits::Zero
 		+ num_traits::One;
 	/// Type of header that we're syncing.
-	type Header: Clone + std::fmt::Debug + SourceHeader<Self::Hash, Self::Number>;
+	type Header: Clone + std::fmt::Debug + PartialEq + SourceHeader<Self::Hash, Self::Number>;
 	/// Type of extra data for the header that we're receiving from the source node:
 	/// 1) extra data is required for some headers;
 	/// 2) target node may answer if it'll require extra data before header is submitted;
@@ -78,7 +80,7 @@ pub trait HeadersSyncPipeline: Clone + Copy {
 	/// 4) header and extra data are submitted in single transaction.
 	///
 	/// Example: Ethereum transactions receipts.
-	type Extra: Clone + std::fmt::Debug;
+	type Extra: Clone + PartialEq + std::fmt::Debug;
 	/// Type of data required to 'complete' header that we're receiving from the source node:
 	/// 1) completion data is required for some headers;
 	/// 2) target node can't answer if it'll require completion data before header is accepted;
@@ -100,20 +102,45 @@ pub trait SourceHeader<Hash, Number> {
 	fn parent_id(&self) -> HeaderId<Hash, Number>;
 }
 
+/// Header how it's stored in the synchronization queue.
+#[derive(Clone, Debug)]
+#[cfg_attr(test, derive(PartialEq))]
+pub struct QueuedHeader<P: HeadersSyncPipeline>(Arc<QueuedHeaderData<P>>);
+
+impl<P: HeadersSyncPipeline> QueuedHeader<P> {
+	/// Creates new queued header.
+	pub fn new(header: P::Header) -> Self {
+		QueuedHeader(Arc::new(QueuedHeaderData { header, extra: None }))
+	}
+
+	/// Set associated extra data.
+	pub fn set_extra(self, extra: P::Extra) -> Self {
+		QueuedHeader(Arc::new(QueuedHeaderData {
+			header: Arc::try_unwrap(self.0)
+				.map(|data| data.header)
+				.unwrap_or_else(|data| data.header.clone()),
+			extra: Some(extra),
+		}))
+	}
+}
+
+impl<P: HeadersSyncPipeline> Deref for QueuedHeader<P> {
+	type Target = QueuedHeaderData<P>;
+
+	fn deref(&self) -> &Self::Target {
+		&self.0
+	}
+}
+
 /// Header how it's stored in the synchronization queue.
 #[derive(Clone, Debug, Default)]
 #[cfg_attr(test, derive(PartialEq))]
-pub struct QueuedHeader<P: HeadersSyncPipeline> {
+pub struct QueuedHeaderData<P: HeadersSyncPipeline> {
 	header: P::Header,
 	extra: Option<P::Extra>,
 }
 
 impl<P: HeadersSyncPipeline> QueuedHeader<P> {
-	/// Creates new queued header.
-	pub fn new(header: P::Header) -> Self {
-		QueuedHeader { header, extra: None }
-	}
-
 	/// Returns ID of header.
 	pub fn id(&self) -> HeaderId<P::Hash, P::Number> {
 		self.header.id()
@@ -133,15 +160,4 @@ impl<P: HeadersSyncPipeline> QueuedHeader<P> {
 	pub fn extra(&self) -> &Option<P::Extra> {
 		&self.extra
 	}
-
-	/// Extract header and extra from self.
-	pub fn extract(self) -> (P::Header, Option<P::Extra>) {
-		(self.header, self.extra)
-	}
-
-	/// Set associated extra data.
-	pub fn set_extra(mut self, extra: P::Extra) -> Self {
-		self.extra = Some(extra);
-		self
-	}
 }