Limit max number of unconfirmed messages at inbound lane (paritytech#545

)

* limit maximal number of unconfirmed messages at inbound lane

* unrewarded_relayer_entries API

* change relay to support max unrewarded relayer entries

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

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

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

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

* removed pub

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

bin/millau/runtime/src/lib.rs

@@ -322,6 +322,8 @@ impl pallet_shift_session_manager::Trait for Runtime {}
 
 parameter_types! {
 	pub const MaxMessagesToPruneAtOnce: bp_message_lane::MessageNonce = 8;
+	pub const MaxUnrewardedRelayerEntriesAtInboundLane: bp_message_lane::MessageNonce =
+		bp_millau::MAX_UNREWARDED_RELAYER_ENTRIES_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 =
@@ -331,6 +333,7 @@ parameter_types! {
 impl pallet_message_lane::Trait for Runtime {
 	type Event = Event;
 	type MaxMessagesToPruneAtOnce = MaxMessagesToPruneAtOnce;
+	type MaxUnrewardedRelayerEntriesAtInboundLane = MaxUnrewardedRelayerEntriesAtInboundLane;
 	type MaxUnconfirmedMessagesAtInboundLane = MaxUnconfirmedMessagesAtInboundLane;
 	type MaxMessagesInDeliveryTransaction = MaxMessagesInDeliveryTransaction;
 
@@ -580,5 +583,9 @@ impl_runtime_apis! {
 		fn latest_confirmed_nonce(lane: bp_message_lane::LaneId) -> bp_message_lane::MessageNonce {
 			BridgeRialtoMessageLane::inbound_latest_confirmed_nonce(lane)
 		}
+
+		fn unrewarded_relayers_state(lane: bp_message_lane::LaneId) -> bp_message_lane::UnrewardedRelayersState {
+			BridgeRialtoMessageLane::inbound_unrewarded_relayers_state(lane)
+		}
 	}
 }

bin/rialto/runtime/src/lib.rs

@@ -429,6 +429,8 @@ impl pallet_shift_session_manager::Trait for Runtime {}
 
 parameter_types! {
 	pub const MaxMessagesToPruneAtOnce: bp_message_lane::MessageNonce = 8;
+	pub const MaxUnrewardedRelayerEntriesAtInboundLane: bp_message_lane::MessageNonce =
+		bp_millau::MAX_UNREWARDED_RELAYER_ENTRIES_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 =
@@ -438,6 +440,7 @@ parameter_types! {
 impl pallet_message_lane::Trait for Runtime {
 	type Event = Event;
 	type MaxMessagesToPruneAtOnce = MaxMessagesToPruneAtOnce;
+	type MaxUnrewardedRelayerEntriesAtInboundLane = MaxUnrewardedRelayerEntriesAtInboundLane;
 	type MaxUnconfirmedMessagesAtInboundLane = MaxUnconfirmedMessagesAtInboundLane;
 	type MaxMessagesInDeliveryTransaction = MaxMessagesInDeliveryTransaction;
 
@@ -743,6 +746,10 @@ impl_runtime_apis! {
 		fn latest_confirmed_nonce(lane: bp_message_lane::LaneId) -> bp_message_lane::MessageNonce {
 			BridgeMillauMessageLane::inbound_latest_confirmed_nonce(lane)
 		}
+
+		fn unrewarded_relayers_state(lane: bp_message_lane::LaneId) -> bp_message_lane::UnrewardedRelayersState {
+			BridgeMillauMessageLane::inbound_unrewarded_relayers_state(lane)
+		}
 	}
 
 	#[cfg(feature = "runtime-benchmarks")]

modules/message-lane/src/inbound_lane.rs

@@ -31,6 +31,8 @@ pub trait InboundLaneStorage {
 
 	/// Lane id.
 	fn id(&self) -> LaneId;
+	/// Return maximal number of unrewarded relayer entries in inbound lane.
+	fn max_unrewarded_relayer_entries(&self) -> MessageNonce;
 	/// Return maximal number of unconfirmed messages in inbound lane.
 	fn max_unconfirmed_messages(&self) -> MessageNonce;
 	/// Get lane data from the storage.
@@ -97,8 +99,14 @@ impl<S: InboundLaneStorage> InboundLane<S> {
 			return false;
 		}
 
+		// if there are more unrewarded relayer entries than we may accept, reject this message
+		if data.relayers.len() as MessageNonce >= self.storage.max_unrewarded_relayer_entries() {
+			return false;
+		}
+
 		// if there are more unconfirmed messages than we may accept, reject this message
-		if self.storage.max_unconfirmed_messages() <= data.relayers.len() as MessageNonce {
+		let unconfirmed_messages_count = nonce.saturating_sub(data.latest_confirmed_nonce);
+		if unconfirmed_messages_count > self.storage.max_unconfirmed_messages() {
 			return false;
 		}
 
@@ -271,10 +279,10 @@ mod tests {
 	}
 
 	#[test]
-	fn fails_to_receive_messages_above_max_limit_per_lane() {
+	fn fails_to_receive_messages_above_unrewarded_relayer_entries_limit_per_lane() {
 		run_test(|| {
 			let mut lane = inbound_lane::<TestRuntime, _>(TEST_LANE_ID);
-			let max_nonce = <TestRuntime as crate::Trait>::MaxUnconfirmedMessagesAtInboundLane::get();
+			let max_nonce = <TestRuntime as crate::Trait>::MaxUnrewardedRelayerEntriesAtInboundLane::get();
 			for current_nonce in 1..max_nonce + 1 {
 				assert!(lane.receive_message::<TestMessageDispatch>(
 					TEST_RELAYER_A + current_nonce,
@@ -303,6 +311,39 @@ mod tests {
 		});
 	}
 
+	#[test]
+	fn fails_to_receive_messages_above_unconfirmed_messages_limit_per_lane() {
+		run_test(|| {
+			let mut lane = inbound_lane::<TestRuntime, _>(TEST_LANE_ID);
+			let max_nonce = <TestRuntime as crate::Trait>::MaxUnconfirmedMessagesAtInboundLane::get();
+			for current_nonce in 1..=max_nonce {
+				assert!(lane.receive_message::<TestMessageDispatch>(
+					TEST_RELAYER_A,
+					current_nonce,
+					message_data(REGULAR_PAYLOAD).into()
+				));
+			}
+			// Fails to dispatch new message from different than latest relayer.
+			assert_eq!(
+				false,
+				lane.receive_message::<TestMessageDispatch>(
+					TEST_RELAYER_B,
+					max_nonce + 1,
+					message_data(REGULAR_PAYLOAD).into()
+				)
+			);
+			// Fails to dispatch new messages from latest relayer.
+			assert_eq!(
+				false,
+				lane.receive_message::<TestMessageDispatch>(
+					TEST_RELAYER_A,
+					max_nonce + 1,
+					message_data(REGULAR_PAYLOAD).into()
+				)
+			);
+		});
+	}
+
 	#[test]
 	fn correctly_receives_following_messages_from_two_relayers_alternately() {
 		run_test(|| {

modules/message-lane/src/lib.rs

@@ -75,16 +75,20 @@ pub trait Trait<I = DefaultInstance>: frame_system::Trait {
 	/// confirmed. The reason is that if you want to use lane, you should be ready to pay
 	/// for it.
 	type MaxMessagesToPruneAtOnce: Get<MessageNonce>;
-	/// Maximal number of "messages" (see note below) in the 'unconfirmed' state at inbound lane.
-	/// Unconfirmed message at inbound lane is the message that has been: sent, delivered and
-	/// dispatched. Its delivery confirmation is still pending. This limit is introduced to bound
-	/// maximal number of relayers-ids in the inbound lane state.
+	/// Maximal number of unrewarded relayer entries at inbound lane. Unrewarded means that the
+	/// relayer has delivered messages, but either confirmations haven't been delivered back to the
+	/// source chain, or we haven't received reward confirmations yet.
 	///
-	/// "Message" in this context does not necessarily mean an individual message, but instead
-	/// continuous range of individual messages, that are delivered by single relayer. So if relayer#1
-	/// has submitted delivery transaction#1 with individual messages [1; 2] and then delivery
-	/// transaction#2 with individual messages [3; 4], this would be treated as single "Message" and
-	/// would occupy single unit of `MaxUnconfirmedMessagesAtInboundLane` limit.
+	/// This constant limits maximal number of entries in the `InboundLaneData::relayers`. Keep
+	/// in mind that the same relayer account may take several (non-consecutive) entries in this
+	/// set.
+	type MaxUnrewardedRelayerEntriesAtInboundLane: Get<MessageNonce>;
+	/// Maximal number of unconfirmed messages at inbound lane. Unconfirmed means that the
+	/// message has been delivered, but either confirmations haven't been delivered back to the
+	/// source chain, or we haven't received reward confirmations for these messages yet.
+	///
+	/// This constant limits difference between last message from last entry of the
+	/// `InboundLaneData::relayers` and first message at the first entry.
 	type MaxUnconfirmedMessagesAtInboundLane: Get<MessageNonce>;
 	/// Maximal number of messages in single delivery transaction. This directly affects the base
 	/// weight of the delivery transaction.
@@ -481,6 +485,17 @@ impl<T: Trait<I>, I: Instance> Module<T, I> {
 	pub fn inbound_latest_confirmed_nonce(lane: LaneId) -> MessageNonce {
 		InboundLanes::<T, I>::get(&lane).latest_confirmed_nonce
 	}
+
+	/// Get state of unrewarded relayers set.
+	pub fn inbound_unrewarded_relayers_state(
+		lane: bp_message_lane::LaneId,
+	) -> bp_message_lane::UnrewardedRelayersState {
+		let relayers = InboundLanes::<T, I>::get(&lane).relayers;
+		bp_message_lane::UnrewardedRelayersState {
+			unrewarded_relayer_entries: relayers.len() as _,
+			messages_in_oldest_entry: relayers.front().map(|(begin, end, _)| 1 + end - begin).unwrap_or(0),
+		}
+	}
 }
 
 /// Getting storage keys for messages and lanes states. These keys are normally used when building
@@ -569,6 +584,10 @@ impl<T: Trait<I>, I: Instance> InboundLaneStorage for RuntimeInboundLaneStorage<
 		self.lane_id
 	}
 
+	fn max_unrewarded_relayer_entries(&self) -> MessageNonce {
+		T::MaxUnrewardedRelayerEntriesAtInboundLane::get()
+	}
+
 	fn max_unconfirmed_messages(&self) -> MessageNonce {
 		T::MaxUnconfirmedMessagesAtInboundLane::get()
 	}
@@ -681,6 +700,7 @@ mod tests {
 		message, run_test, Origin, TestEvent, TestMessageDeliveryAndDispatchPayment, TestMessagesProof, TestRuntime,
 		PAYLOAD_REJECTED_BY_TARGET_CHAIN, REGULAR_PAYLOAD, TEST_LANE_ID, TEST_RELAYER_A, TEST_RELAYER_B,
 	};
+	use bp_message_lane::UnrewardedRelayersState;
 	use frame_support::{assert_noop, assert_ok};
 	use frame_system::{EventRecord, Module as System, Phase};
 	use hex_literal::hex;
@@ -916,6 +936,13 @@ mod tests {
 						.collect(),
 				},
 			);
+			assert_eq!(
+				Module::<TestRuntime>::inbound_unrewarded_relayers_state(TEST_LANE_ID),
+				UnrewardedRelayersState {
+					unrewarded_relayer_entries: 2,
+					messages_in_oldest_entry: 1,
+				},
+			);
 
 			// message proof includes outbound lane state with latest confirmed message updated to 9
 			let mut message_proof: TestMessagesProof = Ok(vec![message(11, REGULAR_PAYLOAD)]).into();
@@ -941,6 +968,13 @@ mod tests {
 					latest_confirmed_nonce: 9,
 				},
 			);
+			assert_eq!(
+				Module::<TestRuntime>::inbound_unrewarded_relayers_state(TEST_LANE_ID),
+				UnrewardedRelayersState {
+					unrewarded_relayer_entries: 2,
+					messages_in_oldest_entry: 1,
+				},
+			);
 		});
 	}
 

modules/message-lane/src/mock.rs

@@ -99,13 +99,15 @@ impl frame_system::Trait for TestRuntime {
 
 parameter_types! {
 	pub const MaxMessagesToPruneAtOnce: u64 = 10;
-	pub const MaxUnconfirmedMessagesAtInboundLane: u64 = 16;
+	pub const MaxUnrewardedRelayerEntriesAtInboundLane: u64 = 16;
+	pub const MaxUnconfirmedMessagesAtInboundLane: u64 = 32;
 	pub const MaxMessagesInDeliveryTransaction: u64 = 128;
 }
 
 impl Trait for TestRuntime {
 	type Event = TestEvent;
 	type MaxMessagesToPruneAtOnce = MaxMessagesToPruneAtOnce;
+	type MaxUnrewardedRelayerEntriesAtInboundLane = MaxUnrewardedRelayerEntriesAtInboundLane;
 	type MaxUnconfirmedMessagesAtInboundLane = MaxUnconfirmedMessagesAtInboundLane;
 	type MaxMessagesInDeliveryTransaction = MaxMessagesInDeliveryTransaction;
 

primitives/message-lane/src/lib.rs

@@ -102,6 +102,16 @@ impl<RelayerId> Default for InboundLaneData<RelayerId> {
 	}
 }
 
+/// Gist of `InboundLaneData::relayers` field used by runtime APIs.
+#[derive(Clone, Encode, Decode, RuntimeDebug, PartialEq, Eq)]
+pub struct UnrewardedRelayersState {
+	/// Number of entries in the `InboundLaneData::relayers` set.
+	pub unrewarded_relayer_entries: MessageNonce,
+	/// Number of messages in the oldest entry of `InboundLaneData::relayers`. This is the
+	/// minimal number of reward proofs required to push out this entry from the set.
+	pub messages_in_oldest_entry: MessageNonce,
+}
+
 /// Outbound lane data.
 #[derive(Encode, Decode, Clone, RuntimeDebug, PartialEq, Eq)]
 pub struct OutboundLaneData {

primitives/millau/src/lib.rs

@@ -22,7 +22,7 @@
 
 mod millau_hash;
 
-use bp_message_lane::{LaneId, MessageNonce};
+use bp_message_lane::{LaneId, MessageNonce, UnrewardedRelayersState};
 use bp_runtime::Chain;
 use frame_support::{weights::Weight, RuntimeDebug};
 use sp_core::Hasher as HasherT;
@@ -83,6 +83,8 @@ pub const MAXIMUM_EXTRINSIC_SIZE: u32 = MAXIMUM_BLOCK_SIZE / 100 * AVAILABLE_BLO
 // 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 unrewarded relayer entries at inbound lane.
+pub const MAX_UNREWARDED_RELAYER_ENTRIES_AT_INBOUND_LANE: MessageNonce = 1024;
 /// Maximal number of unconfirmed messages at inbound lane.
 pub const MAX_UNCONFIRMED_MESSAGES_AT_INBOUND_LANE: MessageNonce = 1024;
 
@@ -129,6 +131,8 @@ pub const TO_MILLAU_LATEST_GENERATED_NONCE_METHOD: &str = "ToMillauOutboundLaneA
 pub const FROM_MILLAU_LATEST_RECEIVED_NONCE_METHOD: &str = "FromMillauInboundLaneApi_latest_received_nonce";
 /// Name of the `FromMillauInboundLaneApi::latest_onfirmed_nonce` runtime method.
 pub const FROM_MILLAU_LATEST_CONFIRMED_NONCE_METHOD: &str = "FromMillauInboundLaneApi_latest_confirmed_nonce";
+/// Name of the `FromMillauInboundLaneApi::unrewarded_relayers_state` runtime method.
+pub const FROM_MILLAU_UNREWARDED_RELAYERS_STATE: &str = "FromMillauInboundLaneApi_unrewarded_relayers_state";
 
 /// Alias to 512-bit hash when used in the context of a transaction signature on the chain.
 pub type Signature = MultiSignature;
@@ -206,5 +210,7 @@ sp_api::decl_runtime_apis! {
 		fn latest_received_nonce(lane: LaneId) -> MessageNonce;
 		/// Nonce of latest message that has been confirmed to the bridged chain.
 		fn latest_confirmed_nonce(lane: LaneId) -> MessageNonce;
+		/// State of the unrewarded relayers set at given lane.
+		fn unrewarded_relayers_state(lane: LaneId) -> UnrewardedRelayersState;
 	}
 }

primitives/rialto/src/lib.rs

@@ -20,7 +20,7 @@
 // Runtime-generated DecodeLimit::decode_all_With_depth_limit
 #![allow(clippy::unnecessary_mut_passed)]
 
-use bp_message_lane::{LaneId, MessageNonce};
+use bp_message_lane::{LaneId, MessageNonce, UnrewardedRelayersState};
 use bp_runtime::Chain;
 use frame_support::{weights::Weight, RuntimeDebug};
 use sp_core::Hasher as HasherT;
@@ -45,6 +45,8 @@ pub const MAXIMUM_EXTRINSIC_SIZE: u32 = MAXIMUM_BLOCK_SIZE / 100 * AVAILABLE_BLO
 // 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 unrewarded relayer entries at inbound lane.
+pub const MAX_UNREWARDED_RELAYER_ENTRIES_AT_INBOUND_LANE: MessageNonce = 128;
 /// Maximal number of unconfirmed messages at inbound lane.
 pub const MAX_UNCONFIRMED_MESSAGES_AT_INBOUND_LANE: MessageNonce = 128;
 
@@ -91,6 +93,8 @@ pub const TO_RIALTO_LATEST_RECEIVED_NONCE_METHOD: &str = "ToRialtoOutboundLaneAp
 pub const FROM_RIALTO_LATEST_RECEIVED_NONCE_METHOD: &str = "FromRialtoInboundLaneApi_latest_received_nonce";
 /// Name of the `FromRialtoInboundLaneApi::latest_onfirmed_nonce` runtime method.
 pub const FROM_RIALTO_LATEST_CONFIRMED_NONCE_METHOD: &str = "FromRialtoInboundLaneApi_latest_confirmed_nonce";
+/// Name of the `FromRialtoInboundLaneApi::unrewarded_relayers_state` runtime method.
+pub const FROM_RIALTO_UNREWARDED_RELAYERS_STATE: &str = "FromRialtoInboundLaneApi_unrewarded_relayers_state";
 
 /// Alias to 512-bit hash when used in the context of a transaction signature on the chain.
 pub type Signature = MultiSignature;
@@ -169,5 +173,7 @@ sp_api::decl_runtime_apis! {
 		fn latest_received_nonce(lane: LaneId) -> MessageNonce;
 		/// Nonce of latest message that has been confirmed to the bridged chain.
 		fn latest_confirmed_nonce(lane: LaneId) -> MessageNonce;
+		/// State of the unrewarded relayers set at given lane.
+		fn unrewarded_relayers_state(lane: LaneId) -> UnrewardedRelayersState;
 	}
 }

relays/messages-relay/src/message_lane_loop.rs

@@ -30,7 +30,7 @@ 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, MessageNonce, Weight};
+use bp_message_lane::{LaneId, MessageNonce, UnrewardedRelayersState, Weight};
 use futures::{channel::mpsc::unbounded, future::FutureExt, stream::StreamExt};
 use relay_utils::{
 	interval,
@@ -59,6 +59,10 @@ pub struct Params {
 /// Message delivery race parameters.
 #[derive(Debug, Clone)]
 pub struct MessageDeliveryParams {
+	/// Maximal number of unconfirmed relayer entries at the inbound lane. If there's that number of entries
+	/// in the `InboundLaneData::relayers` set, all new messages will be rejected until reward payment will
+	/// be proved (by including outbound lane state to the message delivery transaction).
+	pub max_unrewarded_relayer_entries_at_target: MessageNonce,
 	/// 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.
@@ -153,6 +157,11 @@ pub trait TargetClient<P: MessageLane>: Clone + Send + Sync {
 		&self,
 		id: TargetHeaderIdOf<P>,
 	) -> Result<(TargetHeaderIdOf<P>, MessageNonce), Self::Error>;
+	/// Get state of unrewarded relayers set at the inbound lane.
+	async fn unrewarded_relayers_state(
+		&self,
+		id: TargetHeaderIdOf<P>,
+	) -> Result<(TargetHeaderIdOf<P>, UnrewardedRelayersState), Self::Error>;
 
 	/// Prove messages receiving at given block.
 	async fn prove_messages_receiving(
@@ -662,6 +671,19 @@ pub(crate) mod tests {
 			Ok((id, data.target_latest_received_nonce))
 		}
 
+		async fn unrewarded_relayers_state(
+			&self,
+			id: TargetHeaderIdOf<TestMessageLane>,
+		) -> Result<(TargetHeaderIdOf<TestMessageLane>, UnrewardedRelayersState), Self::Error> {
+			Ok((
+				id,
+				UnrewardedRelayersState {
+					unrewarded_relayer_entries: 0,
+					messages_in_oldest_entry: 0,
+				},
+			))
+		}
+
 		async fn latest_confirmed_received_nonce(
 			&self,
 			id: TargetHeaderIdOf<TestMessageLane>,
@@ -728,6 +750,7 @@ pub(crate) mod tests {
 					reconnect_delay: Duration::from_millis(0),
 					stall_timeout: Duration::from_millis(60 * 1000),
 					delivery_params: MessageDeliveryParams {
+						max_unrewarded_relayer_entries_at_target: 4,
 						max_unconfirmed_nonces_at_target: 4,
 						max_messages_in_single_batch: 4,
 						max_messages_weight_in_single_batch: 4,