Limit max number of unconfirmed messages at inbound lane (#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>

bridges/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)
+		}
 	}
 }

bridges/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")]

bridges/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(|| {

bridges/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,
+				},
+			);
 		});
 	}
 

bridges/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;
 

bridges/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 {

bridges/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;
 	}
 }

bridges/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;
 	}
 }

bridges/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,

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

@@ -26,7 +26,7 @@ use crate::message_race_strategy::BasicStrategy;
 use crate::metrics::MessageLaneLoopMetrics;
 
 use async_trait::async_trait;
-use bp_message_lane::{MessageNonce, Weight};
+use bp_message_lane::{MessageNonce, UnrewardedRelayersState, Weight};
 use futures::stream::FusedStream;
 use relay_utils::FailedClient;
 use std::{collections::BTreeMap, marker::PhantomData, ops::RangeInclusive, time::Duration};
@@ -56,6 +56,7 @@ pub async fn run<P: MessageLane>(
 		target_state_updates,
 		stall_timeout,
 		MessageDeliveryStrategy::<P> {
+			max_unrewarded_relayer_entries_at_target: params.max_unrewarded_relayer_entries_at_target,
 			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,
@@ -157,13 +158,15 @@ where
 	C: MessageLaneTargetClient<P>,
 {
 	type Error = C::Error;
+	type TargetNoncesData = DeliveryRaceTargetNoncesData;
 
 	async fn nonces(
 		&self,
 		at_block: TargetHeaderIdOf<P>,
-	) -> Result<(TargetHeaderIdOf<P>, TargetClientNonces), Self::Error> {
+	) -> Result<(TargetHeaderIdOf<P>, TargetClientNonces<DeliveryRaceTargetNoncesData>), Self::Error> {
 		let (at_block, latest_received_nonce) = self.client.latest_received_nonce(at_block).await?;
 		let (at_block, latest_confirmed_nonce) = self.client.latest_confirmed_received_nonce(at_block).await?;
+		let (at_block, unrewarded_relayers) = self.client.unrewarded_relayers_state(at_block).await?;
 
 		if let Some(metrics_msg) = self.metrics_msg.as_ref() {
 			metrics_msg.update_target_latest_received_nonce::<P>(latest_received_nonce);
@@ -174,7 +177,10 @@ where
 			at_block,
 			TargetClientNonces {
 				latest_nonce: latest_received_nonce,
-				confirmed_nonce: Some(latest_confirmed_nonce),
+				nonces_data: DeliveryRaceTargetNoncesData {
+					confirmed_nonce: latest_confirmed_nonce,
+					unrewarded_relayers,
+				},
 			},
 		))
 	}
@@ -191,8 +197,21 @@ where
 	}
 }
 
