Custom relay strategy

relays/bin-substrate/src/chains/kusama_messages_to_polkadot.rs

@@ -19,12 +19,13 @@
 use std::ops::RangeInclusive;
 
 use codec::Encode;
+use frame_support::weights::Weight;
 use sp_core::{Bytes, Pair};
+use sp_runtime::{FixedPointNumber, FixedU128};
 
 use bp_messages::MessageNonce;
 use bridge_runtime_common::messages::target::FromBridgedChainMessagesProof;
-use frame_support::weights::Weight;
-use messages_relay::message_lane::MessageLane;
+use messages_relay::{message_lane::MessageLane, relay_strategy::MixStrategy};
 use relay_kusama_client::{
 	HeaderId as KusamaHeaderId, Kusama, SigningParams as KusamaSigningParams,
 };
@@ -33,7 +34,6 @@ use relay_polkadot_client::{
 };
 use relay_substrate_client::{Chain, Client, TransactionSignScheme, UnsignedTransaction};
 use relay_utils::metrics::MetricsParams;
-use sp_runtime::{FixedPointNumber, FixedU128};
 use substrate_relay_helper::{
 	messages_lane::{
 		select_delivery_transaction_limits, MessagesRelayParams, StandaloneMessagesMetrics,
@@ -177,7 +177,13 @@ type PolkadotTargetClient = SubstrateMessagesTarget<KusamaMessagesToPolkadot>;
 
 /// Run Kusama-to-Polkadot messages sync.
 pub async fn run(
-	params: MessagesRelayParams<Kusama, KusamaSigningParams, Polkadot, PolkadotSigningParams>,
+	params: MessagesRelayParams<
+		Kusama,
+		KusamaSigningParams,
+		Polkadot,
+		PolkadotSigningParams,
+		MixStrategy,
+	>,
 ) -> anyhow::Result<()> {
 	let stall_timeout = relay_substrate_client::bidirectional_transaction_stall_timeout(
 		params.source_transactions_mortality,
@@ -223,14 +229,12 @@ pub async fn run(
 			Max messages in single transaction: {}\n\t\
 			Max messages size in single transaction: {}\n\t\
 			Max messages weight in single transaction: {}\n\t\
-			Relayer mode: {:?}\n\t\
 			Tx mortality: {:?}/{:?}\n\t\
 			Stall timeout: {:?}",
 		lane.message_lane.relayer_id_at_source,
 		max_messages_in_single_batch,
 		max_messages_size_in_single_batch,
 		max_messages_weight_in_single_batch,
-		params.relayer_mode,
 		params.source_transactions_mortality,
 		params.target_transactions_mortality,
 		stall_timeout,
@@ -258,7 +262,7 @@ pub async fn run(
 				max_messages_in_single_batch,
 				max_messages_weight_in_single_batch,
 				max_messages_size_in_single_batch,
-				relayer_mode: params.relayer_mode,
+				relay_strategy: params.relay_strategy,
 			},
 		},
 		KusamaSourceClient::new(
@@ -332,7 +336,7 @@ pub(crate) async fn update_polkadot_to_kusama_conversion_rate(
 						transaction_nonce,
 					),
 				)
-				.encode(),
+					.encode(),
 			)
 		})
 		.await

relays/bin-substrate/src/chains/millau_messages_to_rialto.rs

@@ -25,7 +25,7 @@ use sp_core::{Bytes, Pair};
 use bp_messages::MessageNonce;
 use bridge_runtime_common::messages::target::FromBridgedChainMessagesProof;
 use frame_support::weights::Weight;
-use messages_relay::message_lane::MessageLane;
+use messages_relay::{message_lane::MessageLane, relay_strategy::MixStrategy};
 use relay_millau_client::{
 	HeaderId as MillauHeaderId, Millau, SigningParams as MillauSigningParams,
 };
@@ -174,7 +174,13 @@ type RialtoTargetClient = SubstrateMessagesTarget<MillauMessagesToRialto>;
 
 /// Run Millau-to-Rialto messages sync.
 pub async fn run(
-	params: MessagesRelayParams<Millau, MillauSigningParams, Rialto, RialtoSigningParams>,
+	params: MessagesRelayParams<
+		Millau,
+		MillauSigningParams,
+		Rialto,
+		RialtoSigningParams,
+		MixStrategy,
+	>,
 ) -> anyhow::Result<()> {
 	let stall_timeout = relay_substrate_client::bidirectional_transaction_stall_timeout(
 		params.source_transactions_mortality,
@@ -217,14 +223,12 @@ pub async fn run(
 			Max messages in single transaction: {}\n\t\
 			Max messages size in single transaction: {}\n\t\
 			Max messages weight in single transaction: {}\n\t\
-			Relayer mode: {:?}\n\t\
 			Tx mortality: {:?}/{:?}\n\t\
 			Stall timeout: {:?}",
 		lane.message_lane.relayer_id_at_source,
 		max_messages_in_single_batch,
 		max_messages_size_in_single_batch,
 		max_messages_weight_in_single_batch,
-		params.relayer_mode,
 		params.source_transactions_mortality,
 		params.target_transactions_mortality,
 		stall_timeout,
@@ -252,7 +256,7 @@ pub async fn run(
 				max_messages_in_single_batch,
 				max_messages_weight_in_single_batch,
 				max_messages_size_in_single_batch,
-				relayer_mode: params.relayer_mode,
+				relay_strategy: params.relay_strategy,
 			},
 		},
 		MillauSourceClient::new(

relays/bin-substrate/src/chains/polkadot_messages_to_kusama.rs

@@ -24,7 +24,7 @@ use sp_core::{Bytes, Pair};
 use bp_messages::MessageNonce;
 use bridge_runtime_common::messages::target::FromBridgedChainMessagesProof;
 use frame_support::weights::Weight;
-use messages_relay::message_lane::MessageLane;
+use messages_relay::{message_lane::MessageLane, relay_strategy::MixStrategy};
 use relay_kusama_client::{
 	HeaderId as KusamaHeaderId, Kusama, SigningParams as KusamaSigningParams,
 };
@@ -176,7 +176,13 @@ type KusamaTargetClient = SubstrateMessagesTarget<PolkadotMessagesToKusama>;
 
 /// Run Polkadot-to-Kusama messages sync.
 pub async fn run(
-	params: MessagesRelayParams<Polkadot, PolkadotSigningParams, Kusama, KusamaSigningParams>,
+	params: MessagesRelayParams<
+		Polkadot,
+		PolkadotSigningParams,
+		Kusama,
+		KusamaSigningParams,
+		MixStrategy,
+	>,
 ) -> anyhow::Result<()> {
 	let stall_timeout = relay_substrate_client::bidirectional_transaction_stall_timeout(
 		params.source_transactions_mortality,
@@ -222,14 +228,12 @@ pub async fn run(
 			Max messages in single transaction: {}\n\t\
 			Max messages size in single transaction: {}\n\t\
 			Max messages weight in single transaction: {}\n\t\
-			Relayer mode: {:?}\n\t\
 			Tx mortality: {:?}/{:?}\n\t\
 			Stall timeout: {:?}",
 		lane.message_lane.relayer_id_at_source,
 		max_messages_in_single_batch,
 		max_messages_size_in_single_batch,
 		max_messages_weight_in_single_batch,
-		params.relayer_mode,
 		params.source_transactions_mortality,
 		params.target_transactions_mortality,
 		stall_timeout,
@@ -257,7 +261,7 @@ pub async fn run(
 				max_messages_in_single_batch,
 				max_messages_weight_in_single_batch,
 				max_messages_size_in_single_batch,
-				relayer_mode: params.relayer_mode,
+				relay_strategy: params.relay_strategy,
 			},
 		},
 		PolkadotSourceClient::new(

relays/bin-substrate/src/chains/rialto_messages_to_millau.rs

@@ -25,7 +25,7 @@ use sp_core::{Bytes, Pair};
 use bp_messages::MessageNonce;
 use bridge_runtime_common::messages::target::FromBridgedChainMessagesProof;
 use frame_support::weights::Weight;
-use messages_relay::message_lane::MessageLane;
+use messages_relay::{message_lane::MessageLane, relay_strategy::MixStrategy};
 use relay_millau_client::{
 	HeaderId as MillauHeaderId, Millau, SigningParams as MillauSigningParams,
 };
@@ -174,7 +174,13 @@ type MillauTargetClient = SubstrateMessagesTarget<RialtoMessagesToMillau>;
 
 /// Run Rialto-to-Millau messages sync.
 pub async fn run(
-	params: MessagesRelayParams<Rialto, RialtoSigningParams, Millau, MillauSigningParams>,
+	params: MessagesRelayParams<
+		Rialto,
+		RialtoSigningParams,
+		Millau,
+		MillauSigningParams,
+		MixStrategy,
+	>,
 ) -> anyhow::Result<()> {
 	let stall_timeout = relay_substrate_client::bidirectional_transaction_stall_timeout(
 		params.source_transactions_mortality,
@@ -216,14 +222,12 @@ pub async fn run(
 			Max messages in single transaction: {}\n\t\
 			Max messages size in single transaction: {}\n\t\
 			Max messages weight in single transaction: {}\n\t\
-			Relayer mode: {:?}\n\t\
 			Tx mortality: {:?}/{:?}\n\t\
 			Stall timeout: {:?}",
 		lane.message_lane.relayer_id_at_source,
 		max_messages_in_single_batch,
 		max_messages_size_in_single_batch,
 		max_messages_weight_in_single_batch,
-		params.relayer_mode,
 		params.source_transactions_mortality,
 		params.target_transactions_mortality,
 		stall_timeout,
@@ -251,7 +255,7 @@ pub async fn run(
 				max_messages_in_single_batch,
 				max_messages_weight_in_single_batch,
 				max_messages_size_in_single_batch,
-				relayer_mode: params.relayer_mode,
+				relay_strategy: params.relay_strategy,
 			},
 		},
 		RialtoSourceClient::new(

relays/bin-substrate/src/chains/rococo_messages_to_wococo.rs

@@ -24,7 +24,7 @@ use sp_core::{Bytes, Pair};
 use bp_messages::MessageNonce;
 use bridge_runtime_common::messages::target::FromBridgedChainMessagesProof;
 use frame_support::weights::Weight;
-use messages_relay::message_lane::MessageLane;
+use messages_relay::{message_lane::MessageLane, relay_strategy::MixStrategy};
 use relay_rococo_client::{
 	HeaderId as RococoHeaderId, Rococo, SigningParams as RococoSigningParams,
 };
@@ -174,7 +174,13 @@ type WococoTargetClient = SubstrateMessagesTarget<RococoMessagesToWococo>;
 
 /// Run Rococo-to-Wococo messages sync.
 pub async fn run(
-	params: MessagesRelayParams<Rococo, RococoSigningParams, Wococo, WococoSigningParams>,
+	params: MessagesRelayParams<
+		Rococo,
+		RococoSigningParams,
+		Wococo,
+		WococoSigningParams,
+		MixStrategy,
+	>,
 ) -> anyhow::Result<()> {
 	let stall_timeout = relay_substrate_client::bidirectional_transaction_stall_timeout(
 		params.source_transactions_mortality,
@@ -220,14 +226,12 @@ pub async fn run(
 			Max messages in single transaction: {}\n\t\
 			Max messages size in single transaction: {}\n\t\
 			Max messages weight in single transaction: {}\n\t\
-			Relayer mode: {:?}\n\t\
 			Tx mortality: {:?}/{:?}\n\t\
 			Stall timeout: {:?}",
 		lane.message_lane.relayer_id_at_source,
 		max_messages_in_single_batch,
 		max_messages_size_in_single_batch,
 		max_messages_weight_in_single_batch,
-		params.relayer_mode,
 		params.source_transactions_mortality,
 		params.target_transactions_mortality,
 		stall_timeout,
@@ -255,7 +259,7 @@ pub async fn run(
 				max_messages_in_single_batch,
 				max_messages_weight_in_single_batch,
 				max_messages_size_in_single_batch,
-				relayer_mode: params.relayer_mode,
+				relay_strategy: params.relay_strategy,
 			},
 		},
 		RococoSourceClient::new(

relays/bin-substrate/src/chains/wococo_messages_to_rococo.rs

@@ -24,7 +24,7 @@ use sp_core::{Bytes, Pair};
 use bp_messages::MessageNonce;
 use bridge_runtime_common::messages::target::FromBridgedChainMessagesProof;
 use frame_support::weights::Weight;
-use messages_relay::message_lane::MessageLane;
+use messages_relay::{message_lane::MessageLane, relay_strategy::MixStrategy};
 use relay_rococo_client::{
 	HeaderId as RococoHeaderId, Rococo, SigningParams as RococoSigningParams,
 };
@@ -173,7 +173,13 @@ type RococoTargetClient = SubstrateMessagesTarget<WococoMessagesToRococo>;
 
 /// Run Wococo-to-Rococo messages sync.
 pub async fn run(
-	params: MessagesRelayParams<Wococo, WococoSigningParams, Rococo, RococoSigningParams>,
+	params: MessagesRelayParams<
+		Wococo,
+		WococoSigningParams,
+		Rococo,
+		RococoSigningParams,
+		MixStrategy,
+	>,
 ) -> anyhow::Result<()> {
 	let stall_timeout = relay_substrate_client::bidirectional_transaction_stall_timeout(
 		params.source_transactions_mortality,
@@ -219,14 +225,12 @@ pub async fn run(
 			Max messages in single transaction: {}\n\t\
 			Max messages size in single transaction: {}\n\t\
 			Max messages weight in single transaction: {}\n\t\
-			Relayer mode: {:?}\n\t\
 			Tx mortality: {:?}/{:?}\n\t\
 			Stall timeout: {:?}",
 		lane.message_lane.relayer_id_at_source,
 		max_messages_in_single_batch,
 		max_messages_size_in_single_batch,
 		max_messages_weight_in_single_batch,
-		params.relayer_mode,
 		params.source_transactions_mortality,
 		params.target_transactions_mortality,
 		stall_timeout,
@@ -254,7 +258,7 @@ pub async fn run(
 				max_messages_in_single_batch,
 				max_messages_weight_in_single_batch,
 				max_messages_size_in_single_batch,
-				relayer_mode: params.relayer_mode,
+				relay_strategy: params.relay_strategy,
 			},
 		},
 		WococoSourceClient::new(

relays/bin-substrate/src/cli/relay_headers_and_messages.rs

@@ -27,6 +27,7 @@ use structopt::StructOpt;
 use strum::VariantNames;
 
 use codec::Encode;
+use messages_relay::relay_strategy::MixStrategy;
 use relay_substrate_client::{
 	AccountIdOf, Chain, Client, TransactionSignScheme, UnsignedTransaction,
 };
@@ -375,6 +376,7 @@ impl RelayHeadersAndMessages {
 
 			let lanes = params.shared.lane;
 			let relayer_mode = params.shared.relayer_mode.into();
+			let relay_strategy = MixStrategy::new(relayer_mode);
 
 			const METRIC_IS_SOME_PROOF: &str =
 				"it is `None` when metric has been already registered; \
@@ -512,38 +514,38 @@ impl RelayHeadersAndMessages {
 				let left_to_right_messages = left_to_right_messages(MessagesRelayParams {
 					source_client: left_client.clone(),
 					source_sign: left_sign.clone(),
-					source_transactions_mortality: left_transactions_mortality.clone(),
+					source_transactions_mortality: left_transactions_mortality,
 					target_client: right_client.clone(),
 					target_sign: right_sign.clone(),
-					target_transactions_mortality: right_transactions_mortality.clone(),
+					target_transactions_mortality: right_transactions_mortality,
 					source_to_target_headers_relay: Some(left_to_right_on_demand_headers.clone()),
 					target_to_source_headers_relay: Some(right_to_left_on_demand_headers.clone()),
 					lane_id: lane,
-					relayer_mode,
 					metrics_params: metrics_params.clone().disable().metrics_prefix(
 						messages_relay::message_lane_loop::metrics_prefix::<
 							<LeftToRightMessages as SubstrateMessageLane>::MessageLane,
 						>(&lane),
 					),
+					relay_strategy: relay_strategy.clone(),
 				})
 				.map_err(|e| anyhow::format_err!("{}", e))
 				.boxed();
 				let right_to_left_messages = right_to_left_messages(MessagesRelayParams {
 					source_client: right_client.clone(),
 					source_sign: right_sign.clone(),
-					source_transactions_mortality: right_transactions_mortality.clone(),
+					source_transactions_mortality: right_transactions_mortality,
 					target_client: left_client.clone(),
 					target_sign: left_sign.clone(),
-					target_transactions_mortality: left_transactions_mortality.clone(),
+					target_transactions_mortality: left_transactions_mortality,
 					source_to_target_headers_relay: Some(right_to_left_on_demand_headers.clone()),
 					target_to_source_headers_relay: Some(left_to_right_on_demand_headers.clone()),
 					lane_id: lane,
-					relayer_mode,
 					metrics_params: metrics_params.clone().disable().metrics_prefix(
 						messages_relay::message_lane_loop::metrics_prefix::<
 							<RightToLeftMessages as SubstrateMessageLane>::MessageLane,
 						>(&lane),
 					),
+					relay_strategy: relay_strategy.clone(),
 				})
 				.map_err(|e| anyhow::format_err!("{}", e))
 				.boxed();

relays/bin-substrate/src/cli/relay_messages.rs

@@ -17,6 +17,7 @@
 use structopt::StructOpt;
 use strum::{EnumString, EnumVariantNames, VariantNames};
 
+use messages_relay::relay_strategy::MixStrategy;
 use substrate_relay_helper::messages_lane::MessagesRelayParams;
 
 use crate::{
@@ -80,6 +81,8 @@ impl RelayMessages {
 			let target_client = self.target.to_client::<Target>().await?;
 			let target_sign = self.target_sign.to_keypair::<Target>()?;
 			let target_transactions_mortality = self.target_sign.transactions_mortality()?;
+			let relayer_mode = self.relayer_mode.into();
+			let relay_strategy = MixStrategy::new(relayer_mode);
 
 			relay_messages(MessagesRelayParams {
 				source_client,
@@ -91,8 +94,8 @@ impl RelayMessages {
 				source_to_target_headers_relay: None,
 				target_to_source_headers_relay: None,
 				lane_id: self.lane.into(),
-				relayer_mode: self.relayer_mode.into(),
 				metrics_params: self.prometheus_params.into(),
+				relay_strategy,
 			})
 			.await
 			.map_err(|e| anyhow::format_err!("{}", e))