Ability to send messages from Rialto to Millau (#525)

* relay Rialto messages to Millau

* impl SubmitRialtoToMillauMessage

* fmt

bridges/bin/millau/runtime/src/lib.rs

@@ -54,6 +54,7 @@ pub use frame_support::{
 	StorageValue,
 };
 
+pub use frame_system::Call as SystemCall;
 pub use pallet_balances::Call as BalancesCall;
 pub use pallet_message_lane::Call as MessageLaneCall;
 pub use pallet_substrate_bridge::Call as BridgeRialtoCall;

bridges/relays/substrate/src/cli.rs

@@ -108,6 +108,40 @@ pub enum Command {
 		#[structopt(long)]
 		fee: bp_millau::Balance,
 	},
+	/// Serve given lane of Rialto -> Millau messages.
+	RialtoMessagesToMillau {
+		#[structopt(flatten)]
+		rialto: RialtoConnectionParams,
+		#[structopt(flatten)]
+		rialto_sign: RialtoSigningParams,
+		#[structopt(flatten)]
+		millau: MillauConnectionParams,
+		#[structopt(flatten)]
+		millau_sign: MillauSigningParams,
+		#[structopt(flatten)]
+		prometheus_params: PrometheusParams,
+		/// Hex-encoded id of lane that should be served by relay.
+		#[structopt(long)]
+		lane: HexLaneId,
+	},
+	/// Submit message to given Rialto -> Millau lane.
+	SubmitRialtoToMillauMessage {
+		#[structopt(flatten)]
+		rialto: RialtoConnectionParams,
+		#[structopt(flatten)]
+		rialto_sign: RialtoSigningParams,
+		#[structopt(flatten)]
+		millau_sign: MillauSigningParams,
+		/// Hex-encoded lane id.
+		#[structopt(long)]
+		lane: HexLaneId,
+		/// Message type.
+		#[structopt(long, possible_values = &ToMillauMessage::variants())]
+		message: ToMillauMessage,
+		/// Delivery and dispatch fee.
+		#[structopt(long)]
+		fee: bp_rialto::Balance,
+	},
 }
 
 arg_enum! {
@@ -118,6 +152,14 @@ arg_enum! {
 	}
 }
 
+arg_enum! {
+	#[derive(Debug)]
+	/// All possible messages that may be delivered to the Millau chain.
+	pub enum ToMillauMessage {
+		Remark,
+	}
+}
+
 /// Lane id.
 #[derive(Debug)]
 pub struct HexLaneId(LaneId);

bridges/relays/substrate/src/main.rs

@@ -43,6 +43,7 @@ mod messages_target;
 mod millau_headers_to_rialto;
 mod millau_messages_to_rialto;
 mod rialto_headers_to_millau;