+/// Additional nonces data from the target client used by message delivery race.
+#[derive(Debug, Clone)]
+struct DeliveryRaceTargetNoncesData {
+	/// Latest nonce that we know: (1) has been delivered to us (2) has been confirmed
+	/// back to the source node (by confirmations race) and (3) relayer has received
+	/// reward for (and this has been confirmed by the message delivery race).
+	confirmed_nonce: MessageNonce,
+	/// State of the unrewarded relayers set at the target node.
+	unrewarded_relayers: UnrewardedRelayersState,
+}
+
 /// Messages delivery strategy.
 struct MessageDeliveryStrategy<P: MessageLane> {
+	/// Maximal unrewarded relayer entries at target client.
+	max_unrewarded_relayer_entries_at_target: MessageNonce,
 	/// Maximal unconfirmed nonces at target client.
 	max_unconfirmed_nonces_at_target: MessageNonce,
 	/// Maximal number of messages in the single delivery transaction.
@@ -202,7 +221,7 @@ struct MessageDeliveryStrategy<P: MessageLane> {
 	/// Latest confirmed nonce at the source client.
 	latest_confirmed_nonce_at_source: Option<MessageNonce>,
 	/// Target nonces from the source client.
-	target_nonces: Option<TargetClientNonces>,
+	target_nonces: Option<TargetClientNonces<DeliveryRaceTargetNoncesData>>,
 	/// Basic delivery strategy.
 	strategy: MessageDeliveryStrategyBase<P>,
 }
@@ -221,6 +240,7 @@ impl<P: MessageLane> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::M
 {
 	type SourceNoncesRange = MessageWeightsMap;
 	type ProofParameters = MessageProofParameters;
+	type TargetNoncesData = DeliveryRaceTargetNoncesData;
 
 	fn is_empty(&self) -> bool {
 		self.strategy.is_empty()
@@ -245,22 +265,23 @@ impl<P: MessageLane> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::M
 
 	fn target_nonces_updated(
 		&mut self,
-		nonces: TargetClientNonces,
+		nonces: TargetClientNonces<DeliveryRaceTargetNoncesData>,
 		race_state: &mut RaceState<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::MessagesProof>,
 	) {
 		self.target_nonces = Some(nonces.clone());
-		self.strategy.target_nonces_updated(nonces, race_state)
+		self.strategy.target_nonces_updated(
+			TargetClientNonces {
+				latest_nonce: nonces.latest_nonce,
+				nonces_data: (),
+			},
+			race_state,
+		)
 	}
 
 	fn select_nonces_to_deliver(
 		&mut self,
 		race_state: &RaceState<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::MessagesProof>,
 	) -> Option<(RangeInclusive<MessageNonce>, Self::ProofParameters)> {
-		const CONFIRMED_NONCE_PROOF: &str = "\
-			ClientNonces are crafted by MessageDeliveryRace(Source|Target);\
-			MessageDeliveryRace(Source|Target) always fills confirmed_nonce field;\
-			qed";
-
 		let latest_confirmed_nonce_at_source = self.latest_confirmed_nonce_at_source?;
 		let target_nonces = self.target_nonces.as_ref()?;
 
@@ -295,9 +316,28 @@ impl<P: MessageLane> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::M
 		//
 		// Important note: we're including outbound state lane proof whenever there are unconfirmed nonces
 		// on the target chain. Other strategy is to include it only if it's absolutely necessary.
-		let latest_confirmed_nonce_at_target = target_nonces.confirmed_nonce.expect(CONFIRMED_NONCE_PROOF);
+		let latest_confirmed_nonce_at_target = target_nonces.nonces_data.confirmed_nonce;
 		let outbound_state_proof_required = latest_confirmed_nonce_at_target < latest_confirmed_nonce_at_source;
 
+		// The target node would also reject messages if there are too many entries in the
+		// "unrewarded relayers" set. If we are unable to prove new rewards to the target node, then
+		// we should wait for confirmations race.
+		let unrewarded_relayer_entries_limit_reached =
+			target_nonces.nonces_data.unrewarded_relayers.unrewarded_relayer_entries
+				>= self.max_unrewarded_relayer_entries_at_target;
+		if unrewarded_relayer_entries_limit_reached {
+			// so there are already too many unrewarded relayer entries in the set
+			//
+			// => check if we can prove enough rewards. If not, we should wait for more rewards to be paid
+			let number_of_rewards_being_proved =
+				latest_confirmed_nonce_at_source.saturating_sub(latest_confirmed_nonce_at_target);
+			let enough_rewards_being_proved = number_of_rewards_being_proved
+				>= target_nonces.nonces_data.unrewarded_relayers.messages_in_oldest_entry;
+			if !enough_rewards_being_proved {
+				return None;
+			}
+		}
+
 		// If we're here, then the confirmations race did its job && sending side now knows that messages
 		// have been delivered. Now let's select nonces that we want to deliver.
 		//
@@ -404,13 +444,20 @@ mod tests {
 		};
 
 		let mut race_strategy = TestStrategy {
+			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,
 			latest_confirmed_nonce_at_source: Some(19),
 			target_nonces: Some(TargetClientNonces {
 				latest_nonce: 19,
-				confirmed_nonce: Some(19),
+				nonces_data: DeliveryRaceTargetNoncesData {
+					confirmed_nonce: 19,
+					unrewarded_relayers: UnrewardedRelayersState {
+						unrewarded_relayer_entries: 0,
+						messages_in_oldest_entry: 0,
+					},
+				},
 			}),
 			strategy: BasicStrategy::new(),
 		};
@@ -422,9 +469,13 @@ mod tests {
 				confirmed_nonce: Some(19),
 			},
 		);
-		race_strategy
-			.strategy
-			.target_nonces_updated(race_strategy.target_nonces.clone().unwrap(), &mut race_state);
+		race_strategy.strategy.target_nonces_updated(
+			TargetClientNonces {
+				latest_nonce: 19,
+				nonces_data: (),
+			},
+			&mut race_state,
+		);
 
 		(race_state, race_strategy)
 	}
@@ -483,7 +534,58 @@ mod tests {
 		// if there are new confirmed nonces on source, we want to relay this information
 		// to target to prune rewards queue
 		let prev_confirmed_nonce_at_source = strategy.latest_confirmed_nonce_at_source.unwrap();
-		strategy.target_nonces.as_mut().unwrap().confirmed_nonce = Some(prev_confirmed_nonce_at_source - 1);
+		strategy.target_nonces.as_mut().unwrap().nonces_data.confirmed_nonce = prev_confirmed_nonce_at_source - 1;
+		assert_eq!(
+			strategy.select_nonces_to_deliver(&state),
+			Some(((20..=23), proof_parameters(true, 4)))
+		);
+	}
+
+	#[test]
+	fn message_delivery_strategy_selects_nothing_if_there_are_too_many_unrewarded_relayers() {
+		let (state, mut strategy) = prepare_strategy();
+
+		// if there are already `max_unrewarded_relayer_entries_at_target` entries at target,
+		// we need to wait until rewards will be paid
+		{
+			let mut unrewarded_relayers = &mut strategy.target_nonces.as_mut().unwrap().nonces_data.unrewarded_relayers;
+			unrewarded_relayers.unrewarded_relayer_entries = strategy.max_unrewarded_relayer_entries_at_target;
+			unrewarded_relayers.messages_in_oldest_entry = 4;
+		}
+		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
+	}
+
+	#[test]
+	fn message_delivery_strategy_selects_nothing_if_proved_rewards_is_not_enough_to_remove_oldest_unrewarded_entry() {
+		let (state, mut strategy) = prepare_strategy();
+
+		// if there are already `max_unrewarded_relayer_entries_at_target` entries at target,
+		// we need to prove at least `messages_in_oldest_entry` rewards
+		let prev_confirmed_nonce_at_source = strategy.latest_confirmed_nonce_at_source.unwrap();
+		{
+			let mut nonces_data = &mut strategy.target_nonces.as_mut().unwrap().nonces_data;
+			nonces_data.confirmed_nonce = prev_confirmed_nonce_at_source - 1;
+			let mut unrewarded_relayers = &mut nonces_data.unrewarded_relayers;
+			unrewarded_relayers.unrewarded_relayer_entries = strategy.max_unrewarded_relayer_entries_at_target;
+			unrewarded_relayers.messages_in_oldest_entry = 4;
+		}
+		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
+	}
+
+	#[test]
+	fn message_delivery_strategy_includes_outbound_state_proof_if_proved_rewards_is_enough() {
+		let (state, mut strategy) = prepare_strategy();
+
+		// if there are already `max_unrewarded_relayer_entries_at_target` entries at target,
+		// we need to prove at least `messages_in_oldest_entry` rewards
+		let prev_confirmed_nonce_at_source = strategy.latest_confirmed_nonce_at_source.unwrap();
+		{
+			let mut nonces_data = &mut strategy.target_nonces.as_mut().unwrap().nonces_data;
+			nonces_data.confirmed_nonce = prev_confirmed_nonce_at_source - 3;
+			let mut unrewarded_relayers = &mut nonces_data.unrewarded_relayers;
+			unrewarded_relayers.unrewarded_relayer_entries = strategy.max_unrewarded_relayer_entries_at_target;
+			unrewarded_relayers.messages_in_oldest_entry = 3;
+		}
 		assert_eq!(
 			strategy.select_nonces_to_deliver(&state),
 			Some(((20..=23), proof_parameters(true, 4)))
@@ -522,7 +624,7 @@ mod tests {
 		// relay 3 new messages
 		let prev_confirmed_nonce_at_source = strategy.latest_confirmed_nonce_at_source.unwrap();
 		strategy.latest_confirmed_nonce_at_source = Some(prev_confirmed_nonce_at_source - 1);
-		strategy.target_nonces.as_mut().unwrap().confirmed_nonce = Some(prev_confirmed_nonce_at_source - 1);
+		strategy.target_nonces.as_mut().unwrap().nonces_data.confirmed_nonce = prev_confirmed_nonce_at_source - 1;
 		assert_eq!(
 			strategy.select_nonces_to_deliver(&state),
 			Some(((20..=22), proof_parameters(false, 3)))

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

@@ -83,12 +83,11 @@ pub struct SourceClientNonces<NoncesRange> {
 
 /// Nonces on the race target client.
 #[derive(Debug, Clone)]
-pub struct TargetClientNonces {
+pub struct TargetClientNonces<TargetNoncesData> {
 	/// Latest nonce that is known to the target client.
 	pub latest_nonce: MessageNonce,
-	/// Latest nonce that is confirmed to the bridged client. This nonce only makes
-	/// sense in some races. In other races it is `None`.
-	pub confirmed_nonce: Option<MessageNonce>,
+	/// Additional data from target node that may be used by the race.
+	pub nonces_data: TargetNoncesData,
 }
 
 /// One of message lane clients, which is source client for the race.
@@ -121,10 +120,14 @@ pub trait SourceClient<P: MessageRace> {
 pub trait TargetClient<P: MessageRace> {
 	/// Type of error this clients returns.
 	type Error: std::fmt::Debug + MaybeConnectionError;
+	/// Type of the additional data from the target client, used by the race.
+	type TargetNoncesData: std::fmt::Debug;
 
 	/// Return nonces that are known to the target client.
-	async fn nonces(&self, at_block: P::TargetHeaderId)
-		-> Result<(P::TargetHeaderId, TargetClientNonces), Self::Error>;
+	async fn nonces(
+		&self,
+		at_block: P::TargetHeaderId,
+	) -> Result<(P::TargetHeaderId, TargetClientNonces<Self::TargetNoncesData>), Self::Error>;
 	/// Submit proof to the target client.
 	async fn submit_proof(
 		&self,
@@ -140,6 +143,8 @@ pub trait RaceStrategy<SourceHeaderId, TargetHeaderId, Proof> {
 	type SourceNoncesRange: NoncesRange;
 	/// Additional proof parameters required to generate proof.
 	type ProofParameters;
+	/// Additional data expected from the target client.
+	type TargetNoncesData;
 
 	/// Should return true if nothing has to be synced.
 	fn is_empty(&self) -> bool;
@@ -158,7 +163,7 @@ pub trait RaceStrategy<SourceHeaderId, TargetHeaderId, Proof> {
 	/// Called when nonces are updated at target node of the race.
 	fn target_nonces_updated(
 		&mut self,
-		nonces: TargetClientNonces,
+		nonces: TargetClientNonces<Self::TargetNoncesData>,
 		race_state: &mut RaceState<SourceHeaderId, TargetHeaderId, Proof>,
 	);
 	/// Should return `Some(nonces)` if we need to deliver proof of `nonces` (and associated
@@ -187,10 +192,10 @@ pub struct RaceState<SourceHeaderId, TargetHeaderId, Proof> {
 }
 
 /// Run race loop until connection with target or source node is lost.
-pub async fn run<P: MessageRace, SC: SourceClient<P>>(
+pub async fn run<P: MessageRace, SC: SourceClient<P>, TC: TargetClient<P>>(
 	race_source: SC,
 	race_source_updated: impl FusedStream<Item = SourceClientState<P>>,
-	race_target: impl TargetClient<P>,
+	race_target: TC,
 	race_target_updated: impl FusedStream<Item = TargetClientState<P>>,
 	stall_timeout: Duration,
 	mut strategy: impl RaceStrategy<
@@ -199,6 +204,7 @@ pub async fn run<P: MessageRace, SC: SourceClient<P>>(
 		P::Proof,
 		SourceNoncesRange = SC::NoncesRange,
 		ProofParameters = SC::ProofParameters,
+		TargetNoncesData = TC::TargetNoncesData,
 	>,
 ) -> Result<(), FailedClient> {
 	let mut progress_context = Instant::now();
@@ -524,7 +530,7 @@ mod tests {
 		strategy.target_nonces_updated(
 			TargetClientNonces {
 				latest_nonce: 5u64,
-				confirmed_nonce: None,
+				nonces_data: (),
 			},
 			&mut race_state,
 		);

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

@@ -157,11 +157,12 @@ where
 	C: MessageLaneSourceClient<P>,
 {
 	type Error = C::Error;
+	type TargetNoncesData = ();
 
 	async fn nonces(
 		&self,
 		at_block: SourceHeaderIdOf<P>,
-	) -> Result<(SourceHeaderIdOf<P>, TargetClientNonces), Self::Error> {
+	) -> Result<(SourceHeaderIdOf<P>, TargetClientNonces<()>), Self::Error> {
 		let (at_block, latest_confirmed_nonce) = self.client.latest_confirmed_received_nonce(at_block).await?;
 		if let Some(metrics_msg) = self.metrics_msg.as_ref() {
 			metrics_msg.update_source_latest_confirmed_nonce::<P>(latest_confirmed_nonce);
@@ -170,7 +171,7 @@ where
 			at_block,
 			TargetClientNonces {
 				latest_nonce: latest_confirmed_nonce,
-				confirmed_nonce: None,
+				nonces_data: (),
 			},
 		))
 	}

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

@@ -145,6 +145,7 @@ where
 {
 	type SourceNoncesRange = SourceNoncesRange;
 	type ProofParameters = ();
+	type TargetNoncesData = ();
 
 	fn is_empty(&self) -> bool {
 		self.source_queue.is_empty()
@@ -185,7 +186,7 @@ where
 
 	fn target_nonces_updated(
 		&mut self,
-		nonces: TargetClientNonces,
+		nonces: TargetClientNonces<()>,
 		race_state: &mut RaceState<
 			HeaderId<SourceHeaderHash, SourceHeaderNumber>,
 			HeaderId<TargetHeaderHash, TargetHeaderNumber>,
@@ -269,10 +270,10 @@ mod tests {
 		}
 	}
 
-	fn target_nonces(latest_nonce: MessageNonce) -> TargetClientNonces {
+	fn target_nonces(latest_nonce: MessageNonce) -> TargetClientNonces<()> {
 		TargetClientNonces {
 			latest_nonce,
-			confirmed_nonce: None,
+			nonces_data: (),
 		}
 	}
 

bridges/relays/substrate/src/messages_lane.rs

@@ -35,10 +35,12 @@ pub trait SubstrateMessageLane: MessageLane {
 	/// Name of the runtime method that returns latest received (confirmed) nonce at the the source chain.
 	const OUTBOUND_LANE_LATEST_RECEIVED_NONCE_METHOD: &'static str;
 
-	/// Name of the runtime method that returns latest received nonce at the source chain.
+	/// Name of the runtime method that returns latest received nonce at the target chain.
 	const INBOUND_LANE_LATEST_RECEIVED_NONCE_METHOD: &'static str;
-	/// Name of the runtime method that returns latest confirmed (reward-paid) nonce at the source chain.
+	/// Name of the runtime method that returns latest confirmed (reward-paid) nonce at the target chain.
 	const INBOUND_LANE_LATEST_CONFIRMED_NONCE_METHOD: &'static str;
+	/// Numebr of the runtime method that returns state of "unrewarded relayers" set at the target chain.
+	const INBOUND_LANE_UNREWARDED_RELAYERS_STATE: &'static str;
 
 	/// Name of the runtime method that returns id of best finalized source header at target chain.
 	const BEST_FINALIZED_SOURCE_HEADER_ID_AT_TARGET: &'static str;

bridges/relays/substrate/src/messages_target.rs

@@ -22,7 +22,7 @@ use crate::messages_lane::SubstrateMessageLane;
 use crate::messages_source::read_client_state;
 
 use async_trait::async_trait;
-use bp_message_lane::{LaneId, MessageNonce};
+use bp_message_lane::{LaneId, MessageNonce, UnrewardedRelayersState};
 use bp_runtime::InstanceId;
 use codec::{Decode, Encode};
 use messages_relay::{
@@ -135,6 +135,23 @@ where
 		Ok((id, latest_received_nonce))
 	}
 
+	async fn unrewarded_relayers_state(
+		&self,
+		id: TargetHeaderIdOf<P>,
+	) -> Result<(TargetHeaderIdOf<P>, UnrewardedRelayersState), Self::Error> {
+		let encoded_response = self
+			.client
+			.state_call(
+				P::INBOUND_LANE_UNREWARDED_RELAYERS_STATE.into(),
+				Bytes(self.lane_id.encode()),
+				Some(id.1),
+			)
+			.await?;
+		let unrewarded_relayers_state: UnrewardedRelayersState =
+			Decode::decode(&mut &encoded_response.0[..]).map_err(SubstrateError::ResponseParseFailed)?;
+		Ok((id, unrewarded_relayers_state))
+	}
+
 	async fn prove_messages_receiving(
 		&self,
 		id: TargetHeaderIdOf<P>,

bridges/relays/substrate/src/millau_messages_to_rialto.rs

@@ -46,6 +46,7 @@ impl SubstrateMessageLane for MillauMessagesToRialto {
 	const INBOUND_LANE_LATEST_RECEIVED_NONCE_METHOD: &'static str = bp_millau::FROM_MILLAU_LATEST_RECEIVED_NONCE_METHOD;
 	const INBOUND_LANE_LATEST_CONFIRMED_NONCE_METHOD: &'static str =
 		bp_millau::FROM_MILLAU_LATEST_CONFIRMED_NONCE_METHOD;
+	const INBOUND_LANE_UNREWARDED_RELAYERS_STATE: &'static str = bp_millau::FROM_MILLAU_UNREWARDED_RELAYERS_STATE;
 
 	const BEST_FINALIZED_SOURCE_HEADER_ID_AT_TARGET: &'static str = bp_millau::FINALIZED_MILLAU_BLOCK_METHOD;
 	const BEST_FINALIZED_TARGET_HEADER_ID_AT_SOURCE: &'static str = bp_rialto::FINALIZED_RIALTO_BLOCK_METHOD;
@@ -126,6 +127,7 @@ pub fn run(
 			reconnect_delay,
 			stall_timeout,
 			delivery_params: messages_relay::message_lane_loop::MessageDeliveryParams {
+				max_unrewarded_relayer_entries_at_target: bp_rialto::MAX_UNREWARDED_RELAYER_ENTRIES_AT_INBOUND_LANE,
 				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

bridges/relays/substrate/src/rialto_messages_to_millau.rs

@@ -46,6 +46,7 @@ impl SubstrateMessageLane for RialtoMessagesToMillau {
 	const INBOUND_LANE_LATEST_RECEIVED_NONCE_METHOD: &'static str = bp_rialto::FROM_RIALTO_LATEST_RECEIVED_NONCE_METHOD;
 	const INBOUND_LANE_LATEST_CONFIRMED_NONCE_METHOD: &'static str =
 		bp_rialto::FROM_RIALTO_LATEST_CONFIRMED_NONCE_METHOD;
+	const INBOUND_LANE_UNREWARDED_RELAYERS_STATE: &'static str = bp_rialto::FROM_RIALTO_UNREWARDED_RELAYERS_STATE;
 
 	const BEST_FINALIZED_SOURCE_HEADER_ID_AT_TARGET: &'static str = bp_rialto::FINALIZED_RIALTO_BLOCK_METHOD;
 	const BEST_FINALIZED_TARGET_HEADER_ID_AT_SOURCE: &'static str = bp_millau::FINALIZED_MILLAU_BLOCK_METHOD;
@@ -126,6 +127,7 @@ pub fn run(
 			reconnect_delay,
 			stall_timeout,
 			delivery_params: messages_relay::message_lane_loop::MessageDeliveryParams {
+				max_unrewarded_relayer_entries_at_target: bp_millau::MAX_UNREWARDED_RELAYER_ENTRIES_AT_INBOUND_LANE,
 				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