Limit max number of messages in delivery transaction (#541)

* limit max number of messages in delivery tx

* support max-messages-in-delivery-tx in relayer

* clippy

* clippy

* Update modules/message-lane/src/lib.rs

Co-authored-by: Tomasz Drwięga <tomusdrw@users.noreply.github.com>

Co-authored-by: Tomasz Drwięga <tomusdrw@users.noreply.github.com>

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

@@ -322,13 +322,17 @@ impl pallet_shift_session_manager::Trait for Runtime {}
 
 parameter_types! {
 	pub const MaxMessagesToPruneAtOnce: bp_message_lane::MessageNonce = 8;
-	pub const MaxUnconfirmedMessagesAtInboundLane: bp_message_lane::MessageNonce = bp_millau::MAX_UNCONFIRMED_MESSAGES_AT_INBOUND_LANE;
+	pub const MaxUnconfirmedMessagesAtInboundLane: bp_message_lane::MessageNonce =
+		bp_millau::MAX_UNCONFIRMED_MESSAGES_AT_INBOUND_LANE;
+	pub const MaxMessagesInDeliveryTransaction: bp_message_lane::MessageNonce =
+		bp_millau::MAX_MESSAGES_IN_DELIVERY_TRANSACTION;
 }
 
 impl pallet_message_lane::Trait for Runtime {
 	type Event = Event;
 	type MaxMessagesToPruneAtOnce = MaxMessagesToPruneAtOnce;
 	type MaxUnconfirmedMessagesAtInboundLane = MaxUnconfirmedMessagesAtInboundLane;
+	type MaxMessagesInDeliveryTransaction = MaxMessagesInDeliveryTransaction;
 
 	type OutboundPayload = crate::rialto_messages::ToRialtoMessagePayload;
 	type OutboundMessageFee = Balance;

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

@@ -193,7 +193,8 @@ impl SourceHeaderChain<bp_rialto::Balance> for Rialto {
 
 	fn verify_messages_proof(
 		proof: Self::MessagesProof,
+		max_messages: MessageNonce,
 	) -> Result<ProvedMessages<Message<bp_rialto::Balance>>, Self::Error> {
-		messages::target::verify_messages_proof::<WithRialtoMessageBridge, Runtime>(proof)
+		messages::target::verify_messages_proof::<WithRialtoMessageBridge, Runtime>(proof, max_messages)
 	}
 }

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

@@ -429,13 +429,17 @@ impl pallet_shift_session_manager::Trait for Runtime {}
 
 parameter_types! {
 	pub const MaxMessagesToPruneAtOnce: bp_message_lane::MessageNonce = 8;
-	pub const MaxUnconfirmedMessagesAtInboundLane: bp_message_lane::MessageNonce = bp_rialto::MAX_UNCONFIRMED_MESSAGES_AT_INBOUND_LANE;
+	pub const MaxUnconfirmedMessagesAtInboundLane: bp_message_lane::MessageNonce =
+		bp_rialto::MAX_UNCONFIRMED_MESSAGES_AT_INBOUND_LANE;
+	pub const MaxMessagesInDeliveryTransaction: bp_message_lane::MessageNonce =
+		bp_rialto::MAX_MESSAGES_IN_DELIVERY_TRANSACTION;
 }
 
 impl pallet_message_lane::Trait for Runtime {
 	type Event = Event;
 	type MaxMessagesToPruneAtOnce = MaxMessagesToPruneAtOnce;
 	type MaxUnconfirmedMessagesAtInboundLane = MaxUnconfirmedMessagesAtInboundLane;
+	type MaxMessagesInDeliveryTransaction = MaxMessagesInDeliveryTransaction;
 
 	type OutboundPayload = crate::millau_messages::ToMillauMessagePayload;
 	type OutboundMessageFee = Balance;

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

@@ -193,7 +193,8 @@ impl SourceHeaderChain<bp_millau::Balance> for Millau {
 
 	fn verify_messages_proof(
 		proof: Self::MessagesProof,
+		max_messages: MessageNonce,
 	) -> Result<ProvedMessages<Message<bp_millau::Balance>>, Self::Error> {
-		messages::target::verify_messages_proof::<WithMillauMessageBridge, Runtime>(proof)
+		messages::target::verify_messages_proof::<WithMillauMessageBridge, Runtime>(proof, max_messages)
 	}
 }

bridges/bin/runtime-common/Cargo.toml

@@ -9,6 +9,7 @@ license = "GPL-3.0-or-later WITH Classpath-exception-2.0"
 
 [dependencies]
 codec = { package = "parity-scale-codec", version = "1.3.1", default-features = false, features = ["derive"] }
+hash-db = { version = "0.15.2", default-features = false }
 
 # Bridge dependencies
 
@@ -34,6 +35,7 @@ std = [
 	"bp-runtime/std",
 	"codec/std",
 	"frame-support/std",
+	"hash-db/std",
 	"pallet-bridge-call-dispatch/std",
 	"pallet-message-lane/std",
 	"pallet-substrate-bridge/std",

bridges/bin/runtime-common/src/messages.rs

@@ -29,6 +29,8 @@ use bp_message_lane::{
 use bp_runtime::InstanceId;
 use codec::{Compact, Decode, Input};
 use frame_support::{traits::Instance, RuntimeDebug};
+use hash_db::Hasher;
+use pallet_substrate_bridge::StorageProofChecker;
 use sp_runtime::traits::{CheckedAdd, CheckedDiv, CheckedMul};
 use sp_std::{cmp::PartialOrd, marker::PhantomData, ops::RangeInclusive, vec::Vec};
 use sp_trie::StorageProof;
@@ -344,6 +346,7 @@ pub mod target {
 	/// Verify proof of Bridged -> This chain messages.
 	pub fn verify_messages_proof<B: MessageBridge, ThisRuntime>(
 		proof: FromBridgedChainMessagesProof<B>,
+		max_messages: MessageNonce,
 	) -> Result<ProvedMessages<Message<BalanceOf<BridgedChain<B>>>>, &'static str>
 	where
 		ThisRuntime: pallet_substrate_bridge::Trait,
@@ -351,64 +354,147 @@ pub mod target {
 		HashOf<BridgedChain<B>>:
 			Into<bp_runtime::HashOf<<ThisRuntime as pallet_substrate_bridge::Trait>::BridgedChain>>,
 	{
-		let (bridged_header_hash, bridged_storage_proof, lane_id, begin, end) = proof;
+		verify_messages_proof_with_parser::<B, _, _>(
-		pallet_substrate_bridge::Module::<ThisRuntime>::parse_finalized_storage_proof(
+			proof,
-			bridged_header_hash.into(),
+			max_messages,
-			bridged_storage_proof,
+			|bridged_header_hash, bridged_storage_proof| {
-			|storage| {
+				pallet_substrate_bridge::Module::<ThisRuntime>::parse_finalized_storage_proof(
-				// Read messages first. All messages that are claimed to be in the proof must
+					bridged_header_hash.into(),
-				// be in the proof. So any error in `read_value`, or even missing value is fatal.
+					bridged_storage_proof,
-				//
+					|storage_adapter| storage_adapter,
-				// Mind that we allow proofs with no messages if outbound lane state is proved.
+				)
-				let mut messages = Vec::with_capacity(end.saturating_sub(begin) as _);
+				.map(|storage| StorageProofCheckerAdapter::<_, B, ThisRuntime> {
-				for nonce in begin..=end {
+					storage,
-					let message_key = MessageKey { lane_id, nonce };
+					_dummy: Default::default(),
-					let storage_message_key = pallet_message_lane::storage_keys::message_key::<
+				})
-						ThisRuntime,
+				.map_err(|err| MessageProofError::Custom(err.into()))
-						MessageLaneInstanceOf<BridgedChain<B>>,
-					>(&lane_id, nonce);
-					let raw_message_data = storage
-						.read_value(storage_message_key.0.as_ref())
-						.map_err(|_| "Failed to read message from storage proof")?
-						.ok_or("Message is missing from the messages proof")?;
-					let message_data = MessageData::<BalanceOf<BridgedChain<B>>>::decode(&mut &raw_message_data[..])
-						.map_err(|_| "Failed to decode message from the proof")?;
-					messages.push(Message {
-						key: message_key,
-						data: message_data,
-					});
-				}
-
-				// Now let's check if proof contains outbound lane state proof. It is optional, so we
-				// simply ignore `read_value` errors and missing value.
-				let mut proved_lane_messages = ProvedLaneMessages {
-					lane_state: None,
-					messages,
-				};
-				let storage_outbound_lane_data_key = pallet_message_lane::storage_keys::outbound_lane_data_key::<
-					MessageLaneInstanceOf<BridgedChain<B>>,
-				>(&lane_id);
-				let raw_outbound_lane_data = storage.read_value(storage_outbound_lane_data_key.0.as_ref());
-				if let Ok(Some(raw_outbound_lane_data)) = raw_outbound_lane_data {
-					proved_lane_messages.lane_state = Some(
-						OutboundLaneData::decode(&mut &raw_outbound_lane_data[..])
-							.map_err(|_| "Failed to decode outbound lane data from the proof")?,
-					);
-				}
-
-				// Now we may actually check if the proof is empty or not.
-				if proved_lane_messages.lane_state.is_none() && proved_lane_messages.messages.is_empty() {
-					return Err("Messages proof is empty");
-				}
-
-				// We only support single lane messages in this schema
-				let mut proved_messages = ProvedMessages::new();
-				proved_messages.insert(lane_id, proved_lane_messages);
-
-				Ok(proved_messages)
 			},
 		)
-		.map_err(<&'static str>::from)?
+		.map_err(Into::into)
+	}
+
+	#[derive(Debug, PartialEq)]
+	pub(crate) enum MessageProofError {
+		Empty,
+		TooManyMessages,
+		MissingRequiredMessage,
+		FailedToDecodeMessage,
+		FailedToDecodeOutboundLaneState,
+		Custom(&'static str),
+	}
+
+	impl From<MessageProofError> for &'static str {
+		fn from(err: MessageProofError) -> &'static str {
+			match err {
+				MessageProofError::Empty => "Messages proof is empty",
+				MessageProofError::TooManyMessages => "Too many messages in the proof",
+				MessageProofError::MissingRequiredMessage => "Message is missing from the proof",
+				MessageProofError::FailedToDecodeMessage => "Failed to decode message from the proof",
+				MessageProofError::FailedToDecodeOutboundLaneState => {
+					"Failed to decode outbound lane data from the proof"
+				}
+				MessageProofError::Custom(err) => err,
+			}
+		}
+	}
+
+	pub(crate) trait MessageProofParser {
+		fn read_raw_outbound_lane_data(&self, lane_id: &LaneId) -> Option<Vec<u8>>;
+		fn read_raw_message(&self, message_key: &MessageKey) -> Option<Vec<u8>>;
+	}
+
+	struct StorageProofCheckerAdapter<H: Hasher, B, ThisRuntime> {
+		storage: StorageProofChecker<H>,
+		_dummy: sp_std::marker::PhantomData<(B, ThisRuntime)>,
+	}
+
+	impl<H, B, ThisRuntime> MessageProofParser for StorageProofCheckerAdapter<H, B, ThisRuntime>
+	where
+		H: Hasher,
+		B: MessageBridge,
+		ThisRuntime: pallet_message_lane::Trait<MessageLaneInstanceOf<BridgedChain<B>>>,
+	{
+		fn read_raw_outbound_lane_data(&self, lane_id: &LaneId) -> Option<Vec<u8>> {
+			let storage_outbound_lane_data_key = pallet_message_lane::storage_keys::outbound_lane_data_key::<
+				MessageLaneInstanceOf<BridgedChain<B>>,
+			>(lane_id);
+			self.storage
+				.read_value(storage_outbound_lane_data_key.0.as_ref())
+				.ok()?
+		}
+
+		fn read_raw_message(&self, message_key: &MessageKey) -> Option<Vec<u8>> {
+			let storage_message_key = pallet_message_lane::storage_keys::message_key::<
+				ThisRuntime,
+				MessageLaneInstanceOf<BridgedChain<B>>,
+			>(&message_key.lane_id, message_key.nonce);
+			self.storage.read_value(storage_message_key.0.as_ref()).ok()?
+		}
+	}
+
+	/// Verify proof of Bridged -> This chain messages using given message proof parser.
+	pub(crate) fn verify_messages_proof_with_parser<B: MessageBridge, BuildParser, Parser>(
+		proof: FromBridgedChainMessagesProof<B>,
+		max_messages: MessageNonce,
+		build_parser: BuildParser,
+	) -> Result<ProvedMessages<Message<BalanceOf<BridgedChain<B>>>>, MessageProofError>
+	where
+		BuildParser: FnOnce(HashOf<BridgedChain<B>>, StorageProof) -> Result<Parser, MessageProofError>,
+		Parser: MessageProofParser,
+	{
+		let (bridged_header_hash, bridged_storage_proof, lane_id, begin, end) = proof;
+
+		// receiving proofs where end < begin is ok (if proof includes outbound lane state)
+		// => hence unwrap_or(0)
+		let messages_in_the_proof = end.checked_sub(begin).and_then(|diff| diff.checked_add(1)).unwrap_or(0);
+		if messages_in_the_proof > max_messages {
+			return Err(MessageProofError::TooManyMessages);
+		}
+
+		let parser = build_parser(bridged_header_hash, bridged_storage_proof)?;
+
+		// Read messages first. All messages that are claimed to be in the proof must
+		// be in the proof. So any error in `read_value`, or even missing value is fatal.
+		//
+		// Mind that we allow proofs with no messages if outbound lane state is proved.
+		let mut messages = Vec::with_capacity(end.saturating_sub(begin) as _);
+		for nonce in begin..=end {
+			let message_key = MessageKey { lane_id, nonce };
+			let raw_message_data = parser
+				.read_raw_message(&message_key)
+				.ok_or(MessageProofError::MissingRequiredMessage)?;
+			let message_data = MessageData::<BalanceOf<BridgedChain<B>>>::decode(&mut &raw_message_data[..])
+				.map_err(|_| MessageProofError::FailedToDecodeMessage)?;
+			messages.push(Message {
+				key: message_key,
+				data: message_data,
+			});
+		}
+
+		// Now let's check if proof contains outbound lane state proof. It is optional, so we
+		// simply ignore `read_value` errors and missing value.
+		let mut proved_lane_messages = ProvedLaneMessages {
+			lane_state: None,
+			messages,
+		};
+		let raw_outbound_lane_data = parser.read_raw_outbound_lane_data(&lane_id);
+		if let Some(raw_outbound_lane_data) = raw_outbound_lane_data {
+			proved_lane_messages.lane_state = Some(
+				OutboundLaneData::decode(&mut &raw_outbound_lane_data[..])
+					.map_err(|_| MessageProofError::FailedToDecodeOutboundLaneState)?,
+			);
+		}
+
+		// Now we may actually check if the proof is empty or not.
+		if proved_lane_messages.lane_state.is_none() && proved_lane_messages.messages.is_empty() {
+			return Err(MessageProofError::Empty);
+		}
+
+		// We only support single lane messages in this schema
+		let mut proved_messages = ProvedMessages::new();
+		proved_messages.insert(lane_id, proved_lane_messages);
+
+		Ok(proved_messages)
 	}
 }
 
@@ -417,6 +503,7 @@ mod tests {
 	use super::*;
 	use codec::{Decode, Encode};
 	use frame_support::weights::Weight;
+	use std::ops::RangeInclusive;
 
 	const DELIVERY_TRANSACTION_WEIGHT: Weight = 100;
 	const DELIVERY_CONFIRMATION_TRANSACTION_WEIGHT: Weight = 100;
@@ -685,4 +772,207 @@ mod tests {
 			.is_ok(),
 		);
 	}
+
+	#[derive(Debug)]
+	struct TestMessageProofParser {
+		failing: bool,
+		messages: RangeInclusive<MessageNonce>,
+		outbound_lane_data: Option<OutboundLaneData>,
+	}
+
+	impl target::MessageProofParser for TestMessageProofParser {
+		fn read_raw_outbound_lane_data(&self, _lane_id: &LaneId) -> Option<Vec<u8>> {
+			if self.failing {
+				Some(vec![])
+			} else {
+				self.outbound_lane_data.clone().map(|data| data.encode())
+			}
+		}
+
+		fn read_raw_message(&self, message_key: &MessageKey) -> Option<Vec<u8>> {
+			if self.failing {
+				Some(vec![])
+			} else if self.messages.contains(&message_key.nonce) {
+				Some(
+					MessageData::<BridgedChainBalance> {
+						payload: message_key.nonce.encode(),
+						fee: BridgedChainBalance(0),
+					}
+					.encode(),
+				)
+			} else {
+				None
+			}
+		}
+	}
+
+	#[allow(clippy::reversed_empty_ranges)]
+	fn no_messages_range() -> RangeInclusive<MessageNonce> {
+		1..=0
+	}
+
+	#[test]
+	fn messages_proof_is_rejected_if_there_are_too_many_messages() {
+		assert_eq!(
+			target::verify_messages_proof_with_parser::<OnThisChainBridge, _, TestMessageProofParser>(
+				(Default::default(), StorageProof::new(vec![]), Default::default(), 1, 11),
+				10,
+				|_, _| unreachable!(),
+			),
+			Err(target::MessageProofError::TooManyMessages),
+		);
+	}
+
+	#[test]
+	fn message_proof_is_rejected_if_build_parser_fails() {
+		assert_eq!(
+			target::verify_messages_proof_with_parser::<OnThisChainBridge, _, TestMessageProofParser>(
+				(Default::default(), StorageProof::new(vec![]), Default::default(), 1, 10),
+				10,
+				|_, _| Err(target::MessageProofError::Custom("test")),
+			),
+			Err(target::MessageProofError::Custom("test")),
+		);
+	}
+
+	#[test]
+	fn message_proof_is_rejected_if_required_message_is_missing() {
+		assert_eq!(
+			target::verify_messages_proof_with_parser::<OnThisChainBridge, _, _>(
+				(Default::default(), StorageProof::new(vec![]), Default::default(), 1, 10),
+				10,
+				|_, _| Ok(TestMessageProofParser {
+					failing: false,
+					messages: 1..=5,
+					outbound_lane_data: None,
+				}),
+			),
+			Err(target::MessageProofError::MissingRequiredMessage),
+		);
+	}
+
+	#[test]
+	fn message_proof_is_rejected_if_message_decode_fails() {
+		assert_eq!(
+			target::verify_messages_proof_with_parser::<OnThisChainBridge, _, _>(
+				(Default::default(), StorageProof::new(vec![]), Default::default(), 1, 10),
+				10,
+				|_, _| Ok(TestMessageProofParser {
+					failing: true,
+					messages: 1..=10,
+					outbound_lane_data: None,
+				}),
+			),
+			Err(target::MessageProofError::FailedToDecodeMessage),
+		);
+	}
+
+	#[test]
+	fn message_proof_is_rejected_if_outbound_lane_state_decode_fails() {
+		assert_eq!(
+			target::verify_messages_proof_with_parser::<OnThisChainBridge, _, _>(
+				(Default::default(), StorageProof::new(vec![]), Default::default(), 1, 0),
+				10,
+				|_, _| Ok(TestMessageProofParser {
+					failing: true,
+					messages: no_messages_range(),
+					outbound_lane_data: Some(OutboundLaneData {
+						oldest_unpruned_nonce: 1,
+						latest_received_nonce: 1,
+						latest_generated_nonce: 1,
+					}),
+				}),
+			),
+			Err(target::MessageProofError::FailedToDecodeOutboundLaneState),
+		);
+	}
+
+	#[test]
+	fn message_proof_is_rejected_if_it_is_empty() {
+		assert_eq!(
+			target::verify_messages_proof_with_parser::<OnThisChainBridge, _, _>(
+				(Default::default(), StorageProof::new(vec![]), Default::default(), 1, 0),
+				10,
+				|_, _| Ok(TestMessageProofParser {
+					failing: false,
+					messages: no_messages_range(),
+					outbound_lane_data: None,
+				}),
+			),
+			Err(target::MessageProofError::Empty),
+		);
+	}
+
+	#[test]
+	fn non_empty_message_proof_without_messages_is_accepted() {
+		assert_eq!(
+			target::verify_messages_proof_with_parser::<OnThisChainBridge, _, _>(
+				(Default::default(), StorageProof::new(vec![]), Default::default(), 1, 0),
+				10,
+				|_, _| Ok(TestMessageProofParser {
+					failing: false,
+					messages: no_messages_range(),
+					outbound_lane_data: Some(OutboundLaneData {
+						oldest_unpruned_nonce: 1,
+						latest_received_nonce: 1,
+						latest_generated_nonce: 1,
+					}),
+				}),
+			),
+			Ok(vec![(
+				Default::default(),
+				ProvedLaneMessages {
+					lane_state: Some(OutboundLaneData {
+						oldest_unpruned_nonce: 1,
+						latest_received_nonce: 1,
+						latest_generated_nonce: 1,
+					}),
+					messages: Vec::new(),
+				},
+			)]
+			.into_iter()
+			.collect()),
+		);
+	}
+
+	#[test]
+	fn non_empty_message_proof_is_accepted() {
+		assert_eq!(
+			target::verify_messages_proof_with_parser::<OnThisChainBridge, _, _>(
+				(Default::default(), StorageProof::new(vec![]), Default::default(), 1, 1),
+				10,
+				|_, _| Ok(TestMessageProofParser {
+					failing: false,
+					messages: 1..=1,
+					outbound_lane_data: Some(OutboundLaneData {
+						oldest_unpruned_nonce: 1,
+						latest_received_nonce: 1,
+						latest_generated_nonce: 1,
+					}),
+				}),
+			),
+			Ok(vec![(
+				Default::default(),
+				ProvedLaneMessages {
+					lane_state: Some(OutboundLaneData {
+						oldest_unpruned_nonce: 1,
+						latest_received_nonce: 1,
+						latest_generated_nonce: 1,
+					}),
+					messages: vec![Message {
+						key: MessageKey {
+							lane_id: Default::default(),
+							nonce: 1
+						},
+						data: MessageData {
+							payload: 1u64.encode(),
+							fee: BridgedChainBalance(0)
+						},
+					}],
+				},
+			)]
+			.into_iter()
+			.collect()),
+		);
+	}
 }

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

@@ -57,8 +57,12 @@ pub mod instant_payments;
 mod mock;
 
 // TODO: update me (https://github.com/paritytech/parity-bridges-common/issues/78)
-/// Upper bound of delivery transaction weight.
+/// Weight of message delivery without any code that is touching messages.
-const DELIVERY_BASE_WEIGHT: Weight = 0;
+const DELIVERY_OVERHEAD_WEIGHT: Weight = 0;
+// TODO: update me (https://github.com/paritytech/parity-bridges-common/issues/78)
+/// Single-message delivery weight. This shall not include message dispatch weight and
+/// any delivery transaction code that is not specific to this message.
+const SINGLE_MESSAGE_DELIVERY_WEIGHT: Weight = 0;
 
 /// The module configuration trait
 pub trait Trait<I = DefaultInstance>: frame_system::Trait {
@@ -82,6 +86,11 @@ pub trait Trait<I = DefaultInstance>: frame_system::Trait {
 	/// transaction#2 with individual messages [3; 4], this would be treated as single "Message" and
 	/// would occupy single unit of `MaxUnconfirmedMessagesAtInboundLane` limit.
 	type MaxUnconfirmedMessagesAtInboundLane: Get<MessageNonce>;
+	/// Maximal number of messages in single delivery transaction. This directly affects the base
+	/// weight of the delivery transaction.
+	///
+	/// All transactions that deliver more messages than this number, are rejected.
+	type MaxMessagesInDeliveryTransaction: Get<MessageNonce>;
 
 	/// Payload type of outbound messages. This payload is dispatched on the bridged chain.
 	type OutboundPayload: Parameter;
@@ -305,7 +314,13 @@ decl_module! {
 		}
 
 		/// Receive messages proof from bridged chain.
-		#[weight = DELIVERY_BASE_WEIGHT + dispatch_weight]
+		#[weight = DELIVERY_OVERHEAD_WEIGHT
+			.saturating_add(
+				T::MaxMessagesInDeliveryTransaction::get()
+					.saturating_mul(SINGLE_MESSAGE_DELIVERY_WEIGHT)
+			)
+			.saturating_add(*dispatch_weight)
+		]
 		pub fn receive_messages_proof(
 			origin,
 			relayer_id: T::InboundRelayer,
@@ -316,7 +331,11 @@ decl_module! {
 			let _ = ensure_signed(origin)?;
 
 			// verify messages proof && convert proof into messages
-			let messages = verify_and_decode_messages_proof::<T::SourceHeaderChain, T::InboundMessageFee, T::InboundPayload>(proof)
+			let messages = verify_and_decode_messages_proof::<
+				T::SourceHeaderChain,
+				T::InboundMessageFee,
+				T::InboundPayload,
+			>(proof, T::MaxMessagesInDeliveryTransaction::get())
 				.map_err(|err| {
 					frame_support::debug::trace!(
 						"Rejecting invalid messages proof: {:?}",
@@ -627,8 +646,9 @@ impl<T: Trait<I>, I: Instance> OutboundLaneStorage for RuntimeOutboundLaneStorag
 /// Verify messages proof and return proved messages with decoded payload.
 fn verify_and_decode_messages_proof<Chain: SourceHeaderChain<Fee>, Fee, DispatchPayload: Decode>(
 	proof: Chain::MessagesProof,
+	max_messages: MessageNonce,
 ) -> Result<ProvedMessages<DispatchMessage<DispatchPayload, Fee>>, Chain::Error> {
-	Chain::verify_messages_proof(proof).map(|messages_by_lane| {
+	Chain::verify_messages_proof(proof, max_messages).map(|messages_by_lane| {
 		messages_by_lane
 			.into_iter()
 			.map(|(lane, lane_data)| {

bridges/modules/message-lane/src/mock.rs

@@ -100,12 +100,14 @@ impl frame_system::Trait for TestRuntime {
 parameter_types! {
 	pub const MaxMessagesToPruneAtOnce: u64 = 10;
 	pub const MaxUnconfirmedMessagesAtInboundLane: u64 = 16;
+	pub const MaxMessagesInDeliveryTransaction: u64 = 128;
 }
 
 impl Trait for TestRuntime {
 	type Event = TestEvent;
 	type MaxMessagesToPruneAtOnce = MaxMessagesToPruneAtOnce;
 	type MaxUnconfirmedMessagesAtInboundLane = MaxUnconfirmedMessagesAtInboundLane;
+	type MaxMessagesInDeliveryTransaction = MaxMessagesInDeliveryTransaction;
 
 	type OutboundPayload = TestPayload;
 	type OutboundMessageFee = TestMessageFee;
@@ -279,6 +281,7 @@ impl SourceHeaderChain<TestMessageFee> for TestSourceHeaderChain {
 
 	fn verify_messages_proof(
 		proof: Self::MessagesProof,
+		_max_messages: MessageNonce,
 	) -> Result<ProvedMessages<Message<TestMessageFee>>, Self::Error> {
 		proof
 			.result

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

@@ -16,7 +16,7 @@
 
 //! Primitives of message lane module, that are used on the target chain.
 
-use crate::{LaneId, Message, MessageData, MessageKey, OutboundLaneData};
+use crate::{LaneId, Message, MessageData, MessageKey, MessageNonce, OutboundLaneData};
 
 use codec::{Decode, Encode, Error as CodecError};
 use frame_support::{weights::Weight, Parameter, RuntimeDebug};
@@ -67,9 +67,15 @@ pub trait SourceHeaderChain<Fee> {
 
 	/// Verify messages proof and return proved messages.
 	///
+	/// Returns error if either proof is incorrect, or the number of messages in the proof
+	/// is larger than `max_messages`.
+	///
 	/// Messages vector is required to be sorted by nonce within each lane. Out-of-order
 	/// messages will be rejected.
-	fn verify_messages_proof(proof: Self::MessagesProof) -> Result<ProvedMessages<Message<Fee>>, Self::Error>;
+	fn verify_messages_proof(
+		proof: Self::MessagesProof,
+		max_messages: MessageNonce,
+	) -> Result<ProvedMessages<Message<Fee>>, Self::Error>;
 }
 
 /// Called when inbound message is received.

bridges/primitives/millau/src/lib.rs

@@ -76,6 +76,9 @@ pub const AVAILABLE_BLOCK_RATIO: u32 = 75;
 /// transactions minus 10% for initialization).
 pub const MAXIMUM_EXTRINSIC_WEIGHT: Weight = MAXIMUM_BLOCK_WEIGHT / 100 * (AVAILABLE_BLOCK_RATIO as Weight - 10);
 
+// TODO: may need to be updated after https://github.com/paritytech/parity-bridges-common/issues/78
+/// Maximal number of messages in single delivery transaction.
+pub const MAX_MESSAGES_IN_DELIVERY_TRANSACTION: MessageNonce = 1024;
 /// Maximal number of unconfirmed messages at inbound lane.
 pub const MAX_UNCONFIRMED_MESSAGES_AT_INBOUND_LANE: MessageNonce = 1024;
 

bridges/primitives/rialto/src/lib.rs

@@ -38,6 +38,9 @@ pub const AVAILABLE_BLOCK_RATIO: u32 = 75;
 /// transactions minus 10% for initialization).
 pub const MAXIMUM_EXTRINSIC_WEIGHT: Weight = MAXIMUM_BLOCK_WEIGHT / 100 * (AVAILABLE_BLOCK_RATIO as Weight - 10);
 
+// TODO: may need to be updated after https://github.com/paritytech/parity-bridges-common/issues/78
+/// Maximal number of messages in single delivery transaction.
+pub const MAX_MESSAGES_IN_DELIVERY_TRANSACTION: MessageNonce = 128;
 /// Maximal number of unconfirmed messages at inbound lane.
 pub const MAX_UNCONFIRMED_MESSAGES_AT_INBOUND_LANE: MessageNonce = 128;
 

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

@@ -63,6 +63,8 @@ pub struct MessageDeliveryParams {
 	/// 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 number of relayed messages in single delivery transaction.
+	pub max_messages_in_single_batch: MessageNonce,
 	/// Maximal cumulative dispatch weight of relayed messages in single delivery transaction.
 	pub max_messages_weight_in_single_batch: Weight,
 }
@@ -727,6 +729,7 @@ pub(crate) mod tests {
 					stall_timeout: Duration::from_millis(60 * 1000),
 					delivery_params: MessageDeliveryParams {
 						max_unconfirmed_nonces_at_target: 4,
+						max_messages_in_single_batch: 4,
 						max_messages_weight_in_single_batch: 4,
 					},
 				},

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

@@ -57,6 +57,7 @@ pub async fn run<P: MessageLane>(
 		stall_timeout,
 		MessageDeliveryStrategy::<P> {
 			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,
 			latest_confirmed_nonce_at_source: None,
 			target_nonces: None,
@@ -194,6 +195,8 @@ where
 struct MessageDeliveryStrategy<P: MessageLane> {
 	/// Maximal unconfirmed nonces at target client.
 	max_unconfirmed_nonces_at_target: MessageNonce,
+	/// Maximal number of messages in the single delivery transaction.
+	max_messages_in_single_batch: MessageNonce,
 	/// Maximal cumulative messages weight in the single delivery transaction.
 	max_messages_weight_in_single_batch: Weight,
 	/// Latest confirmed nonce at the source client.
@@ -315,6 +318,7 @@ impl<P: MessageLane> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::M
 			.checked_sub(future_confirmed_nonce_at_target)
 			.and_then(|diff| self.max_unconfirmed_nonces_at_target.checked_sub(diff))
 			.unwrap_or_default();
+		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 mut selected_weight: Weight = 0;
 		let mut selected_count: MessageNonce = 0;
@@ -401,6 +405,7 @@ mod tests {
 
 		let mut race_strategy = TestStrategy {
 			max_unconfirmed_nonces_at_target: 4,
+			max_messages_in_single_batch: 4,
 			max_messages_weight_in_single_batch: 4,
 			latest_confirmed_nonce_at_source: Some(19),
 			target_nonces: Some(TargetClientNonces {
@@ -498,7 +503,19 @@ mod tests {
 	}
 
 	#[test]
-	fn message_delivery_strategy_limits_batch_by_messages_count() {
+	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),
+			Some(((20..=22), proof_parameters(false, 3)))
+		);
+	}
+
+	#[test]
+	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

bridges/relays/substrate/src/millau_messages_to_rialto.rs

@@ -127,6 +127,7 @@ pub fn run(
 			stall_timeout,
 			delivery_params: messages_relay::message_lane_loop::MessageDeliveryParams {
 				max_unconfirmed_nonces_at_target: bp_rialto::MAX_UNCONFIRMED_MESSAGES_AT_INBOUND_LANE,
+				max_messages_in_single_batch: bp_rialto::MAX_MESSAGES_IN_DELIVERY_TRANSACTION,
 				// 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,

bridges/relays/substrate/src/rialto_messages_to_millau.rs

@@ -127,6 +127,7 @@ pub fn run(
 			stall_timeout,
 			delivery_params: messages_relay::message_lane_loop::MessageDeliveryParams {
 				max_unconfirmed_nonces_at_target: bp_millau::MAX_UNCONFIRMED_MESSAGES_AT_INBOUND_LANE,
+				max_messages_in_single_batch: bp_millau::MAX_MESSAGES_IN_DELIVERY_TRANSACTION,
 				// TODO: subtract base weight of delivery from this when it'll be known
 				// https://github.com/paritytech/parity-bridges-common/issues/78
 				max_messages_weight_in_single_batch: bp_millau::MAXIMUM_EXTRINSIC_WEIGHT,