Reward relayers for dispatching messages (#385)

* reward relayers for dispatching messages * clippy * Update modules/message-lane/src/lib.rs Co-authored-by: Hernando Castano <HCastano@users.noreply.github.com> * added comment * Update modules/message-lane/src/inbound_lane.rs Co-authored-by: Tomasz Drwięga <tomusdrw@users.noreply.github.com> * Update modules/message-lane/src/inbound_lane.rs Co-authored-by: Tomasz Drwięga <tomusdrw@users.noreply.github.com> * SubmitterId + RelayerId -> AccountId * add confirmation_relayer arg to pay_relayer_reward * cargo fmt --all * removed verify_and_decode_messages_proof from SourceHeaderChain * &mut self -> RefCell * Optimize max messages at inbound lane (#418) * Add tests for checking messages above max limit Signed-off-by: MaciejBaj <macie.baj@gmail.com> * Extend the relayers entry of inbound lane by additional msg nonce Signed-off-by: MaciejBaj <macie.baj@gmail.com> * Support additional message nonce from inbound relayers Signed-off-by: MaciejBaj <macie.baj@gmail.com> * Code format Signed-off-by: MaciejBaj <macie.baj@gmail.com> * Merge messages range for highest relayers * Change unwrap() to ensure() while accessing relayers * Edit rustdocs for relayers deque at inbound lane data Co-authored-by: Hernando Castano <HCastano@users.noreply.github.com> * Declare additional relayers A & B and use across tests consistently Signed-off-by: MaciejBaj <macie.baj@gmail.com> * Remove duplicates and improve naming for inbound lane tests * Fix test checking max limit per inbound lane * Correct relayers rewards loop after a proof is received * Remove redundant check for messages ahead of received range * Correct grammar at inbound lane tests rustdocs Co-authored-by: Hernando Castano <HCastano@users.noreply.github.com> * Improve code quality of relayers updates 💅 Co-authored-by: Tomasz Drwięga <tomusdrw@users.noreply.github.com> * Test dispatches above max limit from same relayer Co-authored-by: Hernando Castano <HCastano@users.noreply.github.com> Co-authored-by: Tomasz Drwięga <tomusdrw@users.noreply.github.com> * Fix typo. Co-authored-by: Hernando Castano <HCastano@users.noreply.github.com> Co-authored-by: Tomasz Drwięga <tomusdrw@users.noreply.github.com> Co-authored-by: Maciej Baj <macie.baj@gmail.com> Co-authored-by: Tomasz Drwięga <tomasz@parity.io>
2026-05-31 13:21:01 +00:00 · 2020-10-20 13:12:35 +03:00
parent c42023269a
commit bff930d01e
6 changed files with 649 additions and 81 deletions
@@ -18,20 +18,25 @@

 use bp_message_lane::{
 	target_chain::{DispatchMessage, DispatchMessageData, MessageDispatch},
-	InboundLaneData, LaneId, MessageKey, MessageNonce,
+	InboundLaneData, LaneId, MessageKey, MessageNonce, OutboundLaneData,
 };
+use sp_std::prelude::PartialEq;

 /// Inbound lane storage.
 pub trait InboundLaneStorage {
 	/// Delivery and dispatch fee type on source chain.
 	type MessageFee;
+	/// Id of relayer on source chain.
+	type Relayer: PartialEq;

 	/// Lane id.
 	fn id(&self) -> LaneId;
+	/// Return maximal number of unconfirmed messages in inbound lane.
+	fn max_unconfirmed_messages(&self) -> MessageNonce;
 	/// Get lane data from the storage.
-	fn data(&self) -> InboundLaneData;
+	fn data(&self) -> InboundLaneData<Self::Relayer>;
 	/// Update lane data in the storage.
-	fn set_data(&mut self, data: InboundLaneData);
+	fn set_data(&mut self, data: InboundLaneData<Self::Relayer>);
 }

 /// Inbound messages lane.
@@ -45,9 +50,44 @@ impl<S: InboundLaneStorage> InboundLane<S> {
 		InboundLane { storage }
 	}

+	/// Receive state of the corresponding outbound lane.
+	pub fn receive_state_update(&mut self, outbound_lane_data: OutboundLaneData) -> Option<MessageNonce> {
+		let mut data = self.storage.data();
+		if outbound_lane_data.latest_received_nonce > data.latest_received_nonce {
+			// this is something that should never happen if proofs are correct
+			return None;
+		}
+		if outbound_lane_data.latest_received_nonce <= data.latest_confirmed_nonce {
+			return None;
+		}
+
+		data.latest_confirmed_nonce = outbound_lane_data.latest_received_nonce;
+		// Firstly, remove all of the records where higher nonce <= new confirmed nonce
+		while data
+			.relayers
+			.front()
+			.map(|(_, nonce_high, _)| *nonce_high <= data.latest_confirmed_nonce)
+			.unwrap_or(false)
+		{
+			data.relayers.pop_front();
+		}
+		// Secondly, update the next record with lower nonce equal to new confirmed nonce if needed.
+		// Note: There will be max. 1 record to update as we don't allow messages from relayers to overlap.
+		match data.relayers.front_mut() {
+			Some((nonce_low, _, _)) if *nonce_low < data.latest_confirmed_nonce => {
+				*nonce_low = data.latest_confirmed_nonce + 1;
+			}
+			_ => {}
+		}
+
+		self.storage.set_data(data);
+		Some(outbound_lane_data.latest_received_nonce)
+	}
+
 	/// Receive new message.
 	pub fn receive_message<P: MessageDispatch<S::MessageFee>>(
 		&mut self,
+		relayer: S::Relayer,
 		nonce: MessageNonce,
 		message_data: DispatchMessageData<P::DispatchPayload, S::MessageFee>,
 	) -> bool {
@@ -57,7 +97,24 @@ impl<S: InboundLaneStorage> InboundLane<S> {
 			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 {
+			return false;
+		}
+
 		data.latest_received_nonce = nonce;
+
+		let push_new = match data.relayers.back_mut() {
+			Some((_, nonce_high, last_relayer)) if last_relayer == &relayer => {
+				*nonce_high = nonce;
+				false
+			}
+			_ => true,
+		};
+		if push_new {
+			data.relayers.push_back((nonce, nonce, relayer));
+		}
+
 		self.storage.set_data(data);

 		P::dispatch(DispatchMessage {
@@ -77,23 +134,222 @@ mod tests {
 	use super::*;
 	use crate::{
 		inbound_lane,
-		mock::{message_data, run_test, TestMessageDispatch, TestRuntime, REGULAR_PAYLOAD, TEST_LANE_ID},
+		mock::{
+			message_data, run_test, TestMessageDispatch, TestRuntime, REGULAR_PAYLOAD, TEST_LANE_ID, TEST_RELAYER_A,
+			TEST_RELAYER_B, TEST_RELAYER_C,
+		},
+		DefaultInstance, RuntimeInboundLaneStorage,
 	};

+	fn receive_regular_message(
+		lane: &mut InboundLane<RuntimeInboundLaneStorage<TestRuntime, DefaultInstance>>,
+		nonce: MessageNonce,
+	) {
+		assert!(lane.receive_message::<TestMessageDispatch>(
+			TEST_RELAYER_A,
+			nonce,
+			message_data(REGULAR_PAYLOAD).into()
+		));
+	}
+
+	#[test]
+	fn receive_status_update_ignores_status_from_the_future() {
+		run_test(|| {
+			let mut lane = inbound_lane::<TestRuntime, _>(TEST_LANE_ID);
+			receive_regular_message(&mut lane, 1);
+			assert_eq!(
+				lane.receive_state_update(OutboundLaneData {
+					latest_received_nonce: 10,
+					..Default::default()
+				}),
+				None,
+			);
+
+			assert_eq!(lane.storage.data().latest_confirmed_nonce, 0);
+		});
+	}
+
+	#[test]
+	fn receive_status_update_ignores_obsolete_status() {
+		run_test(|| {
+			let mut lane = inbound_lane::<TestRuntime, _>(TEST_LANE_ID);
+			receive_regular_message(&mut lane, 1);
+			receive_regular_message(&mut lane, 2);
+			receive_regular_message(&mut lane, 3);
+			assert_eq!(
+				lane.receive_state_update(OutboundLaneData {
+					latest_received_nonce: 3,
+					..Default::default()
+				}),
+				Some(3),
+			);
+			assert_eq!(lane.storage.data().latest_confirmed_nonce, 3);
+
+			assert_eq!(
+				lane.receive_state_update(OutboundLaneData {
+					latest_received_nonce: 3,
+					..Default::default()
+				}),
+				None,
+			);
+			assert_eq!(lane.storage.data().latest_confirmed_nonce, 3);
+		});
+	}
+
+	#[test]
+	fn receive_status_update_works() {
+		run_test(|| {
+			let mut lane = inbound_lane::<TestRuntime, _>(TEST_LANE_ID);
+			receive_regular_message(&mut lane, 1);
+			receive_regular_message(&mut lane, 2);
+			receive_regular_message(&mut lane, 3);
+			assert_eq!(lane.storage.data().latest_confirmed_nonce, 0);
+			assert_eq!(lane.storage.data().relayers, vec![(1, 3, TEST_RELAYER_A)]);
+
+			assert_eq!(
+				lane.receive_state_update(OutboundLaneData {
+					latest_received_nonce: 2,
+					..Default::default()
+				}),
+				Some(2),
+			);
+			assert_eq!(lane.storage.data().latest_confirmed_nonce, 2);
+			assert_eq!(lane.storage.data().relayers, vec![(3, 3, TEST_RELAYER_A)]);
+
+			assert_eq!(
+				lane.receive_state_update(OutboundLaneData {
+					latest_received_nonce: 3,
+					..Default::default()
+				}),
+				Some(3),
+			);
+			assert_eq!(lane.storage.data().latest_confirmed_nonce, 3);
+			assert_eq!(lane.storage.data().relayers, vec![]);
+		});
+	}
+
+	#[test]
+	fn receive_status_update_works_with_batches_from_relayers() {
+		run_test(|| {
+			let mut lane = inbound_lane::<TestRuntime, _>(TEST_LANE_ID);
+			let mut seed_storage_data = lane.storage.data();
+			// Prepare data
+			seed_storage_data.latest_confirmed_nonce = 0;
+			seed_storage_data.latest_received_nonce = 5;
+			seed_storage_data.relayers.push_back((1, 1, TEST_RELAYER_A));
+			// Simulate messages batch (2, 3, 4) from relayer #2
+			seed_storage_data.relayers.push_back((2, 4, TEST_RELAYER_B));
+			seed_storage_data.relayers.push_back((5, 5, TEST_RELAYER_C));
+			lane.storage.set_data(seed_storage_data);
+			// Check
+			assert_eq!(
+				lane.receive_state_update(OutboundLaneData {
+					latest_received_nonce: 3,
+					..Default::default()
+				}),
+				Some(3),
+			);
+			assert_eq!(lane.storage.data().latest_confirmed_nonce, 3);
+			assert_eq!(
+				lane.storage.data().relayers,
+				vec![(4, 4, TEST_RELAYER_B), (5, 5, TEST_RELAYER_C)]
+			);
+		});
+	}
+
 	#[test]
 	fn fails_to_receive_message_with_incorrect_nonce() {
 		run_test(|| {
 			let mut lane = inbound_lane::<TestRuntime, _>(TEST_LANE_ID);
-			assert!(!lane.receive_message::<TestMessageDispatch>(10, message_data(REGULAR_PAYLOAD).into()));
+			assert!(!lane.receive_message::<TestMessageDispatch>(
+				TEST_RELAYER_A,
+				10,
+				message_data(REGULAR_PAYLOAD).into()
+			));
 			assert_eq!(lane.storage.data().latest_received_nonce, 0);
 		});
 	}

+	#[test]
+	fn fails_to_receive_messages_above_max_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 + 1 {
+				assert!(lane.receive_message::<TestMessageDispatch>(
+					TEST_RELAYER_A + current_nonce,
+					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_A + max_nonce + 1,
+					max_nonce + 1,
+					message_data(REGULAR_PAYLOAD).into()
+				)
+			);
+			// Fails to dispatch new messages from latest relayer. Prevents griefing attacks.
+			assert_eq!(
+				false,
+				lane.receive_message::<TestMessageDispatch>(
+					TEST_RELAYER_A + max_nonce,
+					max_nonce + 1,
+					message_data(REGULAR_PAYLOAD).into()
+				)
+			);
+		});
+	}
+
+	#[test]
+	fn correctly_receives_following_messages_from_two_relayers_alternately() {
+		run_test(|| {
+			let mut lane = inbound_lane::<TestRuntime, _>(TEST_LANE_ID);
+			assert!(lane.receive_message::<TestMessageDispatch>(
+				TEST_RELAYER_A,
+				1,
+				message_data(REGULAR_PAYLOAD).into()
+			));
+			assert!(lane.receive_message::<TestMessageDispatch>(
+				TEST_RELAYER_B,
+				2,
+				message_data(REGULAR_PAYLOAD).into()
+			));
+			assert!(lane.receive_message::<TestMessageDispatch>(
+				TEST_RELAYER_A,
+				3,
+				message_data(REGULAR_PAYLOAD).into()
+			));
+			assert_eq!(
+				lane.storage.data().relayers,
+				vec![(1, 1, TEST_RELAYER_A), (2, 2, TEST_RELAYER_B), (3, 3, TEST_RELAYER_A)]
+			);
+		});
+	}
+
+	#[test]
+	fn rejects_same_message_from_two_different_relayers() {
+		run_test(|| {
+			let mut lane = inbound_lane::<TestRuntime, _>(TEST_LANE_ID);
+			assert!(lane.receive_message::<TestMessageDispatch>(
+				TEST_RELAYER_A,
+				1,
+				message_data(REGULAR_PAYLOAD).into()
+			));
+			assert_eq!(
+				false,
+				lane.receive_message::<TestMessageDispatch>(TEST_RELAYER_B, 1, message_data(REGULAR_PAYLOAD).into())
+			);
+		});
+	}
+
 	#[test]
 	fn correct_message_is_processed_instantly() {
 		run_test(|| {
 			let mut lane = inbound_lane::<TestRuntime, _>(TEST_LANE_ID);
-			assert!(lane.receive_message::<TestMessageDispatch>(1, message_data(REGULAR_PAYLOAD).into()));
+			receive_regular_message(&mut lane, 1);
 			assert_eq!(lane.storage.data().latest_received_nonce, 1);
 		});
 	}
@@ -34,7 +34,7 @@ use crate::outbound_lane::{OutboundLane, OutboundLaneStorage};

 use bp_message_lane::{
 	source_chain::{LaneMessageVerifier, MessageDeliveryAndDispatchPayment, TargetHeaderChain},
-	target_chain::{MessageDispatch, SourceHeaderChain},
+	target_chain::{DispatchMessage, MessageDispatch, ProvedLaneMessages, ProvedMessages, SourceHeaderChain},
 	InboundLaneData, LaneId, MessageData, MessageKey, MessageNonce, OutboundLaneData,
 };
 use codec::{Decode, Encode};
@@ -43,7 +43,7 @@ use frame_support::{
 	Parameter, StorageMap,
 };
 use frame_system::ensure_signed;
-use sp_std::{marker::PhantomData, prelude::*};
+use sp_std::{cell::RefCell, marker::PhantomData, prelude::*};

 mod inbound_lane;
 mod outbound_lane;
@@ -66,6 +66,11 @@ 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 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.
+	type MaxUnconfirmedMessagesAtInboundLane: Get<MessageNonce>;

 	/// Payload type of outbound messages. This payload is dispatched on the bridged chain.
 	type OutboundPayload: Parameter;
@@ -76,11 +81,13 @@ pub trait Trait<I = DefaultInstance>: frame_system::Trait {
 	type InboundPayload: Decode;
 	/// Message fee type of inbound messages. This fee is paid on the bridged chain.
 	type InboundMessageFee: Decode;
+	/// Identifier of relayer that deliver messages to this chain. Relayer reward is paid on the bridged chain.
+	type InboundRelayer: Parameter;

 	// Types that are used by outbound_lane (on source chain).

 	/// Target header chain.
-	type TargetHeaderChain: TargetHeaderChain<Self::OutboundPayload>;
+	type TargetHeaderChain: TargetHeaderChain<Self::OutboundPayload, Self::AccountId>;
 	/// Message payload verifier.
 	type LaneMessageVerifier: LaneMessageVerifier<Self::AccountId, Self::OutboundPayload, Self::OutboundMessageFee>;
 	/// Message delivery payment.
@@ -98,8 +105,10 @@ pub trait Trait<I = DefaultInstance>: frame_system::Trait {
 type MessagesProofOf<T, I> =
 	<<T as Trait<I>>::SourceHeaderChain as SourceHeaderChain<<T as Trait<I>>::InboundMessageFee>>::MessagesProof;
 /// Shortcut to messages delivery proof type for Trait.
-type MessagesDeliveryProofOf<T, I> =
-	<<T as Trait<I>>::TargetHeaderChain as TargetHeaderChain<<T as Trait<I>>::OutboundPayload>>::MessagesDeliveryProof;
+type MessagesDeliveryProofOf<T, I> = <<T as Trait<I>>::TargetHeaderChain as TargetHeaderChain<
+	<T as Trait<I>>::OutboundPayload,
+	<T as frame_system::Trait>::AccountId,
+>>::MessagesDeliveryProof;

 decl_error! {
 	pub enum Error for Module<T: Trait<I>, I: Instance> {
@@ -121,7 +130,7 @@ decl_error! {
 decl_storage! {
 	trait Store for Module<T: Trait<I>, I: Instance = DefaultInstance> as MessageLane {
 		/// Map of lane id => inbound lane data.
-		InboundLanes: map hasher(blake2_128_concat) LaneId => InboundLaneData;
+		InboundLanes: map hasher(blake2_128_concat) LaneId => InboundLaneData<T::InboundRelayer>;
 		/// Map of lane id => outbound lane data.
 		OutboundLanes: map hasher(blake2_128_concat) LaneId => OutboundLaneData;
 		/// All queued outbound messages.
@@ -227,29 +236,33 @@ decl_module! {
 		#[weight = DELIVERY_BASE_WEIGHT + dispatch_weight]
 		pub fn receive_messages_proof(
 			origin,
+			relayer_id: T::InboundRelayer,
 			proof: MessagesProofOf<T, I>,
 			dispatch_weight: Weight,
 		) -> DispatchResult {
 			let _ = ensure_signed(origin)?;

 			// verify messages proof && convert proof into messages
-			let messages = T::SourceHeaderChain::verify_messages_proof(proof).map_err(|err| {
-				frame_support::debug::trace!(
-					target: "runtime",
-					"Rejecting invalid messages proof: {:?}",
-					err,
-				);
+			let messages = verify_and_decode_messages_proof::<T::SourceHeaderChain, T::InboundMessageFee, T::InboundPayload>(proof)
+				.map_err(|err| {
+					frame_support::debug::trace!(
+						target: "runtime",
+						"Rejecting invalid messages proof: {:?}",
+						err,
+					);

-				Error::<T, I>::InvalidMessagesProof
-			})?;
-
-			// try to decode message payloads
-			let messages: Vec<_> = messages.into_iter().map(Into::into).collect();
+					Error::<T, I>::InvalidMessagesProof
+				})?;

 			// verify that relayer is paying actual dispatch weight
 			let actual_dispatch_weight: Weight = messages
-				.iter()
-				.map(T::MessageDispatch::dispatch_weight)
+				.values()
+				.map(|lane_messages| lane_messages
+					.messages
+					.iter()
+					.map(T::MessageDispatch::dispatch_weight)
+					.sum::<Weight>()
+				)
 				.sum();
 			if dispatch_weight < actual_dispatch_weight {
 				frame_support::debug::trace!(
@@ -262,13 +275,31 @@ decl_module! {
 				return Err(Error::<T, I>::InvalidMessagesDispatchWeight.into());
 			}

-			// dispatch messages
-			let total_messages = messages.len();
+			// dispatch messages and (optionally) update lane(s) state(s)
+			let mut total_messages = 0;
 			let mut valid_messages = 0;
-			for message in messages {
-				let mut lane = inbound_lane::<T, I>(message.key.lane_id);
-				if lane.receive_message::<T::MessageDispatch>(message.key.nonce, message.data) {
-					valid_messages += 1;
+			for (lane_id, lane_data) in messages {
+				let mut lane = inbound_lane::<T, I>(lane_id);
+
+				if let Some(lane_state) = lane_data.lane_state {
+					let updated_latest_confirmed_nonce = lane.receive_state_update(lane_state);
+					if let Some(updated_latest_confirmed_nonce) = updated_latest_confirmed_nonce {
+						frame_support::debug::trace!(
+							target: "runtime",
+							"Received lane {:?} state update: latest_confirmed_nonce={}",
+							lane_id,
+							updated_latest_confirmed_nonce,
+						);
+					}
+				}
+
+				for message in lane_data.messages {
+					debug_assert_eq!(message.key.lane_id, lane_id);
+
+					total_messages += 1;
+					if lane.receive_message::<T::MessageDispatch>(relayer_id.clone(), message.key.nonce, message.data) {
+						valid_messages += 1;
+					}
 				}
 			}

@@ -285,8 +316,8 @@ decl_module! {
 		/// Receive messages delivery proof from bridged chain.
 		#[weight = 0] // TODO: update me (https://github.com/paritytech/parity-bridges-common/issues/78)
 		pub fn receive_messages_delivery_proof(origin, proof: MessagesDeliveryProofOf<T, I>) -> DispatchResult {
-			let _ = ensure_signed(origin)?;
-			let (lane_id, nonce) = T::TargetHeaderChain::verify_messages_delivery_proof(proof).map_err(|err| {
+			let confirmation_relayer = ensure_signed(origin)?;
+			let (lane_id, lane_data) = T::TargetHeaderChain::verify_messages_delivery_proof(proof).map_err(|err| {
 				frame_support::debug::trace!(
 					target: "runtime",
 					"Rejecting invalid messages delivery proof: {:?}",
@@ -296,16 +327,37 @@ decl_module! {
 				Error::<T, I>::InvalidMessagesDeliveryProof
 			})?;

+			// mark messages as delivered
 			let mut lane = outbound_lane::<T, I>(lane_id);
-			let received_range = lane.confirm_delivery(nonce);
+			let received_range = lane.confirm_delivery(lane_data.latest_received_nonce);
 			if let Some(received_range) = received_range {
 				Self::deposit_event(RawEvent::MessagesDelivered(lane_id, received_range.0, received_range.1));
+
+				// reward relayers that have delivered messages
+				// this loop is bounded by `T::MaxUnconfirmedMessagesAtInboundLane` on the bridged chain
+				for (nonce_low, nonce_high, relayer) in lane_data.relayers {
+					let nonce_begin = sp_std::cmp::max(nonce_low, received_range.0);
+					let nonce_end = sp_std::cmp::min(nonce_high, received_range.1);
+					// loop won't proceed if current entry is ahead of received range (begin > end).
+					for nonce in nonce_begin..nonce_end + 1 {
+						let message_data = OutboundMessages::<T, I>::get(MessageKey {
+							lane_id,
+							nonce,
+						}).expect("message was just confirmed; we never prune unconfirmed messages; qed");
+
+						<T as Trait<I>>::MessageDeliveryAndDispatchPayment::pay_relayer_reward(
+							&confirmation_relayer,
+							&relayer,
+							&message_data.fee,
+						);
+					}
+				}
 			}

 			frame_support::debug::trace!(
 				target: "runtime",
 				"Received messages delivery proof up to (and including) {} at lane {:?}",
-				nonce,
+				lane_data.latest_received_nonce,
 				lane_id,
 			);

@@ -318,6 +370,7 @@ decl_module! {
 fn inbound_lane<T: Trait<I>, I: Instance>(lane_id: LaneId) -> InboundLane<RuntimeInboundLaneStorage<T, I>> {
 	InboundLane::new(RuntimeInboundLaneStorage {
 		lane_id,
+		cached_data: RefCell::new(None),
 		_phantom: Default::default(),
 	})
 }
@@ -331,24 +384,44 @@ fn outbound_lane<T: Trait<I>, I: Instance>(lane_id: LaneId) -> OutboundLane<Runt
 }

 /// Runtime inbound lane storage.
-struct RuntimeInboundLaneStorage<T, I = DefaultInstance> {
+struct RuntimeInboundLaneStorage<T: Trait<I>, I = DefaultInstance> {
 	lane_id: LaneId,
-	_phantom: PhantomData<(T, I)>,
+	cached_data: RefCell<Option<InboundLaneData<T::InboundRelayer>>>,
+	_phantom: PhantomData<I>,
 }

 impl<T: Trait<I>, I: Instance> InboundLaneStorage for RuntimeInboundLaneStorage<T, I> {
 	type MessageFee = T::InboundMessageFee;
+	type Relayer = T::InboundRelayer;

 	fn id(&self) -> LaneId {
 		self.lane_id
 	}

-	fn data(&self) -> InboundLaneData {
-		InboundLanes::<I>::get(&self.lane_id)
+	fn max_unconfirmed_messages(&self) -> MessageNonce {
+		T::MaxUnconfirmedMessagesAtInboundLane::get()
 	}

-	fn set_data(&mut self, data: InboundLaneData) {
-		InboundLanes::<I>::insert(&self.lane_id, data)
+	fn data(&self) -> InboundLaneData<T::InboundRelayer> {
+		match self.cached_data.clone().into_inner() {
+			Some(data) => data,
+			None => {
+				let data = InboundLanes::<T, I>::get(&self.lane_id);
+				*self.cached_data.try_borrow_mut().expect(
+					"we're in the single-threaded environment;\
+						we have no recursive borrows; qed",
+				) = Some(data.clone());
+				data
+			}
+		}
+	}
+
+	fn set_data(&mut self, data: InboundLaneData<T::InboundRelayer>) {
+		*self.cached_data.try_borrow_mut().expect(
+			"we're in the single-threaded environment;\
+				we have no recursive borrows; qed",
+		) = Some(data.clone());
+		InboundLanes::<T, I>::insert(&self.lane_id, data)
 	}
 }

@@ -399,12 +472,32 @@ impl<T: Trait<I>, I: Instance> OutboundLaneStorage for RuntimeOutboundLaneStorag
 	}
 }

+/// Verify messages proof and return proved messages with decoded payload.
+fn verify_and_decode_messages_proof<Chain: SourceHeaderChain<Fee>, Fee, DispatchPayload: Decode>(
+	proof: Chain::MessagesProof,
+) -> Result<ProvedMessages<DispatchMessage<DispatchPayload, Fee>>, Chain::Error> {
+	Chain::verify_messages_proof(proof).map(|messages_by_lane| {
+		messages_by_lane
+			.into_iter()
+			.map(|(lane, lane_data)| {
+				(
+					lane,
+					ProvedLaneMessages {
+						lane_state: lane_data.lane_state,
+						messages: lane_data.messages.into_iter().map(Into::into).collect(),
+					},
+				)
+			})
+			.collect()
+	})
+}
+
 #[cfg(test)]
 mod tests {
 	use super::*;
 	use crate::mock::{
-		message, run_test, Origin, TestEvent, TestMessageDeliveryAndDispatchPayment, TestRuntime,
-		PAYLOAD_REJECTED_BY_TARGET_CHAIN, REGULAR_PAYLOAD, TEST_LANE_ID,
+		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 frame_support::{assert_noop, assert_ok};
 	use frame_system::{EventRecord, Module as System, Phase};
@@ -440,7 +533,13 @@ mod tests {

 		assert_ok!(Module::<TestRuntime>::receive_messages_delivery_proof(
 			Origin::signed(1),
-			Ok((TEST_LANE_ID, 1)),
+			Ok((
+				TEST_LANE_ID,
+				InboundLaneData {
+					latest_received_nonce: 1,
+					..Default::default()
+				}
+			)),
 		));

 		assert_eq!(
@@ -508,13 +607,53 @@ mod tests {
 		run_test(|| {
 			assert_ok!(Module::<TestRuntime>::receive_messages_proof(
 				Origin::signed(1),
-				Ok(vec![message(1, REGULAR_PAYLOAD)]),
+				TEST_RELAYER_A,
+				Ok(vec![message(1, REGULAR_PAYLOAD)]).into(),
+				REGULAR_PAYLOAD.1,
+			));
+
+			assert_eq!(InboundLanes::<TestRuntime>::get(TEST_LANE_ID).latest_received_nonce, 1);
+		});
+	}
+
+	#[test]
+	fn receive_messages_proof_updates_confirmed_message_nonce() {
+		run_test(|| {
+			// say we have received 10 messages && last confirmed message is 8
+			InboundLanes::<TestRuntime, DefaultInstance>::insert(
+				TEST_LANE_ID,
+				InboundLaneData {
+					latest_confirmed_nonce: 8,
+					latest_received_nonce: 10,
+					relayers: vec![(9, 9, TEST_RELAYER_A), (10, 10, TEST_RELAYER_B)]
+						.into_iter()
+						.collect(),
+				},
+			);
+
+			// 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();
+			message_proof.result.as_mut().unwrap()[0].1.lane_state = Some(OutboundLaneData {
+				latest_received_nonce: 9,
+				..Default::default()
+			});
+
+			assert_ok!(Module::<TestRuntime>::receive_messages_proof(
+				Origin::signed(1),
+				TEST_RELAYER_A,
+				message_proof,
 				REGULAR_PAYLOAD.1,
 			));

 			assert_eq!(
-				InboundLanes::<DefaultInstance>::get(TEST_LANE_ID).latest_received_nonce,
-				1
+				InboundLanes::<TestRuntime>::get(TEST_LANE_ID),
+				InboundLaneData {
+					relayers: vec![(10, 10, TEST_RELAYER_B), (11, 11, TEST_RELAYER_A)]
+						.into_iter()
+						.collect(),
+					latest_received_nonce: 11,
+					latest_confirmed_nonce: 9,
+				},
 			);
 		});
 	}
@@ -525,7 +664,8 @@ mod tests {
 			assert_noop!(
 				Module::<TestRuntime>::receive_messages_proof(
 					Origin::signed(1),
-					Ok(vec![message(1, REGULAR_PAYLOAD)]),
+					TEST_RELAYER_A,
+					Ok(vec![message(1, REGULAR_PAYLOAD)]).into(),
 					REGULAR_PAYLOAD.1 - 1,
 				),
 				Error::<TestRuntime, DefaultInstance>::InvalidMessagesDispatchWeight,
@@ -537,7 +677,12 @@ mod tests {
 	fn receive_messages_proof_rejects_invalid_proof() {
 		run_test(|| {
 			assert_noop!(
-				Module::<TestRuntime, DefaultInstance>::receive_messages_proof(Origin::signed(1), Err(()), 0),
+				Module::<TestRuntime, DefaultInstance>::receive_messages_proof(
+					Origin::signed(1),
+					TEST_RELAYER_A,
+					Err(()).into(),
+					0,
+				),
 				Error::<TestRuntime, DefaultInstance>::InvalidMessagesProof,
 			);
 		});
@@ -556,6 +701,68 @@ mod tests {
 		});
 	}

+	#[test]
+	fn receive_messages_delivery_proof_rewards_relayers() {
+		run_test(|| {
+			assert_ok!(Module::<TestRuntime>::send_message(
+				Origin::signed(1),
+				TEST_LANE_ID,
+				REGULAR_PAYLOAD,
+				1000,
+			));
+			assert_ok!(Module::<TestRuntime>::send_message(
+				Origin::signed(1),
+				TEST_LANE_ID,
+				REGULAR_PAYLOAD,
+				2000,
+			));
+
+			// this reports delivery of message 1 => reward is paid to TEST_RELAYER_A
+			assert_ok!(Module::<TestRuntime>::receive_messages_delivery_proof(
+				Origin::signed(1),
+				Ok((
+					TEST_LANE_ID,
+					InboundLaneData {
+						relayers: vec![(1, 1, TEST_RELAYER_A)].into_iter().collect(),
+						latest_received_nonce: 1,
+						..Default::default()
+					}
+				)),
+			));
+			assert!(TestMessageDeliveryAndDispatchPayment::is_reward_paid(
+				TEST_RELAYER_A,
+				1000
+			));
+			assert!(!TestMessageDeliveryAndDispatchPayment::is_reward_paid(
+				TEST_RELAYER_B,
+				2000
+			));
+
+			// this reports delivery of both message 1 and message 2 => reward is paid only to TEST_RELAYER_B
+			assert_ok!(Module::<TestRuntime>::receive_messages_delivery_proof(
+				Origin::signed(1),
+				Ok((
+					TEST_LANE_ID,
+					InboundLaneData {
+						relayers: vec![(1, 1, TEST_RELAYER_A), (2, 2, TEST_RELAYER_B)]
+							.into_iter()
+							.collect(),
+						latest_received_nonce: 2,
+						..Default::default()
+					}
+				)),
+			));
+			assert!(!TestMessageDeliveryAndDispatchPayment::is_reward_paid(
+				TEST_RELAYER_A,
+				1000
+			));
+			assert!(TestMessageDeliveryAndDispatchPayment::is_reward_paid(
+				TEST_RELAYER_B,
+				2000
+			));
+		});
+	}
+
 	#[test]
 	fn receive_messages_delivery_proof_rejects_invalid_proof() {
 		run_test(|| {
@@ -574,14 +781,12 @@ mod tests {

 			assert_ok!(Module::<TestRuntime, DefaultInstance>::receive_messages_proof(
 				Origin::signed(1),
-				Ok(vec![invalid_message]),
+				TEST_RELAYER_A,
+				Ok(vec![invalid_message]).into(),
 				0, // weight may be zero in this case (all messages are improperly encoded)
 			),);

-			assert_eq!(
-				InboundLanes::<DefaultInstance>::get(&TEST_LANE_ID).latest_received_nonce,
-				1,
-			);
+			assert_eq!(InboundLanes::<TestRuntime>::get(&TEST_LANE_ID).latest_received_nonce, 1,);
 		});
 	}

@@ -593,18 +798,17 @@ mod tests {

 			assert_ok!(Module::<TestRuntime, DefaultInstance>::receive_messages_proof(
 				Origin::signed(1),
+				TEST_RELAYER_A,
 				Ok(vec![
 					message(1, REGULAR_PAYLOAD),
 					invalid_message,
 					message(3, REGULAR_PAYLOAD),
-				]),
+				])
+				.into(),
 				REGULAR_PAYLOAD.1 + REGULAR_PAYLOAD.1,
 			),);

-			assert_eq!(
-				InboundLanes::<DefaultInstance>::get(&TEST_LANE_ID).latest_received_nonce,
-				3,
-			);
+			assert_eq!(InboundLanes::<TestRuntime>::get(&TEST_LANE_ID).latest_received_nonce, 3,);
 		});
 	}
 }
@@ -18,10 +18,10 @@ use crate::Trait;

 use bp_message_lane::{
 	source_chain::{LaneMessageVerifier, MessageDeliveryAndDispatchPayment, TargetHeaderChain},
-	target_chain::{DispatchMessage, MessageDispatch, SourceHeaderChain},
-	LaneId, Message, MessageData, MessageKey, MessageNonce,
+	target_chain::{DispatchMessage, MessageDispatch, ProvedLaneMessages, ProvedMessages, SourceHeaderChain},
+	InboundLaneData, LaneId, Message, MessageData, MessageKey, MessageNonce,
 };
-use codec::Encode;
+use codec::{Decode, Encode};
 use frame_support::{impl_outer_event, impl_outer_origin, parameter_types, weights::Weight};
 use sp_core::H256;
 use sp_runtime::{
@@ -29,10 +29,12 @@ use sp_runtime::{
 	traits::{BlakeTwo256, IdentityLookup},
 	Perbill,
 };
+use std::collections::BTreeMap;

 pub type AccountId = u64;
 pub type TestPayload = (u64, Weight);
 pub type TestMessageFee = u64;
+pub type TestRelayer = u64;

 #[derive(Clone, Eq, PartialEq, Debug)]
 pub struct TestRuntime;
@@ -89,17 +91,20 @@ impl frame_system::Trait for TestRuntime {

 parameter_types! {
 	pub const MaxMessagesToPruneAtOnce: u64 = 10;
+	pub const MaxUnconfirmedMessagesAtInboundLane: u64 = 16;
 }

 impl Trait for TestRuntime {
 	type Event = TestEvent;
 	type MaxMessagesToPruneAtOnce = MaxMessagesToPruneAtOnce;
+	type MaxUnconfirmedMessagesAtInboundLane = MaxUnconfirmedMessagesAtInboundLane;

 	type OutboundPayload = TestPayload;
 	type OutboundMessageFee = TestMessageFee;

 	type InboundPayload = TestPayload;
 	type InboundMessageFee = TestMessageFee;
+	type InboundRelayer = TestRelayer;

 	type TargetHeaderChain = TestTargetHeaderChain;
 	type LaneMessageVerifier = TestLaneMessageVerifier;
@@ -109,6 +114,15 @@ impl Trait for TestRuntime {
 	type MessageDispatch = TestMessageDispatch;
 }

+/// Account id of test relayer.
+pub const TEST_RELAYER_A: AccountId = 100;
+
+/// Account id of additional test relayer - B.
+pub const TEST_RELAYER_B: AccountId = 101;
+
+/// Account id of additional test relayer - C.
+pub const TEST_RELAYER_C: AccountId = 102;
+
 /// Error that is returned by all test implementations.
 pub const TEST_ERROR: &str = "Test error";

@@ -121,14 +135,42 @@ pub const REGULAR_PAYLOAD: TestPayload = (0, 50);
 /// Payload that is rejected by `TestTargetHeaderChain`.
 pub const PAYLOAD_REJECTED_BY_TARGET_CHAIN: TestPayload = (1, 50);

+/// Vec of proved messages, grouped by lane.
+pub type MessagesByLaneVec = Vec<(LaneId, ProvedLaneMessages<Message<TestMessageFee>>)>;
+
+/// Test messages proof.
+#[derive(Debug, Encode, Decode, Clone, PartialEq, Eq)]
+pub struct TestMessagesProof {
+	pub result: Result<MessagesByLaneVec, ()>,
+}
+
+impl From<Result<Vec<Message<TestMessageFee>>, ()>> for TestMessagesProof {
+	fn from(result: Result<Vec<Message<TestMessageFee>>, ()>) -> Self {
+		Self {
+			result: result.map(|messages| {
+				let mut messages_by_lane: BTreeMap<LaneId, ProvedLaneMessages<Message<TestMessageFee>>> =
+					BTreeMap::new();
+				for message in messages {
+					messages_by_lane
+						.entry(message.key.lane_id)
+						.or_default()
+						.messages
+						.push(message);
+				}
+				messages_by_lane.into_iter().collect()
+			}),
+		}
+	}
+}
+
 /// Target header chain that is used in tests.
 #[derive(Debug, Default)]
 pub struct TestTargetHeaderChain;

-impl TargetHeaderChain<TestPayload> for TestTargetHeaderChain {
+impl TargetHeaderChain<TestPayload, TestRelayer> for TestTargetHeaderChain {
 	type Error = &'static str;

-	type MessagesDeliveryProof = Result<(LaneId, MessageNonce), ()>;
+	type MessagesDeliveryProof = Result<(LaneId, InboundLaneData<TestRelayer>), ()>;

 	fn verify_message(payload: &TestPayload) -> Result<(), Self::Error> {
 		if *payload == PAYLOAD_REJECTED_BY_TARGET_CHAIN {
@@ -140,7 +182,7 @@ impl TargetHeaderChain<TestPayload> for TestTargetHeaderChain {

 	fn verify_messages_delivery_proof(
 		proof: Self::MessagesDeliveryProof,
-	) -> Result<(LaneId, MessageNonce), Self::Error> {
+	) -> Result<(LaneId, InboundLaneData<TestRelayer>), Self::Error> {
 		proof.map_err(|_| TEST_ERROR)
 	}
 }
@@ -176,10 +218,17 @@ impl TestMessageDeliveryAndDispatchPayment {
 		frame_support::storage::unhashed::put(b":reject-message-fee:", &true);
 	}

-	/// Returns true if given fee has been paid by given relayer.
+	/// Returns true if given fee has been paid by given submitter.
 	pub fn is_fee_paid(submitter: AccountId, fee: TestMessageFee) -> bool {
 		frame_support::storage::unhashed::get(b":message-fee:") == Some((submitter, fee))
 	}
+
+	/// Returns true if given relayer has been rewarded with given balance. The reward-paid flag is
+	/// cleared after the call.
+	pub fn is_reward_paid(relayer: AccountId, fee: TestMessageFee) -> bool {
+		let key = (b":relayer-reward:", relayer, fee).encode();
+		frame_support::storage::unhashed::take::<bool>(&key).is_some()
+	}
 }

 impl MessageDeliveryAndDispatchPayment<AccountId, TestMessageFee> for TestMessageDeliveryAndDispatchPayment {
@@ -193,6 +242,11 @@ impl MessageDeliveryAndDispatchPayment<AccountId, TestMessageFee> for TestMessag
 		frame_support::storage::unhashed::put(b":message-fee:", &(submitter, fee));
 		Ok(())
 	}
+
+	fn pay_relayer_reward(_confirmation_relayer: &AccountId, relayer: &AccountId, fee: &TestMessageFee) {
+		let key = (b":relayer-reward:", relayer, fee).encode();
+		frame_support::storage::unhashed::put(&key, &true);
+	}
 }

 /// Source header chain that is used in tests.
@@ -202,10 +256,15 @@ pub struct TestSourceHeaderChain;
 impl SourceHeaderChain<TestMessageFee> for TestSourceHeaderChain {
 	type Error = &'static str;

-	type MessagesProof = Result<Vec<Message<TestMessageFee>>, ()>;
+	type MessagesProof = TestMessagesProof;

-	fn verify_messages_proof(proof: Self::MessagesProof) -> Result<Vec<Message<TestMessageFee>>, Self::Error> {
-		proof.map_err(|_| TEST_ERROR)
+	fn verify_messages_proof(
+		proof: Self::MessagesProof,
+	) -> Result<ProvedMessages<Message<TestMessageFee>>, Self::Error> {
+		proof
+			.result
+			.map(|proof| proof.into_iter().collect())
+			.map_err(|_| TEST_ERROR)
 	}
 }