+mod rialto_messages_to_millau;
 
 fn main() {
 	initialize_relay();
@@ -287,7 +288,7 @@ async fn run_command(command: cli::Command) -> Result<(), String> {
 				millau_runtime::Call::BridgeRialtoMessageLane(millau_runtime::MessageLaneCall::send_message(
 					lane.into(),
 					MessagePayload {
-						spec_version: millau_runtime::VERSION.spec_version,
+						spec_version: rialto_runtime::VERSION.spec_version,
 						weight: rialto_call_weight,
 						origin: CallOrigin::RealAccount(
 							millau_sender_public,
@@ -312,6 +313,120 @@ async fn run_command(command: cli::Command) -> Result<(), String> {
 				.submit_extrinsic(Bytes(signed_millau_call.encode()))
 				.await?;
 		}
+		cli::Command::RialtoMessagesToMillau {
+			rialto,
+			rialto_sign,
+			millau,
+			millau_sign,
+			prometheus_params,
+			lane,
+		} => {
+			let rialto_client = RialtoClient::new(ConnectionParams {
+				host: rialto.rialto_host,
+				port: rialto.rialto_port,
+			})
+			.await?;
+			let rialto_sign = RialtoSigningParams::from_suri(
+				&rialto_sign.rialto_signer,
+				rialto_sign.rialto_signer_password.as_deref(),
+			)
+			.map_err(|e| format!("Failed to parse rialto-signer: {:?}", e))?;
+			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_messages_to_millau::run(
+				rialto_client,
+				rialto_sign,
+				millau_client,
+				millau_sign,
+				lane.into(),
+				prometheus_params.into(),
+			);
+		}
+		cli::Command::SubmitRialtoToMillauMessage {
+			rialto,
+			rialto_sign,
+			millau_sign,
+			lane,
+			message,
+			fee,
+		} => {
+			let rialto_client = RialtoClient::new(ConnectionParams {
+				host: rialto.rialto_host,
+				port: rialto.rialto_port,
+			})
+			.await?;
+			let rialto_sign = RialtoSigningParams::from_suri(
+				&rialto_sign.rialto_signer,
+				rialto_sign.rialto_signer_password.as_deref(),
+			)
+			.map_err(|e| format!("Failed to parse rialto-signer: {:?}", e))?;
+			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))?;
+
+			let millau_call = match message {
+				cli::ToMillauMessage::Remark => millau_runtime::Call::System(millau_runtime::SystemCall::remark(
+					format!(
+						"Unix time: {}",
+						std::time::SystemTime::now()
+							.duration_since(std::time::SystemTime::UNIX_EPOCH)
+							.unwrap_or_default()
+							.as_secs(),
+					)
+					.as_bytes()
+					.to_vec(),
+				)),
+			};
+			let millau_call_weight = millau_call.get_dispatch_info().weight;
+
+			let rialto_sender_public: bp_rialto::AccountSigner = rialto_sign.signer.public().clone().into();
+			let millau_origin_public = millau_sign.signer.public();
+
+			let mut millau_origin_signature_message = Vec::new();
+			millau_call.encode_to(&mut millau_origin_signature_message);
+			rialto_sender_public.encode_to(&mut millau_origin_signature_message);
+			let millau_origin_signature = millau_sign.signer.sign(&millau_origin_signature_message);
+
+			let rialto_call =
+				rialto_runtime::Call::BridgeMillauMessageLane(rialto_runtime::MessageLaneCall::send_message(
+					lane.into(),
+					MessagePayload {
+						spec_version: millau_runtime::VERSION.spec_version,
+						weight: millau_call_weight,
+						origin: CallOrigin::RealAccount(
+							rialto_sender_public,
+							millau_origin_public.into(),
+							millau_origin_signature.into(),
+						),
+						call: millau_call.encode(),
+					},
+					fee,
+				));
+
+			let signed_rialto_call = Rialto::sign_transaction(
+				&rialto_client,
+				&rialto_sign.signer,
+				rialto_client
+					.next_account_index(rialto_sign.signer.public().clone().into())
+					.await?,
+				rialto_call,
+			);
+
+			rialto_client
+				.submit_extrinsic(Bytes(signed_rialto_call.encode()))
+				.await?;
+		}
 	}
 
 	Ok(())

bridges/relays/substrate/src/messages_lane.rs

@@ -14,11 +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/>.
 
+use crate::messages_source::SubstrateMessagesProof;
+use crate::messages_target::SubstrateMessagesReceivingProof;
+
 use async_trait::async_trait;
 use bp_message_lane::MessageNonce;
 use codec::Encode;
 use messages_relay::message_lane::{MessageLane, SourceHeaderIdOf, TargetHeaderIdOf};
-use relay_substrate_client::Error as SubstrateError;
+use relay_substrate_client::{BlockNumberOf, Chain, Client, Error as SubstrateError, HashOf};
+use relay_utils::BlockNumberBase;
 use std::ops::RangeInclusive;
 
 /// Message sync pipeline for Substrate <-> Substrate relays.
@@ -61,3 +65,53 @@ pub trait SubstrateMessageLane: MessageLane {
 		proof: Self::MessagesReceivingProof,
 	) -> Result<Self::SourceSignedTransaction, SubstrateError>;
 }
+
+/// Substrate-to-Substrate message lane.
+#[derive(Debug)]
+pub struct SubstrateMessageLaneToSubstrate<Source: Chain, SourceSignParams, Target: Chain, TargetSignParams> {
+	/// Client for the source Substrate chain.
+	pub(crate) source_client: Client<Source>,
+	/// Parameters required to sign transactions for source chain.
+	pub(crate) source_sign: SourceSignParams,
+	/// Client for the target Substrate chain.
+	pub(crate) target_client: Client<Target>,
+	/// Parameters required to sign transactions for target chain.
+	pub(crate) target_sign: TargetSignParams,
+	/// Account id of relayer at the source chain.
+	pub(crate) relayer_id_at_source: Source::AccountId,
+}
+
+impl<Source: Chain, SourceSignParams: Clone, Target: Chain, TargetSignParams: Clone> Clone
+	for SubstrateMessageLaneToSubstrate<Source, SourceSignParams, Target, TargetSignParams>
+{
+	fn clone(&self) -> Self {
+		Self {
+			source_client: self.source_client.clone(),
+			source_sign: self.source_sign.clone(),
+			target_client: self.target_client.clone(),
+			target_sign: self.target_sign.clone(),
+			relayer_id_at_source: self.relayer_id_at_source.clone(),
+		}
+	}
+}
+
+impl<Source: Chain, SourceSignParams, Target: Chain, TargetSignParams> MessageLane
+	for SubstrateMessageLaneToSubstrate<Source, SourceSignParams, Target, TargetSignParams>
+where
+	SourceSignParams: Clone + Send + Sync + 'static,
+	TargetSignParams: Clone + Send + Sync + 'static,
+	BlockNumberOf<Source>: BlockNumberBase,
+	BlockNumberOf<Target>: BlockNumberBase,
+{
+	const SOURCE_NAME: &'static str = Source::NAME;
+	const TARGET_NAME: &'static str = Target::NAME;
+
+	type MessagesProof = SubstrateMessagesProof<Source>;
+	type MessagesReceivingProof = SubstrateMessagesReceivingProof<Target>;
+
+	type SourceHeaderNumber = BlockNumberOf<Source>;
+	type SourceHeaderHash = HashOf<Source>;
+
+	type TargetHeaderNumber = BlockNumberOf<Target>;
+	type TargetHeaderHash = HashOf<Target>;
+}

bridges/relays/substrate/src/messages_target.rs

@@ -36,6 +36,9 @@ use sp_runtime::{traits::Header as HeaderT, DeserializeOwned};
 use sp_trie::StorageProof;
 use std::ops::RangeInclusive;
 
+/// Message receiving proof returned by the target Substrate node.
+pub type SubstrateMessagesReceivingProof<C> = (HashOf<C>, StorageProof, LaneId);
+
 /// Substrate client as Substrate messages target.
 pub struct SubstrateMessagesTarget<C: Chain, P> {
 	client: Client<C>,
@@ -75,7 +78,7 @@ where
 	C::Index: DeserializeOwned,
 	<C::Header as HeaderT>::Number: BlockNumberBase,
 	P: SubstrateMessageLane<
-		MessagesReceivingProof = (HashOf<C>, StorageProof, LaneId),
+		MessagesReceivingProof = SubstrateMessagesReceivingProof<C>,
 		TargetHeaderNumber = <C::Header as HeaderT>::Number,
 		TargetHeaderHash = <C::Header as HeaderT>::Hash,
 	>,

bridges/relays/substrate/src/millau_messages_to_rialto.rs

@@ -16,7 +16,7 @@
 
 //! Millau-to-Rialto messages sync entrypoint.
 
-use crate::messages_lane::SubstrateMessageLane;
+use crate::messages_lane::{SubstrateMessageLane, SubstrateMessageLaneToSubstrate};
 use crate::messages_source::SubstrateMessagesSource;
 use crate::messages_target::SubstrateMessagesTarget;
 use crate::{MillauClient, RialtoClient};
@@ -24,50 +24,16 @@ use crate::{MillauClient, RialtoClient};
 use async_trait::async_trait;
 use bp_message_lane::{LaneId, MessageNonce};
 use bp_runtime::{MILLAU_BRIDGE_INSTANCE, RIALTO_BRIDGE_INSTANCE};
-use frame_support::weights::Weight;
 use messages_relay::message_lane::MessageLane;
 use relay_millau_client::{HeaderId as MillauHeaderId, Millau, SigningParams as MillauSigningParams};
 use relay_rialto_client::{HeaderId as RialtoHeaderId, Rialto, SigningParams as RialtoSigningParams};
-use relay_substrate_client::{BlockNumberOf, Error as SubstrateError, HashOf, TransactionSignScheme};
+use relay_substrate_client::{Chain, Error as SubstrateError, TransactionSignScheme};
 use relay_utils::metrics::MetricsParams;
 use sp_core::Pair;
-use sp_trie::StorageProof;
 use std::{ops::RangeInclusive, time::Duration};
 
-/// Millau -> Rialto messages proof:
+/// Millau-to-Rialto message lane.
-///
+type MillauMessagesToRialto = SubstrateMessageLaneToSubstrate<Millau, MillauSigningParams, Rialto, RialtoSigningParams>;
-/// - cumulative dispatch-weight of messages in the batch;
-/// - proof that we'll actually submit to the Rialto node.
-type FromMillauMessagesProof = (
-	Weight,
-	(HashOf<Millau>, StorageProof, LaneId, MessageNonce, MessageNonce),
-);
-/// Rialto -> Millau messages receiving proof.
-type FromRialtoMessagesReceivingProof = (HashOf<Rialto>, StorageProof, LaneId);
-
-/// Millau-to-Rialto messages pipeline.
-#[derive(Debug, Clone)]
-struct MillauMessagesToRialto {
-	millau_client: MillauClient,
-	millau_sign: MillauSigningParams,
-	rialto_client: RialtoClient,
-	rialto_sign: RialtoSigningParams,
-	relayer_id: bp_millau::AccountId,
-}
-
-impl MessageLane for MillauMessagesToRialto {
-	const SOURCE_NAME: &'static str = "Millau";
-	const TARGET_NAME: &'static str = "Rialto";
-
-	type MessagesProof = FromMillauMessagesProof;
-	type MessagesReceivingProof = FromRialtoMessagesReceivingProof;
-
-	type SourceHeaderNumber = BlockNumberOf<Millau>;
-	type SourceHeaderHash = HashOf<Millau>;
-
-	type TargetHeaderNumber = BlockNumberOf<Rialto>;
-	type TargetHeaderHash = HashOf<Rialto>;
-}
 
 #[async_trait]
 impl SubstrateMessageLane for MillauMessagesToRialto {
@@ -90,12 +56,12 @@ impl SubstrateMessageLane for MillauMessagesToRialto {
 	async fn make_messages_receiving_proof_transaction(
 		&self,
 		_generated_at_block: RialtoHeaderId,
-		proof: FromRialtoMessagesReceivingProof,
+		proof: <Self as MessageLane>::MessagesReceivingProof,
 	) -> Result<Self::SourceSignedTransaction, SubstrateError> {
-		let account_id = self.millau_sign.signer.public().as_array_ref().clone().into();
+		let account_id = self.source_sign.signer.public().as_array_ref().clone().into();
-		let nonce = self.millau_client.next_account_index(account_id).await?;
+		let nonce = self.source_client.next_account_index(account_id).await?;
 		let call = millau_runtime::MessageLaneCall::receive_messages_delivery_proof(proof).into();
-		let transaction = Millau::sign_transaction(&self.millau_client, &self.millau_sign.signer, nonce, call);
+		let transaction = Millau::sign_transaction(&self.source_client, &self.source_sign.signer, nonce, call);
 		Ok(transaction)
 	}
 
@@ -103,15 +69,18 @@ impl SubstrateMessageLane for MillauMessagesToRialto {
 		&self,
 		_generated_at_header: MillauHeaderId,
 		_nonces: RangeInclusive<MessageNonce>,
-		proof: FromMillauMessagesProof,
+		proof: <Self as MessageLane>::MessagesProof,
 	) -> Result<Self::TargetSignedTransaction, SubstrateError> {
 		let (dispatch_weight, proof) = proof;
-		let account_id = self.rialto_sign.signer.public().as_array_ref().clone().into();
+		let account_id = self.target_sign.signer.public().as_array_ref().clone().into();
-		let nonce = self.rialto_client.next_account_index(account_id).await?;
+		let nonce = self.target_client.next_account_index(account_id).await?;
-		let call =
+		let call = rialto_runtime::MessageLaneCall::receive_messages_proof(
-			rialto_runtime::MessageLaneCall::receive_messages_proof(self.relayer_id.clone(), proof, dispatch_weight)
+			self.relayer_id_at_source.clone(),
-				.into();
+			proof,
-		let transaction = Rialto::sign_transaction(&self.rialto_client, &self.rialto_sign.signer, nonce, call);
+			dispatch_weight,
+		)
+		.into();
+		let transaction = Rialto::sign_transaction(&self.target_client, &self.target_sign.signer, nonce, call);
 		Ok(transaction)
 	}
 }
@@ -131,25 +100,23 @@ pub fn run(
 	lane_id: LaneId,
 	metrics_params: Option<MetricsParams>,
 ) {
-	let millau_tick = Duration::from_secs(5);
-	let rialto_tick = Duration::from_secs(5);
 	let reconnect_delay = Duration::from_secs(10);
 	let stall_timeout = Duration::from_secs(5 * 60);
-	let relayer_id = millau_sign.signer.public().as_array_ref().clone().into();
+	let relayer_id_at_millau = millau_sign.signer.public().as_array_ref().clone().into();
 
 	let lane = MillauMessagesToRialto {
-		millau_client: millau_client.clone(),
+		source_client: millau_client.clone(),
-		millau_sign,
+		source_sign: millau_sign,
-		rialto_client: rialto_client.clone(),
+		target_client: rialto_client.clone(),
-		rialto_sign,
+		target_sign: rialto_sign,
-		relayer_id,
+		relayer_id_at_source: relayer_id_at_millau,
 	};
 
 	messages_relay::message_lane_loop::run(
 		messages_relay::message_lane_loop::Params {
 			lane: lane_id,
-			source_tick: millau_tick,
+			source_tick: Millau::AVERAGE_BLOCK_INTERVAL,
-			target_tick: rialto_tick,
+			target_tick: Rialto::AVERAGE_BLOCK_INTERVAL,
 			reconnect_delay,
 			stall_timeout,
 			delivery_params: messages_relay::message_lane_loop::MessageDeliveryParams {

bridges/relays/substrate/src/rialto_messages_to_millau.rs

@@ -0,0 +1,134 @@
+// 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 messages sync entrypoint.
+
+use crate::messages_lane::{SubstrateMessageLane, SubstrateMessageLaneToSubstrate};
+use crate::messages_source::SubstrateMessagesSource;
+use crate::messages_target::SubstrateMessagesTarget;
+use crate::{MillauClient, RialtoClient};
+
+use async_trait::async_trait;
+use bp_message_lane::{LaneId, MessageNonce};
+use bp_runtime::{MILLAU_BRIDGE_INSTANCE, RIALTO_BRIDGE_INSTANCE};
+use messages_relay::message_lane::MessageLane;
+use relay_millau_client::{HeaderId as MillauHeaderId, Millau, SigningParams as MillauSigningParams};
+use relay_rialto_client::{HeaderId as RialtoHeaderId, Rialto, SigningParams as RialtoSigningParams};
+use relay_substrate_client::{Chain, Error as SubstrateError, TransactionSignScheme};
+use relay_utils::metrics::MetricsParams;
+use sp_core::Pair;
+use std::{ops::RangeInclusive, time::Duration};
+
+/// Rialto-to-Millau message lane.
+type RialtoMessagesToMillau = SubstrateMessageLaneToSubstrate<Rialto, RialtoSigningParams, Millau, MillauSigningParams>;
+
+#[async_trait]
+impl SubstrateMessageLane for RialtoMessagesToMillau {
+	const OUTBOUND_LANE_MESSAGES_DISPATCH_WEIGHT_METHOD: &'static str =
+		bp_millau::TO_MILLAU_MESSAGES_DISPATCH_WEIGHT_METHOD;
+	const OUTBOUND_LANE_LATEST_GENERATED_NONCE_METHOD: &'static str =
+		bp_millau::TO_MILLAU_LATEST_GENERATED_NONCE_METHOD;
+	const OUTBOUND_LANE_LATEST_RECEIVED_NONCE_METHOD: &'static str = bp_millau::TO_MILLAU_LATEST_RECEIVED_NONCE_METHOD;
+
+	const INBOUND_LANE_LATEST_RECEIVED_NONCE_METHOD: &'static str = bp_rialto::FROM_RIALTO_LATEST_RECEIVED_NONCE_METHOD;
+	const INBOUND_LANE_LATEST_CONFIRMED_NONCE_METHOD: &'static str =
+		bp_rialto::FROM_RIALTO_LATEST_CONFIRMED_NONCE_METHOD;
+
+	const BEST_FINALIZED_SOURCE_HEADER_ID_AT_TARGET: &'static str = bp_rialto::FINALIZED_RIALTO_BLOCK_METHOD;
+	const BEST_FINALIZED_TARGET_HEADER_ID_AT_SOURCE: &'static str = bp_millau::FINALIZED_MILLAU_BLOCK_METHOD;
+
+	type SourceSignedTransaction = <Rialto as TransactionSignScheme>::SignedTransaction;
+	type TargetSignedTransaction = <Millau as TransactionSignScheme>::SignedTransaction;
+
+	async fn make_messages_receiving_proof_transaction(
+		&self,
+		_generated_at_block: MillauHeaderId,
+		proof: <Self as MessageLane>::MessagesReceivingProof,
+	) -> Result<Self::SourceSignedTransaction, SubstrateError> {
+		let account_id = self.source_sign.signer.public().as_array_ref().clone().into();
+		let nonce = self.source_client.next_account_index(account_id).await?;
+		let call = rialto_runtime::MessageLaneCall::receive_messages_delivery_proof(proof).into();
+		let transaction = Rialto::sign_transaction(&self.source_client, &self.source_sign.signer, nonce, call);
+		Ok(transaction)
+	}
+
+	async fn make_messages_delivery_transaction(
+		&self,
+		_generated_at_header: RialtoHeaderId,
+		_nonces: RangeInclusive<MessageNonce>,
+		proof: <Self as MessageLane>::MessagesProof,
+	) -> Result<Self::TargetSignedTransaction, SubstrateError> {
+		let (dispatch_weight, proof) = proof;
+		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 = millau_runtime::MessageLaneCall::receive_messages_proof(
+			self.relayer_id_at_source.clone(),
+			proof,
+			dispatch_weight,
+		)
+		.into();
+		let transaction = Millau::sign_transaction(&self.target_client, &self.target_sign.signer, nonce, call);
+		Ok(transaction)
+	}
+}
+
+/// Rialto node as messages source.
+type RialtoSourceClient = SubstrateMessagesSource<Rialto, RialtoMessagesToMillau>;
+
+/// Millau node as messages target.
+type MillauTargetClient = SubstrateMessagesTarget<Millau, RialtoMessagesToMillau>;
+
+/// Run Rialto-to-Millau messages sync.
+pub fn run(
+	rialto_client: RialtoClient,
+	rialto_sign: RialtoSigningParams,
+	millau_client: MillauClient,
+	millau_sign: MillauSigningParams,
+	lane_id: LaneId,
+	metrics_params: Option<MetricsParams>,
+) {
+	let reconnect_delay = Duration::from_secs(10);
+	let stall_timeout = Duration::from_secs(5 * 60);
+	let relayer_id_at_rialto = rialto_sign.signer.public().as_array_ref().clone().into();
+
+	let lane = RialtoMessagesToMillau {
+		source_client: rialto_client.clone(),
+		source_sign: rialto_sign,
+		target_client: millau_client.clone(),
+		target_sign: millau_sign,
+		relayer_id_at_source: relayer_id_at_rialto,
+	};
+
+	messages_relay::message_lane_loop::run(
+		messages_relay::message_lane_loop::Params {
+			lane: lane_id,
+			source_tick: Rialto::AVERAGE_BLOCK_INTERVAL,
+			target_tick: Millau::AVERAGE_BLOCK_INTERVAL,
+			reconnect_delay,
+			stall_timeout,
+			delivery_params: messages_relay::message_lane_loop::MessageDeliveryParams {
+				max_unconfirmed_nonces_at_target: bp_millau::MAX_UNCONFIRMED_MESSAGES_AT_INBOUND_LANE,
+				// TODO: subtract base weight of delivery from this when it'll be known
+				// https://github.com/paritytech/parity-bridges-common/issues/78
+				max_messages_weight_in_single_batch: bp_millau::MAXIMUM_EXTRINSIC_WEIGHT,
+			},
+		},
+		RialtoSourceClient::new(rialto_client, lane.clone(), lane_id, MILLAU_BRIDGE_INSTANCE),
+		MillauTargetClient::new(millau_client, lane, lane_id, RIALTO_BRIDGE_INSTANCE),
+		metrics_params,
+		futures::future::pending(),
+	);
+}