Limit messages weight in batch (#496)

* limit messages in the batch by weight/count

* fixed components compilation

* reverted obsolete parts of #469

* implement generated_messages_weights

* actually use computed weight in message proof

* fmt and clippy

* fixed TODO

* clippy

* Update relays/messages-relay/src/message_race_loop.rs

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

* add issue reference

* add assert message

* grumbles

* fmt

* reexport weight from bp-message-lane

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

bridges/bin/millau/node/src/service.rs

@@ -221,7 +221,6 @@ pub fn new_full(config: Configuration) -> Result<TaskManager, ServiceError> {
 				finality_proof_provider.clone(),
 			)));
 			io.extend_with(MessageLaneApi::to_delegate(MessageLaneRpcHandler::new(
-				client.clone(),
 				backend.clone(),
 				Arc::new(MillauMessageLaneKeys),
 			)));

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

@@ -542,12 +542,19 @@ impl_runtime_apis! {
 
 	// TODO: runtime should support several chains (https://github.com/paritytech/parity-bridges-common/issues/457)
 	impl bp_message_lane::OutboundLaneApi<Block> for Runtime {
-		fn messages_dispatch_weight(lane: bp_message_lane::LaneId, begin: bp_message_lane::MessageNonce, end: bp_message_lane::MessageNonce) -> Weight {
+		fn messages_dispatch_weight(
-			(begin..=end)
+			lane: bp_message_lane::LaneId,
-				.filter_map(|nonce| BridgeRialtoMessageLane::outbound_message_payload(lane, nonce))
+			begin: bp_message_lane::MessageNonce,
-				.filter_map(|encoded_payload| rialto_messages::ToRialtoMessagePayload::decode(&mut &encoded_payload[..]).ok())
+			end: bp_message_lane::MessageNonce,
-				.map(|decoded_payload| decoded_payload.weight)
+		) -> Vec<(bp_message_lane::MessageNonce, Weight)> {
-				.fold(0, |sum, weight| sum.saturating_add(weight))
+			(begin..=end).filter_map(|nonce| {
+				let encoded_payload = BridgeRialtoMessageLane::outbound_message_payload(lane, nonce)?;
+				let decoded_payload = rialto_messages::ToRialtoMessagePayload::decode(
+					&mut &encoded_payload[..]
+				).ok()?;
+				Some((nonce, decoded_payload.weight))
+			})
+			.collect()
 		}
 
 		fn latest_received_nonce(lane: bp_message_lane::LaneId) -> bp_message_lane::MessageNonce {

bridges/bin/rialto/node/src/service.rs

@@ -220,7 +220,6 @@ pub fn new_full(config: Configuration) -> Result<TaskManager, ServiceError> {
 				finality_proof_provider.clone(),
 			)));
 			io.extend_with(MessageLaneApi::to_delegate(MessageLaneRpcHandler::new(
-				client.clone(),
 				backend.clone(),
 				Arc::new(RialtoMessageLaneKeys),
 			)));

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

@@ -706,12 +706,19 @@ impl_runtime_apis! {
 
 	// TODO: runtime should support several chains (https://github.com/paritytech/parity-bridges-common/issues/457)
 	impl bp_message_lane::OutboundLaneApi<Block> for Runtime {
-		fn messages_dispatch_weight(lane: bp_message_lane::LaneId, begin: bp_message_lane::MessageNonce, end: bp_message_lane::MessageNonce) -> Weight {
+		fn messages_dispatch_weight(
-			(begin..=end)
+			lane: bp_message_lane::LaneId,
-				.filter_map(|nonce| BridgeMillauMessageLane::outbound_message_payload(lane, nonce))
+			begin: bp_message_lane::MessageNonce,
-				.filter_map(|encoded_payload| millau_messages::ToMillauMessagePayload::decode(&mut &encoded_payload[..]).ok())
+			end: bp_message_lane::MessageNonce,
-				.map(|decoded_payload| decoded_payload.weight)
+		) -> Vec<(bp_message_lane::MessageNonce, Weight)> {
-				.fold(0, |sum, weight| sum.saturating_add(weight))
+			(begin..=end).filter_map(|nonce| {
+				let encoded_payload = BridgeMillauMessageLane::outbound_message_payload(lane, nonce)?;
+				let decoded_payload = millau_messages::ToMillauMessagePayload::decode(
+					&mut &encoded_payload[..]
+				).ok()?;
+				Some((nonce, decoded_payload.weight))
+			})
+			.collect()
 		}
 
 		fn latest_received_nonce(lane: bp_message_lane::LaneId) -> bp_message_lane::MessageNonce {

bridges/modules/message-lane/rpc/Cargo.toml

@@ -20,9 +20,7 @@ bp-message-lane = { path = "../../../primitives/message-lane" }
 
 # Substrate Dependencies
 
-frame-support = "2.0"
 sc-client-api = "2.0"
-sp-api = "2.0"
 sp-blockchain = "2.0"
 sp-core = "2.0"
 sp-runtime = "2.0"

bridges/modules/message-lane/rpc/src/lib.rs

@@ -18,14 +18,12 @@
 
 use crate::error::{Error, FutureResult};
 
-use bp_message_lane::{LaneId, MessageNonce, OutboundLaneApi};
+use bp_message_lane::{LaneId, MessageNonce};
 use bp_runtime::InstanceId;
-use frame_support::weights::Weight;
 use futures::{FutureExt, TryFutureExt};
 use jsonrpc_core::futures::Future as _;
 use jsonrpc_derive::rpc;
 use sc_client_api::Backend as BackendT;
-use sp_api::ProvideRuntimeApi;
 use sp_blockchain::{Error as BlockchainError, HeaderBackend};
 use sp_core::{storage::StorageKey, Bytes};
 use sp_runtime::{codec::Encode, generic::BlockId, traits::Block as BlockT};
@@ -56,8 +54,8 @@ pub trait Runtime: Send + Sync + 'static {
 /// Provides RPC methods for interacting with message-lane pallet.
 #[rpc]
 pub trait MessageLaneApi<BlockHash> {
-	/// Returns cumulative dispatch weight of messages in given inclusive range and their storage proof.
+	/// Returns storage proof of messages in given inclusive range. The state of outbound
-	/// The state of outbound lane is included in the proof if `include_outbound_lane_state` is true.
+	/// lane is included in the proof if `include_outbound_lane_state` is true.
 	#[rpc(name = "messageLane_proveMessages")]
 	fn prove_messages(
 		&self,
@@ -67,7 +65,7 @@ pub trait MessageLaneApi<BlockHash> {
 		end: MessageNonce,
 		include_outbound_lane_state: bool,
 		block: Option<BlockHash>,
-	) -> FutureResult<(Weight, MessagesProof)>;
+	) -> FutureResult<MessagesProof>;
 
 	/// Returns proof-of-message(s) delivery.
 	#[rpc(name = "messageLane_proveMessagesDelivery")]
@@ -80,18 +78,16 @@ pub trait MessageLaneApi<BlockHash> {
 }
 
 /// Implements the MessageLaneApi trait for interacting with message lanes.
-pub struct MessageLaneRpcHandler<Block, Client, Backend, R> {
+pub struct MessageLaneRpcHandler<Block, Backend, R> {
-	client: Arc<Client>,
 	backend: Arc<Backend>,
 	runtime: Arc<R>,
 	_phantom: std::marker::PhantomData<Block>,
 }
 
-impl<Block, Client, Backend, R> MessageLaneRpcHandler<Block, Client, Backend, R> {
+impl<Block, Backend, R> MessageLaneRpcHandler<Block, Backend, R> {
 	/// Creates new mesage lane RPC handler.
-	pub fn new(client: Arc<Client>, backend: Arc<Backend>, runtime: Arc<R>) -> Self {
+	pub fn new(backend: Arc<Backend>, runtime: Arc<R>) -> Self {
 		Self {
-			client,
 			backend,
 			runtime,
 			_phantom: Default::default(),
@@ -99,11 +95,9 @@ impl<Block, Client, Backend, R> MessageLaneRpcHandler<Block, Client, Backend, R>
 	}
 }
 
-impl<Block, Client, Backend, R> MessageLaneApi<Block::Hash> for MessageLaneRpcHandler<Block, Client, Backend, R>
+impl<Block, Backend, R> MessageLaneApi<Block::Hash> for MessageLaneRpcHandler<Block, Backend, R>
 where
 	Block: BlockT,
-	Client: ProvideRuntimeApi<Block> + Send + Sync + 'static,
-	Client::Api: OutboundLaneApi<Block>,
 	Backend: BackendT<Block> + 'static,
 	R: Runtime,
 {
@@ -115,22 +109,7 @@ where
 		end: MessageNonce,
 		include_outbound_lane_state: bool,
 		block: Option<Block::Hash>,
-	) -> FutureResult<(Weight, MessagesProof)> {
+	) -> FutureResult<MessagesProof> {
-		let block = unwrap_or_best(&*self.backend, block);
-
-		let messages_dispatch_weight_result =
-			self.client
-				.runtime_api()
-				.messages_dispatch_weight(&BlockId::Hash(block), lane, begin, end);
-		let messages_dispatch_weight = match messages_dispatch_weight_result {
-			Ok(messages_dispatch_weight) => messages_dispatch_weight,
-			Err(error) => {
-				return Box::new(jsonrpc_core::futures::future::err(
-					blockchain_err(BlockchainError::Execution(Box::new(format!("{:?}", error)))).into(),
-				))
-			}
-		};
-
 		let runtime = self.runtime.clone();
 		let outbound_lane_data_key = if include_outbound_lane_state {
 			Some(runtime.inbound_lane_data_key(&instance, &lane))
@@ -140,14 +119,14 @@ where
 		Box::new(
 			prove_keys_read(
 				self.backend.clone(),
-				Some(block),
+				block,
 				(begin..=end)
 					.map(move |nonce| runtime.message_key(&instance, &lane, nonce))
 					.chain(outbound_lane_data_key.into_iter()),
 			)
 			.boxed()
 			.compat()
-			.map(move |proof| (messages_dispatch_weight, serialize_storage_proof(proof)))
+			.map(serialize_storage_proof)
 			.map_err(Into::into),
 		)
 	}

bridges/primitives/message-lane/src/lib.rs

@@ -23,13 +23,16 @@
 #![allow(clippy::unnecessary_mut_passed)]
 
 use codec::{Decode, Encode};
-use frame_support::{weights::Weight, RuntimeDebug};
+use frame_support::RuntimeDebug;
 use sp_api::decl_runtime_apis;
 use sp_std::{collections::vec_deque::VecDeque, prelude::*};
 
 pub mod source_chain;
 pub mod target_chain;
 
+// Weight is reexported to avoid additional frame-support dependencies in message-lane related crates.
+pub use frame_support::weights::Weight;
+
 /// Lane identifier.
 pub type LaneId = [u8; 4];
 
@@ -127,7 +130,14 @@ decl_runtime_apis! {
 	/// Outbound message lane API.
 	pub trait OutboundLaneApi {
 		/// Returns dispatch weight of all messages in given inclusive range.
-		fn messages_dispatch_weight(lane: LaneId, begin: MessageNonce, end: MessageNonce) -> Weight;
+		///
+		/// If some (or all) messages are missing from the storage, they'll also will
+		/// be missing from the resulting vector. The vector is ordered by the nonce.
+		fn messages_dispatch_weight(
+			lane: LaneId,
+			begin: MessageNonce,
+			end: MessageNonce,
+		) -> Vec<(MessageNonce, Weight)>;
 		/// Returns nonce of the latest message, received by bridged chain.
 		fn latest_received_nonce(lane: LaneId) -> MessageNonce;
 		/// Returns nonce of the latest message, generated by given lane.

bridges/relays/messages-relay/Cargo.toml

@@ -11,7 +11,6 @@ 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-relay/src/message_lane.rs

@@ -21,7 +21,6 @@
 
 use relay_utils::HeaderId;
 
-use num_traits::{CheckedSub, One, Zero};
 use std::fmt::Debug;
 
 /// One-way message lane.
@@ -31,21 +30,6 @@ pub trait MessageLane: Clone + Send + Sync {
 	/// Name of the messages target.
 	const TARGET_NAME: &'static str;
 
-	/// Message nonce type.
-	type MessageNonce: Clone
-		+ Send
-		+ Sync
-		+ Copy
-		+ Debug
-		+ Default
-		+ From<u32>
-		+ Into<u64>
-		+ Ord
-		+ CheckedSub
-		+ std::ops::Add<Output = Self::MessageNonce>
-		+ One
-		+ Zero;
-
 	/// Messages proof.
 	type MessagesProof: Clone + Send + Sync;
 	/// Messages receiving proof.

bridges/relays/messages-relay/src/message_lane_loop.rs

@@ -30,18 +30,18 @@ use crate::message_race_receiving::run as run_message_receiving_race;
 use crate::metrics::MessageLaneLoopMetrics;
 
 use async_trait::async_trait;
-use bp_message_lane::LaneId;
+use bp_message_lane::{LaneId, MessageNonce, Weight};
 use futures::{channel::mpsc::unbounded, future::FutureExt, stream::StreamExt};
 use relay_utils::{
 	interval,
 	metrics::{start as metrics_start, GlobalMetrics, MetricsParams},
 	process_future_result, retry_backoff, FailedClient, MaybeConnectionError,
 };
-use std::{fmt::Debug, future::Future, ops::RangeInclusive, time::Duration};
+use std::{collections::BTreeMap, fmt::Debug, future::Future, ops::RangeInclusive, time::Duration};
 
 /// Message lane loop configuration params.
 #[derive(Debug, Clone)]
-pub struct Params<MessageNonce> {
+pub struct Params {
 	/// Id of lane this loop is servicing.
 	pub lane: LaneId,
 	/// Interval at which we ask target node about its updates.
@@ -52,10 +52,31 @@ pub struct Params<MessageNonce> {
 	pub reconnect_delay: Duration,
 	/// The loop will auto-restart if there has been no updates during this period.
 	pub stall_timeout: Duration,
+	/// Message delivery race parameters.
+	pub delivery_params: MessageDeliveryParams,
+}
+
+/// Message delivery race parameters.
+#[derive(Debug, Clone)]
+pub struct MessageDeliveryParams {
 	/// Message delivery race will stop delivering messages if there are `max_unconfirmed_nonces_at_target`
 	/// unconfirmed nonces on the target node. The race would continue once they're confirmed by the
 	/// receiving race.
 	pub max_unconfirmed_nonces_at_target: MessageNonce,
+	/// Maximal cumulative dispatch weight of relayed messages in single delivery transaction.
+	pub max_messages_weight_in_single_batch: Weight,
+}
+
+/// Messages weights map.
+pub type MessageWeightsMap = BTreeMap<MessageNonce, Weight>;
+
+/// Message delivery race proof parameters.
+#[derive(Debug, PartialEq)]
+pub struct MessageProofParameters {
+	/// Include outbound lane state proof?
+	pub outbound_state_proof_required: bool,
+	/// Cumulative dispatch weight of messages that we're building proof for.
+	pub dispatch_weight: Weight,
 }
 
 /// Source client trait.
@@ -74,20 +95,27 @@ pub trait SourceClient<P: MessageLane>: Clone + Send + Sync {
 	async fn latest_generated_nonce(
 		&self,
 		id: SourceHeaderIdOf<P>,
-	) -> Result<(SourceHeaderIdOf<P>, P::MessageNonce), Self::Error>;
+	) -> Result<(SourceHeaderIdOf<P>, MessageNonce), Self::Error>;
 	/// Get nonce of the latest message, which receiving has been confirmed by the target chain.
 	async fn latest_confirmed_received_nonce(
 		&self,
 		id: SourceHeaderIdOf<P>,
-	) -> Result<(SourceHeaderIdOf<P>, P::MessageNonce), Self::Error>;
+	) -> Result<(SourceHeaderIdOf<P>, MessageNonce), Self::Error>;
+
+	/// Returns mapping of message nonces, generated on this client, to their weights.
+	async fn generated_messages_weights(
+		&self,
+		id: SourceHeaderIdOf<P>,
+		nonces: RangeInclusive<MessageNonce>,
+	) -> Result<MessageWeightsMap, Self::Error>;
 
 	/// Prove messages in inclusive range [begin; end].
 	async fn prove_messages(
 		&self,
 		id: SourceHeaderIdOf<P>,
-		nonces: RangeInclusive<P::MessageNonce>,
+		nonces: RangeInclusive<MessageNonce>,
-		include_outbound_lane_state: bool,
+		proof_parameters: MessageProofParameters,
-	) -> Result<(SourceHeaderIdOf<P>, RangeInclusive<P::MessageNonce>, P::MessagesProof), Self::Error>;
+	) -> Result<(SourceHeaderIdOf<P>, RangeInclusive<MessageNonce>, P::MessagesProof), Self::Error>;
 
 	/// Submit messages receiving proof.
 	async fn submit_messages_receiving_proof(
@@ -113,13 +141,13 @@ pub trait TargetClient<P: MessageLane>: Clone + Send + Sync {
 	async fn latest_received_nonce(
 		&self,
 		id: TargetHeaderIdOf<P>,
-	) -> Result<(TargetHeaderIdOf<P>, P::MessageNonce), Self::Error>;
+	) -> Result<(TargetHeaderIdOf<P>, MessageNonce), Self::Error>;
 
 	/// Get nonce of latest confirmed message.
 	async fn latest_confirmed_received_nonce(
 		&self,
 		id: TargetHeaderIdOf<P>,
-	) -> Result<(TargetHeaderIdOf<P>, P::MessageNonce), Self::Error>;
+	) -> Result<(TargetHeaderIdOf<P>, MessageNonce), Self::Error>;
 
 	/// Prove messages receiving at given block.
 	async fn prove_messages_receiving(
@@ -131,9 +159,9 @@ pub trait TargetClient<P: MessageLane>: Clone + Send + Sync {
 	async fn submit_messages_proof(
 		&self,
 		generated_at_header: SourceHeaderIdOf<P>,
-		nonces: RangeInclusive<P::MessageNonce>,
+		nonces: RangeInclusive<MessageNonce>,
 		proof: P::MessagesProof,
-	) -> Result<RangeInclusive<P::MessageNonce>, Self::Error>;
+	) -> Result<RangeInclusive<MessageNonce>, Self::Error>;
 }
 
 /// State of the client.
@@ -162,7 +190,7 @@ pub struct ClientsState<P: MessageLane> {
 
 /// Run message lane service loop.
 pub fn run<P: MessageLane>(
-	params: Params<P::MessageNonce>,
+	params: Params,
 	mut source_client: impl SourceClient<P>,
 	mut target_client: impl TargetClient<P>,
 	metrics_params: Option<MetricsParams>,
@@ -257,7 +285,7 @@ pub fn run<P: MessageLane>(
 
 /// Run one-way message delivery loop until connection with target or source node is lost, or exit signal is received.
 async fn run_until_connection_lost<P: MessageLane, SC: SourceClient<P>, TC: TargetClient<P>>(
-	params: Params<P::MessageNonce>,
+	params: Params,
 	source_client: SC,
 	target_client: TC,
 	mut metrics_global: Option<&mut GlobalMetrics>,
@@ -289,7 +317,7 @@ async fn run_until_connection_lost<P: MessageLane, SC: SourceClient<P>, TC: Targ
 		delivery_target_state_receiver,
 		params.stall_timeout,
 		metrics_msg.clone(),
-		params.max_unconfirmed_nonces_at_target,
+		params.delivery_params,
 	)
 	.fuse();
 
@@ -430,13 +458,15 @@ pub(crate) mod tests {
 	use relay_utils::HeaderId;
 	use std::sync::Arc;
 
-	pub fn header_id(number: TestSourceHeaderNumber) -> HeaderId<TestSourceHeaderNumber, TestSourceHeaderHash> {
+	pub fn header_id(number: TestSourceHeaderNumber) -> TestSourceHeaderId {
 		HeaderId(number, number)
 	}
 
-	pub type TestMessageNonce = u64;
+	pub type TestSourceHeaderId = HeaderId<TestSourceHeaderNumber, TestSourceHeaderHash>;
-	pub type TestMessagesProof = (RangeInclusive<TestMessageNonce>, Option<TestMessageNonce>);
+	pub type TestTargetHeaderId = HeaderId<TestTargetHeaderNumber, TestTargetHeaderHash>;
-	pub type TestMessagesReceivingProof = TestMessageNonce;
+
+	pub type TestMessagesProof = (RangeInclusive<MessageNonce>, Option<MessageNonce>);
+	pub type TestMessagesReceivingProof = MessageNonce;
 
 	pub type TestSourceHeaderNumber = u64;
 	pub type TestSourceHeaderHash = u64;
@@ -460,8 +490,6 @@ pub(crate) mod tests {
 		const SOURCE_NAME: &'static str = "TestSource";
 		const TARGET_NAME: &'static str = "TestTarget";
 
-		type MessageNonce = TestMessageNonce;
-
 		type MessagesProof = TestMessagesProof;
 		type MessagesReceivingProof = TestMessagesReceivingProof;
 
@@ -477,14 +505,14 @@ pub(crate) mod tests {
 		is_source_fails: bool,
 		is_source_reconnected: bool,
 		source_state: SourceClientState<TestMessageLane>,
-		source_latest_generated_nonce: TestMessageNonce,
+		source_latest_generated_nonce: MessageNonce,
-		source_latest_confirmed_received_nonce: TestMessageNonce,
+		source_latest_confirmed_received_nonce: MessageNonce,
 		submitted_messages_receiving_proofs: Vec<TestMessagesReceivingProof>,
 		is_target_fails: bool,
 		is_target_reconnected: bool,
 		target_state: SourceClientState<TestMessageLane>,
-		target_latest_received_nonce: TestMessageNonce,
+		target_latest_received_nonce: MessageNonce,
-		target_latest_confirmed_received_nonce: TestMessageNonce,
+		target_latest_confirmed_received_nonce: MessageNonce,
 		submitted_messages_proofs: Vec<TestMessagesProof>,
 	}
 
@@ -519,7 +547,7 @@ pub(crate) mod tests {
 		async fn latest_generated_nonce(
 			&self,
 			id: SourceHeaderIdOf<TestMessageLane>,
-		) -> Result<(SourceHeaderIdOf<TestMessageLane>, TestMessageNonce), Self::Error> {
+		) -> Result<(SourceHeaderIdOf<TestMessageLane>, MessageNonce), Self::Error> {
 			let mut data = self.data.lock();
 			(self.tick)(&mut *data);
 			if data.is_source_fails {
@@ -531,21 +559,29 @@ pub(crate) mod tests {
 		async fn latest_confirmed_received_nonce(
 			&self,
 			id: SourceHeaderIdOf<TestMessageLane>,
-		) -> Result<(SourceHeaderIdOf<TestMessageLane>, TestMessageNonce), Self::Error> {
+		) -> Result<(SourceHeaderIdOf<TestMessageLane>, MessageNonce), Self::Error> {
 			let mut data = self.data.lock();
 			(self.tick)(&mut *data);
 			Ok((id, data.source_latest_confirmed_received_nonce))
 		}
 
+		async fn generated_messages_weights(
+			&self,
+			_id: SourceHeaderIdOf<TestMessageLane>,
+			nonces: RangeInclusive<MessageNonce>,
+		) -> Result<MessageWeightsMap, Self::Error> {
+			Ok(nonces.map(|nonce| (nonce, 1)).collect())
+		}
+
 		async fn prove_messages(
 			&self,
 			id: SourceHeaderIdOf<TestMessageLane>,
-			nonces: RangeInclusive<TestMessageNonce>,
+			nonces: RangeInclusive<MessageNonce>,
-			include_outbound_lane_state: bool,
+			proof_parameters: MessageProofParameters,
 		) -> Result<
 			(
 				SourceHeaderIdOf<TestMessageLane>,
-				RangeInclusive<TestMessageNonce>,
+				RangeInclusive<MessageNonce>,
 				TestMessagesProof,
 			),
 			Self::Error,
@@ -557,7 +593,7 @@ pub(crate) mod tests {
 				nonces.clone(),
 				(
 					nonces,
-					if include_outbound_lane_state {
+					if proof_parameters.outbound_state_proof_required {
 						Some(data.source_latest_confirmed_received_nonce)
 					} else {
 						None
@@ -610,7 +646,7 @@ pub(crate) mod tests {
 		async fn latest_received_nonce(
 			&self,
 			id: TargetHeaderIdOf<TestMessageLane>,
-		) -> Result<(TargetHeaderIdOf<TestMessageLane>, TestMessageNonce), Self::Error> {
+		) -> Result<(TargetHeaderIdOf<TestMessageLane>, MessageNonce), Self::Error> {
 			let mut data = self.data.lock();
 			(self.tick)(&mut *data);
 			if data.is_target_fails {
@@ -622,7 +658,7 @@ pub(crate) mod tests {
 		async fn latest_confirmed_received_nonce(
 			&self,
 			id: TargetHeaderIdOf<TestMessageLane>,
-		) -> Result<(TargetHeaderIdOf<TestMessageLane>, TestMessageNonce), Self::Error> {
+		) -> Result<(TargetHeaderIdOf<TestMessageLane>, MessageNonce), Self::Error> {
 			let mut data = self.data.lock();
 			(self.tick)(&mut *data);
 			if data.is_target_fails {
@@ -641,9 +677,9 @@ pub(crate) mod tests {
 		async fn submit_messages_proof(
 			&self,
 			_generated_at_header: SourceHeaderIdOf<TestMessageLane>,
-			nonces: RangeInclusive<TestMessageNonce>,
+			nonces: RangeInclusive<MessageNonce>,
 			proof: TestMessagesProof,
-		) -> Result<RangeInclusive<TestMessageNonce>, Self::Error> {
+		) -> Result<RangeInclusive<MessageNonce>, Self::Error> {
 			let mut data = self.data.lock();
 			(self.tick)(&mut *data);
 			if data.is_target_fails {
@@ -683,8 +719,11 @@ pub(crate) mod tests {
 					source_tick: Duration::from_millis(100),
 					target_tick: Duration::from_millis(100),
 					reconnect_delay: Duration::from_millis(0),
-					stall_timeout: Duration::from_millis(60),
+					stall_timeout: Duration::from_millis(60 * 1000),
-					max_unconfirmed_nonces_at_target: 100,
+					delivery_params: MessageDeliveryParams {
+						max_unconfirmed_nonces_at_target: 4,
+						max_messages_weight_in_single_batch: 4,
+					},
 				},
 				source_client,
 				target_client,
@@ -743,8 +782,6 @@ pub(crate) mod tests {
 
 	#[test]
 	fn message_lane_loop_works() {
-		// with this configuration, target client must first sync headers [1; 10] and
-		// then submit proof-of-messages [0; 10] at once
 		let (exit_sender, exit_receiver) = unbounded();
 		let result = run_loop_test(
 			TestClientData {
@@ -762,18 +799,22 @@ pub(crate) mod tests {
 			},
 			Arc::new(|_: &mut TestClientData| {}),
 			Arc::new(move |data: &mut TestClientData| {
-				// syncing source headers -> target chain (by one)
-				if data.target_state.best_peer.0 < data.source_state.best_self.0 {
-					data.target_state.best_peer =
-						HeaderId(data.target_state.best_peer.0 + 1, data.target_state.best_peer.0 + 1);
-				}
 				// syncing source headers -> target chain (all at once)
+				if data.target_state.best_peer.0 < data.source_state.best_self.0 {
+					data.target_state.best_peer = data.source_state.best_self;
+				}
+				// syncing target headers -> source chain (all at once)
 				if data.source_state.best_peer.0 < data.target_state.best_self.0 {
 					data.source_state.best_peer = data.target_state.best_self;
 				}
-				// if target has received all messages => increase target block so that confirmations may be sent
+				// if target has received messages batch => increase blocks so that confirmations may be sent
-				if data.target_latest_received_nonce == 10 {
+				if data.target_latest_received_nonce == 4
+					|| data.target_latest_received_nonce == 8
+					|| data.target_latest_received_nonce == 10
+				{
 					data.target_state.best_self =
+						HeaderId(data.target_state.best_self.0 + 1, data.target_state.best_self.0 + 1);
+					data.source_state.best_self =
 						HeaderId(data.source_state.best_self.0 + 1, data.source_state.best_self.0 + 1);
 				}
 				// if source has received all messages receiving confirmations => increase source block so that confirmations may be sent

bridges/relays/messages-relay/src/message_race_delivery.rs

@@ -15,21 +15,21 @@
 
 use crate::message_lane::{MessageLane, SourceHeaderIdOf, TargetHeaderIdOf};
 use crate::message_lane_loop::{
-	SourceClient as MessageLaneSourceClient, SourceClientState, TargetClient as MessageLaneTargetClient,
+	MessageDeliveryParams, MessageProofParameters, MessageWeightsMap, SourceClient as MessageLaneSourceClient,
-	TargetClientState,
+	SourceClientState, TargetClient as MessageLaneTargetClient, TargetClientState,
+};
+use crate::message_race_loop::{
+	MessageRace, NoncesRange, RaceState, RaceStrategy, SourceClient, SourceClientNonces, TargetClient,
+	TargetClientNonces,
 };
-use crate::message_race_loop::{ClientNonces, MessageRace, RaceState, RaceStrategy, SourceClient, TargetClient};
 use crate::message_race_strategy::BasicStrategy;
 use crate::metrics::MessageLaneLoopMetrics;
 
 use async_trait::async_trait;
+use bp_message_lane::{MessageNonce, Weight};
 use futures::stream::FusedStream;
-use num_traits::CheckedSub;
 use relay_utils::FailedClient;
-use std::{marker::PhantomData, ops::RangeInclusive, time::Duration};
+use std::{collections::BTreeMap, marker::PhantomData, ops::RangeInclusive, time::Duration};
-
-/// Maximal number of messages to relay in single transaction.
-const MAX_MESSAGES_TO_RELAY_IN_SINGLE_TX: u32 = 4;
 
 /// Run message delivery race.
 pub async fn run<P: MessageLane>(
@@ -39,7 +39,7 @@ pub async fn run<P: MessageLane>(
 	target_state_updates: impl FusedStream<Item = TargetClientState<P>>,
 	stall_timeout: Duration,
 	metrics_msg: Option<MessageLaneLoopMetrics>,
-	max_unconfirmed_nonces_at_target: P::MessageNonce,
+	params: MessageDeliveryParams,
 ) -> Result<(), FailedClient> {
 	crate::message_race_loop::run(
 		MessageDeliveryRaceSource {
@@ -56,10 +56,11 @@ pub async fn run<P: MessageLane>(
 		target_state_updates,
 		stall_timeout,
 		MessageDeliveryStrategy::<P> {
-			max_unconfirmed_nonces_at_target,
+			max_unconfirmed_nonces_at_target: params.max_unconfirmed_nonces_at_target,
-			source_nonces: None,
+			max_messages_weight_in_single_batch: params.max_messages_weight_in_single_batch,
+			latest_confirmed_nonce_at_source: None,
 			target_nonces: None,
-			strategy: BasicStrategy::new(MAX_MESSAGES_TO_RELAY_IN_SINGLE_TX.into()),
+			strategy: BasicStrategy::new(),
 		},
 	)
 	.await
@@ -72,7 +73,7 @@ impl<P: MessageLane> MessageRace for MessageDeliveryRace<P> {
 	type SourceHeaderId = SourceHeaderIdOf<P>;
 	type TargetHeaderId = TargetHeaderIdOf<P>;
 
-	type MessageNonce = P::MessageNonce;
+	type MessageNonce = MessageNonce;
 	type Proof = P::MessagesProof;
 
 	fn source_name() -> String {
@@ -98,12 +99,14 @@ where
 	C: MessageLaneSourceClient<P>,
 {
 	type Error = C::Error;
-	type ProofParameters = bool;
+	type NoncesRange = MessageWeightsMap;
+	type ProofParameters = MessageProofParameters;
 
 	async fn nonces(
 		&self,
 		at_block: SourceHeaderIdOf<P>,
-	) -> Result<(SourceHeaderIdOf<P>, ClientNonces<P::MessageNonce>), Self::Error> {
+		prev_latest_nonce: MessageNonce,
+	) -> Result<(SourceHeaderIdOf<P>, SourceClientNonces<Self::NoncesRange>), Self::Error> {
 		let (at_block, latest_generated_nonce) = self.client.latest_generated_nonce(at_block).await?;
 		let (at_block, latest_confirmed_nonce) = self.client.latest_confirmed_received_nonce(at_block).await?;
 
@@ -112,10 +115,18 @@ where
 			metrics_msg.update_source_latest_confirmed_nonce::<P>(latest_confirmed_nonce);
 		}
 
+		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)
+				.await?
+		} else {
+			MessageWeightsMap::new()
+		};
+
 		Ok((
 			at_block,
-			ClientNonces {
+			SourceClientNonces {
-				latest_nonce: latest_generated_nonce,
+				new_nonces,
 				confirmed_nonce: Some(latest_confirmed_nonce),
 			},
 		))
@@ -124,13 +135,10 @@ where
 	async fn generate_proof(
 		&self,
 		at_block: SourceHeaderIdOf<P>,
-		nonces: RangeInclusive<P::MessageNonce>,
+		nonces: RangeInclusive<MessageNonce>,
 		proof_parameters: Self::ProofParameters,
-	) -> Result<(SourceHeaderIdOf<P>, RangeInclusive<P::MessageNonce>, P::MessagesProof), Self::Error> {
+	) -> Result<(SourceHeaderIdOf<P>, RangeInclusive<MessageNonce>, P::MessagesProof), Self::Error> {
-		let outbound_state_proof_required = proof_parameters;
+		self.client.prove_messages(at_block, nonces, proof_parameters).await
-		self.client
-			.prove_messages(at_block, nonces, outbound_state_proof_required)
-			.await
 	}
 }
 
@@ -152,7 +160,7 @@ where
 	async fn nonces(
 		&self,
 		at_block: TargetHeaderIdOf<P>,
-	) -> Result<(TargetHeaderIdOf<P>, ClientNonces<P::MessageNonce>), Self::Error> {
+	) -> Result<(TargetHeaderIdOf<P>, TargetClientNonces), Self::Error> {
 		let (at_block, latest_received_nonce) = self.client.latest_received_nonce(at_block).await?;
 		let (at_block, latest_confirmed_nonce) = self.client.latest_confirmed_received_nonce(at_block).await?;
 
@@ -163,7 +171,7 @@ where
 
 		Ok((
 			at_block,
-			ClientNonces {
+			TargetClientNonces {
 				latest_nonce: latest_received_nonce,
 				confirmed_nonce: Some(latest_confirmed_nonce),
 			},
@@ -173,9 +181,9 @@ where
 	async fn submit_proof(
 		&self,
 		generated_at_block: SourceHeaderIdOf<P>,
-		nonces: RangeInclusive<P::MessageNonce>,
+		nonces: RangeInclusive<MessageNonce>,
 		proof: P::MessagesProof,
-	) -> Result<RangeInclusive<P::MessageNonce>, Self::Error> {
+	) -> Result<RangeInclusive<MessageNonce>, Self::Error> {
 		self.client
 			.submit_messages_proof(generated_at_block, nonces, proof)
 			.await
@@ -185,11 +193,13 @@ where
 /// Messages delivery strategy.
 struct MessageDeliveryStrategy<P: MessageLane> {
 	/// Maximal unconfirmed nonces at target client.
-	max_unconfirmed_nonces_at_target: P::MessageNonce,
+	max_unconfirmed_nonces_at_target: MessageNonce,
-	/// Latest nonces from the source client.
+	/// Maximal cumulative messages weight in the single delivery transaction.
-	source_nonces: Option<ClientNonces<P::MessageNonce>>,
+	max_messages_weight_in_single_batch: Weight,
+	/// Latest confirmed nonce at the source client.
+	latest_confirmed_nonce_at_source: Option<MessageNonce>,
 	/// Target nonces from the source client.
-	target_nonces: Option<ClientNonces<P::MessageNonce>>,
+	target_nonces: Option<TargetClientNonces>,
 	/// Basic delivery strategy.
 	strategy: MessageDeliveryStrategyBase<P>,
 }
@@ -199,36 +209,41 @@ type MessageDeliveryStrategyBase<P> = BasicStrategy<
 	<P as MessageLane>::SourceHeaderHash,
 	<P as MessageLane>::TargetHeaderNumber,
 	<P as MessageLane>::TargetHeaderHash,
-	<P as MessageLane>::MessageNonce,
+	MessageWeightsMap,
 	<P as MessageLane>::MessagesProof,
 >;
 
-impl<P: MessageLane> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::MessageNonce, P::MessagesProof>
+impl<P: MessageLane> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::MessagesProof>
 	for MessageDeliveryStrategy<P>
 {
-	type ProofParameters = bool;
+	type SourceNoncesRange = MessageWeightsMap;
+	type ProofParameters = MessageProofParameters;
 
 	fn is_empty(&self) -> bool {
 		self.strategy.is_empty()
 	}
 
-	fn best_at_source(&self) -> P::MessageNonce {
+	fn best_at_source(&self) -> MessageNonce {
 		self.strategy.best_at_source()
 	}
 
-	fn best_at_target(&self) -> P::MessageNonce {
+	fn best_at_target(&self) -> MessageNonce {
 		self.strategy.best_at_target()
 	}
 
-	fn source_nonces_updated(&mut self, at_block: SourceHeaderIdOf<P>, nonces: ClientNonces<P::MessageNonce>) {
+	fn source_nonces_updated(
-		self.source_nonces = Some(nonces.clone());
+		&mut self,
+		at_block: SourceHeaderIdOf<P>,
+		nonces: SourceClientNonces<Self::SourceNoncesRange>,
+	) {
+		self.latest_confirmed_nonce_at_source = nonces.confirmed_nonce;
 		self.strategy.source_nonces_updated(at_block, nonces)
 	}
 
 	fn target_nonces_updated(
 		&mut self,
-		nonces: ClientNonces<P::MessageNonce>,
+		nonces: TargetClientNonces,
-		race_state: &mut RaceState<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::MessageNonce, P::MessagesProof>,
+		race_state: &mut RaceState<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::MessagesProof>,
 	) {
 		self.target_nonces = Some(nonces.clone());
 		self.strategy.target_nonces_updated(nonces, race_state)
@@ -236,14 +251,14 @@ impl<P: MessageLane> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::M
 
 	fn select_nonces_to_deliver(
 		&mut self,
-		race_state: &RaceState<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::MessageNonce, P::MessagesProof>,
+		race_state: &RaceState<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::MessagesProof>,
-	) -> Option<(RangeInclusive<P::MessageNonce>, Self::ProofParameters)> {
+	) -> Option<(RangeInclusive<MessageNonce>, Self::ProofParameters)> {
 		const CONFIRMED_NONCE_PROOF: &str = "\
 			ClientNonces are crafted by MessageDeliveryRace(Source|Target);\
 			MessageDeliveryRace(Source|Target) always fills confirmed_nonce field;\
 			qed";
 
-		let source_nonces = self.source_nonces.as_ref()?;
+		let latest_confirmed_nonce_at_source = self.latest_confirmed_nonce_at_source?;
 		let target_nonces = self.target_nonces.as_ref()?;
 
 		// There's additional condition in the message delivery race: target would reject messages
@@ -257,10 +272,9 @@ impl<P: MessageLane> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::M
 		// The receiving race is responsible to deliver confirmations back to the source chain. So if
 		// there's a lot of unconfirmed messages, let's wait until it'll be able to do its job.
 		let latest_received_nonce_at_target = target_nonces.latest_nonce;
-		let latest_confirmed_nonce_at_source = source_nonces.confirmed_nonce.expect(CONFIRMED_NONCE_PROOF);
+		let confirmations_missing = latest_received_nonce_at_target.checked_sub(latest_confirmed_nonce_at_source);
-		let confirmations_missing = latest_received_nonce_at_target.checked_sub(&latest_confirmed_nonce_at_source);
 		match confirmations_missing {
-			Some(confirmations_missing) if confirmations_missing > self.max_unconfirmed_nonces_at_target => {
+			Some(confirmations_missing) if confirmations_missing >= self.max_unconfirmed_nonces_at_target => {
 				log::debug!(
 					target: "bridge",
 					"Cannot deliver any more messages from {} to {}. Too many unconfirmed nonces \
@@ -277,11 +291,7 @@ impl<P: MessageLane> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::M
 			_ => (),
 		}
 
-		// If we're here, then the confirmations race did it job && sending side now knows that messages
+		// Ok - we may have new nonces to deliver. But target may still reject new messages, because we haven't
-		// have been delivered. Now let's select nonces that we want to deliver.
-		let selected_nonces = self.strategy.select_nonces_to_deliver(race_state)?.0;
-
-		// Ok - we have new nonces to deliver. But target may still reject new messages, because we haven't
 		// notified it that (some) messages have been confirmed. So we may want to include updated
 		// `source.latest_confirmed` in the proof.
 		//
@@ -290,11 +300,226 @@ impl<P: MessageLane> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::M
 		let latest_confirmed_nonce_at_target = target_nonces.confirmed_nonce.expect(CONFIRMED_NONCE_PROOF);
 		let outbound_state_proof_required = latest_confirmed_nonce_at_target < latest_confirmed_nonce_at_source;
 
-		// https://github.com/paritytech/parity-bridges-common/issues/432
+		// If we're here, then the confirmations race did its job && sending side now knows that messages
-		// https://github.com/paritytech/parity-bridges-common/issues/433
+		// have been delivered. Now let's select nonces that we want to deliver.
-		// TODO: number of messages must be no larger than:
+		//
-		// `max_unconfirmed_nonces_at_target - (latest_received_nonce_at_target - latest_confirmed_nonce_at_target)`
+		// We may deliver at most:
+		//
+		// max_unconfirmed_nonces_at_target - (latest_received_nonce_at_target - latest_confirmed_nonce_at_target)
+		//
+		// messages in the batch. But since we're including outbound state proof in the batch, then it
+		// may be increased to:
+		//
+		// max_unconfirmed_nonces_at_target - (latest_received_nonce_at_target - latest_confirmed_nonce_at_source)
+		let future_confirmed_nonce_at_target = if outbound_state_proof_required {
+			latest_confirmed_nonce_at_source
+		} else {
+			latest_confirmed_nonce_at_target
+		};
+		let max_nonces = latest_received_nonce_at_target
+			.checked_sub(future_confirmed_nonce_at_target)
+			.and_then(|diff| self.max_unconfirmed_nonces_at_target.checked_sub(diff))
+			.unwrap_or_default();
+		let max_messages_weight_in_single_batch = self.max_messages_weight_in_single_batch;
+		let mut selected_weight: Weight = 0;
+		let mut selected_count: MessageNonce = 0;
 
-		Some((selected_nonces, outbound_state_proof_required))
+		let selected_nonces = self
+			.strategy
+			.select_nonces_to_deliver_with_selector(race_state, |range| {
+				let to_requeue = range
+					.into_iter()
+					.skip_while(|(_, weight)| {
+						// limit messages in the batch by weight
+						let new_selected_weight = match selected_weight.checked_add(*weight) {
+							Some(new_selected_weight) if new_selected_weight <= max_messages_weight_in_single_batch => {
+								new_selected_weight
+							}
+							_ => 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_count = new_selected_count;
+						true
+					})
+					.collect::<BTreeMap<_, _>>();
+				if to_requeue.is_empty() {
+					None
+				} else {
+					Some(to_requeue)
+				}
+			})?;
+
+		Some((
+			selected_nonces,
+			MessageProofParameters {
+				outbound_state_proof_required,
+				dispatch_weight: selected_weight,
+			},
+		))
+	}
+}
+
+impl NoncesRange for MessageWeightsMap {
+	fn begin(&self) -> MessageNonce {
+		self.keys().next().cloned().unwrap_or_default()
+	}
+
+	fn end(&self) -> MessageNonce {
+		self.keys().next_back().cloned().unwrap_or_default()
+	}
+
+	fn greater_than(mut self, nonce: MessageNonce) -> Option<Self> {
+		let gte = self.split_off(&(nonce + 1));
+		if gte.is_empty() {
+			None
+		} else {
+			Some(gte)
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::message_lane_loop::{
+		tests::{header_id, TestMessageLane, TestMessagesProof, TestSourceHeaderId, TestTargetHeaderId},
+		ClientState,
+	};
+
+	type TestRaceState = RaceState<TestSourceHeaderId, TestTargetHeaderId, TestMessagesProof>;
+	type TestStrategy = MessageDeliveryStrategy<TestMessageLane>;
+
+	fn prepare_strategy() -> (TestRaceState, TestStrategy) {
+		let mut race_state = RaceState {
+			source_state: Some(ClientState {
+				best_self: header_id(1),
+				best_peer: header_id(1),
+			}),
+			target_state: Some(ClientState {
+				best_self: header_id(1),
+				best_peer: header_id(1),
+			}),
+			nonces_to_submit: None,
+			nonces_submitted: None,
+		};
+
+		let mut race_strategy = TestStrategy {
+			max_unconfirmed_nonces_at_target: 4,
+			max_messages_weight_in_single_batch: 4,
+			latest_confirmed_nonce_at_source: Some(19),
+			target_nonces: Some(TargetClientNonces {
+				latest_nonce: 19,
+				confirmed_nonce: Some(19),
+			}),
+			strategy: BasicStrategy::new(),
+		};
+
+		race_strategy.strategy.source_nonces_updated(
+			header_id(1),
+			SourceClientNonces {
+				new_nonces: vec![(20, 1), (21, 1), (22, 1), (23, 1)].into_iter().collect(),
+				confirmed_nonce: Some(19),
+			},
+		);
+		race_strategy
+			.strategy
+			.target_nonces_updated(race_strategy.target_nonces.clone().unwrap(), &mut race_state);
+
+		(race_state, race_strategy)
+	}
+
+	fn proof_parameters(state_required: bool, weight: Weight) -> MessageProofParameters {
+		MessageProofParameters {
+			outbound_state_proof_required: state_required,
+			dispatch_weight: weight,
+		}
+	}
+
+	#[test]
+	fn weights_map_works_as_nonces_range() {
+		fn build_map(range: RangeInclusive<MessageNonce>) -> MessageWeightsMap {
+			range.map(|idx| (idx, idx)).collect()
+		}
+
+		let map = build_map(20..=30);
+
+		assert_eq!(map.begin(), 20);
+		assert_eq!(map.end(), 30);
+		assert_eq!(map.clone().greater_than(10), Some(build_map(20..=30)));
+		assert_eq!(map.clone().greater_than(19), Some(build_map(20..=30)));
+		assert_eq!(map.clone().greater_than(20), Some(build_map(21..=30)));
+		assert_eq!(map.clone().greater_than(25), Some(build_map(26..=30)));
+		assert_eq!(map.clone().greater_than(29), Some(build_map(30..=30)));
+		assert_eq!(map.greater_than(30), None);
+	}
+
+	#[test]
+	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),
+			Some(((20..=23), proof_parameters(false, 4)))
+		);
+	}
+
+	#[test]
+	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,
+		// we need to wait until confirmations will be delivered by receiving race
+		strategy.latest_confirmed_nonce_at_source =
+			Some(strategy.target_nonces.as_ref().unwrap().latest_nonce - strategy.max_unconfirmed_nonces_at_target);
+		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
+	}
+
+	#[test]
+	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
+		// to target to prune rewards queue
+		let prev_confirmed_nonce_at_source = strategy.latest_confirmed_nonce_at_source.unwrap();
+		strategy.target_nonces.as_mut().unwrap().confirmed_nonce = Some(prev_confirmed_nonce_at_source - 1);
+		assert_eq!(
+			strategy.select_nonces_to_deliver(&state),
+			Some(((20..=23), proof_parameters(true, 4)))
+		);
+	}
+
+	#[test]
+	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),
+			Some(((20..=22), proof_parameters(false, 3)))
+		);
+	}
+
+	#[test]
+	fn message_delivery_strategy_limits_batch_by_messages_count() {
+		let (state, mut strategy) = prepare_strategy();
+
+		// 1 delivery confirmation from target to source is still missing, so we may only
+		// relay 3 new messages
+		let prev_confirmed_nonce_at_source = strategy.latest_confirmed_nonce_at_source.unwrap();
+		strategy.latest_confirmed_nonce_at_source = Some(prev_confirmed_nonce_at_source - 1);
+		strategy.target_nonces.as_mut().unwrap().confirmed_nonce = Some(prev_confirmed_nonce_at_source - 1);
+		assert_eq!(
+			strategy.select_nonces_to_deliver(&state),
+			Some(((20..=22), proof_parameters(false, 3)))
+		);
 	}
 }

bridges/relays/messages-relay/src/message_race_loop.rs

@@ -23,6 +23,7 @@
 use crate::message_lane_loop::ClientState;
 
 use async_trait::async_trait;
+use bp_message_lane::MessageNonce;
 use futures::{
 	future::FutureExt,
 	stream::{FusedStream, StreamExt},
@@ -58,10 +59,32 @@ type SourceClientState<P> = ClientState<<P as MessageRace>::SourceHeaderId, <P a
 /// State of race target client.
 type TargetClientState<P> = ClientState<<P as MessageRace>::TargetHeaderId, <P as MessageRace>::SourceHeaderId>;
 
-/// Nonces on the race client.
+/// Inclusive nonces range.
+pub trait NoncesRange: Debug + Sized {
+	/// Get begin of the range.
+	fn begin(&self) -> MessageNonce;
+	/// Get end of the range.
+	fn end(&self) -> MessageNonce;
+	/// Returns new range with current range nonces that are greater than the passed `nonce`.
+	/// If there are no such nonces, `None` is returned.
+	fn greater_than(self, nonce: MessageNonce) -> Option<Self>;
+}
+
+/// Nonces on the race source client.
 #[derive(Debug, Clone)]
-pub struct ClientNonces<MessageNonce> {
+pub struct SourceClientNonces<NoncesRange> {
-	/// Latest nonce that is known to the client.
+	/// New nonces range known to the client. `New` here means all nonces generated after
+	/// `prev_latest_nonce` passed to the `SourceClient::nonces` method.
+	pub new_nonces: NoncesRange,
+	/// Latest nonce that is confirmed to the bridged client. This nonce only makes
+	/// sense in some races. In other races it is `None`.
+	pub confirmed_nonce: Option<MessageNonce>,
+}
+
+/// Nonces on the race target client.
+#[derive(Debug, Clone)]
+pub struct TargetClientNonces {
+	/// Latest nonce that is known to the target client.
 	pub latest_nonce: MessageNonce,
 	/// Latest nonce that is confirmed to the bridged client. This nonce only makes
 	/// sense in some races. In other races it is `None`.
@@ -73,6 +96,8 @@ pub struct ClientNonces<MessageNonce> {
 pub trait SourceClient<P: MessageRace> {
 	/// Type of error this clients returns.
 	type Error: std::fmt::Debug + MaybeConnectionError;
+	/// Type of nonces range returned by the source client.
+	type NoncesRange: NoncesRange;
 	/// Additional proof parameters required to generate proof.
 	type ProofParameters;
 
@@ -80,14 +105,15 @@ pub trait SourceClient<P: MessageRace> {
 	async fn nonces(
 		&self,
 		at_block: P::SourceHeaderId,
-	) -> Result<(P::SourceHeaderId, ClientNonces<P::MessageNonce>), Self::Error>;
+		prev_latest_nonce: MessageNonce,
+	) -> Result<(P::SourceHeaderId, SourceClientNonces<Self::NoncesRange>), Self::Error>;
 	/// Generate proof for delivering to the target client.
 	async fn generate_proof(
 		&self,
 		at_block: P::SourceHeaderId,
-		nonces: RangeInclusive<P::MessageNonce>,
+		nonces: RangeInclusive<MessageNonce>,
 		proof_parameters: Self::ProofParameters,
-	) -> Result<(P::SourceHeaderId, RangeInclusive<P::MessageNonce>, P::Proof), Self::Error>;
+	) -> Result<(P::SourceHeaderId, RangeInclusive<MessageNonce>, P::Proof), Self::Error>;
 }
 
 /// One of message lane clients, which is target client for the race.
@@ -97,21 +123,21 @@ pub trait TargetClient<P: MessageRace> {
 	type Error: std::fmt::Debug + MaybeConnectionError;
 
 	/// Return nonces that are known to the target client.
-	async fn nonces(
+	async fn nonces(&self, at_block: P::TargetHeaderId)
-		&self,
+		-> Result<(P::TargetHeaderId, TargetClientNonces), Self::Error>;
-		at_block: P::TargetHeaderId,
-	) -> Result<(P::TargetHeaderId, ClientNonces<P::MessageNonce>), Self::Error>;
 	/// Submit proof to the target client.
 	async fn submit_proof(
 		&self,
 		generated_at_block: P::SourceHeaderId,
-		nonces: RangeInclusive<P::MessageNonce>,
+		nonces: RangeInclusive<MessageNonce>,
 		proof: P::Proof,
-	) -> Result<RangeInclusive<P::MessageNonce>, Self::Error>;
+	) -> Result<RangeInclusive<MessageNonce>, Self::Error>;
 }
 
 /// Race strategy.
-pub trait RaceStrategy<SourceHeaderId, TargetHeaderId, MessageNonce, Proof> {
+pub trait RaceStrategy<SourceHeaderId, TargetHeaderId, Proof> {
+	/// Type of nonces range expected from the source client.
+	type SourceNoncesRange: NoncesRange;
 	/// Additional proof parameters required to generate proof.
 	type ProofParameters;
 
@@ -123,25 +149,25 @@ pub trait RaceStrategy<SourceHeaderId, TargetHeaderId, MessageNonce, Proof> {
 	fn best_at_target(&self) -> MessageNonce;
 
 	/// Called when nonces are updated at source node of the race.
-	fn source_nonces_updated(&mut self, at_block: SourceHeaderId, nonce: ClientNonces<MessageNonce>);
+	fn source_nonces_updated(&mut self, at_block: SourceHeaderId, nonces: SourceClientNonces<Self::SourceNoncesRange>);
 	/// Called when nonces are updated at target node of the race.
 	fn target_nonces_updated(
 		&mut self,
-		nonces: ClientNonces<MessageNonce>,
+		nonces: TargetClientNonces,
-		race_state: &mut RaceState<SourceHeaderId, TargetHeaderId, MessageNonce, Proof>,
+		race_state: &mut RaceState<SourceHeaderId, TargetHeaderId, Proof>,
 	);
 	/// 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(
 		&mut self,
-		race_state: &RaceState<SourceHeaderId, TargetHeaderId, MessageNonce, Proof>,
+		race_state: &RaceState<SourceHeaderId, TargetHeaderId, Proof>,
 	) -> Option<(RangeInclusive<MessageNonce>, Self::ProofParameters)>;
 }
 
 /// State of the race.
 #[derive(Debug)]
-pub struct RaceState<SourceHeaderId, TargetHeaderId, MessageNonce, Proof> {
+pub struct RaceState<SourceHeaderId, TargetHeaderId, Proof> {
 	/// Source state, if known.
 	pub source_state: Option<ClientState<SourceHeaderId, TargetHeaderId>>,
 	/// Target state, if known.
@@ -162,8 +188,8 @@ pub async fn run<P: MessageRace, SC: SourceClient<P>>(
 	mut strategy: impl RaceStrategy<
 		P::SourceHeaderId,
 		P::TargetHeaderId,
-		P::MessageNonce,
 		P::Proof,
+		SourceNoncesRange = SC::NoncesRange,
 		ProofParameters = SC::ProofParameters,
 	>,
 ) -> Result<(), FailedClient> {
@@ -337,7 +363,7 @@ pub async fn run<P: MessageRace, SC: SourceClient<P>>(
 					.expect("source_nonces_required is only true when source_state is Some; qed")
 					.best_self
 					.clone();
-				source_nonces.set(race_source.nonces(at_block).fuse());
+				source_nonces.set(race_source.nonces(at_block, strategy.best_at_source()).fuse());
 			} else {
 				source_client_is_online = true;
 			}
@@ -375,9 +401,7 @@ pub async fn run<P: MessageRace, SC: SourceClient<P>>(
 	}
 }
 
-impl<SourceHeaderId, TargetHeaderId, MessageNonce, Proof> Default
+impl<SourceHeaderId, TargetHeaderId, Proof> Default for RaceState<SourceHeaderId, TargetHeaderId, Proof> {
-	for RaceState<SourceHeaderId, TargetHeaderId, MessageNonce, Proof>
-{
 	fn default() -> Self {
 		RaceState {
 			source_state: None,
@@ -392,7 +416,7 @@ impl<SourceHeaderId, TargetHeaderId, MessageNonce, Proof> Default
 fn print_race_progress<P, S>(prev_time: Instant, strategy: &S) -> Instant
 where
 	P: MessageRace,
-	S: RaceStrategy<P::SourceHeaderId, P::TargetHeaderId, P::MessageNonce, P::Proof>,
+	S: RaceStrategy<P::SourceHeaderId, P::TargetHeaderId, P::Proof>,
 {
 	let now_time = Instant::now();
 
@@ -414,13 +438,13 @@ where
 	now_time
 }
 
-fn select_nonces_to_deliver<SourceHeaderId, TargetHeaderId, MessageNonce, Proof, Strategy>(
+fn select_nonces_to_deliver<SourceHeaderId, TargetHeaderId, Proof, Strategy>(
-	race_state: &RaceState<SourceHeaderId, TargetHeaderId, MessageNonce, Proof>,
+	race_state: &RaceState<SourceHeaderId, TargetHeaderId, Proof>,
 	strategy: &mut Strategy,
 ) -> Option<(SourceHeaderId, RangeInclusive<MessageNonce>, Strategy::ProofParameters)>
 where
 	SourceHeaderId: Clone,
-	Strategy: RaceStrategy<SourceHeaderId, TargetHeaderId, MessageNonce, Proof>,
+	Strategy: RaceStrategy<SourceHeaderId, TargetHeaderId, Proof>,
 {
 	race_state.target_state.as_ref().and_then(|target_state| {
 		strategy
@@ -442,8 +466,8 @@ mod tests {
 		const BEST_AT_TARGET: u64 = 8;
 
 		// target node only knows about source' BEST_AT_TARGET block
-		// source node has BEST_AT_SOURCE > BEST_AT_SOURCE block
+		// source node has BEST_AT_SOURCE > BEST_AT_TARGET block
-		let mut race_state = RaceState::<_, _, _, ()> {
+		let mut race_state = RaceState::<_, _, ()> {
 			source_state: Some(ClientState {
 				best_self: HeaderId(BEST_AT_SOURCE, BEST_AT_SOURCE),
 				best_peer: HeaderId(0, 0),
@@ -457,16 +481,16 @@ mod tests {
 		};
 
 		// we have some nonces to deliver and they're generated at GENERATED_AT < BEST_AT_SOURCE
-		let mut strategy = BasicStrategy::new(100);
+		let mut strategy = BasicStrategy::new();
 		strategy.source_nonces_updated(
 			HeaderId(GENERATED_AT, GENERATED_AT),
-			ClientNonces {
+			SourceClientNonces {
-				latest_nonce: 10u64,
+				new_nonces: 0..=10,
 				confirmed_nonce: None,
 			},
 		);
 		strategy.target_nonces_updated(
-			ClientNonces {
+			TargetClientNonces {
 				latest_nonce: 5u64,
 				confirmed_nonce: None,
 			},

bridges/relays/messages-relay/src/message_race_receiving.rs

@@ -18,11 +18,14 @@ use crate::message_lane_loop::{
 	SourceClient as MessageLaneSourceClient, SourceClientState, TargetClient as MessageLaneTargetClient,
 	TargetClientState,
 };
-use crate::message_race_loop::{ClientNonces, MessageRace, SourceClient, TargetClient};
+use crate::message_race_loop::{
+	MessageRace, NoncesRange, SourceClient, SourceClientNonces, TargetClient, TargetClientNonces,
+};
 use crate::message_race_strategy::BasicStrategy;
 use crate::metrics::MessageLaneLoopMetrics;
 
 use async_trait::async_trait;
+use bp_message_lane::MessageNonce;
 use futures::stream::FusedStream;
 use relay_utils::FailedClient;
 use std::{marker::PhantomData, ops::RangeInclusive, time::Duration};
@@ -33,7 +36,7 @@ type ReceivingConfirmationsBasicStrategy<P> = BasicStrategy<
 	<P as MessageLane>::TargetHeaderHash,
 	<P as MessageLane>::SourceHeaderNumber,
 	<P as MessageLane>::SourceHeaderHash,
-	<P as MessageLane>::MessageNonce,
+	RangeInclusive<MessageNonce>,
 	<P as MessageLane>::MessagesReceivingProof,
 >;
 
@@ -60,7 +63,7 @@ pub async fn run<P: MessageLane>(
 		},
 		source_state_updates,
 		stall_timeout,
-		ReceivingConfirmationsBasicStrategy::<P>::new(std::u32::MAX.into()),
+		ReceivingConfirmationsBasicStrategy::<P>::new(),
 	)
 	.await
 }
@@ -72,7 +75,7 @@ impl<P: MessageLane> MessageRace for ReceivingConfirmationsRace<P> {
 	type SourceHeaderId = TargetHeaderIdOf<P>;
 	type TargetHeaderId = SourceHeaderIdOf<P>;
 
-	type MessageNonce = P::MessageNonce;
+	type MessageNonce = MessageNonce;
 	type Proof = P::MessagesReceivingProof;
 
 	fn source_name() -> String {
@@ -98,20 +101,22 @@ where
 	C: MessageLaneTargetClient<P>,
 {
 	type Error = C::Error;
+	type NoncesRange = RangeInclusive<MessageNonce>;
 	type ProofParameters = ();
 
 	async fn nonces(
 		&self,
 		at_block: TargetHeaderIdOf<P>,
-	) -> Result<(TargetHeaderIdOf<P>, ClientNonces<P::MessageNonce>), Self::Error> {
+		prev_latest_nonce: MessageNonce,
+	) -> Result<(TargetHeaderIdOf<P>, SourceClientNonces<Self::NoncesRange>), Self::Error> {
 		let (at_block, latest_received_nonce) = self.client.latest_received_nonce(at_block).await?;
 		if let Some(metrics_msg) = self.metrics_msg.as_ref() {
 			metrics_msg.update_target_latest_received_nonce::<P>(latest_received_nonce);
 		}
 		Ok((
 			at_block,
-			ClientNonces {
+			SourceClientNonces {
-				latest_nonce: latest_received_nonce,
+				new_nonces: prev_latest_nonce + 1..=latest_received_nonce,
 				confirmed_nonce: None,
 			},
 		))
@@ -121,12 +126,12 @@ where
 	async fn generate_proof(
 		&self,
 		at_block: TargetHeaderIdOf<P>,
-		nonces: RangeInclusive<P::MessageNonce>,
+		nonces: RangeInclusive<MessageNonce>,
 		_proof_parameters: Self::ProofParameters,
 	) -> Result<
 		(
 			TargetHeaderIdOf<P>,
-			RangeInclusive<P::MessageNonce>,
+			RangeInclusive<MessageNonce>,
 			P::MessagesReceivingProof,
 		),
 		Self::Error,
@@ -156,14 +161,14 @@ where
 	async fn nonces(
 		&self,
 		at_block: SourceHeaderIdOf<P>,
-	) -> Result<(SourceHeaderIdOf<P>, ClientNonces<P::MessageNonce>), Self::Error> {
+	) -> Result<(SourceHeaderIdOf<P>, TargetClientNonces), Self::Error> {
 		let (at_block, latest_confirmed_nonce) = self.client.latest_confirmed_received_nonce(at_block).await?;
 		if let Some(metrics_msg) = self.metrics_msg.as_ref() {
 			metrics_msg.update_source_latest_confirmed_nonce::<P>(latest_confirmed_nonce);
 		}
 		Ok((
 			at_block,
-			ClientNonces {
+			TargetClientNonces {
 				latest_nonce: latest_confirmed_nonce,
 				confirmed_nonce: None,
 			},
@@ -173,12 +178,51 @@ where
 	async fn submit_proof(
 		&self,
 		generated_at_block: TargetHeaderIdOf<P>,
-		nonces: RangeInclusive<P::MessageNonce>,
+		nonces: RangeInclusive<MessageNonce>,
 		proof: P::MessagesReceivingProof,
-	) -> Result<RangeInclusive<P::MessageNonce>, Self::Error> {
+	) -> Result<RangeInclusive<MessageNonce>, Self::Error> {
 		self.client
 			.submit_messages_receiving_proof(generated_at_block, proof)
 			.await?;
 		Ok(nonces)
 	}
 }
+
+impl NoncesRange for RangeInclusive<MessageNonce> {
+	fn begin(&self) -> MessageNonce {
+		*RangeInclusive::<MessageNonce>::start(self)
+	}
+
+	fn end(&self) -> MessageNonce {
+		*RangeInclusive::<MessageNonce>::end(self)
+	}
+
+	fn greater_than(self, nonce: MessageNonce) -> Option<Self> {
+		let next_nonce = nonce + 1;
+		let end = *self.end();
+		if next_nonce > end {
+			None
+		} else {
+			Some(std::cmp::max(self.begin(), next_nonce)..=end)
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn range_inclusive_works_as_nonces_range() {
+		let range = 20..=30;
+
+		assert_eq!(NoncesRange::begin(&range), 20);
+		assert_eq!(NoncesRange::end(&range), 30);
+		assert_eq!(range.clone().greater_than(10), Some(20..=30));
+		assert_eq!(range.clone().greater_than(19), Some(20..=30));
+		assert_eq!(range.clone().greater_than(20), Some(21..=30));
+		assert_eq!(range.clone().greater_than(25), Some(26..=30));
+		assert_eq!(range.clone().greater_than(29), Some(30..=30));
+		assert_eq!(range.greater_than(30), None);
+	}
+}

bridges/relays/messages-relay/src/message_race_strategy.rs

@@ -17,95 +17,172 @@
 //! 2) new nonces may be proved to target node (i.e. they have appeared at the
 //!    block, which is known to the target node).
 
-use crate::message_race_loop::{ClientNonces, RaceState, RaceStrategy};
+use crate::message_race_loop::{NoncesRange, RaceState, RaceStrategy, SourceClientNonces, TargetClientNonces};
 
-use num_traits::{One, Zero};
+use bp_message_lane::MessageNonce;
 use relay_utils::HeaderId;
 use std::{collections::VecDeque, marker::PhantomData, ops::RangeInclusive};
 
 /// Nonces delivery strategy.
 #[derive(Debug)]
-pub struct BasicStrategy<SourceHeaderNumber, SourceHeaderHash, TargetHeaderNumber, TargetHeaderHash, Nonce, Proof> {
+pub struct BasicStrategy<
+	SourceHeaderNumber,
+	SourceHeaderHash,
+	TargetHeaderNumber,
+	TargetHeaderHash,
+	SourceNoncesRange,
+	Proof,
+> {
 	/// All queued nonces.
-	source_queue: VecDeque<(HeaderId<SourceHeaderHash, SourceHeaderNumber>, Nonce)>,
+	source_queue: VecDeque<(HeaderId<SourceHeaderHash, SourceHeaderNumber>, SourceNoncesRange)>,
 	/// Best nonce known to target node.
-	target_nonce: Nonce,
+	target_nonce: MessageNonce,
-	/// Max nonces to relay in single transaction.
-	max_nonces_to_relay_in_single_tx: Nonce,
 	/// Unused generic types dump.
 	_phantom: PhantomData<(TargetHeaderNumber, TargetHeaderHash, Proof)>,
 }
 
-impl<SourceHeaderNumber, SourceHeaderHash, TargetHeaderNumber, TargetHeaderHash, Nonce: Default, Proof>
+impl<SourceHeaderNumber, SourceHeaderHash, TargetHeaderNumber, TargetHeaderHash, SourceNoncesRange, Proof>
-	BasicStrategy<SourceHeaderNumber, SourceHeaderHash, TargetHeaderNumber, TargetHeaderHash, Nonce, Proof>
+	BasicStrategy<SourceHeaderNumber, SourceHeaderHash, TargetHeaderNumber, TargetHeaderHash, SourceNoncesRange, Proof>
-{
-	/// Create new delivery strategy.
-	pub fn new(max_nonces_to_relay_in_single_tx: Nonce) -> Self {
-		BasicStrategy {
-			source_queue: VecDeque::new(),
-			target_nonce: Default::default(),
-			max_nonces_to_relay_in_single_tx,
-			_phantom: Default::default(),
-		}
-	}
-}
-
-impl<SourceHeaderNumber, SourceHeaderHash, TargetHeaderNumber, TargetHeaderHash, Nonce, Proof>
-	RaceStrategy<
-		HeaderId<SourceHeaderHash, SourceHeaderNumber>,
-		HeaderId<TargetHeaderHash, TargetHeaderNumber>,
-		Nonce,
-		Proof,
-	> for BasicStrategy<SourceHeaderNumber, SourceHeaderHash, TargetHeaderNumber, TargetHeaderHash, Nonce, Proof>
 where
 	SourceHeaderHash: Clone,
 	SourceHeaderNumber: Clone + Ord,
-	Nonce: Clone + Copy + From<u32> + Ord + std::ops::Add<Output = Nonce> + One + Zero,
+	SourceNoncesRange: NoncesRange,
 {
+	/// Create new delivery strategy.
+	pub fn new() -> Self {
+		BasicStrategy {
+			source_queue: VecDeque::new(),
+			target_nonce: Default::default(),
+			_phantom: Default::default(),
+		}
+	}
+
+	/// Should return `Some(nonces)` if we need to deliver proof of `nonces` (and associated
+	/// data) from source to target node.
+	///
+	/// The `selector` function receives range of nonces and should return `None` if the whole
+	/// range needs to be delivered. If there are some nonces in the range that can't be delivered
+	/// 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(
+		&mut self,
+		race_state: &RaceState<
+			HeaderId<SourceHeaderHash, SourceHeaderNumber>,
+			HeaderId<TargetHeaderHash, TargetHeaderNumber>,
+			Proof,
+		>,
+		mut selector: impl FnMut(SourceNoncesRange) -> Option<SourceNoncesRange>,
+	) -> Option<RangeInclusive<MessageNonce>> {
+		// if we have already selected nonces that we want to submit, do nothing
+		if race_state.nonces_to_submit.is_some() {
+			return None;
+		}
+
+		// if we already submitted some nonces, do nothing
+		if race_state.nonces_submitted.is_some() {
+			return None;
+		}
+
+		// 1) we want to deliver all nonces, starting from `target_nonce + 1`
+		// 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.target_state.as_ref()?.best_peer;
+		let mut nonces_end = None;
+
+		while let Some((queued_at, queued_range)) = self.source_queue.pop_front() {
+			// select (sub) range to deliver
+			let queued_range_begin = queued_range.begin();
+			let queued_range_end = queued_range.end();
+			let range_to_requeue = if queued_at.0 > best_header_at_target.0 {
+				// if header that has queued the range is not yet finalized at bridged chain,
+				// we can't prove anything
+				Some(queued_range)
+			} else {
+				// selector returns `Some(range)` if this `range` needs to be requeued
+				selector(queued_range)
+			};
+
+			// requeue (sub) range and update range to deliver
+			match range_to_requeue {
+				Some(range_to_requeue) => {
+					assert!(
+						range_to_requeue.begin() <= range_to_requeue.end()
+							&& range_to_requeue.begin() >= queued_range_begin
+							&& range_to_requeue.end() == queued_range_end,
+						"Incorrect implementation of internal `selector` function. Expected original\
+						range {:?} to end with returned range {:?}",
+						queued_range_begin..=queued_range_end,
+						range_to_requeue,
+					);
+
+					if range_to_requeue.begin() != queued_range_begin {
+						nonces_end = Some(range_to_requeue.begin() - 1);
+					}
+					self.source_queue.push_front((queued_at, range_to_requeue));
+					break;
+				}
+				None => {
+					nonces_end = Some(queued_range_end);
+				}
+			}
+		}
+
+		nonces_end.map(|nonces_end| RangeInclusive::new(self.target_nonce + 1, nonces_end))
+	}
+}
+
+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,
+	SourceHeaderNumber: Clone + Ord,
+	SourceNoncesRange: NoncesRange,
+{
+	type SourceNoncesRange = SourceNoncesRange;
 	type ProofParameters = ();
 
 	fn is_empty(&self) -> bool {
 		self.source_queue.is_empty()
 	}
 
-	fn best_at_source(&self) -> Nonce {
+	fn best_at_source(&self) -> MessageNonce {
-		self.source_queue
+		std::cmp::max(
-			.back()
+			self.source_queue
-			.map(|(_, nonce)| *nonce)
+				.back()
-			.unwrap_or_else(Zero::zero)
+				.map(|(_, range)| range.end())
+				.unwrap_or(self.target_nonce),
+			self.target_nonce,
+		)
 	}
 
-	fn best_at_target(&self) -> Nonce {
+	fn best_at_target(&self) -> MessageNonce {
 		self.target_nonce
 	}
 
 	fn source_nonces_updated(
 		&mut self,
 		at_block: HeaderId<SourceHeaderHash, SourceHeaderNumber>,
-		nonces: ClientNonces<Nonce>,
+		nonces: SourceClientNonces<SourceNoncesRange>,
 	) {
-		let nonce = nonces.latest_nonce;
+		let prev_best_at_source = self.best_at_source();
-
+		self.source_queue.extend(
-		if nonce <= self.target_nonce {
+			nonces
-			return;
+				.new_nonces
-		}
+				.greater_than(prev_best_at_source)
-
+				.into_iter()
-		match self.source_queue.back() {
+				.map(move |range| (at_block.clone(), range)),
-			Some((_, prev_nonce)) if *prev_nonce < nonce => (),
+		)
-			Some(_) => return,
-			None => (),
-		}
-
-		self.source_queue.push_back((at_block, nonce))
 	}
 
 	fn target_nonces_updated(
 		&mut self,
-		nonces: ClientNonces<Nonce>,
+		nonces: TargetClientNonces,
 		race_state: &mut RaceState<
 			HeaderId<SourceHeaderHash, SourceHeaderNumber>,
 			HeaderId<TargetHeaderHash, TargetHeaderNumber>,
-			Nonce,
 			Proof,
 		>,
 	) {
@@ -115,12 +192,15 @@ where
 			return;
 		}
 
-		while let Some(true) = self
+		while let Some(true) = self.source_queue.front().map(|(_, range)| range.begin() <= nonce) {
-			.source_queue
+			let maybe_subrange = self
-			.front()
+				.source_queue
-			.map(|(_, source_nonce)| *source_nonce <= nonce)
+				.pop_front()
-		{
+				.and_then(|(at_block, range)| range.greater_than(nonce).map(|subrange| (at_block, subrange)));
-			self.source_queue.pop_front();
+			if let Some((at_block, subrange)) = maybe_subrange {
+				self.source_queue.push_front((at_block, subrange));
+				break;
+			}
 		}
 
 		let need_to_select_new_nonces = race_state
@@ -149,60 +229,11 @@ where
 		race_state: &RaceState<
 			HeaderId<SourceHeaderHash, SourceHeaderNumber>,
 			HeaderId<TargetHeaderHash, TargetHeaderNumber>,
-			Nonce,
 			Proof,
 		>,
-	) -> Option<(RangeInclusive<Nonce>, Self::ProofParameters)> {
+	) -> Option<(RangeInclusive<MessageNonce>, Self::ProofParameters)> {
-		// if we have already selected nonces that we want to submit, do nothing
+		self.select_nonces_to_deliver_with_selector(race_state, |_| None)
-		if race_state.nonces_to_submit.is_some() {
+			.map(|range| (range, ()))
-			return None;
-		}
-
-		// if we already submitted some nonces, do nothing
-		if race_state.nonces_submitted.is_some() {
-			return None;
-		}
-
-		// 1) we want to deliver all nonces, starting from `target_nonce + 1`
-		// 2) we want to deliver at most `self.max_nonces_to_relay_in_single_tx` nonces in this batch
-		// 3) we can't deliver new nonce until header, that has emitted this nonce, is finalized
-		// by target client
-		let nonces_begin = self.target_nonce + 1.into();
-		let best_header_at_target = &race_state.target_state.as_ref()?.best_peer;
-		let mut nonces_end = None;
-		let mut i = Zero::zero();
-
-		// https://github.com/paritytech/parity-bridges-common/issues/433
-		// TODO: instead of limiting number of messages by number, provide custom limit callback here.
-		// In delivery race it'll be weight-based callback. In receiving race it'll be unlimited callback.
-
-		while i < self.max_nonces_to_relay_in_single_tx {
-			let nonce = nonces_begin + i;
-
-			// if queue is empty, we don't need to prove anything
-			let (first_queued_at, first_queued_nonce) = match self.source_queue.front() {
-				Some((first_queued_at, first_queued_nonce)) => ((*first_queued_at).clone(), *first_queued_nonce),
-				None => break,
-			};
-
-			// if header that has queued the message is not yet finalized at bridged chain,
-			// we can't prove anything
-			if first_queued_at.0 > best_header_at_target.0 {
-				break;
-			}
-
-			// ok, we may deliver this nonce
-			nonces_end = Some(nonce);
-
-			// probably remove it from the queue?
-			if nonce == first_queued_nonce {
-				self.source_queue.pop_front();
-			}
-
-			i = i + One::one();
-		}
-
-		nonces_end.map(|nonces_end| (RangeInclusive::new(nonces_begin, nonces_end), ()))
 	}
 }
 
@@ -211,21 +242,30 @@ mod tests {
 	use super::*;
 	use crate::message_lane::MessageLane;
 	use crate::message_lane_loop::{
-		tests::{header_id, TestMessageLane, TestMessageNonce, TestMessagesProof},
+		tests::{header_id, TestMessageLane, TestMessagesProof},
 		ClientState,
 	};
 
+	type SourceNoncesRange = RangeInclusive<MessageNonce>;
+
 	type BasicStrategy<P> = super::BasicStrategy<
 		<P as MessageLane>::SourceHeaderNumber,
 		<P as MessageLane>::SourceHeaderHash,
 		<P as MessageLane>::TargetHeaderNumber,
 		<P as MessageLane>::TargetHeaderHash,
-		<P as MessageLane>::MessageNonce,
+		SourceNoncesRange,
 		<P as MessageLane>::MessagesProof,
 	>;
 
-	fn nonces(latest_nonce: TestMessageNonce) -> ClientNonces<TestMessageNonce> {
+	fn source_nonces(new_nonces: SourceNoncesRange) -> SourceClientNonces<SourceNoncesRange> {
-		ClientNonces {
+		SourceClientNonces {
+			new_nonces,
+			confirmed_nonce: None,
+		}
+	}
+
+	fn target_nonces(latest_nonce: MessageNonce) -> TargetClientNonces {
+		TargetClientNonces {
 			latest_nonce,
 			confirmed_nonce: None,
 		}
@@ -233,105 +273,116 @@ mod tests {
 
 	#[test]
 	fn strategy_is_empty_works() {
-		let mut strategy = BasicStrategy::<TestMessageLane>::new(4);
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
 		assert_eq!(strategy.is_empty(), true);
-		strategy.source_nonces_updated(header_id(1), nonces(1));
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=1));
 		assert_eq!(strategy.is_empty(), false);
 	}
 
+	#[test]
+	fn best_at_source_is_never_lower_than_target_nonce() {
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
+		assert_eq!(strategy.best_at_source(), 0);
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=5));
+		assert_eq!(strategy.best_at_source(), 5);
+		strategy.target_nonces_updated(target_nonces(10), &mut Default::default());
+		assert_eq!(strategy.source_queue, vec![]);
+		assert_eq!(strategy.best_at_source(), 10);
+	}
+
 	#[test]
 	fn source_nonce_is_never_lower_than_known_target_nonce() {
-		let mut strategy = BasicStrategy::<TestMessageLane>::new(4);
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
-		strategy.target_nonces_updated(nonces(10), &mut Default::default());
+		strategy.target_nonces_updated(target_nonces(10), &mut Default::default());
-		strategy.source_nonces_updated(header_id(1), nonces(5));
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=5));
 		assert_eq!(strategy.source_queue, vec![]);
 	}
 
 	#[test]
 	fn source_nonce_is_never_lower_than_latest_known_source_nonce() {
-		let mut strategy = BasicStrategy::<TestMessageLane>::new(4);
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
-		strategy.source_nonces_updated(header_id(1), nonces(5));
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=5));
-		strategy.source_nonces_updated(header_id(2), nonces(3));
+		strategy.source_nonces_updated(header_id(2), source_nonces(1..=3));
-		strategy.source_nonces_updated(header_id(2), nonces(5));
+		strategy.source_nonces_updated(header_id(2), source_nonces(1..=5));
-		assert_eq!(strategy.source_queue, vec![(header_id(1), 5)]);
+		assert_eq!(strategy.source_queue, vec![(header_id(1), 1..=5)]);
 	}
 
 	#[test]
 	fn target_nonce_is_never_lower_than_latest_known_target_nonce() {
-		let mut strategy = BasicStrategy::<TestMessageLane>::new(4);
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
-		strategy.target_nonces_updated(nonces(10), &mut Default::default());
+		strategy.target_nonces_updated(target_nonces(10), &mut Default::default());
-		strategy.target_nonces_updated(nonces(5), &mut Default::default());
+		strategy.target_nonces_updated(target_nonces(5), &mut Default::default());
 		assert_eq!(strategy.target_nonce, 10);
 	}
 
 	#[test]
 	fn updated_target_nonce_removes_queued_entries() {
-		let mut strategy = BasicStrategy::<TestMessageLane>::new(4);
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
-		strategy.source_nonces_updated(header_id(1), nonces(5));
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=5));
-		strategy.source_nonces_updated(header_id(2), nonces(10));
+		strategy.source_nonces_updated(header_id(2), source_nonces(6..=10));
-		strategy.source_nonces_updated(header_id(3), nonces(15));
+		strategy.source_nonces_updated(header_id(3), source_nonces(11..=15));
-		strategy.source_nonces_updated(header_id(4), nonces(20));
+		strategy.source_nonces_updated(header_id(4), source_nonces(16..=20));
-		strategy.target_nonces_updated(nonces(15), &mut Default::default());
+		strategy.target_nonces_updated(target_nonces(15), &mut Default::default());
-		assert_eq!(strategy.source_queue, vec![(header_id(4), 20)]);
+		assert_eq!(strategy.source_queue, vec![(header_id(4), 16..=20)]);
+		strategy.target_nonces_updated(target_nonces(17), &mut Default::default());
+		assert_eq!(strategy.source_queue, vec![(header_id(4), 18..=20)]);
 	}
 
 	#[test]
 	fn selected_nonces_are_dropped_on_target_nonce_update() {
 		let mut state = RaceState::default();
-		let mut strategy = BasicStrategy::<TestMessageLane>::new(4);
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
 		state.nonces_to_submit = Some((header_id(1), 5..=10, (5..=10, None)));
-		strategy.target_nonces_updated(nonces(7), &mut state);
+		strategy.target_nonces_updated(target_nonces(7), &mut state);
 		assert!(state.nonces_to_submit.is_some());
-		strategy.target_nonces_updated(nonces(10), &mut state);
+		strategy.target_nonces_updated(target_nonces(10), &mut state);
 		assert!(state.nonces_to_submit.is_none());
 	}
 
 	#[test]
 	fn submitted_nonces_are_dropped_on_target_nonce_update() {
 		let mut state = RaceState::default();
-		let mut strategy = BasicStrategy::<TestMessageLane>::new(4);
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
 		state.nonces_submitted = Some(5..=10);
-		strategy.target_nonces_updated(nonces(7), &mut state);
+		strategy.target_nonces_updated(target_nonces(7), &mut state);
 		assert!(state.nonces_submitted.is_some());
-		strategy.target_nonces_updated(nonces(10), &mut state);
+		strategy.target_nonces_updated(target_nonces(10), &mut state);
 		assert!(state.nonces_submitted.is_none());
 	}
 
 	#[test]
 	fn nothing_is_selected_if_something_is_already_selected() {
 		let mut state = RaceState::default();
-		let mut strategy = BasicStrategy::<TestMessageLane>::new(4);
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
 		state.nonces_to_submit = Some((header_id(1), 1..=10, (1..=10, None)));
-		strategy.source_nonces_updated(header_id(1), nonces(10));
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=10));
 		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
 	}
 
 	#[test]
 	fn nothing_is_selected_if_something_is_already_submitted() {
 		let mut state = RaceState::default();
-		let mut strategy = BasicStrategy::<TestMessageLane>::new(4);
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
 		state.nonces_submitted = Some(1..=10);
-		strategy.source_nonces_updated(header_id(1), nonces(10));
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=10));
 		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
 	}
 
 	#[test]
 	fn select_nonces_to_deliver_works() {
-		let mut state = RaceState::<_, _, TestMessageNonce, TestMessagesProof>::default();
+		let mut state = RaceState::<_, _, TestMessagesProof>::default();
-		let mut strategy = BasicStrategy::<TestMessageLane>::new(4);
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
-		strategy.source_nonces_updated(header_id(1), nonces(1));
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=1));
-		strategy.source_nonces_updated(header_id(2), nonces(2));
+		strategy.source_nonces_updated(header_id(2), source_nonces(2..=2));
-		strategy.source_nonces_updated(header_id(3), nonces(6));
+		strategy.source_nonces_updated(header_id(3), source_nonces(6..=6));
-		strategy.source_nonces_updated(header_id(5), nonces(8));
+		strategy.source_nonces_updated(header_id(5), source_nonces(8..=8));
 
 		state.target_state = Some(ClientState {
 			best_self: header_id(0),
 			best_peer: header_id(4),
 		});
-		assert_eq!(strategy.select_nonces_to_deliver(&state), Some((1..=4, ())));
+		assert_eq!(strategy.select_nonces_to_deliver(&state), Some((1..=6, ())));
-		strategy.target_nonces_updated(nonces(4), &mut state);
+		strategy.target_nonces_updated(target_nonces(6), &mut state);
-		assert_eq!(strategy.select_nonces_to_deliver(&state), Some((5..=6, ())));
-		strategy.target_nonces_updated(nonces(6), &mut state);
 		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
 
 		state.target_state = Some(ClientState {
@@ -339,7 +390,57 @@ mod tests {
 			best_peer: header_id(5),
 		});
 		assert_eq!(strategy.select_nonces_to_deliver(&state), Some((7..=8, ())));
-		strategy.target_nonces_updated(nonces(8), &mut state);
+		strategy.target_nonces_updated(target_nonces(8), &mut state);
 		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
 	}
+
+	#[test]
+	fn select_nonces_to_deliver_able_to_split_ranges_with_selector() {
+		let mut state = RaceState::<_, _, TestMessagesProof>::default();
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=100));
+
+		state.target_state = Some(ClientState {
+			best_self: header_id(0),
+			best_peer: header_id(1),
+		});
+		assert_eq!(
+			strategy.select_nonces_to_deliver_with_selector(&state, |_| Some(50..=100)),
+			Some(1..=49),
+		);
+	}
+
+	fn run_panic_test_for_incorrect_selector(
+		invalid_selector: impl Fn(SourceNoncesRange) -> Option<SourceNoncesRange>,
+	) {
+		let mut state = RaceState::<_, _, TestMessagesProof>::default();
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=100));
+		strategy.target_nonces_updated(target_nonces(50), &mut state);
+		state.target_state = Some(ClientState {
+			best_self: header_id(0),
+			best_peer: header_id(1),
+		});
+
+		strategy.select_nonces_to_deliver_with_selector(&state, invalid_selector);
+	}
+
+	#[test]
+	#[should_panic]
+	fn select_nonces_to_deliver_panics_if_selector_returns_empty_range() {
+		#[allow(clippy::reversed_empty_ranges)]
+		run_panic_test_for_incorrect_selector(|_| Some(2..=1))
+	}
+
+	#[test]
+	#[should_panic]
+	fn select_nonces_to_deliver_panics_if_selector_returns_range_that_starts_before_passed_range() {
+		run_panic_test_for_incorrect_selector(|range| Some(range.begin() - 1..=*range.end()))
+	}
+
+	#[test]
+	#[should_panic]
+	fn select_nonces_to_deliver_panics_if_selector_returns_range_with_mismatched_end() {
+		run_panic_test_for_incorrect_selector(|range| Some(range.begin()..=*range.end() + 1))
+	}
 }

bridges/relays/messages-relay/src/metrics.rs

@@ -19,6 +19,7 @@
 use crate::message_lane::MessageLane;
 use crate::message_lane_loop::{SourceClientState, TargetClientState};
 
+use bp_message_lane::MessageNonce;
 use relay_utils::metrics::{register, GaugeVec, Metrics, Opts, Registry, U64};
 
 /// Message lane relay metrics.
@@ -77,30 +78,30 @@ impl MessageLaneLoopMetrics {
 	}
 
 	/// Update latest generated nonce at source.
-	pub fn update_source_latest_generated_nonce<P: MessageLane>(&self, source_latest_generated_nonce: P::MessageNonce) {
+	pub fn update_source_latest_generated_nonce<P: MessageLane>(&self, source_latest_generated_nonce: MessageNonce) {
 		self.lane_state_nonces
 			.with_label_values(&["source_latest_generated"])
-			.set(source_latest_generated_nonce.into());
+			.set(source_latest_generated_nonce);
 	}
 
 	/// Update latest confirmed nonce at source.
-	pub fn update_source_latest_confirmed_nonce<P: MessageLane>(&self, source_latest_confirmed_nonce: P::MessageNonce) {
+	pub fn update_source_latest_confirmed_nonce<P: MessageLane>(&self, source_latest_confirmed_nonce: MessageNonce) {
 		self.lane_state_nonces
 			.with_label_values(&["source_latest_confirmed"])
-			.set(source_latest_confirmed_nonce.into());
+			.set(source_latest_confirmed_nonce);
 	}
 
 	/// Update latest received nonce at target.
-	pub fn update_target_latest_received_nonce<P: MessageLane>(&self, target_latest_generated_nonce: P::MessageNonce) {
+	pub fn update_target_latest_received_nonce<P: MessageLane>(&self, target_latest_generated_nonce: MessageNonce) {
 		self.lane_state_nonces
 			.with_label_values(&["target_latest_received"])
-			.set(target_latest_generated_nonce.into());
+			.set(target_latest_generated_nonce);
 	}
 
 	/// Update latest confirmed nonce at target.
-	pub fn update_target_latest_confirmed_nonce<P: MessageLane>(&self, target_latest_confirmed_nonce: P::MessageNonce) {
+	pub fn update_target_latest_confirmed_nonce<P: MessageLane>(&self, target_latest_confirmed_nonce: MessageNonce) {
 		self.lane_state_nonces
 			.with_label_values(&["target_latest_confirmed"])
-			.set(target_latest_confirmed_nonce.into());
+			.set(target_latest_confirmed_nonce);
 	}
 }

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

@@ -23,7 +23,6 @@ use crate::{ConnectionParams, Error, Result};
 use bp_message_lane::{LaneId, MessageNonce};
 use bp_runtime::InstanceId;
 use codec::Decode;
-use frame_support::weights::Weight;
 use frame_system::AccountInfo;
 use jsonrpsee::common::DeserializeOwned;
 use jsonrpsee::raw::RawClient;
@@ -215,8 +214,8 @@ impl<C: Chain> Client<C> {
 		range: RangeInclusive<MessageNonce>,
 		include_outbound_lane_state: bool,
 		at_block: C::Hash,
-	) -> Result<(Weight, StorageProof)> {
+	) -> Result<StorageProof> {
-		let (dispatch_weight, encoded_trie_nodes) = SubstrateMessageLane::<C, _, _>::prove_messages(
+		let encoded_trie_nodes = SubstrateMessageLane::<C, _, _>::prove_messages(
 			&self.client,
 			instance,
 			lane,
@@ -229,7 +228,7 @@ impl<C: Chain> Client<C> {
 		.map_err(Error::Request)?;
 		let decoded_trie_nodes: Vec<Vec<u8>> =
 			Decode::decode(&mut &encoded_trie_nodes[..]).map_err(Error::ResponseParseFailed)?;
-		Ok((dispatch_weight, StorageProof::new(decoded_trie_nodes)))
+		Ok(StorageProof::new(decoded_trie_nodes))
 	}
 
 	/// Returns proof-of-message(s) delivery.

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

@@ -36,6 +36,8 @@ pub enum Error {
 	ResponseParseFailed(codec::Error),
 	/// Account does not exist on the chain.
 	AccountDoesNotExist,
+	/// Custom logic error.
+	Custom(String),
 }
 
 impl From<WsNewDnsError> for Error {
@@ -69,6 +71,7 @@ impl ToString for Error {
 			Self::Request(e) => e.to_string(),
 			Self::ResponseParseFailed(e) => e.what().to_string(),
 			Self::AccountDoesNotExist => "Account does not exist on the chain".into(),
+			Self::Custom(e) => e.clone(),
 		}
 	}
 }

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

@@ -25,7 +25,6 @@ use crate::chain::Chain;
 
 use bp_message_lane::{LaneId, MessageNonce};
 use bp_runtime::InstanceId;
-use frame_support::weights::Weight;
 use sp_core::{
 	storage::{StorageData, StorageKey},
 	Bytes,
@@ -63,7 +62,7 @@ jsonrpsee::rpc_api! {
 			end: MessageNonce,
 			include_outbound_lane_state: bool,
 			block: Option<C::Hash>,
-		) -> (Weight, Bytes);
+		) -> Bytes;
 
 		#[rpc(method = "messageLane_proveMessagesDelivery", positional_params)]
 		fn prove_messages_delivery(

bridges/relays/substrate/src/messages_source.rs

@@ -25,7 +25,7 @@ use codec::{Decode, Encode};
 use frame_support::weights::Weight;
 use messages_relay::{
 	message_lane::{MessageLane, SourceHeaderIdOf, TargetHeaderIdOf},
-	message_lane_loop::{ClientState, SourceClient, SourceClientState},
+	message_lane_loop::{ClientState, MessageProofParameters, MessageWeightsMap, SourceClient, SourceClientState},
 };
 use relay_substrate_client::{Chain, Client, Error as SubstrateError, HashOf, HeaderIdOf};
 use relay_utils::HeaderId;
@@ -95,7 +95,6 @@ where
 	C::Index: DeserializeOwned,
 	<C::Header as HeaderT>::Number: Into<u64>,
 	P: MessageLane<
-		MessageNonce = MessageNonce,
 		MessagesProof = SubstrateMessagesProof<C>,
 		SourceHeaderNumber = <C::Header as HeaderT>::Number,
 		SourceHeaderHash = <C::Header as HeaderT>::Hash,
@@ -119,7 +118,7 @@ where
 	async fn latest_generated_nonce(
 		&self,
 		id: SourceHeaderIdOf<P>,
-	) -> Result<(SourceHeaderIdOf<P>, P::MessageNonce), Self::Error> {
+	) -> Result<(SourceHeaderIdOf<P>, MessageNonce), Self::Error> {
 		let encoded_response = self
 			.client
 			.state_call(
@@ -129,7 +128,7 @@ where
 				Some(id.1),
 			)
 			.await?;
-		let latest_generated_nonce: P::MessageNonce =
+		let latest_generated_nonce: MessageNonce =
 			Decode::decode(&mut &encoded_response.0[..]).map_err(SubstrateError::ResponseParseFailed)?;
 		Ok((id, latest_generated_nonce))
 	}
@@ -137,7 +136,7 @@ where
 	async fn latest_confirmed_received_nonce(
 		&self,
 		id: SourceHeaderIdOf<P>,
-	) -> Result<(SourceHeaderIdOf<P>, P::MessageNonce), Self::Error> {
+	) -> Result<(SourceHeaderIdOf<P>, MessageNonce), Self::Error> {
 		let encoded_response = self
 			.client
 			.state_call(
@@ -147,29 +146,62 @@ where
 				Some(id.1),
 			)
 			.await?;
-		let latest_received_nonce: P::MessageNonce =
+		let latest_received_nonce: MessageNonce =
 			Decode::decode(&mut &encoded_response.0[..]).map_err(SubstrateError::ResponseParseFailed)?;
 		Ok((id, latest_received_nonce))
 	}
 
+	async fn generated_messages_weights(
+		&self,
+		id: SourceHeaderIdOf<P>,
+		nonces: RangeInclusive<MessageNonce>,
+	) -> Result<MessageWeightsMap, Self::Error> {
+		let encoded_response = self
+			.client
+			.state_call(
+				// TODO: https://github.com/paritytech/parity-bridges-common/issues/457
+				"OutboundLaneApi_messages_dispatch_weight".into(),
+				Bytes((self.lane, nonces.start(), nonces.end()).encode()),
+				Some(id.1),
+			)
+			.await?;
+		let weights: Vec<(MessageNonce, Weight)> =
+			Decode::decode(&mut &encoded_response.0[..]).map_err(SubstrateError::ResponseParseFailed)?;
+
+		let mut expected_nonce = *nonces.start();
+		let mut weights_map = MessageWeightsMap::new();
+		for (nonce, weight) in weights {
+			if nonce != expected_nonce {
+				return Err(SubstrateError::Custom(format!(
+					"Unexpected nonce in messages_dispatch_weight call result. Expected {}, got {}",
+					expected_nonce, nonce
+				)));
+			}
+
+			weights_map.insert(nonce, weight);
+			expected_nonce += 1;
+		}
+		Ok(weights_map)
+	}
+
 	async fn prove_messages(
 		&self,
 		id: SourceHeaderIdOf<P>,
-		nonces: RangeInclusive<P::MessageNonce>,
+		nonces: RangeInclusive<MessageNonce>,
-		include_outbound_lane_state: bool,
+		proof_parameters: MessageProofParameters,
-	) -> Result<(SourceHeaderIdOf<P>, RangeInclusive<P::MessageNonce>, P::MessagesProof), Self::Error> {
+	) -> Result<(SourceHeaderIdOf<P>, RangeInclusive<MessageNonce>, P::MessagesProof), Self::Error> {
-		let (weight, proof) = self
+		let proof = self
 			.client
 			.prove_messages(
 				self.instance,
 				self.lane,
 				nonces.clone(),
-				include_outbound_lane_state,
+				proof_parameters.outbound_state_proof_required,
 				id.1,
 			)
 			.await?;
 		let proof = (id.1, proof, self.lane, *nonces.start(), *nonces.end());
-		Ok((id, nonces, (weight, proof)))
+		Ok((id, nonces, (proof_parameters.dispatch_weight, proof)))
 	}
 
 	async fn submit_messages_receiving_proof(

bridges/relays/substrate/src/messages_target.rs

@@ -53,7 +53,7 @@ pub trait SubstrateTransactionMaker<C: Chain, P: MessageLane>: Clone + Send + Sy
 	async fn make_messages_delivery_transaction(
 		&self,
 		generated_at_header: SourceHeaderIdOf<P>,
-		nonces: RangeInclusive<P::MessageNonce>,
+		nonces: RangeInclusive<MessageNonce>,
 		proof: P::MessagesProof,
 	) -> Result<Self::SignedTransaction, SubstrateError>;
 }
@@ -91,7 +91,6 @@ where
 	C::Index: DeserializeOwned,
 	<C::Header as HeaderT>::Number: Into<u64>,
 	P: MessageLane<
-		MessageNonce = MessageNonce,
 		MessagesReceivingProof = (HashOf<C>, StorageProof, LaneId),
 		TargetHeaderNumber = <C::Header as HeaderT>::Number,
 		TargetHeaderHash = <C::Header as HeaderT>::Hash,
@@ -115,7 +114,7 @@ where
 	async fn latest_received_nonce(
 		&self,
 		id: TargetHeaderIdOf<P>,
-	) -> Result<(TargetHeaderIdOf<P>, P::MessageNonce), Self::Error> {
+	) -> Result<(TargetHeaderIdOf<P>, MessageNonce), Self::Error> {
 		let encoded_response = self
 			.client
 			.state_call(
@@ -125,7 +124,7 @@ where
 				Some(id.1),
 			)
 			.await?;
-		let latest_received_nonce: P::MessageNonce =
+		let latest_received_nonce: MessageNonce =
 			Decode::decode(&mut &encoded_response.0[..]).map_err(SubstrateError::ResponseParseFailed)?;
 		Ok((id, latest_received_nonce))
 	}
@@ -133,7 +132,7 @@ where
 	async fn latest_confirmed_received_nonce(
 		&self,
 		id: TargetHeaderIdOf<P>,
-	) -> Result<(TargetHeaderIdOf<P>, P::MessageNonce), Self::Error> {
+	) -> Result<(TargetHeaderIdOf<P>, MessageNonce), Self::Error> {
 		let encoded_response = self
 			.client
 			.state_call(
@@ -143,7 +142,7 @@ where
 				Some(id.1),
 			)
 			.await?;
-		let latest_received_nonce: P::MessageNonce =
+		let latest_received_nonce: MessageNonce =
 			Decode::decode(&mut &encoded_response.0[..]).map_err(SubstrateError::ResponseParseFailed)?;
 		Ok((id, latest_received_nonce))
 	}
@@ -163,9 +162,9 @@ where
 	async fn submit_messages_proof(
 		&self,
 		generated_at_header: SourceHeaderIdOf<P>,
-		nonces: RangeInclusive<P::MessageNonce>,
+		nonces: RangeInclusive<MessageNonce>,
 		proof: P::MessagesProof,
-	) -> Result<RangeInclusive<P::MessageNonce>, Self::Error> {
+	) -> Result<RangeInclusive<MessageNonce>, Self::Error> {
 		let tx = self
 			.tx_maker
 			.make_messages_delivery_transaction(generated_at_header, nonces.clone(), proof)

bridges/relays/substrate/src/millau_messages_to_rialto.rs

@@ -52,7 +52,6 @@ impl MessageLane for MillauMessagesToRialto {
 	const SOURCE_NAME: &'static str = "Millau";
 	const TARGET_NAME: &'static str = "Rialto";
 
-	type MessageNonce = MessageNonce;
 	type MessagesProof = FromMillauMessagesProof;
 	type MessagesReceivingProof = FromRialtoMessagesReceivingProof;
 
@@ -144,7 +143,12 @@ pub fn run(
 			target_tick: rialto_tick,
 			reconnect_delay,
 			stall_timeout,
-			max_unconfirmed_nonces_at_target: bp_rialto::MAX_UNCONFIRMED_MESSAGES_AT_INBOUND_LANE,
+			delivery_params: messages_relay::message_lane_loop::MessageDeliveryParams {
+				max_unconfirmed_nonces_at_target: bp_rialto::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_rialto::MAXIMUM_EXTRINSIC_WEIGHT,
+			},
 		},
 		MillauSourceClient::new(
 			millau_client.clone(),