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 <[email protected]>

* add issue reference

* add assert message

* grumbles

* fmt

* reexport weight from bp-message-lane

Co-authored-by: Hernando Castano <[email protected]>
Co-authored-by: Hernando Castano <[email protected]>

Author: svyatonik

Cargo.lock

@@ -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),
 			)));

bin/millau/node/src/service.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 {
-			(begin..=end)
-				.filter_map(|nonce| BridgeRialtoMessageLane::outbound_message_payload(lane, nonce))
-				.filter_map(|encoded_payload| rialto_messages::ToRialtoMessagePayload::decode(&mut &encoded_payload[..]).ok())
-				.map(|decoded_payload| decoded_payload.weight)
-				.fold(0, |sum, weight| sum.saturating_add(weight))
+		fn messages_dispatch_weight(
+			lane: bp_message_lane::LaneId,
+			begin: bp_message_lane::MessageNonce,
+			end: bp_message_lane::MessageNonce,
+		) -> Vec<(bp_message_lane::MessageNonce, 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 {

bin/millau/runtime/src/lib.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),
 			)));

bin/rialto/node/src/service.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 {
-			(begin..=end)
-				.filter_map(|nonce| BridgeMillauMessageLane::outbound_message_payload(lane, nonce))
-				.filter_map(|encoded_payload| millau_messages::ToMillauMessagePayload::decode(&mut &encoded_payload[..]).ok())
-				.map(|decoded_payload| decoded_payload.weight)
-				.fold(0, |sum, weight| sum.saturating_add(weight))
+		fn messages_dispatch_weight(
+			lane: bp_message_lane::LaneId,
+			begin: bp_message_lane::MessageNonce,
+			end: bp_message_lane::MessageNonce,
+		) -> Vec<(bp_message_lane::MessageNonce, 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 {

bin/rialto/runtime/src/lib.rs

@@ -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"

modules/message-lane/rpc/Cargo.toml

@@ -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.
-	/// The state of outbound lane is included in the proof if `include_outbound_lane_state` is true.
+	/// Returns storage proof of messages in given inclusive range. The state of outbound
+	/// 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,30 +78,26 @@ pub trait MessageLaneApi<BlockHash> {
 }
 
 /// Implements the MessageLaneApi trait for interacting with message lanes.
-pub struct MessageLaneRpcHandler<Block, Client, Backend, R> {
-	client: Arc<Client>,
+pub struct MessageLaneRpcHandler<Block, Backend, R> {
 	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(),
 		}
 	}
 }
 
-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)> {
-		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(),
-				))
-			}
-		};
-
+	) -> FutureResult<MessagesProof> {
 		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),
 		)
 	}

modules/message-lane/rpc/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.

primitives/message-lane/src/lib.rs

@@ -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

relays/messages-relay/Cargo.toml

@@ -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.