Rialto -> Millau headers relay (#477)

* Rialto -> Millau headers relay

* removed more constraints

* removed file from other PR

* Update primitives/rialto/src/lib.rs

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

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

bridges/relays/ethereum/src/ethereum_exchange.rs

@@ -36,15 +36,15 @@ use relay_ethereum_client::{
 	Client as EthereumClient, ConnectionParams as EthereumConnectionParams,
 };
 use relay_rialto_client::{Rialto, SigningParams as RialtoSigningParams};
-use relay_substrate_client::{Client as SubstrateClient, ConnectionParams as SubstrateConnectionParams};
+use relay_substrate_client::{
+	Chain as SubstrateChain, Client as SubstrateClient, ConnectionParams as SubstrateConnectionParams,
+};
 use relay_utils::{metrics::MetricsParams, HeaderId};
 use rialto_runtime::exchange::EthereumTransactionInclusionProof;
 use std::{sync::Arc, 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);
 
 /// Exchange relay mode.
 #[derive(Debug)]
@@ -210,7 +210,7 @@ impl TargetClient<EthereumToSubstrateExchange> for SubstrateTransactionsTarget {
 	type Error = RpcError;
 
 	async fn tick(&self) {
-		async_std::task::sleep(SUBSTRATE_TICK_INTERVAL).await;
+		async_std::task::sleep(Rialto::AVERAGE_BLOCK_INTERVAL).await;
 	}
 
 	async fn is_header_known(&self, id: &EthereumHeaderId) -> Result<bool, Self::Error> {

bridges/relays/ethereum/src/ethereum_sync_loop.rs

@@ -34,7 +34,9 @@ use relay_ethereum_client::{
 	Client as EthereumClient, ConnectionParams as EthereumConnectionParams,
 };
 use relay_rialto_client::{Rialto, SigningParams as RialtoSigningParams};
-use relay_substrate_client::{Client as SubstrateClient, ConnectionParams as SubstrateConnectionParams};
+use relay_substrate_client::{
+	Chain as SubstrateChain, Client as SubstrateClient, ConnectionParams as SubstrateConnectionParams,
+};
 use relay_utils::metrics::MetricsParams;
 
 use std::fmt::Debug;
@@ -45,8 +47,6 @@ pub mod consts {
 
 	/// Interval at which we check new Ethereum headers when we are synced/almost synced.
 	pub const ETHEREUM_TICK_INTERVAL: Duration = Duration::from_secs(10);
-	/// Interval at which we check new Substrate blocks.
-	pub const SUBSTRATE_TICK_INTERVAL: Duration = Duration::from_secs(5);
 	/// Max number of headers in single submit transaction.
 	pub const MAX_HEADERS_IN_SINGLE_SUBMIT: usize = 32;
 	/// Max total size of headers in single submit transaction. This only affects signed
@@ -253,7 +253,7 @@ pub fn run(params: EthereumSyncParams) -> Result<(), RpcError> {
 		source,
 		consts::ETHEREUM_TICK_INTERVAL,
 		target,
-		consts::SUBSTRATE_TICK_INTERVAL,
+		Rialto::AVERAGE_BLOCK_INTERVAL,
 		(),
 		sync_params,
 		metrics_params,

bridges/relays/ethereum/src/substrate_sync_loop.rs

@@ -32,7 +32,8 @@ use relay_ethereum_client::{
 };
 use relay_rialto_client::{HeaderId as RialtoHeaderId, Rialto, SyncHeader as RialtoSyncHeader};
 use relay_substrate_client::{
-	headers_source::HeadersSource, Client as SubstrateClient, ConnectionParams as SubstrateConnectionParams,
+	headers_source::HeadersSource, Chain as SubstrateChain, Client as SubstrateClient,
+	ConnectionParams as SubstrateConnectionParams,
 };
 use relay_utils::metrics::MetricsParams;
 use sp_runtime::Justification;
@@ -43,8 +44,6 @@ use std::{collections::HashSet, time::Duration};
 pub mod consts {
 	use super::*;
 
-	/// Interval at which we check new Substrate headers when we are synced/almost synced.
-	pub const SUBSTRATE_TICK_INTERVAL: Duration = Duration::from_secs(10);
 	/// Interval at which we check new Ethereum blocks.
 	pub const ETHEREUM_TICK_INTERVAL: Duration = Duration::from_secs(5);
 	/// Max Ethereum headers we want to have in all 'before-submitted' states.
@@ -174,7 +173,7 @@ pub fn run(params: SubstrateSyncParams) -> Result<(), RpcError> {
 
 	headers_relay::sync_loop::run(
 		source,
-		consts::SUBSTRATE_TICK_INTERVAL,
+		Rialto::AVERAGE_BLOCK_INTERVAL,
 		target,
 		consts::ETHEREUM_TICK_INTERVAL,
 		(),

bridges/relays/headers-relay/src/sync_types.rs

@@ -68,7 +68,7 @@ pub trait HeadersSyncPipeline: Clone + Send + Sync {
 		+ num_traits::One
 		+ Into<u64>;
 	/// Type of header that we're syncing.
-	type Header: Clone + std::fmt::Debug + PartialEq + SourceHeader<Self::Hash, Self::Number> + Send + Sync;
+	type Header: 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;
@@ -94,7 +94,7 @@ pub trait HeadersSyncPipeline: Clone + Send + Sync {
 pub type HeaderIdOf<P> = HeaderId<<P as HeadersSyncPipeline>::Hash, <P as HeadersSyncPipeline>::Number>;
 
 /// Header that we're receiving from source node.
-pub trait SourceHeader<Hash, Number> {
+pub trait SourceHeader<Hash, Number>: Clone + std::fmt::Debug + PartialEq + Send + Sync {
 	/// Returns ID of header.
 	fn id(&self) -> HeaderId<Hash, Number>;
 	/// Returns ID of parent header.

bridges/relays/millau-client/src/lib.rs

@@ -26,6 +26,8 @@ use sp_runtime::{
 };
 use std::time::Duration;
 
+pub use millau_runtime::BridgeRialtoCall;
+
 /// Millau header id.
 pub type HeaderId = relay_utils::HeaderId<millau_runtime::Hash, millau_runtime::BlockNumber>;
 

bridges/relays/rialto-client/src/lib.rs

@@ -151,6 +151,12 @@ impl From<rialto_runtime::Header> for SyncHeader {
 	}
 }
 
+impl From<SyncHeader> for rialto_runtime::Header {
+	fn from(header: SyncHeader) -> Self {
+		header.0
+	}
+}
+
 impl SourceHeader<rialto_runtime::Hash, rialto_runtime::BlockNumber> for SyncHeader {
 	fn id(&self) -> HeaderId {
 		relay_utils::HeaderId(*self.number(), self.hash())

bridges/relays/substrate-client/src/chain.rs

@@ -34,14 +34,23 @@ pub trait Chain: ChainBase {
 	const NAME: &'static str;
 	/// Average block interval.
 	///
-	/// How often blocks are produced on that chain. It's suggested to set this value to match the block time of the chain.
+	/// How often blocks are produced on that chain. It's suggested to set this value
+	/// to match the block time of the chain.
 	const AVERAGE_BLOCK_INTERVAL: Duration;
 
 	/// The user account identifier type for the runtime.
 	type AccountId: Parameter + Member + MaybeSerializeDeserialize + Debug + MaybeDisplay + Ord + Default;
 	/// Account index (aka nonce) type. This stores the number of previous transactions associated
 	/// with a sender account.
-	type Index: Parameter + Member + MaybeSerialize + Debug + Default + MaybeDisplay + AtLeast32Bit + Copy;
+	type Index: Parameter
+		+ Member
+		+ MaybeSerialize
+		+ Debug
+		+ Default
+		+ MaybeDisplay
+		+ DeserializeOwned
+		+ AtLeast32Bit
+		+ Copy;
 	/// Block type.
 	type SignedBlock: Member + Serialize + DeserializeOwned + BlockWithJustification;
 	/// The aggregated `Call` type.

bridges/relays/substrate-client/src/client.rs

@@ -138,10 +138,7 @@ impl<C: Chain> Client<C> {
 	/// Get the nonce of the given Substrate account.
 	///
 	/// Note: It's the caller's responsibility to make sure `account` is a valid ss58 address.
-	pub async fn next_account_index(&self, account: C::AccountId) -> Result<C::Index>
-	where
-		C::Index: DeserializeOwned,
-	{
+	pub async fn next_account_index(&self, account: C::AccountId) -> Result<C::Index> {
 		Ok(Substrate::<C, _, _>::system_account_next_index(&self.client, account).await?)
 	}
 

bridges/relays/substrate-client/src/headers_source.rs

@@ -25,7 +25,6 @@ use headers_relay::{
 	sync_loop::SourceClient,
 	sync_types::{HeaderIdOf, HeadersSyncPipeline, QueuedHeader, SourceHeader},
 };
-use jsonrpsee::common::DeserializeOwned;
 use num_traits::Saturating;
 use sp_runtime::{traits::Header as HeaderT, Justification};
 use std::marker::PhantomData;
@@ -51,8 +50,7 @@ impl<C, P> SourceClient<P> for HeadersSource<C, P>
 where
 	C: Chain,
 	C::BlockNumber: Into<u64> + Saturating,
-	C::Header: DeserializeOwned + Into<P::Header>,
-	C::Index: DeserializeOwned,
+	C::Header: Into<P::Header>,
 	P: HeadersSyncPipeline<Extra = (), Completion = Justification, Hash = C::Hash, Number = C::BlockNumber>,
 	P::Header: SourceHeader<C::Hash, C::BlockNumber>,
 {

bridges/relays/substrate/Cargo.toml

@@ -12,6 +12,7 @@ codec = { package = "parity-scale-codec", version = "1.3.4" }
 futures = "0.3.7"
 hex = "0.4"
 log = "0.4.11"
+num-traits = "0.2"
 paste = "1.0"
 structopt = "0.3"
 

bridges/relays/substrate/src/cli.rs

@@ -39,6 +39,17 @@ pub enum Command {
 		#[structopt(flatten)]
 		prometheus_params: PrometheusParams,
 	},
+	/// Relay Rialto headers to Millau.
+	RialtoHeadersToMillau {
+		#[structopt(flatten)]
+		rialto: RialtoConnectionParams,
+		#[structopt(flatten)]
+		millau: MillauConnectionParams,
+		#[structopt(flatten)]
+		millau_sign: MillauSigningParams,
+		#[structopt(flatten)]
+		prometheus_params: PrometheusParams,
+	},
 	/// Submit message to given Rialto -> Millau lane.
 	SubmitMillauToRialtoMessage {
 		#[structopt(flatten)]

bridges/relays/substrate/src/headers_maintain.rs

@@ -43,7 +43,7 @@ use headers_relay::{
 use relay_substrate_client::{Chain, Client, Error as SubstrateError, JustificationsSubscription};
 use relay_utils::HeaderId;
 use sp_core::Bytes;
-use sp_runtime::{traits::Header as HeaderT, DeserializeOwned, Justification};
+use sp_runtime::{traits::Header as HeaderT, Justification};
 use std::{collections::VecDeque, task::Poll};
 
 /// Substrate-to-Substrate headers synchronization maintain procedure.
@@ -80,8 +80,6 @@ impl<P: SubstrateHeadersSyncPipeline, C: Chain> SubstrateHeadersToSubstrateMaint
 impl<P, C> SyncMaintain<P> for SubstrateHeadersToSubstrateMaintain<P, C>
 where
 	C: Chain,
-	C::Header: DeserializeOwned,
-	C::Index: DeserializeOwned,
 	P::Number: Decode + From<C::BlockNumber>,
 	P::Hash: Decode + From<C::Hash>,
 	P: SubstrateHeadersSyncPipeline<Completion = Justification, Extra = ()>,
@@ -271,8 +269,6 @@ where
 	P::Number: Decode + From<C::BlockNumber>,
 	P::Hash: Decode + From<C::Hash>,
 	C: Chain,
-	C::Header: DeserializeOwned,
-	C::Index: DeserializeOwned,
 {
 	let call = P::FINALIZED_BLOCK_METHOD.into();
 	let data = Bytes(Vec::new());
@@ -288,7 +284,8 @@ where
 #[cfg(test)]
 mod tests {
 	use super::*;
-	use crate::millau_headers_to_rialto::{sync_params, MillauHeadersToRialto};
+	use crate::headers_pipeline::sync_params;
+	use crate::millau_headers_to_rialto::MillauHeadersToRialto;
 
 	fn parent_hash(index: u8) -> bp_millau::Hash {
 		if index == 1 {

bridges/relays/substrate/src/headers_pipeline.rs

@@ -0,0 +1,140 @@
+// 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/>.
+
+//! Substrate-to-Substrate headers sync entrypoint.
+
+use crate::{
+	headers_maintain::SubstrateHeadersToSubstrateMaintain,
+	headers_target::{SubstrateHeadersSyncPipeline, SubstrateHeadersTarget},
+};
+
+use codec::Encode;
+use headers_relay::{
+	sync::{HeadersSyncParams, TargetTransactionMode},
+	sync_types::{HeadersSyncPipeline, QueuedHeader, SourceHeader},
+};
+use num_traits::Saturating;
+use relay_substrate_client::{headers_source::HeadersSource, BlockNumberOf, Chain, Client, HashOf};
+use sp_runtime::Justification;
+use std::marker::PhantomData;
+
+/// Substrate-to-Substrate headers pipeline.
+#[derive(Debug, Clone)]
+pub struct SubstrateHeadersToSubstrate<SourceChain, SourceSyncHeader, TargetChain: Chain, TargetSign> {
+	/// Client for the target chain.
+	pub(crate) target_client: Client<TargetChain>,
+	/// Data required to sign target chain transactions.
+	pub(crate) target_sign: TargetSign,
+	/// Unused generic arguments dump.
+	_marker: PhantomData<(SourceChain, SourceSyncHeader)>,
+}
+
+impl<SourceChain, SourceSyncHeader, TargetChain: Chain, TargetSign>
+	SubstrateHeadersToSubstrate<SourceChain, SourceSyncHeader, TargetChain, TargetSign>
+{
+	/// Create new Substrate-to-Substrate headers pipeline.
+	pub fn new(target_client: Client<TargetChain>, target_sign: TargetSign) -> Self {
+		SubstrateHeadersToSubstrate {
+			target_client,
+			target_sign,
+			_marker: Default::default(),
+		}
+	}
+}
+
+impl<SourceChain, SourceSyncHeader, TargetChain, TargetSign> HeadersSyncPipeline
+	for SubstrateHeadersToSubstrate<SourceChain, SourceSyncHeader, TargetChain, TargetSign>
+where
+	SourceChain: Clone + Chain,
+	BlockNumberOf<SourceChain>: Saturating + Into<u64>,
+	SourceSyncHeader:
+		SourceHeader<HashOf<SourceChain>, BlockNumberOf<SourceChain>> + std::ops::Deref<Target = SourceChain::Header>,
+	TargetChain: Clone + Chain,
+	TargetSign: Clone + Send + Sync,
+{
+	const SOURCE_NAME: &'static str = SourceChain::NAME;
+	const TARGET_NAME: &'static str = TargetChain::NAME;
+
+	type Hash = HashOf<SourceChain>;
+	type Number = BlockNumberOf<SourceChain>;
+	type Header = SourceSyncHeader;
+	type Extra = ();
+	type Completion = Justification;
+
+	fn estimate_size(source: &QueuedHeader<Self>) -> usize {
+		source.header().encode().len()
+	}
+}
+
+/// Return sync parameters for Substrate-to-Substrate headers sync.
+pub fn sync_params() -> HeadersSyncParams {
+	HeadersSyncParams {
+		max_future_headers_to_download: 32,
+		max_headers_in_submitted_status: 8,
+		max_headers_in_single_submit: 1,
+		max_headers_size_in_single_submit: 1024 * 1024,
+		prune_depth: 256,
+		target_tx_mode: TargetTransactionMode::Signed,
+	}
+}
+
+/// Run Substrate-to-Substrate headers sync.
+pub async fn run<SourceChain, TargetChain, P>(
+	pipeline: P,
+	source_client: Client<SourceChain>,
+	target_client: Client<TargetChain>,
+	metrics_params: Option<relay_utils::metrics::MetricsParams>,
+) where
+	P: SubstrateHeadersSyncPipeline<
+		Hash = HashOf<SourceChain>,
+		Number = BlockNumberOf<SourceChain>,
+		Completion = Justification,
+		Extra = (),
+	>,
+	P::Header: SourceHeader<HashOf<SourceChain>, BlockNumberOf<SourceChain>>,
+	SourceChain: Clone + Chain,
+	SourceChain::Header: Into<P::Header>,
+	BlockNumberOf<SourceChain>: Into<u64> + Saturating,
+	TargetChain: Clone + Chain,
+{
+	let source_justifications = match source_client.clone().subscribe_justifications().await {
+		Ok(source_justifications) => source_justifications,
+		Err(error) => {
+			log::warn!(
+				target: "bridge",
+				"Failed to subscribe to {} justifications: {:?}",
+				SourceChain::NAME,
+				error,
+			);
+
+			return;
+		}
+	};
+
+	let sync_maintain =
+		SubstrateHeadersToSubstrateMaintain::new(pipeline.clone(), source_client.clone(), source_justifications);
+
+	headers_relay::sync_loop::run(
+		HeadersSource::new(source_client),
+		SourceChain::AVERAGE_BLOCK_INTERVAL,
+		SubstrateHeadersTarget::new(target_client, pipeline),
+		TargetChain::AVERAGE_BLOCK_INTERVAL,
+		sync_maintain,
+		sync_params(),
+		metrics_params,
+		futures::future::pending(),
+	);
+}

bridges/relays/substrate/src/headers_target.rs

@@ -28,7 +28,7 @@ use headers_relay::{
 use relay_substrate_client::{Chain, Client, Error as SubstrateError};
 use relay_utils::HeaderId;
 use sp_core::Bytes;
-use sp_runtime::{DeserializeOwned, Justification};
+use sp_runtime::Justification;
 use std::collections::HashSet;
 
 /// Headers sync pipeline for Substrate <-> Substrate relays.
@@ -77,8 +77,6 @@ impl<C: Chain, P> SubstrateHeadersTarget<C, P> {
 impl<C, P> TargetClient<P> for SubstrateHeadersTarget<C, P>
 where
 	C: Chain,
-	C::Header: DeserializeOwned,
-	C::Index: DeserializeOwned,
 	P::Number: Decode,
 	P::Hash: Decode + Encode,
 	P: SubstrateHeadersSyncPipeline<Completion = Justification, Extra = ()>,

bridges/relays/substrate/src/main.rs

@@ -34,8 +34,10 @@ pub type RialtoClient = relay_substrate_client::Client<Rialto>;
 
 mod cli;
 mod headers_maintain;
+mod headers_pipeline;
 mod headers_target;
 mod millau_headers_to_rialto;
+mod rialto_headers_to_millau;
 
 fn main() {
 	initialize_relay();
@@ -71,6 +73,29 @@ async fn run_command(command: cli::Command) -> Result<(), String> {
 			.map_err(|e| format!("Failed to parse rialto-signer: {:?}", e))?;
 			millau_headers_to_rialto::run(millau_client, rialto_client, rialto_sign, prometheus_params.into()).await;
 		}
+		cli::Command::RialtoHeadersToMillau {
+			rialto,
+			millau,
+			millau_sign,
+			prometheus_params,
+		} => {
+			let rialto_client = RialtoClient::new(ConnectionParams {
+				host: rialto.rialto_host,
+				port: rialto.rialto_port,
+			})
+			.await?;
+			let millau_client = MillauClient::new(ConnectionParams {
+				host: millau.millau_host,
+				port: millau.millau_port,
+			})
+			.await?;
+			let millau_sign = MillauSigningParams::from_suri(
+				&millau_sign.millau_signer,
+				millau_sign.millau_signer_password.as_deref(),
+			)
+			.map_err(|e| format!("Failed to parse millau-signer: {:?}", e))?;
+			rialto_headers_to_millau::run(rialto_client, millau_client, millau_sign, prometheus_params.into()).await;
+		}
 		cli::Command::SubmitMillauToRialtoMessage {
 			millau,
 			millau_sign,

bridges/relays/substrate/src/millau_headers_to_rialto.rs

@@ -17,9 +17,8 @@
 //! Millau-to-Rialto headers sync entrypoint.
 
 use crate::{
-	headers_maintain::SubstrateHeadersToSubstrateMaintain,
-	headers_target::{SubstrateHeadersSyncPipeline, SubstrateHeadersTarget},
-	MillauClient, RialtoClient,
+	headers_pipeline::SubstrateHeadersToSubstrate, headers_target::SubstrateHeadersSyncPipeline, MillauClient,
+	RialtoClient,
 };
 
 use async_trait::async_trait;
@@ -27,41 +26,18 @@ use bp_millau::{
 	BEST_MILLAU_BLOCKS_METHOD, FINALIZED_MILLAU_BLOCK_METHOD, INCOMPLETE_MILLAU_HEADERS_METHOD,
 	IS_KNOWN_MILLAU_BLOCK_METHOD,
 };
-use codec::Encode;
-use headers_relay::{
-	sync::{HeadersSyncParams, TargetTransactionMode},
-	sync_types::{HeadersSyncPipeline, QueuedHeader},
-};
+use headers_relay::sync_types::QueuedHeader;
 use relay_millau_client::{HeaderId as MillauHeaderId, Millau, SyncHeader as MillauSyncHeader};
 use relay_rialto_client::{BridgeMillauCall, Rialto, SigningParams as RialtoSigningParams};
-use relay_substrate_client::{
-	headers_source::HeadersSource, BlockNumberOf, Error as SubstrateError, HashOf, TransactionSignScheme,
-};
+use relay_substrate_client::{Error as SubstrateError, TransactionSignScheme};
 use sp_core::Pair;
 use sp_runtime::Justification;
-use std::time::Duration;
 
-/// Millau-to-Rialto headers pipeline.
-#[derive(Debug, Clone)]
-pub struct MillauHeadersToRialto {
-	client: RialtoClient,
-	sign: RialtoSigningParams,
-}
-
-impl HeadersSyncPipeline for MillauHeadersToRialto {
-	const SOURCE_NAME: &'static str = "Millau";
-	const TARGET_NAME: &'static str = "Rialto";
-
-	type Hash = HashOf<Millau>;
-	type Number = BlockNumberOf<Millau>;
-	type Header = MillauSyncHeader;
-	type Extra = ();
-	type Completion = Justification;
-
-	fn estimate_size(source: &QueuedHeader<Self>) -> usize {
-		source.header().encode().len()
-	}
-}
+/// Millau-to-Rialto headers sync pipeline.
+pub(crate) type MillauHeadersToRialto =
+	SubstrateHeadersToSubstrate<Millau, MillauSyncHeader, Rialto, RialtoSigningParams>;
+/// Millau header in-the-queue.
+type QueuedMillauHeader = QueuedHeader<MillauHeadersToRialto>;
 
 #[async_trait]
 impl SubstrateHeadersSyncPipeline for MillauHeadersToRialto {
@@ -76,10 +52,10 @@ impl SubstrateHeadersSyncPipeline for MillauHeadersToRialto {
 		&self,
 		header: QueuedMillauHeader,
 	) -> Result<Self::SignedTransaction, SubstrateError> {
-		let account_id = self.sign.signer.public().as_array_ref().clone().into();
-		let nonce = self.client.next_account_index(account_id).await?;
+		let account_id = self.target_sign.signer.public().as_array_ref().clone().into();
+		let nonce = self.target_client.next_account_index(account_id).await?;
 		let call = BridgeMillauCall::import_signed_header(header.header().clone().into()).into();
-		let transaction = Rialto::sign_transaction(&self.client, &self.sign.signer, nonce, call);
+		let transaction = Rialto::sign_transaction(&self.target_client, &self.target_sign.signer, nonce, call);
 		Ok(transaction)
 	}
 
@@ -88,35 +64,14 @@ impl SubstrateHeadersSyncPipeline for MillauHeadersToRialto {
 		id: MillauHeaderId,
 		completion: Justification,
 	) -> Result<Self::SignedTransaction, SubstrateError> {
-		let account_id = self.sign.signer.public().as_array_ref().clone().into();
-		let nonce = self.client.next_account_index(account_id).await?;
+		let account_id = self.target_sign.signer.public().as_array_ref().clone().into();
+		let nonce = self.target_client.next_account_index(account_id).await?;
 		let call = BridgeMillauCall::finalize_header(id.1, completion).into();
-		let transaction = Rialto::sign_transaction(&self.client, &self.sign.signer, nonce, call);
+		let transaction = Rialto::sign_transaction(&self.target_client, &self.target_sign.signer, nonce, call);
 		Ok(transaction)
 	}
 }
 
-/// Millau header in-the-queue.
-type QueuedMillauHeader = QueuedHeader<MillauHeadersToRialto>;
-
-/// Millau node as headers source.
-type MillauSourceClient = HeadersSource<Millau, MillauHeadersToRialto>;
-
-/// Rialto node as headers target.
-type RialtoTargetClient = SubstrateHeadersTarget<Rialto, MillauHeadersToRialto>;
-
-/// Return sync parameters for Millau-to-Rialto headers sync.
-pub fn sync_params() -> HeadersSyncParams {
-	HeadersSyncParams {
-		max_future_headers_to_download: 32,
-		max_headers_in_submitted_status: 8,
-		max_headers_in_single_submit: 1,
-		max_headers_size_in_single_submit: 1024 * 1024,
-		prune_depth: 256,
-		target_tx_mode: TargetTransactionMode::Signed,
-	}
-}
-
 /// Run Millau-to-Rialto headers sync.
 pub async fn run(
 	millau_client: MillauClient,
@@ -124,37 +79,11 @@ pub async fn run(
 	rialto_sign: RialtoSigningParams,
 	metrics_params: Option<relay_utils::metrics::MetricsParams>,
 ) {
-	let millau_tick = Duration::from_secs(5);
-	let rialto_tick = Duration::from_secs(5);
-
-	let millau_justifications = match millau_client.clone().subscribe_justifications().await {
-		Ok(millau_justifications) => millau_justifications,
-		Err(error) => {
-			log::warn!(
-				target: "bridge",
-				"Failed to subscribe to Millau justifications: {:?}",
-				error,
-			);
-
-			return;
-		}
-	};
-
-	let pipeline = MillauHeadersToRialto {
-		client: rialto_client.clone(),
-		sign: rialto_sign,
-	};
-	let sync_maintain =
-		SubstrateHeadersToSubstrateMaintain::new(pipeline.clone(), rialto_client.clone(), millau_justifications);
-
-	headers_relay::sync_loop::run(
-		MillauSourceClient::new(millau_client),
-		millau_tick,
-		RialtoTargetClient::new(rialto_client, pipeline),
-		rialto_tick,
-		sync_maintain,
-		sync_params(),
+	crate::headers_pipeline::run(
+		MillauHeadersToRialto::new(rialto_client.clone(), rialto_sign),
+		millau_client,
+		rialto_client,
 		metrics_params,
-		futures::future::pending(),
-	);
+	)
+	.await;
 }

bridges/relays/substrate/src/rialto_headers_to_millau.rs

@@ -0,0 +1,88 @@
+// 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/>.
+
+//! Rialto-to-Millau headers sync entrypoint.
+
+use crate::{
+	headers_pipeline::SubstrateHeadersToSubstrate, headers_target::SubstrateHeadersSyncPipeline, MillauClient,
+	RialtoClient,
+};
+
+use async_trait::async_trait;
+use bp_rialto::{
+	BEST_RIALTO_BLOCKS_METHOD, FINALIZED_RIALTO_BLOCK_METHOD, INCOMPLETE_RIALTO_HEADERS_METHOD,
+	IS_KNOWN_RIALTO_BLOCK_METHOD,
+};
+use headers_relay::sync_types::QueuedHeader;
+use relay_millau_client::{BridgeRialtoCall, Millau, SigningParams as MillauSigningParams};
+use relay_rialto_client::{HeaderId as RialtoHeaderId, Rialto, SyncHeader as RialtoSyncHeader};
+use relay_substrate_client::{Error as SubstrateError, TransactionSignScheme};
+use sp_core::Pair;
+use sp_runtime::Justification;
+
+/// Rialto-to-Millau headers sync pipeline.
+type RialtoHeadersToMillau = SubstrateHeadersToSubstrate<Rialto, RialtoSyncHeader, Millau, MillauSigningParams>;
+/// Rialto header in-the-queue.
+type QueuedRialtoHeader = QueuedHeader<RialtoHeadersToMillau>;
+
+#[async_trait]
+impl SubstrateHeadersSyncPipeline for RialtoHeadersToMillau {
+	const BEST_BLOCK_METHOD: &'static str = BEST_RIALTO_BLOCKS_METHOD;
+	const FINALIZED_BLOCK_METHOD: &'static str = FINALIZED_RIALTO_BLOCK_METHOD;
+	const IS_KNOWN_BLOCK_METHOD: &'static str = IS_KNOWN_RIALTO_BLOCK_METHOD;
+	const INCOMPLETE_HEADERS_METHOD: &'static str = INCOMPLETE_RIALTO_HEADERS_METHOD;
+
+	type SignedTransaction = <Millau as TransactionSignScheme>::SignedTransaction;
+
+	async fn make_submit_header_transaction(
+		&self,
+		header: QueuedRialtoHeader,
+	) -> Result<Self::SignedTransaction, SubstrateError> {
+		let account_id = self.target_sign.signer.public().as_array_ref().clone().into();
+		let nonce = self.target_client.next_account_index(account_id).await?;
+		let call = BridgeRialtoCall::import_signed_header(header.header().clone().into()).into();
+		let transaction = Millau::sign_transaction(&self.target_client, &self.target_sign.signer, nonce, call);
+		Ok(transaction)
+	}
+
+	async fn make_complete_header_transaction(
+		&self,
+		id: RialtoHeaderId,
+		completion: Justification,
+	) -> Result<Self::SignedTransaction, SubstrateError> {
+		let account_id = self.target_sign.signer.public().as_array_ref().clone().into();
+		let nonce = self.target_client.next_account_index(account_id).await?;
+		let call = BridgeRialtoCall::finalize_header(id.1, completion).into();
+		let transaction = Millau::sign_transaction(&self.target_client, &self.target_sign.signer, nonce, call);
+		Ok(transaction)
+	}
+}
+
+/// Run Rialto-to-Millau headers sync.
+pub async fn run(
+	rialto_client: RialtoClient,
+	millau_client: MillauClient,
+	millau_sign: MillauSigningParams,
+	metrics_params: Option<relay_utils::metrics::MetricsParams>,
+) {
+	crate::headers_pipeline::run(
+		RialtoHeadersToMillau::new(millau_client.clone(), millau_sign),
+		rialto_client,
+		millau_client,
+		metrics_params,
+	)
+	.await;
+}