Limit size of message delivery transaction (#575)

* limit messages size in delivery transaction

* docs

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

@@ -555,13 +555,13 @@ impl_runtime_apis! {
 			lane: bp_message_lane::LaneId,
 			begin: bp_message_lane::MessageNonce,
 			end: bp_message_lane::MessageNonce,
-		) -> Vec<(bp_message_lane::MessageNonce, Weight)> {
+		) -> Vec<(bp_message_lane::MessageNonce, Weight, u32)> {
 			(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))
+				Some((nonce, decoded_payload.weight, encoded_payload.len() as _))
 			})
 			.collect()
 		}

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

@@ -719,13 +719,13 @@ impl_runtime_apis! {
 			lane: bp_message_lane::LaneId,
 			begin: bp_message_lane::MessageNonce,
 			end: bp_message_lane::MessageNonce,
-		) -> Vec<(bp_message_lane::MessageNonce, Weight)> {
+		) -> Vec<(bp_message_lane::MessageNonce, Weight, u32)> {
 			(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))
+				Some((nonce, decoded_payload.weight, encoded_payload.len() as _))
 			})
 			.collect()
 		}

bridges/primitives/kusama/src/lib.rs

@@ -122,7 +122,7 @@ sp_api::decl_runtime_apis! {
 	/// This API is implemented by runtimes that are sending messages to Kusama chain, not the
 	/// Kusama runtime itself.
 	pub trait ToKusamaOutboundLaneApi {
-		/// Returns dispatch weight of all messages in given inclusive range.
+		/// Returns total dispatch weight and encoded payload size of all messages in given inclusive range.
 		///
 		/// 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.
@@ -130,7 +130,7 @@ sp_api::decl_runtime_apis! {
 			lane: LaneId,
 			begin: MessageNonce,
 			end: MessageNonce,
-		) -> Vec<(MessageNonce, Weight)>;
+		) -> Vec<(MessageNonce, Weight, u32)>;
 		/// 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/primitives/millau/src/lib.rs

@@ -186,7 +186,7 @@ sp_api::decl_runtime_apis! {
 	/// This API is implemented by runtimes that are sending messages to Millau chain, not the
 	/// Millau runtime itself.
 	pub trait ToMillauOutboundLaneApi {
-		/// Returns dispatch weight of all messages in given inclusive range.
+		/// Returns total dispatch weight and encoded payload size of all messages in given inclusive range.
 		///
 		/// 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.
@@ -194,7 +194,7 @@ sp_api::decl_runtime_apis! {
 			lane: LaneId,
 			begin: MessageNonce,
 			end: MessageNonce,
-		) -> Vec<(MessageNonce, Weight)>;
+		) -> Vec<(MessageNonce, Weight, u32)>;
 		/// 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/primitives/polkadot/src/lib.rs

@@ -122,7 +122,7 @@ sp_api::decl_runtime_apis! {
 	/// This API is implemented by runtimes that are sending messages to Polkadot chain, not the
 	/// Polkadot runtime itself.
 	pub trait ToPolkadotOutboundLaneApi {
-		/// Returns dispatch weight of all messages in given inclusive range.
+		/// Returns total dispatch weight and encoded payload size of all messages in given inclusive range.
 		///
 		/// 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.
@@ -130,7 +130,7 @@ sp_api::decl_runtime_apis! {
 			lane: LaneId,
 			begin: MessageNonce,
 			end: MessageNonce,
-		) -> Vec<(MessageNonce, Weight)>;
+		) -> Vec<(MessageNonce, Weight, u32)>;
 		/// 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/primitives/rialto/src/lib.rs

@@ -162,7 +162,7 @@ sp_api::decl_runtime_apis! {
 	/// This API is implemented by runtimes that are sending messages to Rialto chain, not the
 	/// Rialto runtime itself.
 	pub trait ToRialtoOutboundLaneApi {
-		/// Returns dispatch weight of all messages in given inclusive range.
+		/// Returns total dispatch weight and encoded payload size of all messages in given inclusive range.
 		///
 		/// 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.
@@ -170,7 +170,7 @@ sp_api::decl_runtime_apis! {
 			lane: LaneId,
 			begin: MessageNonce,
 			end: MessageNonce,
-		) -> Vec<(MessageNonce, Weight)>;
+		) -> Vec<(MessageNonce, Weight, u32)>;
 		/// 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/src/message_lane_loop.rs

@@ -71,10 +71,21 @@ pub struct MessageDeliveryParams {
 	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,
+	/// Maximal cumulative size of relayed messages in single delivery transaction.
+	pub max_messages_size_in_single_batch: usize,
+}
+
+/// Message weights.
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub struct MessageWeights {
+	/// Message dispatch weight.
+	pub weight: Weight,
+	/// Message size (number of bytes in encoded payload).
+	pub size: usize,
 }
 
 /// Messages weights map.
-pub type MessageWeightsMap = BTreeMap<MessageNonce, Weight>;
+pub type MessageWeightsMap = BTreeMap<MessageNonce, MessageWeights>;
 
 /// Message delivery race proof parameters.
 #[derive(Debug, PartialEq)]
@@ -586,7 +597,9 @@ pub(crate) mod tests {
 			_id: SourceHeaderIdOf<TestMessageLane>,
 			nonces: RangeInclusive<MessageNonce>,
 		) -> Result<MessageWeightsMap, Self::Error> {
-			Ok(nonces.map(|nonce| (nonce, 1)).collect())
+			Ok(nonces
+				.map(|nonce| (nonce, MessageWeights { weight: 1, size: 1 }))
+				.collect())
 		}
 
 		async fn prove_messages(
@@ -754,6 +767,7 @@ pub(crate) mod tests {
 						max_unconfirmed_nonces_at_target: 4,
 						max_messages_in_single_batch: 4,
 						max_messages_weight_in_single_batch: 4,
+						max_messages_size_in_single_batch: 4,
 					},
 				},
 				source_client,

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

@@ -60,6 +60,7 @@ pub async fn run<P: MessageLane>(
 			max_unconfirmed_nonces_at_target: params.max_unconfirmed_nonces_at_target,
 			max_messages_in_single_batch: params.max_messages_in_single_batch,
 			max_messages_weight_in_single_batch: params.max_messages_weight_in_single_batch,
+			max_messages_size_in_single_batch: params.max_messages_size_in_single_batch,
 			latest_confirmed_nonce_at_source: None,
 			target_nonces: None,
 			strategy: BasicStrategy::new(),
@@ -218,6 +219,8 @@ struct MessageDeliveryStrategy<P: MessageLane> {
 	max_messages_in_single_batch: MessageNonce,
 	/// Maximal cumulative messages weight in the single delivery transaction.
 	max_messages_weight_in_single_batch: Weight,
+	/// Maximal messages size in the single delivery transaction.
+	max_messages_size_in_single_batch: usize,
 	/// Latest confirmed nonce at the source client.
 	latest_confirmed_nonce_at_source: Option<MessageNonce>,
 	/// Target nonces from the source client.
@@ -360,7 +363,9 @@ impl<P: MessageLane> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::M
 			.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 max_messages_size_in_single_batch = self.max_messages_size_in_single_batch;
 		let mut selected_weight: Weight = 0;
+		let mut selected_size: usize = 0;
 		let mut selected_count: MessageNonce = 0;
 
 		let selected_nonces = self
@@ -369,11 +374,44 @@ impl<P: MessageLane> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::M
 				let to_requeue = range
 					.into_iter()
 					.skip_while(|(_, weight)| {
+						// Since we (hopefully) have some reserves in `max_messages_weight_in_single_batch`
+						// and `max_messages_size_in_single_batch`, we may still try to submit transaction
+						// with single message if message overflows these limits. The worst case would be if
+						// transaction will be rejected by the target runtime, but at least we have tried.
+
 						// limit messages in the batch by weight
-						let new_selected_weight = match selected_weight.checked_add(*weight) {
+						let new_selected_weight = match selected_weight.checked_add(weight.weight) {
 							Some(new_selected_weight) if new_selected_weight <= max_messages_weight_in_single_batch => {
 								new_selected_weight
 							}
+							new_selected_weight if selected_count == 0 => {
+								log::warn!(
+									target: "bridge",
+									"Going to submit message delivery transaction with declared dispatch \
+									weight {:?} that overflows maximal configured weight {}",
+									new_selected_weight,
+									max_messages_weight_in_single_batch,
+								);
+								new_selected_weight.unwrap_or(Weight::MAX)
+							}
+							_ => return false,
+						};
+
+						// limit messages in the batch by size
+						let new_selected_size = match selected_size.checked_add(weight.size) {
+							Some(new_selected_size) if new_selected_size <= max_messages_size_in_single_batch => {
+								new_selected_size
+							}
+							new_selected_size if selected_count == 0 => {
+								log::warn!(
+									target: "bridge",
+									"Going to submit message delivery transaction with message \
+									size {:?} that overflows maximal configured size {}",
+									new_selected_size,
+									max_messages_size_in_single_batch,
+								);
+								new_selected_size.unwrap_or(usize::MAX)
+							}
 							_ => return false,
 						};
 
@@ -384,6 +422,7 @@ impl<P: MessageLane> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::M
 						}
 
 						selected_weight = new_selected_weight;
+						selected_size = new_selected_size;
 						selected_count = new_selected_count;
 						true
 					})
@@ -427,8 +466,9 @@ impl NoncesRange for MessageWeightsMap {
 #[cfg(test)]
 mod tests {
 	use super::*;
-	use crate::message_lane_loop::tests::{
+	use crate::message_lane_loop::{
-		header_id, TestMessageLane, TestMessagesProof, TestSourceHeaderId, TestTargetHeaderId,
+		tests::{header_id, TestMessageLane, TestMessagesProof, TestSourceHeaderId, TestTargetHeaderId},
+		MessageWeights,
 	};
 
 	type TestRaceState = RaceState<TestSourceHeaderId, TestTargetHeaderId, TestMessagesProof>;
@@ -448,6 +488,7 @@ mod tests {
 			max_unconfirmed_nonces_at_target: 4,
 			max_messages_in_single_batch: 4,
 			max_messages_weight_in_single_batch: 4,
+			max_messages_size_in_single_batch: 4,
 			latest_confirmed_nonce_at_source: Some(19),
 			target_nonces: Some(TargetClientNonces {
 				latest_nonce: 19,
@@ -465,7 +506,14 @@ mod tests {
 		race_strategy.strategy.source_nonces_updated(
 			header_id(1),
 			SourceClientNonces {
-				new_nonces: vec![(20, 1), (21, 1), (22, 1), (23, 1)].into_iter().collect(),
+				new_nonces: vec![
+					(20, MessageWeights { weight: 1, size: 1 }),
+					(21, MessageWeights { weight: 1, size: 1 }),
+					(22, MessageWeights { weight: 1, size: 1 }),
+					(23, MessageWeights { weight: 1, size: 1 }),
+				]
+				.into_iter()
+				.collect(),
 				confirmed_nonce: Some(19),
 			},
 		);
@@ -490,7 +538,17 @@ mod tests {
 	#[test]
 	fn weights_map_works_as_nonces_range() {
 		fn build_map(range: RangeInclusive<MessageNonce>) -> MessageWeightsMap {
-			range.map(|idx| (idx, idx)).collect()
+			range
+				.map(|idx| {
+					(
+						idx,
+						MessageWeights {
+							weight: idx,
+							size: idx as _,
+						},
+					)
+				})
+				.collect()
 		}
 
 		let map = build_map(20..=30);
@@ -604,6 +662,42 @@ mod tests {
 		);
 	}
 
+	#[test]
+	fn message_delivery_strategy_accepts_single_message_even_if_its_weight_overflows_maximal_weight() {
+		let (state, mut strategy) = prepare_strategy();
+
+		// first message doesn't fit in the batch, because it has weight (10) that overflows max weight (4)
+		strategy.strategy.source_queue_mut()[0].1.get_mut(&20).unwrap().weight = 10;
+		assert_eq!(
+			strategy.select_nonces_to_deliver(&state),
+			Some(((20..=20), proof_parameters(false, 10)))
+		);
+	}
+
+	#[test]
+	fn message_delivery_strategy_limits_batch_by_messages_size() {
+		let (state, mut strategy) = prepare_strategy();
+
+		// not all queued messages may fit in the batch, because batch has max weight
+		strategy.max_messages_size_in_single_batch = 3;
+		assert_eq!(
+			strategy.select_nonces_to_deliver(&state),
+			Some(((20..=22), proof_parameters(false, 3)))
+		);
+	}
+
+	#[test]
+	fn message_delivery_strategy_accepts_single_message_even_if_its_weight_overflows_maximal_size() {
+		let (state, mut strategy) = prepare_strategy();
+
+		// first message doesn't fit in the batch, because it has weight (10) that overflows max weight (4)
+		strategy.strategy.source_queue_mut()[0].1.get_mut(&20).unwrap().size = 10;
+		assert_eq!(
+			strategy.select_nonces_to_deliver(&state),
+			Some(((20..=20), proof_parameters(false, 1)))
+		);
+	}
+
 	#[test]
 	fn message_delivery_strategy_limits_batch_by_messages_count_when_there_is_upper_limit() {
 		let (state, mut strategy) = prepare_strategy();

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

@@ -57,6 +57,14 @@ where
 		}
 	}
 
+	/// Mutable reference to source queue to use in tests.
+	#[cfg(test)]
+	pub(crate) fn source_queue_mut(
+		&mut self,
+	) -> &mut VecDeque<(HeaderId<SourceHeaderHash, SourceHeaderNumber>, SourceNoncesRange)> {
+		&mut self.source_queue
+	}
+
 	/// Should return `Some(nonces)` if we need to deliver proof of `nonces` (and associated
 	/// data) from source to target node.
 	///

bridges/relays/substrate/src/messages_source.rs

@@ -27,7 +27,9 @@ use codec::{Decode, Encode};
 use frame_support::weights::Weight;
 use messages_relay::{
 	message_lane::{SourceHeaderIdOf, TargetHeaderIdOf},
-	message_lane_loop::{ClientState, MessageProofParameters, MessageWeightsMap, SourceClient, SourceClientState},
+	message_lane_loop::{
+		ClientState, MessageProofParameters, MessageWeights, MessageWeightsMap, SourceClient, SourceClientState,
+	},
 };
 use relay_substrate_client::{Chain, Client, Error as SubstrateError, HashOf, HeaderIdOf};
 use relay_utils::{BlockNumberBase, HeaderId};
@@ -235,7 +237,7 @@ where
 }
 
 fn make_message_weights_map<C: Chain>(
-	weights: Vec<(MessageNonce, Weight)>,
+	weights: Vec<(MessageNonce, Weight, u32)>,
 	nonces: RangeInclusive<MessageNonce>,
 ) -> Result<MessageWeightsMap, SubstrateError> {
 	let make_missing_nonce_error = |expected_nonce| {
@@ -255,7 +257,7 @@ fn make_message_weights_map<C: Chain>(
 	// check if last nonce is missing - loop below is not checking this
 	let last_nonce_is_missing = weights
 		.last()
-		.map(|(last_nonce, _)| last_nonce != nonces.end())
+		.map(|(last_nonce, _, _)| last_nonce != nonces.end())
 		.unwrap_or(true);
 	if last_nonce_is_missing {
 		return make_missing_nonce_error(*nonces.end());
@@ -264,7 +266,7 @@ fn make_message_weights_map<C: Chain>(
 	let mut expected_nonce = *nonces.start();
 	let mut is_at_head = true;
 
-	for (nonce, weight) in weights {
+	for (nonce, weight, size) in weights {
 		match (nonce == expected_nonce, is_at_head) {
 			(true, _) => (),
 			(false, true) => {
@@ -286,7 +288,13 @@ fn make_message_weights_map<C: Chain>(
 			}
 		}
 
-		weights_map.insert(nonce, weight);
+		weights_map.insert(
+			nonce,
+			MessageWeights {
+				weight,
+				size: size as _,
+			},
+		);
 		expected_nonce = nonce + 1;
 		is_at_head = false;
 	}
@@ -301,23 +309,35 @@ mod tests {
 	#[test]
 	fn make_message_weights_map_succeeds_if_no_messages_are_missing() {
 		assert_eq!(
-			make_message_weights_map::<relay_rialto_client::Rialto>(vec![(1, 0), (2, 0), (3, 0)], 1..=3).unwrap(),
+			make_message_weights_map::<relay_rialto_client::Rialto>(vec![(1, 0, 0), (2, 0, 0), (3, 0, 0)], 1..=3,)
-			vec![(1, 0), (2, 0), (3, 0)].into_iter().collect(),
+				.unwrap(),
+			vec![
+				(1, MessageWeights { weight: 0, size: 0 }),
+				(2, MessageWeights { weight: 0, size: 0 }),
+				(3, MessageWeights { weight: 0, size: 0 }),
+			]
+			.into_iter()
+			.collect(),
 		);
 	}
 
 	#[test]
 	fn make_message_weights_map_succeeds_if_head_messages_are_missing() {
 		assert_eq!(
-			make_message_weights_map::<relay_rialto_client::Rialto>(vec![(2, 0), (3, 0)], 1..=3).unwrap(),
+			make_message_weights_map::<relay_rialto_client::Rialto>(vec![(2, 0, 0), (3, 0, 0)], 1..=3,).unwrap(),
-			vec![(2, 0), (3, 0)].into_iter().collect(),
+			vec![
+				(2, MessageWeights { weight: 0, size: 0 }),
+				(3, MessageWeights { weight: 0, size: 0 }),
+			]
+			.into_iter()
+			.collect(),
 		);
 	}
 
 	#[test]
 	fn make_message_weights_map_fails_if_mid_messages_are_missing() {
 		assert!(matches!(
-			make_message_weights_map::<relay_rialto_client::Rialto>(vec![(1, 0), (3, 0)], 1..=3),
+			make_message_weights_map::<relay_rialto_client::Rialto>(vec![(1, 0, 0), (3, 0, 0)], 1..=3,),
 			Err(SubstrateError::Custom(_))
 		));
 	}
@@ -325,7 +345,7 @@ mod tests {
 	#[test]
 	fn make_message_weights_map_fails_if_tail_messages_are_missing() {
 		assert!(matches!(
-			make_message_weights_map::<relay_rialto_client::Rialto>(vec![(1, 0), (2, 0)], 1..=3),
+			make_message_weights_map::<relay_rialto_client::Rialto>(vec![(1, 0, 0), (2, 0, 0)], 1..=3,),
 			Err(SubstrateError::Custom(_))
 		));
 	}

bridges/relays/substrate/src/millau_messages_to_rialto.rs

@@ -133,6 +133,8 @@ pub fn run(
 				// 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,
+				// 2/3 is reserved for proofs and tx overhead
+				max_messages_size_in_single_batch: bp_rialto::MAXIMUM_EXTRINSIC_SIZE as usize / 3,
 			},
 		},
 		MillauSourceClient::new(millau_client, lane.clone(), lane_id, RIALTO_BRIDGE_INSTANCE),

bridges/relays/substrate/src/rialto_messages_to_millau.rs

@@ -133,6 +133,8 @@ pub fn run(
 				// 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,
+				// 2/3 is reserved for proofs and tx overhead
+				max_messages_size_in_single_batch: bp_millau::MAXIMUM_EXTRINSIC_SIZE as usize / 3,
 			},
 		},
 		RialtoSourceClient::new(rialto_client, lane.clone(), lane_id, MILLAU_BRIDGE_INSTANCE),