Messages relayer operating mode (#995)

* introduce relayer mode enum

* removed MaximalReward mode

* TODO

* Saturating -> SaturatingAdd

* ref issue in TODOs

* Update relays/messages/src/message_race_delivery.rs

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

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

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

@@ -159,7 +159,7 @@ pub async fn run(
 	};
 
 	// 2/3 is reserved for proofs and tx overhead
-	let max_messages_size_in_single_batch = bp_rialto::max_extrinsic_size() as usize / 3;
+	let max_messages_size_in_single_batch = bp_rialto::max_extrinsic_size() / 3;
 	// TODO: use Millau weights after https://github.com/paritytech/parity-bridges-common/issues/390
 	let (max_messages_in_single_batch, max_messages_weight_in_single_batch) =
 		select_delivery_transaction_limits::<pallet_bridge_messages::weights::RialtoWeight<millau_runtime::Runtime>>(
@@ -193,6 +193,7 @@ pub async fn run(
 				max_messages_in_single_batch,
 				max_messages_weight_in_single_batch,
 				max_messages_size_in_single_batch,
+				relayer_mode: messages_relay::message_lane_loop::RelayerMode::Altruistic,
 			},
 		},
 		MillauSourceClient::new(

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

@@ -159,7 +159,7 @@ pub async fn run(
 	};
 
 	// 2/3 is reserved for proofs and tx overhead
-	let max_messages_size_in_single_batch = bp_millau::max_extrinsic_size() as usize / 3;
+	let max_messages_size_in_single_batch = bp_millau::max_extrinsic_size() / 3;
 	let (max_messages_in_single_batch, max_messages_weight_in_single_batch) =
 		select_delivery_transaction_limits::<pallet_bridge_messages::weights::RialtoWeight<rialto_runtime::Runtime>>(
 			bp_millau::max_extrinsic_weight(),
@@ -192,6 +192,7 @@ pub async fn run(
 				max_messages_in_single_batch,
 				max_messages_weight_in_single_batch,
 				max_messages_size_in_single_batch,
+				relayer_mode: messages_relay::message_lane_loop::RelayerMode::Altruistic,
 			},
 		},
 		RialtoSourceClient::new(

bridges/relays/bin-substrate/src/messages_lane.rs

@@ -139,6 +139,7 @@ where
 	type MessagesProof = SubstrateMessagesProof<Source>;
 	type MessagesReceivingProof = SubstrateMessagesReceivingProof<Target>;
 
+	type SourceChainBalance = Source::Balance;
 	type SourceHeaderNumber = BlockNumberOf<Source>;
 	type SourceHeaderHash = HashOf<Source>;
 

bridges/relays/bin-substrate/src/messages_source.rs

@@ -30,7 +30,7 @@ use frame_support::{traits::Instance, weights::Weight};
 use messages_relay::{
 	message_lane::{SourceHeaderIdOf, TargetHeaderIdOf},
 	message_lane_loop::{
-		ClientState, MessageProofParameters, MessageWeights, MessageWeightsMap, SourceClient, SourceClientState,
+		ClientState, MessageDetails, MessageDetailsMap, MessageProofParameters, SourceClient, SourceClientState,
 	},
 };
 use pallet_bridge_messages::Config as MessagesConfig;
@@ -112,6 +112,7 @@ where
 	C::BlockNumber: BlockNumberBase,
 	P: SubstrateMessageLane<
 		MessagesProof = SubstrateMessagesProof<C>,
+		SourceChainBalance = C::Balance,
 		SourceHeaderNumber = <C::Header as HeaderT>::Number,
 		SourceHeaderHash = <C::Header as HeaderT>::Hash,
 		SourceChain = C,
@@ -168,11 +169,11 @@ where
 		Ok((id, latest_received_nonce))
 	}
 
-	async fn generated_messages_weights(
+	async fn generated_message_details(
 		&self,
 		id: SourceHeaderIdOf<P>,
 		nonces: RangeInclusive<MessageNonce>,
-	) -> Result<MessageWeightsMap, SubstrateError> {
+	) -> Result<MessageDetailsMap<P::SourceChainBalance>, SubstrateError> {
 		let encoded_response = self
 			.client
 			.state_call(
@@ -242,6 +243,10 @@ where
 			target_to_source_headers_relay.require_finalized_header(id).await;
 		}
 	}
+
+	async fn estimate_confirmation_transaction(&self) -> P::SourceChainBalance {
+		num_traits::Zero::zero() // TODO: https://github.com/paritytech/parity-bridges-common/issues/997
+	}
 }
 
 pub async fn read_client_state<SelfChain, BridgedHeaderHash, BridgedHeaderNumber>(
@@ -290,7 +295,7 @@ where
 fn make_message_details_map<C: Chain>(
 	weights: Vec<bp_messages::MessageDetails<C::Balance>>,
 	nonces: RangeInclusive<MessageNonce>,
-) -> Result<MessageWeightsMap, SubstrateError> {
+) -> Result<MessageDetailsMap<C::Balance>, SubstrateError> {
 	let make_missing_nonce_error = |expected_nonce| {
 		Err(SubstrateError::Custom(format!(
 			"Missing nonce {} in messages_dispatch_weight call result. Expected all nonces from {:?}",
@@ -298,7 +303,7 @@ fn make_message_details_map<C: Chain>(
 		)))
 	};
 
-	let mut weights_map = MessageWeightsMap::new();
+	let mut weights_map = MessageDetailsMap::new();
 
 	// this is actually prevented by external logic
 	if nonces.is_empty() {
@@ -341,9 +346,11 @@ fn make_message_details_map<C: Chain>(
 
 		weights_map.insert(
 			details.nonce,
-			MessageWeights {
-				weight: details.dispatch_weight,
+			MessageDetails {
+				dispatch_weight: details.dispatch_weight,
 				size: details.size as _,
+				// TODO: https://github.com/paritytech/parity-bridges-common/issues/997
+				reward: num_traits::Zero::zero(),
 			},
 		);
 		expected_nonce = details.nonce + 1;
@@ -376,9 +383,30 @@ mod tests {
 		assert_eq!(
 			make_message_details_map::<relay_rialto_client::Rialto>(message_details_from_rpc(1..=3), 1..=3,).unwrap(),
 			vec![
-				(1, MessageWeights { weight: 0, size: 0 }),
-				(2, MessageWeights { weight: 0, size: 0 }),
-				(3, MessageWeights { weight: 0, size: 0 }),
+				(
+					1,
+					MessageDetails {
+						dispatch_weight: 0,
+						size: 0,
+						reward: 0
+					}
+				),
+				(
+					2,
+					MessageDetails {
+						dispatch_weight: 0,
+						size: 0,
+						reward: 0
+					}
+				),
+				(
+					3,
+					MessageDetails {
+						dispatch_weight: 0,
+						size: 0,
+						reward: 0
+					}
+				),
 			]
 			.into_iter()
 			.collect(),
@@ -390,8 +418,22 @@ mod tests {
 		assert_eq!(
 			make_message_details_map::<relay_rialto_client::Rialto>(message_details_from_rpc(2..=3), 1..=3,).unwrap(),
 			vec![
-				(2, MessageWeights { weight: 0, size: 0 }),
-				(3, MessageWeights { weight: 0, size: 0 }),
+				(
+					2,
+					MessageDetails {
+						dispatch_weight: 0,
+						size: 0,
+						reward: 0
+					}
+				),
+				(
+					3,
+					MessageDetails {
+						dispatch_weight: 0,
+						size: 0,
+						reward: 0
+					}
+				),
 			]
 			.into_iter()
 			.collect(),

bridges/relays/bin-substrate/src/messages_target.rs

@@ -27,7 +27,7 @@ use bp_messages::{LaneId, MessageNonce, UnrewardedRelayersState};
 use bp_runtime::ChainId;
 use bridge_runtime_common::messages::source::FromBridgedChainMessagesDeliveryProof;
 use codec::{Decode, Encode};
-use frame_support::traits::Instance;
+use frame_support::{traits::Instance, weights::Weight};
 use messages_relay::{
 	message_lane::{SourceHeaderIdOf, TargetHeaderIdOf},
 	message_lane_loop::{TargetClient, TargetClientState},
@@ -229,4 +229,13 @@ where
 			source_to_target_headers_relay.require_finalized_header(id).await;
 		}
 	}
+
+	async fn estimate_delivery_transaction_in_source_tokens(
+		&self,
+		_nonces: RangeInclusive<MessageNonce>,
+		_total_dispatch_weight: Weight,
+		_total_size: u32,
+	) -> P::SourceChainBalance {
+		num_traits::Zero::zero() // TODO: https://github.com/paritytech/parity-bridges-common/issues/997
+	}
 }

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

@@ -17,7 +17,7 @@
 use bp_runtime::Chain as ChainBase;
 use frame_support::Parameter;
 use jsonrpsee_ws_client::{DeserializeOwned, Serialize};
-use num_traits::{CheckedSub, Zero};
+use num_traits::{CheckedSub, SaturatingAdd, Zero};
 use sp_core::{storage::StorageKey, Pair};
 use sp_runtime::{
 	generic::SignedBlock,
@@ -58,7 +58,7 @@ pub trait Chain: ChainBase + Clone {
 	///
 	/// The chain may suport multiple tokens, but this particular type is for token that is used
 	/// to pay for transaction dispatch, to reward different relayers (headers, messages), etc.
-	type Balance: Parameter + Member + DeserializeOwned + Clone + Copy + CheckedSub + PartialOrd + Zero;
+	type Balance: Parameter + Member + DeserializeOwned + Clone + Copy + CheckedSub + PartialOrd + SaturatingAdd + Zero;
 }
 
 /// Substrate-based chain with `frame_system::Config::AccountData` set to

bridges/relays/messages/Cargo.toml

@@ -6,11 +6,12 @@ edition = "2018"
 license = "GPL-3.0-or-later WITH Classpath-exception-2.0"
 
 [dependencies]
-async-std = "1.6.5"
+async-std = { version = "1.6.5", features = ["attributes"] }
 async-trait = "0.1.40"
 futures = "0.3.5"
 hex = "0.4"
 log = "0.4.11"
+num-traits = "0.2"
 parking_lot = "0.11.0"
 
 # Bridge Dependencies

bridges/relays/messages/src/message_lane.rs

@@ -19,6 +19,7 @@
 //! 1) relay new messages from source to target node;
 //! 2) relay proof-of-delivery from target to source node.
 
+use num_traits::{SaturatingAdd, Zero};
 use relay_utils::{BlockNumberBase, HeaderId};
 use std::fmt::Debug;
 
@@ -34,6 +35,12 @@ pub trait MessageLane: 'static + Clone + Send + Sync {
 	/// Messages receiving proof.
 	type MessagesReceivingProof: Clone + Debug + Send + Sync;
 
+	/// The type of the source chain token balance, that is used to:
+	///
+	/// 1) pay transaction fees;
+	/// 2) pay message delivery and dispatch fee;
+	/// 3) pay relayer rewards.
+	type SourceChainBalance: Clone + Copy + Debug + PartialOrd + SaturatingAdd + Zero + Send + Sync;
 	/// Number of the source header.
 	type SourceHeaderNumber: BlockNumberBase;
 	/// Hash of the source header.

bridges/relays/messages/src/message_lane_loop.rs

@@ -58,6 +58,15 @@ pub struct Params {
 	pub delivery_params: MessageDeliveryParams,
 }
 
+/// Relayer operating mode.
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub enum RelayerMode {
+	/// The relayer doesn't care about rewards.
+	Altruistic,
+	/// The relayer will deliver all messages and confirmations as long as he's not losing any funds.
+	NoLosses,
+}
+
 /// Message delivery race parameters.
 #[derive(Debug, Clone)]
 pub struct MessageDeliveryParams {
@@ -74,20 +83,24 @@ pub struct MessageDeliveryParams {
 	/// Maximal cumulative dispatch weight of relayed messages in single delivery transaction.
 	pub max_messages_weight_in_single_batch: Weight,
 	/// Maximal cumulative size of relayed messages in single delivery transaction.
-	pub max_messages_size_in_single_batch: usize,
+	pub max_messages_size_in_single_batch: u32,
+	/// Relayer operating mode.
+	pub relayer_mode: RelayerMode,
 }
 
-/// Message weights.
+/// Message details.
 #[derive(Debug, Clone, Copy, PartialEq)]
-pub struct MessageWeights {
+pub struct MessageDetails<SourceChainBalance> {
 	/// Message dispatch weight.
-	pub weight: Weight,
+	pub dispatch_weight: Weight,
 	/// Message size (number of bytes in encoded payload).
-	pub size: usize,
+	pub size: u32,
+	/// The relayer reward paid in the source chain tokens.
+	pub reward: SourceChainBalance,
 }
 
-/// Messages weights map.
-pub type MessageWeightsMap = BTreeMap<MessageNonce, MessageWeights>;
+/// Messages details map.
+pub type MessageDetailsMap<SourceChainBalance> = BTreeMap<MessageNonce, MessageDetails<SourceChainBalance>>;
 
 /// Message delivery race proof parameters.
 #[derive(Debug, PartialEq)]
@@ -117,13 +130,13 @@ pub trait SourceClient<P: MessageLane>: RelayClient {
 
 	/// Returns mapping of message nonces, generated on this client, to their weights.
 	///
-	/// Some weights may be missing from returned map, if corresponding messages were pruned at
+	/// Some messages may be missing from returned map, if corresponding messages were pruned at
 	/// the source chain.
-	async fn generated_messages_weights(
+	async fn generated_message_details(
 		&self,
 		id: SourceHeaderIdOf<P>,
 		nonces: RangeInclusive<MessageNonce>,
-	) -> Result<MessageWeightsMap, Self::Error>;
+	) -> Result<MessageDetailsMap<P::SourceChainBalance>, Self::Error>;
 
 	/// Prove messages in inclusive range [begin; end].
 	async fn prove_messages(
@@ -142,6 +155,9 @@ pub trait SourceClient<P: MessageLane>: RelayClient {
 
 	/// We need given finalized target header on source to continue synchronization.
 	async fn require_target_header_on_source(&self, id: TargetHeaderIdOf<P>);
+
+	/// Estimate cost of single message confirmation transaction in source chain tokens.
+	async fn estimate_confirmation_transaction(&self) -> P::SourceChainBalance;
 }
 
 /// Target client trait.
@@ -183,6 +199,17 @@ pub trait TargetClient<P: MessageLane>: RelayClient {
 
 	/// We need given finalized source header on target to continue synchronization.
 	async fn require_source_header_on_target(&self, id: SourceHeaderIdOf<P>);
+
+	/// Estimate cost of messages delivery transaction in source chain tokens.
+	///
+	/// Please keep in mind that the returned cost must be converted to the source chain
+	/// tokens, even though the transaction fee will be paid in the target chain tokens.
+	async fn estimate_delivery_transaction_in_source_tokens(
+		&self,
+		nonces: RangeInclusive<MessageNonce>,
+		total_dispatch_weight: Weight,
+		total_size: u32,
+	) -> P::SourceChainBalance;
 }
 
 /// State of the client.
@@ -426,6 +453,10 @@ pub(crate) mod tests {
 		HeaderId(number, number)
 	}
 
+	pub const CONFIRMATION_TRANSACTION_COST: TestSourceChainBalance = 1;
+	pub const DELIVERY_TRANSACTION_COST: TestSourceChainBalance = 1;
+
+	pub type TestSourceChainBalance = u64;
 	pub type TestSourceHeaderId = HeaderId<TestSourceHeaderNumber, TestSourceHeaderHash>;
 	pub type TestTargetHeaderId = HeaderId<TestTargetHeaderNumber, TestTargetHeaderHash>;
 
@@ -457,6 +488,7 @@ pub(crate) mod tests {
 		type MessagesProof = TestMessagesProof;
 		type MessagesReceivingProof = TestMessagesReceivingProof;
 
+		type SourceChainBalance = TestSourceChainBalance;
 		type SourceHeaderNumber = TestSourceHeaderNumber;
 		type SourceHeaderHash = TestSourceHeaderHash;
 
@@ -490,6 +522,15 @@ pub(crate) mod tests {
 		tick: Arc<dyn Fn(&mut TestClientData) + Send + Sync>,
 	}
 
+	impl Default for TestSourceClient {
+		fn default() -> Self {
+			TestSourceClient {
+				data: Arc::new(Mutex::new(TestClientData::default())),
+				tick: Arc::new(|_| {}),
+			}
+		}
+	}
+
 	#[async_trait]
 	impl RelayClient for TestSourceClient {
 		type Error = TestError;
@@ -536,13 +577,22 @@ pub(crate) mod tests {
 			Ok((id, data.source_latest_confirmed_received_nonce))
 		}
 
-		async fn generated_messages_weights(
+		async fn generated_message_details(
 			&self,
 			_id: SourceHeaderIdOf<TestMessageLane>,
 			nonces: RangeInclusive<MessageNonce>,
-		) -> Result<MessageWeightsMap, TestError> {
+		) -> Result<MessageDetailsMap<TestSourceChainBalance>, TestError> {
 			Ok(nonces
-				.map(|nonce| (nonce, MessageWeights { weight: 1, size: 1 }))
+				.map(|nonce| {
+					(
+						nonce,
+						MessageDetails {
+							dispatch_weight: 1,
+							size: 1,
+							reward: 1,
+						},
+					)
+				})
 				.collect())
 		}
 
@@ -596,6 +646,10 @@ pub(crate) mod tests {
 			data.target_to_source_header_requirements.push(id);
 			(self.tick)(&mut *data);
 		}
+
+		async fn estimate_confirmation_transaction(&self) -> TestSourceChainBalance {
+			CONFIRMATION_TRANSACTION_COST
+		}
 	}
 
 	#[derive(Clone)]
@@ -604,6 +658,15 @@ pub(crate) mod tests {
 		tick: Arc<dyn Fn(&mut TestClientData) + Send + Sync>,
 	}
 
+	impl Default for TestTargetClient {
+		fn default() -> Self {
+			TestTargetClient {
+				data: Arc::new(Mutex::new(TestClientData::default())),
+				tick: Arc::new(|_| {}),
+			}
+		}
+	}
+
 	#[async_trait]
 	impl RelayClient for TestTargetClient {
 		type Error = TestError;
@@ -702,6 +765,15 @@ pub(crate) mod tests {
 			data.source_to_target_header_requirements.push(id);
 			(self.tick)(&mut *data);
 		}
+
+		async fn estimate_delivery_transaction_in_source_tokens(
+			&self,
+			_nonces: RangeInclusive<MessageNonce>,
+			_total_dispatch_weight: Weight,
+			total_size: u32,
+		) -> TestSourceChainBalance {
+			DELIVERY_TRANSACTION_COST * (total_size as TestSourceChainBalance)
+		}
 	}
 
 	fn run_loop_test(
@@ -734,6 +806,7 @@ pub(crate) mod tests {
 						max_messages_in_single_batch: 4,
 						max_messages_weight_in_single_batch: 4,
 						max_messages_size_in_single_batch: 4,
+						relayer_mode: RelayerMode::Altruistic,
 					},
 				},
 				source_client,

bridges/relays/messages/src/message_race_delivery.rs

@@ -15,8 +15,9 @@
 
 use crate::message_lane::{MessageLane, SourceHeaderIdOf, TargetHeaderIdOf};
 use crate::message_lane_loop::{
-	MessageDeliveryParams, MessageProofParameters, MessageWeightsMap, SourceClient as MessageLaneSourceClient,
-	SourceClientState, TargetClient as MessageLaneTargetClient, TargetClientState,
+	MessageDeliveryParams, MessageDetailsMap, MessageProofParameters, RelayerMode,
+	SourceClient as MessageLaneSourceClient, SourceClientState, TargetClient as MessageLaneTargetClient,
+	TargetClientState,
 };
 use crate::message_race_loop::{
 	MessageRace, NoncesRange, RaceState, RaceStrategy, SourceClient, SourceClientNonces, TargetClient,
@@ -28,13 +29,9 @@ use crate::metrics::MessageLaneLoopMetrics;
 use async_trait::async_trait;
 use bp_messages::{MessageNonce, UnrewardedRelayersState, Weight};
 use futures::stream::FusedStream;
+use num_traits::{SaturatingAdd, Zero};
 use relay_utils::FailedClient;
-use std::{
-	collections::{BTreeMap, VecDeque},
-	marker::PhantomData,
-	ops::RangeInclusive,
-	time::Duration,
-};
+use std::{collections::VecDeque, marker::PhantomData, ops::RangeInclusive, time::Duration};
 
 /// Run message delivery race.
 pub async fn run<P: MessageLane>(
@@ -48,24 +45,27 @@ pub async fn run<P: MessageLane>(
 ) -> Result<(), FailedClient> {
 	crate::message_race_loop::run(
 		MessageDeliveryRaceSource {
-			client: source_client,
+			client: source_client.clone(),
 			metrics_msg: metrics_msg.clone(),
 			_phantom: Default::default(),
 		},
 		source_state_updates,
 		MessageDeliveryRaceTarget {
-			client: target_client,
+			client: target_client.clone(),
 			metrics_msg,
 			_phantom: Default::default(),
 		},
 		target_state_updates,
 		stall_timeout,
-		MessageDeliveryStrategy::<P> {
+		MessageDeliveryStrategy::<P, _, _> {
+			lane_source_client: source_client,
+			lane_target_client: target_client,
 			max_unrewarded_relayer_entries_at_target: params.max_unrewarded_relayer_entries_at_target,
 			max_unconfirmed_nonces_at_target: params.max_unconfirmed_nonces_at_target,
 			max_messages_in_single_batch: params.max_messages_in_single_batch,
 			max_messages_weight_in_single_batch: params.max_messages_weight_in_single_batch,
 			max_messages_size_in_single_batch: params.max_messages_size_in_single_batch,
+			relayer_mode: params.relayer_mode,
 			latest_confirmed_nonces_at_source: VecDeque::new(),
 			target_nonces: None,
 			strategy: BasicStrategy::new(),
@@ -107,7 +107,7 @@ where
 	C: MessageLaneSourceClient<P>,
 {
 	type Error = C::Error;
-	type NoncesRange = MessageWeightsMap;
+	type NoncesRange = MessageDetailsMap<P::SourceChainBalance>;
 	type ProofParameters = MessageProofParameters;
 
 	async fn nonces(
@@ -125,10 +125,10 @@ where
 
 		let new_nonces = if latest_generated_nonce > prev_latest_nonce {
 			self.client
-				.generated_messages_weights(at_block.clone(), prev_latest_nonce + 1..=latest_generated_nonce)
+				.generated_message_details(at_block.clone(), prev_latest_nonce + 1..=latest_generated_nonce)
 				.await?
 		} else {
-			MessageWeightsMap::new()
+			MessageDetailsMap::new()
 		};
 
 		Ok((
@@ -222,7 +222,11 @@ struct DeliveryRaceTargetNoncesData {
 }
 
 /// Messages delivery strategy.
-struct MessageDeliveryStrategy<P: MessageLane> {
+struct MessageDeliveryStrategy<P: MessageLane, SC, TC> {
+	/// The client that is connected to the message lane source node.
+	lane_source_client: SC,
+	/// The client that is connected to the message lane target node.
+	lane_target_client: TC,
 	/// Maximal unrewarded relayer entries at target client.
 	max_unrewarded_relayer_entries_at_target: MessageNonce,
 	/// Maximal unconfirmed nonces at target client.
@@ -232,7 +236,9 @@ struct MessageDeliveryStrategy<P: MessageLane> {
 	/// Maximal cumulative messages weight in the single delivery transaction.
 	max_messages_weight_in_single_batch: Weight,
 	/// Maximal messages size in the single delivery transaction.
-	max_messages_size_in_single_batch: usize,
+	max_messages_size_in_single_batch: u32,
+	/// Relayer operating mode.
+	relayer_mode: RelayerMode,
 	/// Latest confirmed nonces at the source client + the header id where we have first met this nonce.
 	latest_confirmed_nonces_at_source: VecDeque<(SourceHeaderIdOf<P>, MessageNonce)>,
 	/// Target nonces from the source client.
@@ -246,11 +252,11 @@ type MessageDeliveryStrategyBase<P> = BasicStrategy<
 	<P as MessageLane>::SourceHeaderHash,
 	<P as MessageLane>::TargetHeaderNumber,
 	<P as MessageLane>::TargetHeaderHash,
-	MessageWeightsMap,
+	MessageDetailsMap<<P as MessageLane>::SourceChainBalance>,
 	<P as MessageLane>::MessagesProof,
 >;
 
-impl<P: MessageLane> std::fmt::Debug for MessageDeliveryStrategy<P> {
+impl<P: MessageLane, SC, TC> std::fmt::Debug for MessageDeliveryStrategy<P, SC, TC> {
 	fn fmt(&self, fmt: &mut std::fmt::Formatter) -> std::fmt::Result {
 		fmt.debug_struct("MessageDeliveryStrategy")
 			.field(
@@ -280,10 +286,26 @@ impl<P: MessageLane> std::fmt::Debug for MessageDeliveryStrategy<P> {
 	}
 }
 
-impl<P: MessageLane> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::MessagesProof>
-	for MessageDeliveryStrategy<P>
+impl<P: MessageLane, SC, TC> MessageDeliveryStrategy<P, SC, TC> {
+	/// Returns total weight of all undelivered messages.
+	fn total_queued_dispatch_weight(&self) -> Weight {
+		self.strategy
+			.source_queue()
+			.iter()
+			.flat_map(|(_, range)| range.values().map(|details| details.dispatch_weight))
+			.fold(0, |total, weight| total.saturating_add(weight))
+	}
+}
+
+#[async_trait]
+impl<P, SC, TC> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::MessagesProof>
+	for MessageDeliveryStrategy<P, SC, TC>
+where
+	P: MessageLane,
+	SC: MessageLaneSourceClient<P>,
+	TC: MessageLaneTargetClient<P>,
 {
-	type SourceNoncesRange = MessageWeightsMap;
+	type SourceNoncesRange = MessageDetailsMap<P::SourceChainBalance>;
 	type ProofParameters = MessageProofParameters;
 	type TargetNoncesData = DeliveryRaceTargetNoncesData;
 
@@ -383,9 +405,9 @@ impl<P: MessageLane> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::M
 		)
 	}
 
-	fn select_nonces_to_deliver(
+	async fn select_nonces_to_deliver(
 		&mut self,
-		race_state: &RaceState<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::MessagesProof>,
+		race_state: RaceState<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::MessagesProof>,
 	) -> Option<(RangeInclusive<MessageNonce>, Self::ProofParameters)> {
 		let best_finalized_source_header_id_at_best_target =
 			race_state.best_finalized_source_header_id_at_best_target.clone()?;
@@ -473,87 +495,205 @@ impl<P: MessageLane> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::M
 		let max_nonces = std::cmp::min(max_nonces, self.max_messages_in_single_batch);
 		let max_messages_weight_in_single_batch = self.max_messages_weight_in_single_batch;
 		let max_messages_size_in_single_batch = self.max_messages_size_in_single_batch;
-		let mut selected_weight: Weight = 0;
-		let mut selected_size: usize = 0;
-		let mut selected_count: MessageNonce = 0;
+		let relayer_mode = self.relayer_mode;
+		let lane_source_client = self.lane_source_client.clone();
+		let lane_target_client = self.lane_target_client.clone();
 
+		let previous_total_dispatch_weight = self.total_queued_dispatch_weight();
 		let selected_nonces = self
 			.strategy
-			.select_nonces_to_deliver_with_selector(race_state, |range| {
-				let to_requeue = range
-					.into_iter()
-					.skip_while(|(_, weight)| {
-						// Since we (hopefully) have some reserves in `max_messages_weight_in_single_batch`
-						// and `max_messages_size_in_single_batch`, we may still try to submit transaction
-						// with single message if message overflows these limits. The worst case would be if
-						// transaction will be rejected by the target runtime, but at least we have tried.
-
-						// limit messages in the batch by weight
-						let new_selected_weight = match selected_weight.checked_add(weight.weight) {
-							Some(new_selected_weight) if new_selected_weight <= max_messages_weight_in_single_batch => {
-								new_selected_weight
-							}
-							new_selected_weight if selected_count == 0 => {
-								log::warn!(
-									target: "bridge",
-									"Going to submit message delivery transaction with declared dispatch \
-									weight {:?} that overflows maximal configured weight {}",
-									new_selected_weight,
-									max_messages_weight_in_single_batch,
-								);
-								new_selected_weight.unwrap_or(Weight::MAX)
-							}
-							_ => return false,
-						};
-
-						// limit messages in the batch by size
-						let new_selected_size = match selected_size.checked_add(weight.size) {
-							Some(new_selected_size) if new_selected_size <= max_messages_size_in_single_batch => {
-								new_selected_size
-							}
-							new_selected_size if selected_count == 0 => {
-								log::warn!(
-									target: "bridge",
-									"Going to submit message delivery transaction with message \
-									size {:?} that overflows maximal configured size {}",
-									new_selected_size,
-									max_messages_size_in_single_batch,
-								);
-								new_selected_size.unwrap_or(usize::MAX)
-							}
-							_ => return false,
-						};
-
-						// limit number of messages in the batch
-						let new_selected_count = selected_count + 1;
-						if new_selected_count > max_nonces {
-							return false;
-						}
-
-						selected_weight = new_selected_weight;
-						selected_size = new_selected_size;
-						selected_count = new_selected_count;
-						true
-					})
-					.collect::<BTreeMap<_, _>>();
-				if to_requeue.is_empty() {
-					None
-				} else {
-					Some(to_requeue)
-				}
-			})?;
+			.select_nonces_to_deliver_with_selector(race_state.clone(), |range| async {
+				select_nonces_for_delivery_transaction(
+					relayer_mode,
+					max_nonces,
+					max_messages_weight_in_single_batch,
+					max_messages_size_in_single_batch,
+					lane_source_client.clone(),
+					lane_target_client.clone(),
+					range,
+				)
+				.await
+			})
+			.await?;
+		let new_total_dispatch_weight = self.total_queued_dispatch_weight();
+		let dispatch_weight = previous_total_dispatch_weight - new_total_dispatch_weight;
 
 		Some((
 			selected_nonces,
 			MessageProofParameters {
 				outbound_state_proof_required,
-				dispatch_weight: selected_weight,
+				dispatch_weight,
 			},
 		))
 	}
 }
 
-impl NoncesRange for MessageWeightsMap {
+/// From given set of source nonces, that are ready to be delivered, select nonces
+/// to fit into single delivery transaction.
+///
+/// The function returns nonces that are NOT selected for current batch and will be
+/// delivered later.
+async fn select_nonces_for_delivery_transaction<P: MessageLane>(
+	relayer_mode: RelayerMode,
+	max_messages_in_this_batch: MessageNonce,
+	max_messages_weight_in_single_batch: Weight,
+	max_messages_size_in_single_batch: u32,
+	lane_source_client: impl MessageLaneSourceClient<P>,
+	lane_target_client: impl MessageLaneTargetClient<P>,
+	ready_nonces: MessageDetailsMap<P::SourceChainBalance>,
+) -> Option<MessageDetailsMap<P::SourceChainBalance>> {
+	let mut hard_selected_count = 0;
+	let mut soft_selected_count = 0;
+
+	let mut selected_weight: Weight = 0;
+	let mut selected_size: u32 = 0;
+	let mut selected_count: MessageNonce = 0;
+
+	let mut total_reward = P::SourceChainBalance::zero();
+	let mut total_confirmations_cost = P::SourceChainBalance::zero();
+	let mut total_cost = P::SourceChainBalance::zero();
+
+	// technically, multiple confirmations will be delivered in a single transaction,
+	// meaning less loses for relayer. But here we don't know the final relayer yet, so
+	// we're adding a separate transaction for every message. Normally, this cost is covered
+	// by the message sender. Probably reconsider this?
+	let confirmation_transaction_cost = if relayer_mode != RelayerMode::Altruistic {
+		lane_source_client.estimate_confirmation_transaction().await
+	} else {
+		Zero::zero()
+	};
+
+	for (index, (nonce, details)) in ready_nonces.iter().enumerate() {
+		// Since we (hopefully) have some reserves in `max_messages_weight_in_single_batch`
+		// and `max_messages_size_in_single_batch`, we may still try to submit transaction
+		// with single message if message overflows these limits. The worst case would be if
+		// transaction will be rejected by the target runtime, but at least we have tried.
+
+		// limit messages in the batch by weight
+		let new_selected_weight = match selected_weight.checked_add(details.dispatch_weight) {
+			Some(new_selected_weight) if new_selected_weight <= max_messages_weight_in_single_batch => {
+				new_selected_weight
+			}
+			new_selected_weight if selected_count == 0 => {
+				log::warn!(
+					target: "bridge",
+					"Going to submit message delivery transaction with declared dispatch \
+					weight {:?} that overflows maximal configured weight {}",
+					new_selected_weight,
+					max_messages_weight_in_single_batch,
+				);
+				new_selected_weight.unwrap_or(Weight::MAX)
+			}
+			_ => break,
+		};
+
+		// limit messages in the batch by size
+		let new_selected_size = match selected_size.checked_add(details.size) {
+			Some(new_selected_size) if new_selected_size <= max_messages_size_in_single_batch => new_selected_size,
+			new_selected_size if selected_count == 0 => {
+				log::warn!(
+					target: "bridge",
+					"Going to submit message delivery transaction with message \
+					size {:?} that overflows maximal configured size {}",
+					new_selected_size,
+					max_messages_size_in_single_batch,
+				);
+				new_selected_size.unwrap_or(u32::MAX)
+			}
+			_ => break,
+		};
+
+		// limit number of messages in the batch
+		let new_selected_count = selected_count + 1;
+		if new_selected_count > max_messages_in_this_batch {
+			break;
+		}
+
+		// now the message has passed all 'strong' checks, and we CAN deliver it. But do we WANT
+		// to deliver it? It depends on the relayer strategy.
+		match relayer_mode {
+			RelayerMode::Altruistic => {
+				soft_selected_count = index + 1;
+			}
+			RelayerMode::NoLosses => {
+				let delivery_transaction_cost = lane_target_client
+					.estimate_delivery_transaction_in_source_tokens(
+						0..=(new_selected_count as MessageNonce - 1),
+						new_selected_weight,
+						new_selected_size as u32,
+					)
+					.await;
+
+				// if it is the first message that makes reward less than cost, let's log it
+				// if this message makes batch profitable again, let's log it
+				let is_total_reward_less_than_cost = total_reward < total_cost;
+				let prev_total_cost = total_cost;
+				let prev_total_reward = total_reward;
+				total_confirmations_cost = total_confirmations_cost.saturating_add(&confirmation_transaction_cost);
+				total_reward = total_reward.saturating_add(&details.reward);
+				total_cost = total_confirmations_cost.saturating_add(&delivery_transaction_cost);
+				if !is_total_reward_less_than_cost && total_reward < total_cost {
+					log::debug!(
+						target: "bridge",
+						"Message with nonce {} (reward = {:?}) changes total cost {:?}->{:?} and makes it larger than \
+						total reward {:?}->{:?}",
+						nonce,
+						details.reward,
+						prev_total_cost,
+						total_cost,
+						prev_total_reward,
+						total_reward,
+					);
+				} else if is_total_reward_less_than_cost && total_reward >= total_cost {
+					log::debug!(
+						target: "bridge",
+						"Message with nonce {} (reward = {:?}) changes total cost {:?}->{:?} and makes it less than or \
+						equal to the total reward {:?}->{:?} (again)",
+						nonce,
+						details.reward,
+						prev_total_cost,
+						total_cost,
+						prev_total_reward,
+						total_reward,
+					);
+				}
+
+				// NoLosses relayer never want to lose his funds
+				if total_reward >= total_cost {
+					soft_selected_count = index + 1;
+				}
+			}
+		}
+
+		hard_selected_count = index + 1;
+		selected_weight = new_selected_weight;
+		selected_size = new_selected_size;
+		selected_count = new_selected_count;
+	}
+
+	if hard_selected_count != soft_selected_count {
+		log::warn!(
+			target: "bridge",
+			"Relayer may deliver nonces [{:?}; {:?}], but because of its strategy ({:?}) it has selected \
+			nonces [{:?}; {:?}].",
+			ready_nonces.keys().next(),
+			ready_nonces.keys().next().map(|begin| begin + (hard_selected_count as MessageNonce) - 1),
+			relayer_mode,
+			ready_nonces.keys().next(),
+			ready_nonces.keys().next().map(|begin| begin + (soft_selected_count as MessageNonce) - 1),
+
+		);
+
+		hard_selected_count = soft_selected_count;
+	}
+	if hard_selected_count != ready_nonces.len() {
+		Some(ready_nonces.into_iter().skip(hard_selected_count).collect())
+	} else {
+		None
+	}
+}
+
+impl<SourceChainBalance: std::fmt::Debug> NoncesRange for MessageDetailsMap<SourceChainBalance> {
 	fn begin(&self) -> MessageNonce {
 		self.keys().next().cloned().unwrap_or_default()
 	}
@@ -576,12 +716,42 @@ impl NoncesRange for MessageWeightsMap {
 mod tests {
 	use super::*;
 	use crate::message_lane_loop::{
-		tests::{header_id, TestMessageLane, TestMessagesProof, TestSourceHeaderId, TestTargetHeaderId},
-		MessageWeights,
+		tests::{
+			header_id, TestMessageLane, TestMessagesProof, TestSourceChainBalance, TestSourceClient,
+			TestSourceHeaderId, TestTargetClient, TestTargetHeaderId, CONFIRMATION_TRANSACTION_COST,
+			DELIVERY_TRANSACTION_COST,
+		},
+		MessageDetails,
 	};
 
+	const DEFAULT_REWARD: TestSourceChainBalance = CONFIRMATION_TRANSACTION_COST + DELIVERY_TRANSACTION_COST;
+
 	type TestRaceState = RaceState<TestSourceHeaderId, TestTargetHeaderId, TestMessagesProof>;
-	type TestStrategy = MessageDeliveryStrategy<TestMessageLane>;
+	type TestStrategy = MessageDeliveryStrategy<TestMessageLane, TestSourceClient, TestTargetClient>;
+
+	fn source_nonces(
+		new_nonces: RangeInclusive<MessageNonce>,
+		confirmed_nonce: MessageNonce,
+		reward: TestSourceChainBalance,
+	) -> SourceClientNonces<MessageDetailsMap<TestSourceChainBalance>> {
+		SourceClientNonces {
+			new_nonces: new_nonces
+				.into_iter()
+				.map(|nonce| {
+					(
+						nonce,
+						MessageDetails {
+							dispatch_weight: 1,
+							size: 1,
+							reward,
+						},
+					)
+				})
+				.into_iter()
+				.collect(),
+			confirmed_nonce: Some(confirmed_nonce),
+		}
+	}
 
 	fn prepare_strategy() -> (TestRaceState, TestStrategy) {
 		let mut race_state = RaceState {
@@ -594,12 +764,15 @@ mod tests {
 		};
 
 		let mut race_strategy = TestStrategy {
+			relayer_mode: RelayerMode::Altruistic,
 			max_unrewarded_relayer_entries_at_target: 4,
 			max_unconfirmed_nonces_at_target: 4,
 			max_messages_in_single_batch: 4,
 			max_messages_weight_in_single_batch: 4,
 			max_messages_size_in_single_batch: 4,
 			latest_confirmed_nonces_at_source: vec![(header_id(1), 19)].into_iter().collect(),
+			lane_source_client: TestSourceClient::default(),
+			lane_target_client: TestTargetClient::default(),
 			target_nonces: Some(TargetClientNonces {
 				latest_nonce: 19,
 				nonces_data: DeliveryRaceTargetNoncesData {
@@ -614,20 +787,9 @@ mod tests {
 			strategy: BasicStrategy::new(),
 		};
 
-		race_strategy.strategy.source_nonces_updated(
-			header_id(1),
-			SourceClientNonces {
-				new_nonces: vec![
-					(20, MessageWeights { weight: 1, size: 1 }),
-					(21, MessageWeights { weight: 1, size: 1 }),
-					(22, MessageWeights { weight: 1, size: 1 }),
-					(23, MessageWeights { weight: 1, size: 1 }),
-				]
-				.into_iter()
-				.collect(),
-				confirmed_nonce: Some(19),
-			},
-		);
+		race_strategy
+			.strategy
+			.source_nonces_updated(header_id(1), source_nonces(20..=23, 19, DEFAULT_REWARD));
 
 		let target_nonces = TargetClientNonces {
 			latest_nonce: 19,
@@ -652,14 +814,15 @@ mod tests {
 
 	#[test]
 	fn weights_map_works_as_nonces_range() {
-		fn build_map(range: RangeInclusive<MessageNonce>) -> MessageWeightsMap {
+		fn build_map(range: RangeInclusive<MessageNonce>) -> MessageDetailsMap<TestSourceChainBalance> {
 			range
 				.map(|idx| {
 					(
 						idx,
-						MessageWeights {
-							weight: idx,
+						MessageDetails {
+							dispatch_weight: idx,
 							size: idx as _,
+							reward: idx as _,
 						},
 					)
 				})
@@ -678,19 +841,19 @@ mod tests {
 		assert_eq!(map.greater_than(30), None);
 	}
 
-	#[test]
-	fn message_delivery_strategy_selects_messages_to_deliver() {
+	#[async_std::test]
+	async fn message_delivery_strategy_selects_messages_to_deliver() {
 		let (state, mut strategy) = prepare_strategy();
 
 		// both sides are ready to relay new messages
 		assert_eq!(
-			strategy.select_nonces_to_deliver(&state),
+			strategy.select_nonces_to_deliver(state).await,
 			Some(((20..=23), proof_parameters(false, 4)))
 		);
 	}
 
-	#[test]
-	fn message_delivery_strategy_selects_nothing_if_too_many_confirmations_missing() {
+	#[async_std::test]
+	async fn message_delivery_strategy_selects_nothing_if_too_many_confirmations_missing() {
 		let (state, mut strategy) = prepare_strategy();
 
 		// if there are already `max_unconfirmed_nonces_at_target` messages on target,
@@ -701,11 +864,11 @@ mod tests {
 		)]
 		.into_iter()
 		.collect();
-		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
+		assert_eq!(strategy.select_nonces_to_deliver(state).await, None);
 	}
 
-	#[test]
-	fn message_delivery_strategy_includes_outbound_state_proof_when_new_nonces_are_available() {
+	#[async_std::test]
+	async fn message_delivery_strategy_includes_outbound_state_proof_when_new_nonces_are_available() {
 		let (state, mut strategy) = prepare_strategy();
 
 		// if there are new confirmed nonces on source, we want to relay this information
@@ -713,13 +876,13 @@ mod tests {
 		let prev_confirmed_nonce_at_source = strategy.latest_confirmed_nonces_at_source.back().unwrap().1;
 		strategy.target_nonces.as_mut().unwrap().nonces_data.confirmed_nonce = prev_confirmed_nonce_at_source - 1;
 		assert_eq!(
-			strategy.select_nonces_to_deliver(&state),
+			strategy.select_nonces_to_deliver(state).await,
 			Some(((20..=23), proof_parameters(true, 4)))
 		);
 	}
 
-	#[test]
-	fn message_delivery_strategy_selects_nothing_if_there_are_too_many_unrewarded_relayers() {
+	#[async_std::test]
+	async fn message_delivery_strategy_selects_nothing_if_there_are_too_many_unrewarded_relayers() {
 		let (state, mut strategy) = prepare_strategy();
 
 		// if there are already `max_unrewarded_relayer_entries_at_target` entries at target,
@@ -729,11 +892,12 @@ mod tests {
 			unrewarded_relayers.unrewarded_relayer_entries = strategy.max_unrewarded_relayer_entries_at_target;
 			unrewarded_relayers.messages_in_oldest_entry = 4;
 		}
-		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
+		assert_eq!(strategy.select_nonces_to_deliver(state).await, None);
 	}
 
-	#[test]
-	fn message_delivery_strategy_selects_nothing_if_proved_rewards_is_not_enough_to_remove_oldest_unrewarded_entry() {
+	#[async_std::test]
+	async fn message_delivery_strategy_selects_nothing_if_proved_rewards_is_not_enough_to_remove_oldest_unrewarded_entry(
+	) {
 		let (state, mut strategy) = prepare_strategy();
 
 		// if there are already `max_unrewarded_relayer_entries_at_target` entries at target,
@@ -746,11 +910,11 @@ mod tests {
 			unrewarded_relayers.unrewarded_relayer_entries = strategy.max_unrewarded_relayer_entries_at_target;
 			unrewarded_relayers.messages_in_oldest_entry = 4;
 		}
-		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
+		assert_eq!(strategy.select_nonces_to_deliver(state).await, None);
 	}
 
-	#[test]
-	fn message_delivery_strategy_includes_outbound_state_proof_if_proved_rewards_is_enough() {
+	#[async_std::test]
+	async fn message_delivery_strategy_includes_outbound_state_proof_if_proved_rewards_is_enough() {
 		let (state, mut strategy) = prepare_strategy();
 
 		// if there are already `max_unrewarded_relayer_entries_at_target` entries at target,
@@ -764,73 +928,77 @@ mod tests {
 			unrewarded_relayers.messages_in_oldest_entry = 3;
 		}
 		assert_eq!(
-			strategy.select_nonces_to_deliver(&state),
+			strategy.select_nonces_to_deliver(state).await,
 			Some(((20..=23), proof_parameters(true, 4)))
 		);
 	}
 
-	#[test]
-	fn message_delivery_strategy_limits_batch_by_messages_weight() {
+	#[async_std::test]
+	async fn message_delivery_strategy_limits_batch_by_messages_weight() {
 		let (state, mut strategy) = prepare_strategy();
 
 		// not all queued messages may fit in the batch, because batch has max weight
 		strategy.max_messages_weight_in_single_batch = 3;
 		assert_eq!(
-			strategy.select_nonces_to_deliver(&state),
+			strategy.select_nonces_to_deliver(state).await,
 			Some(((20..=22), proof_parameters(false, 3)))
 		);
 	}
 
-	#[test]
-	fn message_delivery_strategy_accepts_single_message_even_if_its_weight_overflows_maximal_weight() {
+	#[async_std::test]
+	async fn message_delivery_strategy_accepts_single_message_even_if_its_weight_overflows_maximal_weight() {
 		let (state, mut strategy) = prepare_strategy();
 
 		// first message doesn't fit in the batch, because it has weight (10) that overflows max weight (4)
-		strategy.strategy.source_queue_mut()[0].1.get_mut(&20).unwrap().weight = 10;
+		strategy.strategy.source_queue_mut()[0]
+			.1
+			.get_mut(&20)
+			.unwrap()
+			.dispatch_weight = 10;
 		assert_eq!(
-			strategy.select_nonces_to_deliver(&state),
+			strategy.select_nonces_to_deliver(state).await,
 			Some(((20..=20), proof_parameters(false, 10)))
 		);
 	}
 
-	#[test]
-	fn message_delivery_strategy_limits_batch_by_messages_size() {
+	#[async_std::test]
+	async fn message_delivery_strategy_limits_batch_by_messages_size() {
 		let (state, mut strategy) = prepare_strategy();
 
 		// not all queued messages may fit in the batch, because batch has max weight
 		strategy.max_messages_size_in_single_batch = 3;
 		assert_eq!(
-			strategy.select_nonces_to_deliver(&state),
+			strategy.select_nonces_to_deliver(state).await,
 			Some(((20..=22), proof_parameters(false, 3)))
 		);
 	}
 
-	#[test]
-	fn message_delivery_strategy_accepts_single_message_even_if_its_weight_overflows_maximal_size() {
+	#[async_std::test]
+	async fn message_delivery_strategy_accepts_single_message_even_if_its_weight_overflows_maximal_size() {
 		let (state, mut strategy) = prepare_strategy();
 
 		// first message doesn't fit in the batch, because it has weight (10) that overflows max weight (4)
 		strategy.strategy.source_queue_mut()[0].1.get_mut(&20).unwrap().size = 10;
 		assert_eq!(
-			strategy.select_nonces_to_deliver(&state),
+			strategy.select_nonces_to_deliver(state).await,
 			Some(((20..=20), proof_parameters(false, 1)))
 		);
 	}
 
-	#[test]
-	fn message_delivery_strategy_limits_batch_by_messages_count_when_there_is_upper_limit() {
+	#[async_std::test]
+	async fn message_delivery_strategy_limits_batch_by_messages_count_when_there_is_upper_limit() {
 		let (state, mut strategy) = prepare_strategy();
 
 		// not all queued messages may fit in the batch, because batch has max number of messages limit
 		strategy.max_messages_in_single_batch = 3;
 		assert_eq!(
-			strategy.select_nonces_to_deliver(&state),
+			strategy.select_nonces_to_deliver(state).await,
 			Some(((20..=22), proof_parameters(false, 3)))
 		);
 	}
 
-	#[test]
-	fn message_delivery_strategy_limits_batch_by_messages_count_when_there_are_unconfirmed_nonces() {
+	#[async_std::test]
+	async fn message_delivery_strategy_limits_batch_by_messages_count_when_there_are_unconfirmed_nonces() {
 		let (state, mut strategy) = prepare_strategy();
 
 		// 1 delivery confirmation from target to source is still missing, so we may only
@@ -841,13 +1009,13 @@ mod tests {
 			.collect();
 		strategy.target_nonces.as_mut().unwrap().nonces_data.confirmed_nonce = prev_confirmed_nonce_at_source - 1;
 		assert_eq!(
-			strategy.select_nonces_to_deliver(&state),
+			strategy.select_nonces_to_deliver(state).await,
 			Some(((20..=22), proof_parameters(false, 3)))
 		);
 	}
 
-	#[test]
-	fn message_delivery_strategy_waits_for_confirmed_nonce_header_to_appear_on_target() {
+	#[async_std::test]
+	async fn message_delivery_strategy_waits_for_confirmed_nonce_header_to_appear_on_target() {
 		// 1 delivery confirmation from target to source is still missing, so we may deliver
 		// reward confirmation with our message delivery transaction. But the problem is that
 		// the reward has been paid at header 2 && this header is still unknown to target node.
@@ -864,7 +1032,7 @@ mod tests {
 		strategy.target_nonces.as_mut().unwrap().nonces_data.confirmed_nonce = prev_confirmed_nonce_at_source - 1;
 		state.best_finalized_source_header_id_at_best_target = Some(header_id(1));
 		assert_eq!(
-			strategy.select_nonces_to_deliver(&state),
+			strategy.select_nonces_to_deliver(state).await,
 			Some(((20..=22), proof_parameters(false, 3)))
 		);
 
@@ -881,13 +1049,13 @@ mod tests {
 		state.best_finalized_source_header_id_at_source = Some(header_id(2));
 		state.best_finalized_source_header_id_at_best_target = Some(header_id(2));
 		assert_eq!(
-			strategy.select_nonces_to_deliver(&state),
+			strategy.select_nonces_to_deliver(state).await,
 			Some(((20..=23), proof_parameters(true, 4)))
 		);
 	}
 
-	#[test]
-	fn source_header_is_requied_when_confirmations_are_required() {
+	#[async_std::test]
+	async fn source_header_is_required_when_confirmations_are_required() {
 		// let's prepare situation when:
 		// - all messages [20; 23] have been generated at source block#1;
 		let (mut state, mut strategy) = prepare_strategy();
@@ -895,7 +1063,7 @@ mod tests {
 		//   relayers vector capacity;
 		strategy.max_unconfirmed_nonces_at_target = 2;
 		assert_eq!(
-			strategy.select_nonces_to_deliver(&state),
+			strategy.select_nonces_to_deliver(state.clone()).await,
 			Some(((20..=21), proof_parameters(false, 2)))
 		);
 		strategy.finalized_target_nonces_updated(
@@ -912,12 +1080,12 @@ mod tests {
 			},
 			&mut state,
 		);
-		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
+		assert_eq!(strategy.select_nonces_to_deliver(state).await, None);
 		// - messages [1; 10] receiving confirmation has been delivered at source block#2;
 		strategy.source_nonces_updated(
 			header_id(2),
 			SourceClientNonces {
-				new_nonces: BTreeMap::new(),
+				new_nonces: MessageDetailsMap::new(),
 				confirmed_nonce: Some(21),
 			},
 		);
@@ -927,4 +1095,36 @@ mod tests {
 			Some(header_id(2))
 		);
 	}
+
+	#[async_std::test]
+	async fn no_losses_relayer_is_delivering_messages_if_cost_is_equal_to_reward() {
+		let (state, mut strategy) = prepare_strategy();
+		strategy.relayer_mode = RelayerMode::NoLosses;
+
+		// so now we have:
+		// - 20..=23 with reward = cost
+		// => strategy shall select all 20..=23
+		assert_eq!(
+			strategy.select_nonces_to_deliver(state).await,
+			Some(((20..=23), proof_parameters(false, 4)))
+		);
+	}
+
+	#[async_std::test]
+	async fn no_losses_relayer_is_not_delivering_messages_if_cost_is_larger_than_reward() {
+		let (mut state, mut strategy) = prepare_strategy();
+		let nonces = source_nonces(24..=25, 19, DEFAULT_REWARD - DELIVERY_TRANSACTION_COST);
+		strategy.strategy.source_nonces_updated(header_id(2), nonces);
+		state.best_finalized_source_header_id_at_best_target = Some(header_id(2));
+		strategy.relayer_mode = RelayerMode::NoLosses;
+
+		// so now we have:
+		// - 20..=23 with reward = cost
+		// - 24..=25 with reward less than cost
+		// => strategy shall only select 20..=23
+		assert_eq!(
+			strategy.select_nonces_to_deliver(state).await,
+			Some(((20..=23), proof_parameters(false, 4)))
+		);
+	}
 }

bridges/relays/messages/src/message_race_loop.rs

@@ -143,6 +143,7 @@ pub trait TargetClient<P: MessageRace> {
 }
 
 /// Race strategy.
+#[async_trait]
 pub trait RaceStrategy<SourceHeaderId, TargetHeaderId, Proof>: Debug {
 	/// Type of nonces range expected from the source client.
 	type SourceNoncesRange: NoncesRange;
@@ -182,14 +183,14 @@ pub trait RaceStrategy<SourceHeaderId, TargetHeaderId, Proof>: Debug {
 	/// Should return `Some(nonces)` if we need to deliver proof of `nonces` (and associated
 	/// data) from source to target node.
 	/// Additionally, parameters required to generate proof are returned.
-	fn select_nonces_to_deliver(
+	async fn select_nonces_to_deliver(
 		&mut self,
-		race_state: &RaceState<SourceHeaderId, TargetHeaderId, Proof>,
+		race_state: RaceState<SourceHeaderId, TargetHeaderId, Proof>,
 	) -> Option<(RangeInclusive<MessageNonce>, Self::ProofParameters)>;
 }
 
 /// State of the race.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct RaceState<SourceHeaderId, TargetHeaderId, Proof> {
 	/// Best finalized source header id at the source client.
 	pub best_finalized_source_header_id_at_source: Option<SourceHeaderId>,
@@ -438,7 +439,7 @@ pub async fn run<P: MessageRace, SC: SourceClient<P>, TC: TargetClient<P>>(
 		if source_client_is_online {
 			source_client_is_online = false;
 
-			let nonces_to_deliver = select_nonces_to_deliver(&race_state, &mut strategy);
+			let nonces_to_deliver = select_nonces_to_deliver(race_state.clone(), &mut strategy).await;
 			let best_at_source = strategy.best_at_source();
 
 			if let Some((at_block, nonces_range, proof_parameters)) = nonces_to_deliver {
@@ -554,27 +555,25 @@ where
 	now_time
 }
 
-fn select_nonces_to_deliver<SourceHeaderId, TargetHeaderId, Proof, Strategy>(
-	race_state: &RaceState<SourceHeaderId, TargetHeaderId, Proof>,
+async fn select_nonces_to_deliver<SourceHeaderId, TargetHeaderId, Proof, Strategy>(
+	race_state: RaceState<SourceHeaderId, TargetHeaderId, Proof>,
 	strategy: &mut Strategy,
 ) -> Option<(SourceHeaderId, RangeInclusive<MessageNonce>, Strategy::ProofParameters)>
 where
 	SourceHeaderId: Clone,
 	Strategy: RaceStrategy<SourceHeaderId, TargetHeaderId, Proof>,
 {
-	race_state
-		.best_finalized_source_header_id_at_best_target
-		.as_ref()
-		.and_then(|best_finalized_source_header_id_at_best_target| {
-			strategy
-				.select_nonces_to_deliver(race_state)
-				.map(|(nonces_range, proof_parameters)| {
-					(
-						best_finalized_source_header_id_at_best_target.clone(),
-						nonces_range,
-						proof_parameters,
-					)
-				})
+	let best_finalized_source_header_id_at_best_target =
+		race_state.best_finalized_source_header_id_at_best_target.clone()?;
+	strategy
+		.select_nonces_to_deliver(race_state)
+		.await
+		.map(|(nonces_range, proof_parameters)| {
+			(
+				best_finalized_source_header_id_at_best_target,
+				nonces_range,
+				proof_parameters,
+			)
 		})
 }
 
@@ -584,8 +583,8 @@ mod tests {
 	use crate::message_race_strategy::BasicStrategy;
 	use relay_utils::HeaderId;
 
-	#[test]
-	fn proof_is_generated_at_best_block_known_to_target_node() {
+	#[async_std::test]
+	async fn proof_is_generated_at_best_block_known_to_target_node() {
 		const GENERATED_AT: u64 = 6;
 		const BEST_AT_SOURCE: u64 = 10;
 		const BEST_AT_TARGET: u64 = 8;
@@ -620,7 +619,7 @@ mod tests {
 
 		// the proof will be generated on source, but using BEST_AT_TARGET block
 		assert_eq!(
-			select_nonces_to_deliver(&race_state, &mut strategy),
+			select_nonces_to_deliver(race_state, &mut strategy).await,
 			Some((HeaderId(BEST_AT_TARGET, BEST_AT_TARGET), 6..=10, (),))
 		);
 	}

bridges/relays/messages/src/message_race_strategy.rs

@@ -19,9 +19,10 @@
 
 use crate::message_race_loop::{NoncesRange, RaceState, RaceStrategy, SourceClientNonces, TargetClientNonces};
 
+use async_trait::async_trait;
 use bp_messages::MessageNonce;
 use relay_utils::HeaderId;
-use std::{collections::VecDeque, fmt::Debug, marker::PhantomData, ops::RangeInclusive};
+use std::{collections::VecDeque, fmt::Debug, future::Future, marker::PhantomData, ops::RangeInclusive};
 
 /// Nonces delivery strategy.
 #[derive(Debug)]
@@ -57,6 +58,13 @@ where
 		}
 	}
 
+	/// Reference to source queue.
+	pub(crate) fn source_queue(
+		&self,
+	) -> &VecDeque<(HeaderId<SourceHeaderHash, SourceHeaderNumber>, SourceNoncesRange)> {
+		&self.source_queue
+	}
+
 	/// Mutable reference to source queue to use in tests.
 	#[cfg(test)]
 	pub(crate) fn source_queue_mut(
@@ -73,14 +81,14 @@ where
 	/// right now, it should return `Some` with 'undeliverable' nonces. Please keep in mind that
 	/// this should be the sub-range that the passed range ends with, because nonces are always
 	/// delivered in-order. Otherwise the function will panic.
-	pub fn select_nonces_to_deliver_with_selector(
+	pub async fn select_nonces_to_deliver_with_selector<F: Future<Output = Option<SourceNoncesRange>>>(
 		&mut self,
-		race_state: &RaceState<
+		race_state: RaceState<
 			HeaderId<SourceHeaderHash, SourceHeaderNumber>,
 			HeaderId<TargetHeaderHash, TargetHeaderNumber>,
 			Proof,
 		>,
-		mut selector: impl FnMut(SourceNoncesRange) -> Option<SourceNoncesRange>,
+		selector: impl Fn(SourceNoncesRange) -> F,
 	) -> Option<RangeInclusive<MessageNonce>> {
 		// if we do not know best nonce at target node, we can't select anything
 		let target_nonce = self.best_target_nonce?;
@@ -99,7 +107,7 @@ where
 		// 2) we can't deliver new nonce until header, that has emitted this nonce, is finalized
 		// by target client
 		// 3) selector is used for more complicated logic
-		let best_header_at_target = &race_state.best_finalized_source_header_id_at_best_target.as_ref()?;
+		let best_header_at_target = race_state.best_finalized_source_header_id_at_best_target.clone()?;
 		let mut nonces_end = None;
 		while let Some((queued_at, queued_range)) = self.source_queue.pop_front() {
 			// select (sub) range to deliver
@@ -111,7 +119,7 @@ where
 				Some(queued_range)
 			} else {
 				// selector returns `Some(range)` if this `range` needs to be requeued
-				selector(queued_range)
+				selector(queued_range).await
 			};
 
 			// requeue (sub) range and update range to deliver
@@ -143,16 +151,17 @@ where
 	}
 }
 
+#[async_trait]
 impl<SourceHeaderNumber, SourceHeaderHash, TargetHeaderNumber, TargetHeaderHash, SourceNoncesRange, Proof>
 	RaceStrategy<HeaderId<SourceHeaderHash, SourceHeaderNumber>, HeaderId<TargetHeaderHash, TargetHeaderNumber>, Proof>
 	for BasicStrategy<SourceHeaderNumber, SourceHeaderHash, TargetHeaderNumber, TargetHeaderHash, SourceNoncesRange, Proof>
 where
-	SourceHeaderHash: Clone + Debug,
-	SourceHeaderNumber: Clone + Ord + Debug,
-	SourceNoncesRange: NoncesRange + Debug,
-	TargetHeaderHash: Debug,
-	TargetHeaderNumber: Debug,
-	Proof: Debug,
+	SourceHeaderHash: Clone + Debug + Send,
+	SourceHeaderNumber: Clone + Ord + Debug + Send,
+	SourceNoncesRange: NoncesRange + Debug + Send,
+	TargetHeaderHash: Debug + Send,
+	TargetHeaderNumber: Debug + Send,
+	Proof: Debug + Send,
 {
 	type SourceNoncesRange = SourceNoncesRange;
 	type ProofParameters = ();
@@ -271,15 +280,16 @@ where
 		));
 	}
 
-	fn select_nonces_to_deliver(
+	async fn select_nonces_to_deliver(
 		&mut self,
-		race_state: &RaceState<
+		race_state: RaceState<
 			HeaderId<SourceHeaderHash, SourceHeaderNumber>,
 			HeaderId<TargetHeaderHash, TargetHeaderNumber>,
 			Proof,
 		>,
 	) -> Option<(RangeInclusive<MessageNonce>, Self::ProofParameters)> {
-		self.select_nonces_to_deliver_with_selector(race_state, |_| None)
+		self.select_nonces_to_deliver_with_selector(race_state, |_| async { None })
+			.await
 			.map(|range| (range, ()))
 	}
 }
@@ -396,28 +406,28 @@ mod tests {
 		assert!(state.nonces_submitted.is_none());
 	}
 
-	#[test]
-	fn nothing_is_selected_if_something_is_already_selected() {
+	#[async_std::test]
+	async fn nothing_is_selected_if_something_is_already_selected() {
 		let mut state = RaceState::default();
 		let mut strategy = BasicStrategy::<TestMessageLane>::new();
 		state.nonces_to_submit = Some((header_id(1), 1..=10, (1..=10, None)));
 		strategy.best_target_nonces_updated(target_nonces(0), &mut state);
 		strategy.source_nonces_updated(header_id(1), source_nonces(1..=10));
-		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
+		assert_eq!(strategy.select_nonces_to_deliver(state.clone()).await, None);
 	}
 
-	#[test]
-	fn nothing_is_selected_if_something_is_already_submitted() {
+	#[async_std::test]
+	async fn nothing_is_selected_if_something_is_already_submitted() {
 		let mut state = RaceState::default();
 		let mut strategy = BasicStrategy::<TestMessageLane>::new();
 		state.nonces_submitted = Some(1..=10);
 		strategy.best_target_nonces_updated(target_nonces(0), &mut state);
 		strategy.source_nonces_updated(header_id(1), source_nonces(1..=10));
-		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
+		assert_eq!(strategy.select_nonces_to_deliver(state.clone()).await, None);
 	}
 
-	#[test]
-	fn select_nonces_to_deliver_works() {
+	#[async_std::test]
+	async fn select_nonces_to_deliver_works() {
 		let mut state = RaceState::<_, _, TestMessagesProof>::default();
 		let mut strategy = BasicStrategy::<TestMessageLane>::new();
 		strategy.best_target_nonces_updated(target_nonces(0), &mut state);
@@ -427,14 +437,20 @@ mod tests {
 		strategy.source_nonces_updated(header_id(5), source_nonces(7..=8));
 
 		state.best_finalized_source_header_id_at_best_target = Some(header_id(4));
-		assert_eq!(strategy.select_nonces_to_deliver(&state), Some((1..=6, ())));
+		assert_eq!(
+			strategy.select_nonces_to_deliver(state.clone()).await,
+			Some((1..=6, ()))
+		);
 		strategy.best_target_nonces_updated(target_nonces(6), &mut state);
-		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
+		assert_eq!(strategy.select_nonces_to_deliver(state.clone()).await, None);
 
 		state.best_finalized_source_header_id_at_best_target = Some(header_id(5));
-		assert_eq!(strategy.select_nonces_to_deliver(&state), Some((7..=8, ())));
+		assert_eq!(
+			strategy.select_nonces_to_deliver(state.clone()).await,
+			Some((7..=8, ()))
+		);
 		strategy.best_target_nonces_updated(target_nonces(8), &mut state);
-		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
+		assert_eq!(strategy.select_nonces_to_deliver(state.clone()).await, None);
 	}
 
 	#[test]
@@ -449,7 +465,9 @@ mod tests {
 		state.best_target_header_id = Some(header_id(1));
 
 		assert_eq!(
-			strategy.select_nonces_to_deliver_with_selector(&state, |_| Some(50..=100)),
+			async_std::task::block_on(
+				strategy.select_nonces_to_deliver_with_selector(state, |_| async { Some(50..=100) })
+			),
 			Some(1..=49),
 		);
 	}
@@ -464,7 +482,11 @@ mod tests {
 		state.best_finalized_source_header_id_at_source = Some(header_id(1));
 		state.best_finalized_source_header_id_at_best_target = Some(header_id(1));
 		state.best_target_header_id = Some(header_id(1));
-		strategy.select_nonces_to_deliver_with_selector(&state, invalid_selector);
+		async_std::task::block_on(async move {
+			strategy
+				.select_nonces_to_deliver_with_selector(state, |range| async { invalid_selector(range) })
+				.await;
+		});
 	}
 
 	#[test]