Call To*InboundLaneApi::message_details() with batched messages (#1545)

* generated_message_details() -> Simplifications

- avoid using a HashMap for `messages_to_refine`. It seems that a vec is
  enough
- minimize the number of conversions between `OutboundMessageDetails` and
  `MessageDetails`
- use references where possible in order to minimize the number of
  intermediary Vecs
- simplify `make_message_details_map()` logic, reduce its scope and rename
  it to `validate_out_msgs_details()`

Signed-off-by: Serban Iorga <serban@parity.io>

* Define typed_state_call()

Signed-off-by: Serban Iorga <serban@parity.io>

* Call To*InboundLaneApi::message_details() with single messages

Signed-off-by: Serban Iorga <serban@parity.io>

* Call To*InboundLaneApi::message_details() with batched messages

Signed-off-by: Serban Iorga <serban@parity.io>

* validate_out_msgs_details() -> change check

* Define split_msgs_to_refine()

Signed-off-by: Serban Iorga <serban@parity.io>

Signed-off-by: Serban Iorga <serban@parity.io>

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

@@ -545,6 +545,18 @@ impl<C: Chain> Client<C> {
 		.await
 	}
 
+	/// Execute runtime call at given block, provided the input and output types.
+	/// It also performs the input encode and output decode.
+	pub async fn typed_state_call<Input: codec::Encode, Output: codec::Decode>(
+		&self,
+		method_name: String,
+		input: Input,
+		at_block: Option<C::Hash>,
+	) -> Result<Output> {
+		let encoded_output = self.state_call(method_name, Bytes(input.encode()), at_block).await?;
+		Output::decode(&mut &encoded_output.0[..]).map_err(Error::ResponseParseFailed)
+	}
+
 	/// Execute runtime call at given block.
 	pub async fn state_call(
 		&self,

bridges/relays/lib-substrate-relay/src/messages_source.rs

@@ -31,8 +31,8 @@ use async_std::sync::Arc;
 use async_trait::async_trait;
 use bp_messages::{
 	storage_keys::{operating_mode_key, outbound_lane_data_key},
-	InboundMessageDetails, LaneId, MessageData, MessageNonce, MessagesOperatingMode,
+	InboundMessageDetails, LaneId, MessageData, MessageNonce, MessagePayload,
-	OutboundLaneData, OutboundMessageDetails, UnrewardedRelayersState,
+	MessagesOperatingMode, OutboundLaneData, OutboundMessageDetails, UnrewardedRelayersState,
 };
 use bp_runtime::{messages::DispatchFeePayment, BasicOperatingMode, HeaderIdProvider};
 use bridge_runtime_common::messages::{
@@ -56,12 +56,13 @@ use relay_substrate_client::{
 use relay_utils::{relay_loop::Client as RelayClient, HeaderId};
 use sp_core::{Bytes, Pair};
 use sp_runtime::{traits::Header as HeaderT, DeserializeOwned};
-use std::{collections::HashMap, ops::RangeInclusive};
+use std::ops::RangeInclusive;
 
 /// Intermediate message proof returned by the source Substrate node. Includes everything
 /// required to submit to the target node: cumulative dispatch weight of bundled messages and
 /// the proof itself.
 pub type SubstrateMessagesProof<C> = (Weight, FromBridgedChainMessagesProof<HashOf<C>>);
+type MessagesToRefine<'a, Balance> = Vec<(MessagePayload, &'a mut OutboundMessageDetails<Balance>)>;
 
 /// Substrate client as Substrate messages source.
 pub struct SubstrateMessagesSource<P: SubstrateMessageLane> {
@@ -192,110 +193,97 @@ where
 		&self,
 		id: SourceHeaderIdOf<MessageLaneAdapter<P>>,
 		nonces: RangeInclusive<MessageNonce>,
-	) -> Result<
+	) -> Result<MessageDetailsMap<BalanceOf<P::SourceChain>>, SubstrateError> {
-		MessageDetailsMap<<MessageLaneAdapter<P> as MessageLane>::SourceChainBalance>,
+		let mut out_msgs_details = self
-		SubstrateError,
-	> {
-		let encoded_response = self
 			.source_client
-			.state_call(
+			.typed_state_call::<_, Vec<_>>(
 				P::TargetChain::TO_CHAIN_MESSAGE_DETAILS_METHOD.into(),
-				Bytes((self.lane_id, nonces.start(), nonces.end()).encode()),
+				(self.lane_id, *nonces.start(), *nonces.end()),
 				Some(id.1),
 			)
 			.await?;
-
+		validate_out_msgs_details::<P::SourceChain>(&out_msgs_details, nonces)?;
-		let mut messages = make_message_details_map::<P::SourceChain>(
-			Decode::decode(&mut &encoded_response.0[..])
-				.map_err(SubstrateError::ResponseParseFailed)?,
-			nonces,
-		)?;
 
 		// prepare arguments of the inbound message details call (if we need it)
-		let mut messages_to_refine = HashMap::new();
+		let mut msgs_to_refine = vec![];
-		for (message_nonce, message) in &messages {
+		for out_msg_details in out_msgs_details.iter_mut() {
-			if message.dispatch_fee_payment != DispatchFeePayment::AtTargetChain {
+			if out_msg_details.dispatch_fee_payment != DispatchFeePayment::AtTargetChain {
 				continue
 			}
 
 			// for pay-at-target messages we may want to ask target chain for
 			// refined dispatch weight
-			let message_key = bp_messages::storage_keys::message_key(
+			let msg_key = bp_messages::storage_keys::message_key(
 				P::TargetChain::WITH_CHAIN_MESSAGES_PALLET_NAME,
 				&self.lane_id,
-				*message_nonce,
+				out_msg_details.nonce,
-			);
-			let message_data: MessageData<BalanceOf<P::SourceChain>> =
-				self.source_client.storage_value(message_key, Some(id.1)).await?.ok_or_else(
-					|| {
-						SubstrateError::Custom(format!(
-							"Message to {} {:?}/{} is missing from runtime the storage of {} at {:?}",
-							P::TargetChain::NAME,
-							self.lane_id,
-							message_nonce,
-							P::SourceChain::NAME,
-							id,
-						))
-					},
-				)?;
-			let message_payload = message_data.payload;
-			messages_to_refine.insert(
-				*message_nonce,
-				(
-					message_payload,
-					OutboundMessageDetails {
-						nonce: *message_nonce,
-						dispatch_weight: message.dispatch_weight,
-						size: message.size,
-						delivery_and_dispatch_fee: message.reward,
-						dispatch_fee_payment: DispatchFeePayment::AtTargetChain,
-					},
-				),
 			);
+			let msg_data: MessageData<BalanceOf<P::SourceChain>> =
+				self.source_client.storage_value(msg_key, Some(id.1)).await?.ok_or_else(|| {
+					SubstrateError::Custom(format!(
+						"Message to {} {:?}/{} is missing from runtime the storage of {} at {:?}",
+						P::TargetChain::NAME,
+						self.lane_id,
+						out_msg_details.nonce,
+						P::SourceChain::NAME,
+						id,
+					))
+				})?;
+
+			msgs_to_refine.push((msg_data.payload, out_msg_details));
 		}
 
-		// request inbound message details from the target client
+		for mut msgs_to_refine_batch in
-		if !messages_to_refine.is_empty() {
+			split_msgs_to_refine::<P::SourceChain, P::TargetChain>(self.lane_id, msgs_to_refine)?
-			let refined_messages_encoded = self
+		{
+			let in_msgs_details = self
 				.target_client
-				.state_call(
+				.typed_state_call::<_, Vec<InboundMessageDetails>>(
 					P::SourceChain::FROM_CHAIN_MESSAGE_DETAILS_METHOD.into(),
-					Bytes((self.lane_id, messages_to_refine.values().collect::<Vec<_>>()).encode()),
+					(self.lane_id, &msgs_to_refine_batch),
 					None,
 				)
 				.await?;
-			let refined_messages =
+			if in_msgs_details.len() != msgs_to_refine_batch.len() {
-				Vec::<InboundMessageDetails>::decode(&mut &refined_messages_encoded.0[..])
-					.map_err(SubstrateError::ResponseParseFailed)?;
-			if refined_messages.len() != messages_to_refine.len() {
 				return Err(SubstrateError::Custom(format!(
 					"Call of {} at {} has returned {} entries instead of expected {}",
 					P::SourceChain::FROM_CHAIN_MESSAGE_DETAILS_METHOD,
 					P::TargetChain::NAME,
-					refined_messages.len(),
+					in_msgs_details.len(),
-					messages_to_refine.len(),
+					msgs_to_refine_batch.len(),
 				)))
 			}
-
+			for ((_, out_msg_details), in_msg_details) in
-			for (nonce, refined_message) in messages_to_refine.keys().zip(refined_messages) {
+				msgs_to_refine_batch.iter_mut().zip(in_msgs_details)
-				let message = messages
+			{
-					.get_mut(nonce)
-					.expect("`messages_to_refine` is a subset of `messages`; qed");
 				log::trace!(
 					target: "bridge",
 					"Refined weight of {}->{} message {:?}/{}: at-source: {}, at-target: {}",
 					P::SourceChain::NAME,
 					P::TargetChain::NAME,
 					self.lane_id,
-					nonce,
+					out_msg_details.nonce,
-					message.dispatch_weight,
+					out_msg_details.dispatch_weight,
-					refined_message.dispatch_weight,
+					in_msg_details.dispatch_weight,
 				);
-				message.dispatch_weight = refined_message.dispatch_weight;
+				out_msg_details.dispatch_weight = in_msg_details.dispatch_weight;
 			}
 		}
 
-		Ok(messages)
+		let mut msgs_details_map = MessageDetailsMap::new();
+		for out_msg_details in out_msgs_details {
+			msgs_details_map.insert(
+				out_msg_details.nonce,
+				MessageDetails {
+					dispatch_weight: out_msg_details.dispatch_weight,
+					size: out_msg_details.size as _,
+					reward: out_msg_details.delivery_and_dispatch_fee,
+					dispatch_fee_payment: out_msg_details.dispatch_fee_payment,
+				},
+			);
+		}
+
+		Ok(msgs_details_map)
 	}
 
 	async fn prove_messages(
@@ -571,10 +559,10 @@ where
 	.unwrap_or(Err(SubstrateError::BridgePalletIsNotInitialized))
 }
 
-fn make_message_details_map<C: Chain>(
+fn validate_out_msgs_details<C: Chain>(
-	weights: Vec<bp_messages::OutboundMessageDetails<C::Balance>>,
+	out_msgs_details: &[OutboundMessageDetails<C::Balance>],
 	nonces: RangeInclusive<MessageNonce>,
-) -> Result<MessageDetailsMap<C::Balance>, SubstrateError> {
+) -> Result<(), SubstrateError> {
 	let make_missing_nonce_error = |expected_nonce| {
 		Err(SubstrateError::Custom(format!(
 			"Missing nonce {} in message_details call result. Expected all nonces from {:?}",
@@ -582,73 +570,88 @@ fn make_message_details_map<C: Chain>(
 		)))
 	};
 
-	let mut weights_map = MessageDetailsMap::new();
+	if out_msgs_details.len() > nonces.clone().count() {
-
+		return Err(SubstrateError::Custom(
-	// this is actually prevented by external logic
+			"More messages than requested returned by the message_details call.".into(),
-	if nonces.is_empty() {
+		))
-		return Ok(weights_map)
 	}
 
-	// check if last nonce is missing - loop below is not checking this
+	// Check if last nonce is missing. The loop below is not checking this.
-	let last_nonce_is_missing =
+	if out_msgs_details.is_empty() && !nonces.is_empty() {
-		weights.last().map(|details| details.nonce != *nonces.end()).unwrap_or(true);
-	if last_nonce_is_missing {
 		return make_missing_nonce_error(*nonces.end())
 	}
 
-	let mut expected_nonce = *nonces.start();
+	let mut nonces_iter = nonces.clone().rev().peekable();
-	let mut is_at_head = true;
+	let mut out_msgs_details_iter = out_msgs_details.iter().rev();
-
+	while let Some((out_msg_details, &nonce)) = out_msgs_details_iter.next().zip(nonces_iter.peek())
-	for details in weights {
+	{
-		match (details.nonce == expected_nonce, is_at_head) {
+		nonces_iter.next();
-			(true, _) => (),
+		if out_msg_details.nonce != nonce {
-			(false, true) => {
+			// Some nonces are missing from the middle/tail of the range. This is critical error.
-				// this may happen if some messages were already pruned from the source node
+			return make_missing_nonce_error(nonce)
-				//
-				// this is not critical error and will be auto-resolved by messages lane (and target
-				// node)
-				log::info!(
-					target: "bridge",
-					"Some messages are missing from the {} node: {:?}. Target node may be out of sync?",
-					C::NAME,
-					expected_nonce..details.nonce,
-				);
-			},
-			(false, false) => {
-				// some nonces are missing from the middle/tail of the range
-				//
-				// this is critical error, because we can't miss any nonces
-				return make_missing_nonce_error(expected_nonce)
-			},
 		}
-
-		weights_map.insert(
-			details.nonce,
-			MessageDetails {
-				dispatch_weight: details.dispatch_weight,
-				size: details.size as _,
-				reward: details.delivery_and_dispatch_fee,
-				dispatch_fee_payment: details.dispatch_fee_payment,
-			},
-		);
-		expected_nonce = details.nonce + 1;
-		is_at_head = false;
 	}
 
-	Ok(weights_map)
+	// Check if some nonces from the beginning of the range are missing. This may happen if
+	// some messages were already pruned from the source node. This is not a critical error
+	// and will be auto-resolved by messages lane (and target node).
+	if nonces_iter.peek().is_some() {
+		log::info!(
+			target: "bridge",
+			"Some messages are missing from the {} node: {:?}. Target node may be out of sync?",
+			C::NAME,
+			nonces_iter.rev().collect::<Vec<_>>(),
+		);
+	}
+
+	Ok(())
+}
+
+fn split_msgs_to_refine<Source: Chain + ChainWithMessages, Target: Chain>(
+	lane_id: LaneId,
+	msgs_to_refine: MessagesToRefine<Source::Balance>,
+) -> Result<Vec<MessagesToRefine<Source::Balance>>, SubstrateError> {
+	let max_batch_size = Target::max_extrinsic_size() as usize;
+	let mut batches = vec![];
+
+	let mut current_msgs_batch = msgs_to_refine;
+	while !current_msgs_batch.is_empty() {
+		let mut next_msgs_batch = vec![];
+		while (lane_id, &current_msgs_batch).encoded_size() > max_batch_size {
+			if current_msgs_batch.len() <= 1 {
+				return Err(SubstrateError::Custom(format!(
+					"Call of {} at {} can't be executed even if only one message is supplied. \
+						max_extrinsic_size(): {}",
+					Source::FROM_CHAIN_MESSAGE_DETAILS_METHOD,
+					Target::NAME,
+					Target::max_extrinsic_size(),
+				)))
+			}
+
+			if let Some(msg) = current_msgs_batch.pop() {
+				next_msgs_batch.insert(0, msg);
+			}
+		}
+
+		batches.push(current_msgs_batch);
+		current_msgs_batch = next_msgs_batch;
+	}
+
+	Ok(batches)
 }
 
 #[cfg(test)]
 mod tests {
 	use super::*;
-	use bp_runtime::messages::DispatchFeePayment;
+	use bp_runtime::{messages::DispatchFeePayment, Chain as ChainBase};
 	use codec::MaxEncodedLen;
+	use relay_rialto_client::Rialto;
 	use relay_rococo_client::Rococo;
 	use relay_wococo_client::Wococo;
 
 	fn message_details_from_rpc(
 		nonces: RangeInclusive<MessageNonce>,
-	) -> Vec<bp_messages::OutboundMessageDetails<bp_wococo::Balance>> {
+	) -> Vec<OutboundMessageDetails<bp_wococo::Balance>> {
 		nonces
 			.into_iter()
 			.map(|nonce| bp_messages::OutboundMessageDetails {
@@ -662,94 +665,49 @@ mod tests {
 	}
 
 	#[test]
-	fn make_message_details_map_succeeds_if_no_messages_are_missing() {
+	fn validate_out_msgs_details_succeeds_if_no_messages_are_missing() {
-		assert_eq!(
+		assert!(
-			make_message_details_map::<Wococo>(message_details_from_rpc(1..=3), 1..=3,).unwrap(),
+			validate_out_msgs_details::<Wococo>(&message_details_from_rpc(1..=3), 1..=3,).is_ok()
-			vec![
-				(
-					1,
-					MessageDetails {
-						dispatch_weight: 0,
-						size: 0,
-						reward: 0,
-						dispatch_fee_payment: DispatchFeePayment::AtSourceChain,
-					}
-				),
-				(
-					2,
-					MessageDetails {
-						dispatch_weight: 0,
-						size: 0,
-						reward: 0,
-						dispatch_fee_payment: DispatchFeePayment::AtSourceChain,
-					}
-				),
-				(
-					3,
-					MessageDetails {
-						dispatch_weight: 0,
-						size: 0,
-						reward: 0,
-						dispatch_fee_payment: DispatchFeePayment::AtSourceChain,
-					}
-				),
-			]
-			.into_iter()
-			.collect(),
 		);
 	}
 
 	#[test]
-	fn make_message_details_map_succeeds_if_head_messages_are_missing() {
+	fn validate_out_msgs_details_succeeds_if_head_messages_are_missing() {
-		assert_eq!(
+		assert!(
-			make_message_details_map::<Wococo>(message_details_from_rpc(2..=3), 1..=3,).unwrap(),
+			validate_out_msgs_details::<Wococo>(&message_details_from_rpc(2..=3), 1..=3,).is_ok()
-			vec![
+		)
-				(
-					2,
-					MessageDetails {
-						dispatch_weight: 0,
-						size: 0,
-						reward: 0,
-						dispatch_fee_payment: DispatchFeePayment::AtSourceChain,
-					}
-				),
-				(
-					3,
-					MessageDetails {
-						dispatch_weight: 0,
-						size: 0,
-						reward: 0,
-						dispatch_fee_payment: DispatchFeePayment::AtSourceChain,
-					}
-				),
-			]
-			.into_iter()
-			.collect(),
-		);
 	}
 
 	#[test]
-	fn make_message_details_map_fails_if_mid_messages_are_missing() {
+	fn validate_out_msgs_details_fails_if_mid_messages_are_missing() {
 		let mut message_details_from_rpc = message_details_from_rpc(1..=3);
 		message_details_from_rpc.remove(1);
 		assert!(matches!(
-			make_message_details_map::<Wococo>(message_details_from_rpc, 1..=3,),
+			validate_out_msgs_details::<Wococo>(&message_details_from_rpc, 1..=3,),
 			Err(SubstrateError::Custom(_))
 		));
 	}
 
 	#[test]
-	fn make_message_details_map_fails_if_tail_messages_are_missing() {
+	fn validate_out_msgs_details_map_fails_if_tail_messages_are_missing() {
 		assert!(matches!(
-			make_message_details_map::<Wococo>(message_details_from_rpc(1..=2), 1..=3,),
+			validate_out_msgs_details::<Wococo>(&message_details_from_rpc(1..=2), 1..=3,),
 			Err(SubstrateError::Custom(_))
 		));
 	}
 
 	#[test]
-	fn make_message_details_map_fails_if_all_messages_are_missing() {
+	fn validate_out_msgs_details_fails_if_all_messages_are_missing() {
 		assert!(matches!(
-			make_message_details_map::<Wococo>(vec![], 1..=3),
+			validate_out_msgs_details::<Wococo>(&[], 1..=3),
+			Err(SubstrateError::Custom(_))
+		));
+	}
+
+	#[test]
+	fn validate_out_msgs_details_fails_if_more_messages_than_nonces() {
+		assert!(matches!(
+			validate_out_msgs_details::<Wococo>(&message_details_from_rpc(1..=5), 2..=5,),
 			Err(SubstrateError::Custom(_))
 		));
 	}
@@ -766,4 +724,101 @@ mod tests {
 			dummy_proof.1.encode().len(),
 		);
 	}
+
+	fn check_split_msgs_to_refine(
+		payload_sizes: Vec<usize>,
+		expected_batches: Result<Vec<usize>, ()>,
+	) {
+		let mut out_msgs_details = vec![];
+		for (idx, _) in payload_sizes.iter().enumerate() {
+			out_msgs_details.push(OutboundMessageDetails::<BalanceOf<Rialto>> {
+				nonce: idx as MessageNonce,
+				dispatch_weight: 0,
+				size: 0,
+				delivery_and_dispatch_fee: 0,
+				dispatch_fee_payment: DispatchFeePayment::AtTargetChain,
+			});
+		}
+
+		let mut msgs_to_refine = vec![];
+		for (&payload_size, out_msg_details) in
+			payload_sizes.iter().zip(out_msgs_details.iter_mut())
+		{
+			let payload = vec![1u8; payload_size];
+			msgs_to_refine.push((payload, out_msg_details));
+		}
+
+		let maybe_batches = split_msgs_to_refine::<Rialto, Rococo>([0, 0, 0, 0], msgs_to_refine);
+		match expected_batches {
+			Ok(expected_batches) => {
+				let batches = maybe_batches.unwrap();
+				let mut idx = 0;
+				assert_eq!(batches.len(), expected_batches.len());
+				for (batch, &expected_batch_size) in batches.iter().zip(expected_batches.iter()) {
+					assert_eq!(batch.len(), expected_batch_size);
+					for msg_to_refine in batch {
+						assert_eq!(msg_to_refine.0.len(), payload_sizes[idx]);
+						idx += 1;
+					}
+				}
+			},
+			Err(_) => {
+				matches!(maybe_batches, Err(SubstrateError::Custom(_)));
+			},
+		}
+	}
+
+	#[test]
+	fn test_split_msgs_to_refine() {
+		let max_extrinsic_size = Rococo::max_extrinsic_size() as usize;
+
+		// Check that an error is returned when one of the messages is too big.
+		check_split_msgs_to_refine(vec![max_extrinsic_size], Err(()));
+		check_split_msgs_to_refine(vec![50, 100, max_extrinsic_size, 200], Err(()));
+
+		// Otherwise check that the split is valid.
+		check_split_msgs_to_refine(vec![100, 200, 300, 400], Ok(vec![4]));
+		check_split_msgs_to_refine(
+			vec![
+				50,
+				100,
+				max_extrinsic_size - 500,
+				500,
+				1000,
+				1500,
+				max_extrinsic_size - 3500,
+				5000,
+				10000,
+			],
+			Ok(vec![3, 4, 2]),
+		);
+		check_split_msgs_to_refine(
+			vec![
+				50,
+				100,
+				max_extrinsic_size - 150,
+				500,
+				1000,
+				1500,
+				max_extrinsic_size - 3000,
+				5000,
+				10000,
+			],
+			Ok(vec![2, 1, 3, 1, 2]),
+		);
+		check_split_msgs_to_refine(
+			vec![
+				5000,
+				10000,
+				max_extrinsic_size - 3500,
+				500,
+				1000,
+				1500,
+				max_extrinsic_size - 500,
+				50,
+				100,
+			],
+			Ok(vec![2, 4, 3]),
+		);
+	}
 }