Merge commit 'e5bed7ac380b6adb54b60a2a72a2a8f07f50d6c1' as 'bridges'

polkadot/bridges/relays/messages/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "messages-relay"
+version = "0.1.0"
+authors = ["Parity Technologies <admin@parity.io>"]
+edition = "2018"
+license = "GPL-3.0-or-later WITH Classpath-exception-2.0"
+
+[dependencies]
+async-std = "1.6.5"
+async-trait = "0.1.40"
+futures = "0.3.5"
+hex = "0.4"
+log = "0.4.11"
+parking_lot = "0.11.0"
+
+# Bridge Dependencies
+
+bp-messages = { path = "../../primitives/messages" }
+relay-utils = { path = "../utils" }

polkadot/bridges/relays/messages/src/lib.rs

@@ -0,0 +1,36 @@
+// Copyright 2019-2021 Parity Technologies (UK) Ltd.
+// This file is part of Parity Bridges Common.
+
+// Parity Bridges Common is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity Bridges Common is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Relaying [`pallet-bridge-messages`](../pallet_bridge_messages/index.html) application specific
+//! data. Message lane allows sending arbitrary messages between bridged chains. This
+//! module provides entrypoint that starts reading messages from given message lane
+//! of source chain and submits proof-of-message-at-source-chain transactions to the
+//! target chain. Additionaly, proofs-of-messages-delivery are sent back from the
+//! target chain to the source chain.
+
+// required for futures::select!
+#![recursion_limit = "1024"]
+#![warn(missing_docs)]
+
+mod metrics;
+
+pub mod message_lane;
+pub mod message_lane_loop;
+
+mod message_race_delivery;
+mod message_race_loop;
+mod message_race_receiving;
+mod message_race_strategy;

polkadot/bridges/relays/messages/src/message_lane.rs

@@ -0,0 +1,52 @@
+// Copyright 2019-2021 Parity Technologies (UK) Ltd.
+// This file is part of Parity Bridges Common.
+
+// Parity Bridges Common is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity Bridges Common is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
+
+//! One-way message lane types. Within single one-way lane we have three 'races' where we try to:
+//!
+//! 1) relay new messages from source to target node;
+//! 2) relay proof-of-delivery from target to source node.
+
+use relay_utils::{BlockNumberBase, HeaderId};
+use std::fmt::Debug;
+
+/// One-way message lane.
+pub trait MessageLane: Clone + Send + Sync {
+	/// Name of the messages source.
+	const SOURCE_NAME: &'static str;
+	/// Name of the messages target.
+	const TARGET_NAME: &'static str;
+
+	/// Messages proof.
+	type MessagesProof: Clone + Debug + Send + Sync;
+	/// Messages receiving proof.
+	type MessagesReceivingProof: Clone + Debug + Send + Sync;
+
+	/// Number of the source header.
+	type SourceHeaderNumber: BlockNumberBase;
+	/// Hash of the source header.
+	type SourceHeaderHash: Clone + Debug + Default + PartialEq + Send + Sync;
+
+	/// Number of the target header.
+	type TargetHeaderNumber: BlockNumberBase;
+	/// Hash of the target header.
+	type TargetHeaderHash: Clone + Debug + Default + PartialEq + Send + Sync;
+}
+
+/// Source header id within given one-way message lane.
+pub type SourceHeaderIdOf<P> = HeaderId<<P as MessageLane>::SourceHeaderHash, <P as MessageLane>::SourceHeaderNumber>;
+
+/// Target header id within given one-way message lane.
+pub type TargetHeaderIdOf<P> = HeaderId<<P as MessageLane>::TargetHeaderHash, <P as MessageLane>::TargetHeaderNumber>;

polkadot/bridges/relays/messages/src/message_lane_loop.rs

@@ -0,0 +1,865 @@
+// Copyright 2019-2021 Parity Technologies (UK) Ltd.
+// This file is part of Parity Bridges Common.
+
+// Parity Bridges Common is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity Bridges Common is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Message delivery loop. Designed to work with messages pallet.
+//!
+//! Single relay instance delivers messages of single lane in single direction.
+//! To serve two-way lane, you would need two instances of relay.
+//! To serve N two-way lanes, you would need N*2 instances of relay.
+//!
+//! Please keep in mind that the best header in this file is actually best
+//! finalized header. I.e. when talking about headers in lane context, we
+//! only care about finalized headers.
+
+use crate::message_lane::{MessageLane, SourceHeaderIdOf, TargetHeaderIdOf};
+use crate::message_race_delivery::run as run_message_delivery_race;
+use crate::message_race_receiving::run as run_message_receiving_race;
+use crate::metrics::MessageLaneLoopMetrics;
+
+use async_trait::async_trait;
+use bp_messages::{LaneId, MessageNonce, UnrewardedRelayersState, Weight};
+use futures::{channel::mpsc::unbounded, future::FutureExt, stream::StreamExt};
+use relay_utils::{
+	interval,
+	metrics::{GlobalMetrics, MetricsParams},
+	process_future_result,
+	relay_loop::Client as RelayClient,
+	retry_backoff, FailedClient,
+};
+use std::{collections::BTreeMap, fmt::Debug, future::Future, ops::RangeInclusive, time::Duration};
+
+/// Message lane loop configuration params.
+#[derive(Debug, Clone)]
+pub struct Params {
+	/// Id of lane this loop is servicing.
+	pub lane: LaneId,
+	/// Interval at which we ask target node about its updates.
+	pub source_tick: Duration,
+	/// Interval at which we ask target node about its updates.
+	pub target_tick: Duration,
+	/// Delay between moments when connection error happens and our reconnect attempt.
+	pub reconnect_delay: Duration,
+	/// The loop will auto-restart if there has been no updates during this period.
+	pub stall_timeout: Duration,
+	/// Message delivery race parameters.
+	pub delivery_params: MessageDeliveryParams,
+}
+
+/// 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.
+	pub max_unconfirmed_nonces_at_target: MessageNonce,
+	/// Maximal number of relayed messages in single delivery transaction.
+	pub max_messages_in_single_batch: MessageNonce,
+	/// Maximal cumulative dispatch weight of relayed messages in single delivery transaction.
+	pub max_messages_weight_in_single_batch: Weight,
+	/// Maximal cumulative size of relayed messages in single delivery transaction.
+	pub max_messages_size_in_single_batch: usize,
+}
+
+/// Message weights.
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub struct MessageWeights {
+	/// Message dispatch weight.
+	pub weight: Weight,
+	/// Message size (number of bytes in encoded payload).
+	pub size: usize,
+}
+
+/// Messages weights map.
+pub type MessageWeightsMap = BTreeMap<MessageNonce, MessageWeights>;
+
+/// Message delivery race proof parameters.
+#[derive(Debug, PartialEq)]
+pub struct MessageProofParameters {
+	/// Include outbound lane state proof?
+	pub outbound_state_proof_required: bool,
+	/// Cumulative dispatch weight of messages that we're building proof for.
+	pub dispatch_weight: Weight,
+}
+
+/// Source client trait.
+#[async_trait]
+pub trait SourceClient<P: MessageLane>: RelayClient {
+	/// Returns state of the client.
+	async fn state(&self) -> Result<SourceClientState<P>, Self::Error>;
+
+	/// Get nonce of instance of latest generated message.
+	async fn latest_generated_nonce(
+		&self,
+		id: SourceHeaderIdOf<P>,
+	) -> Result<(SourceHeaderIdOf<P>, MessageNonce), Self::Error>;
+	/// Get nonce of the latest message, which receiving has been confirmed by the target chain.
+	async fn latest_confirmed_received_nonce(
+		&self,
+		id: SourceHeaderIdOf<P>,
+	) -> Result<(SourceHeaderIdOf<P>, MessageNonce), Self::Error>;
+
+	/// Returns mapping of message nonces, generated on this client, to their weights.
+	///
+	/// Some weights may be missing from returned map, if corresponding messages were pruned at
+	/// the source chain.
+	async fn generated_messages_weights(
+		&self,
+		id: SourceHeaderIdOf<P>,
+		nonces: RangeInclusive<MessageNonce>,
+	) -> Result<MessageWeightsMap, Self::Error>;
+
+	/// Prove messages in inclusive range [begin; end].
+	async fn prove_messages(
+		&self,
+		id: SourceHeaderIdOf<P>,
+		nonces: RangeInclusive<MessageNonce>,
+		proof_parameters: MessageProofParameters,
+	) -> Result<(SourceHeaderIdOf<P>, RangeInclusive<MessageNonce>, P::MessagesProof), Self::Error>;
+
+	/// Submit messages receiving proof.
+	async fn submit_messages_receiving_proof(
+		&self,
+		generated_at_block: TargetHeaderIdOf<P>,
+		proof: P::MessagesReceivingProof,
+	) -> Result<(), Self::Error>;
+
+	/// We need given finalized target header on source to continue synchronization.
+	async fn require_target_header_on_source(&self, id: TargetHeaderIdOf<P>);
+}
+
+/// Target client trait.
+#[async_trait]
+pub trait TargetClient<P: MessageLane>: RelayClient {
+	/// Returns state of the client.
+	async fn state(&self) -> Result<TargetClientState<P>, Self::Error>;
+
+	/// Get nonce of latest received message.
+	async fn latest_received_nonce(
+		&self,
+		id: TargetHeaderIdOf<P>,
+	) -> Result<(TargetHeaderIdOf<P>, MessageNonce), Self::Error>;
+
+	/// Get nonce of latest confirmed message.
+	async fn latest_confirmed_received_nonce(
+		&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(
+		&self,
+		id: TargetHeaderIdOf<P>,
+	) -> Result<(TargetHeaderIdOf<P>, P::MessagesReceivingProof), Self::Error>;
+
+	/// Submit messages proof.
+	async fn submit_messages_proof(
+		&self,
+		generated_at_header: SourceHeaderIdOf<P>,
+		nonces: RangeInclusive<MessageNonce>,
+		proof: P::MessagesProof,
+	) -> Result<RangeInclusive<MessageNonce>, Self::Error>;
+
+	/// We need given finalized source header on target to continue synchronization.
+	async fn require_source_header_on_target(&self, id: SourceHeaderIdOf<P>);
+}
+
+/// State of the client.
+#[derive(Clone, Debug, Default, PartialEq)]
+pub struct ClientState<SelfHeaderId, PeerHeaderId> {
+	/// Best header id of this chain.
+	pub best_self: SelfHeaderId,
+	/// Best finalized header id of this chain.
+	pub best_finalized_self: SelfHeaderId,
+	/// Best finalized header id of the peer chain read at the best block of this chain (at `best_finalized_self`).
+	pub best_finalized_peer_at_best_self: PeerHeaderId,
+}
+
+/// State of source client in one-way message lane.
+pub type SourceClientState<P> = ClientState<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>>;
+
+/// State of target client in one-way message lane.
+pub type TargetClientState<P> = ClientState<TargetHeaderIdOf<P>, SourceHeaderIdOf<P>>;
+
+/// Both clients state.
+#[derive(Debug, Default)]
+pub struct ClientsState<P: MessageLane> {
+	/// Source client state.
+	pub source: Option<SourceClientState<P>>,
+	/// Target client state.
+	pub target: Option<TargetClientState<P>>,
+}
+
+/// Return prefix that will be used by default to expose Prometheus metrics of the finality proofs sync loop.
+pub fn metrics_prefix<P: MessageLane>(lane: &LaneId) -> String {
+	format!(
+		"{}_to_{}_MessageLane_{}",
+		P::SOURCE_NAME,
+		P::TARGET_NAME,
+		hex::encode(lane)
+	)
+}
+
+/// Run message lane service loop.
+pub async fn run<P: MessageLane>(
+	params: Params,
+	source_client: impl SourceClient<P>,
+	target_client: impl TargetClient<P>,
+	metrics_params: MetricsParams,
+	exit_signal: impl Future<Output = ()>,
+) -> Result<(), String> {
+	let exit_signal = exit_signal.shared();
+	relay_utils::relay_loop(source_client, target_client)
+		.reconnect_delay(params.reconnect_delay)
+		.with_metrics(Some(metrics_prefix::<P>(&params.lane)), metrics_params)
+		.loop_metric(|registry, prefix| MessageLaneLoopMetrics::new(registry, prefix))?
+		.standalone_metric(|registry, prefix| GlobalMetrics::new(registry, prefix))?
+		.expose()
+		.await?
+		.run(|source_client, target_client, metrics| {
+			run_until_connection_lost(
+				params.clone(),
+				source_client,
+				target_client,
+				metrics,
+				exit_signal.clone(),
+			)
+		})
+		.await
+}
+
+/// Run one-way message delivery loop until connection with target or source node is lost, or exit signal is received.
+async fn run_until_connection_lost<P: MessageLane, SC: SourceClient<P>, TC: TargetClient<P>>(
+	params: Params,
+	source_client: SC,
+	target_client: TC,
+	metrics_msg: Option<MessageLaneLoopMetrics>,
+	exit_signal: impl Future<Output = ()>,
+) -> Result<(), FailedClient> {
+	let mut source_retry_backoff = retry_backoff();
+	let mut source_client_is_online = false;
+	let mut source_state_required = true;
+	let source_state = source_client.state().fuse();
+	let source_go_offline_future = futures::future::Fuse::terminated();
+	let source_tick_stream = interval(params.source_tick).fuse();
+
+	let mut target_retry_backoff = retry_backoff();
+	let mut target_client_is_online = false;
+	let mut target_state_required = true;
+	let target_state = target_client.state().fuse();
+	let target_go_offline_future = futures::future::Fuse::terminated();
+	let target_tick_stream = interval(params.target_tick).fuse();
+
+	let (
+		(delivery_source_state_sender, delivery_source_state_receiver),
+		(delivery_target_state_sender, delivery_target_state_receiver),
+	) = (unbounded(), unbounded());
+	let delivery_race_loop = run_message_delivery_race(
+		source_client.clone(),
+		delivery_source_state_receiver,
+		target_client.clone(),
+		delivery_target_state_receiver,
+		params.stall_timeout,
+		metrics_msg.clone(),
+		params.delivery_params,
+	)
+	.fuse();
+
+	let (
+		(receiving_source_state_sender, receiving_source_state_receiver),
+		(receiving_target_state_sender, receiving_target_state_receiver),
+	) = (unbounded(), unbounded());
+	let receiving_race_loop = run_message_receiving_race(
+		source_client.clone(),
+		receiving_source_state_receiver,
+		target_client.clone(),
+		receiving_target_state_receiver,
+		params.stall_timeout,
+		metrics_msg.clone(),
+	)
+	.fuse();
+
+	let exit_signal = exit_signal.fuse();
+
+	futures::pin_mut!(
+		source_state,
+		source_go_offline_future,
+		source_tick_stream,
+		target_state,
+		target_go_offline_future,
+		target_tick_stream,
+		delivery_race_loop,
+		receiving_race_loop,
+		exit_signal
+	);
+
+	loop {
+		futures::select! {
+			new_source_state = source_state => {
+				source_state_required = false;
+
+				source_client_is_online = process_future_result(
+					new_source_state,
+					&mut source_retry_backoff,
+					|new_source_state| {
+						log::debug!(
+							target: "bridge",
+							"Received state from {} node: {:?}",
+							P::SOURCE_NAME,
+							new_source_state,
+						);
+						let _ = delivery_source_state_sender.unbounded_send(new_source_state.clone());
+						let _ = receiving_source_state_sender.unbounded_send(new_source_state.clone());
+
+						if let Some(metrics_msg) = metrics_msg.as_ref() {
+							metrics_msg.update_source_state::<P>(new_source_state);
+						}
+					},
+					&mut source_go_offline_future,
+					async_std::task::sleep,
+					|| format!("Error retrieving state from {} node", P::SOURCE_NAME),
+				).fail_if_connection_error(FailedClient::Source)?;
+			},
+			_ = source_go_offline_future => {
+				source_client_is_online = true;
+			},
+			_ = source_tick_stream.next() => {
+				source_state_required = true;
+			},
+			new_target_state = target_state => {
+				target_state_required = false;
+
+				target_client_is_online = process_future_result(
+					new_target_state,
+					&mut target_retry_backoff,
+					|new_target_state| {
+						log::debug!(
+							target: "bridge",
+							"Received state from {} node: {:?}",
+							P::TARGET_NAME,
+							new_target_state,
+						);
+						let _ = delivery_target_state_sender.unbounded_send(new_target_state.clone());
+						let _ = receiving_target_state_sender.unbounded_send(new_target_state.clone());
+
+						if let Some(metrics_msg) = metrics_msg.as_ref() {
+							metrics_msg.update_target_state::<P>(new_target_state);
+						}
+					},
+					&mut target_go_offline_future,
+					async_std::task::sleep,
+					|| format!("Error retrieving state from {} node", P::TARGET_NAME),
+				).fail_if_connection_error(FailedClient::Target)?;
+			},
+			_ = target_go_offline_future => {
+				target_client_is_online = true;
+			},
+			_ = target_tick_stream.next() => {
+				target_state_required = true;
+			},
+
+			delivery_error = delivery_race_loop => {
+				match delivery_error {
+					Ok(_) => unreachable!("only ends with error; qed"),
+					Err(err) => return Err(err),
+				}
+			},
+			receiving_error = receiving_race_loop => {
+				match receiving_error {
+					Ok(_) => unreachable!("only ends with error; qed"),
+					Err(err) => return Err(err),
+				}
+			},
+
+			() = exit_signal => {
+				return Ok(());
+			}
+		}
+
+		if source_client_is_online && source_state_required {
+			log::debug!(target: "bridge", "Asking {} node about its state", P::SOURCE_NAME);
+			source_state.set(source_client.state().fuse());
+			source_client_is_online = false;
+		}
+
+		if target_client_is_online && target_state_required {
+			log::debug!(target: "bridge", "Asking {} node about its state", P::TARGET_NAME);
+			target_state.set(target_client.state().fuse());
+			target_client_is_online = false;
+		}
+	}
+}
+
+#[cfg(test)]
+pub(crate) mod tests {
+	use super::*;
+	use futures::stream::StreamExt;
+	use parking_lot::Mutex;
+	use relay_utils::{HeaderId, MaybeConnectionError};
+	use std::sync::Arc;
+
+	pub fn header_id(number: TestSourceHeaderNumber) -> TestSourceHeaderId {
+		HeaderId(number, number)
+	}
+
+	pub type TestSourceHeaderId = HeaderId<TestSourceHeaderNumber, TestSourceHeaderHash>;
+	pub type TestTargetHeaderId = HeaderId<TestTargetHeaderNumber, TestTargetHeaderHash>;
+
+	pub type TestMessagesProof = (RangeInclusive<MessageNonce>, Option<MessageNonce>);
+	pub type TestMessagesReceivingProof = MessageNonce;
+
+	pub type TestSourceHeaderNumber = u64;
+	pub type TestSourceHeaderHash = u64;
+
+	pub type TestTargetHeaderNumber = u64;
+	pub type TestTargetHeaderHash = u64;
+
+	#[derive(Debug)]
+	pub struct TestError;
+
+	impl MaybeConnectionError for TestError {
+		fn is_connection_error(&self) -> bool {
+			true
+		}
+	}
+
+	#[derive(Clone)]
+	pub struct TestMessageLane;
+
+	impl MessageLane for TestMessageLane {
+		const SOURCE_NAME: &'static str = "TestSource";
+		const TARGET_NAME: &'static str = "TestTarget";
+
+		type MessagesProof = TestMessagesProof;
+		type MessagesReceivingProof = TestMessagesReceivingProof;
+
+		type SourceHeaderNumber = TestSourceHeaderNumber;
+		type SourceHeaderHash = TestSourceHeaderHash;
+
+		type TargetHeaderNumber = TestTargetHeaderNumber;
+		type TargetHeaderHash = TestTargetHeaderHash;
+	}
+
+	#[derive(Debug, Default, Clone)]
+	pub struct TestClientData {
+		is_source_fails: bool,
+		is_source_reconnected: bool,
+		source_state: SourceClientState<TestMessageLane>,
+		source_latest_generated_nonce: MessageNonce,
+		source_latest_confirmed_received_nonce: MessageNonce,
+		submitted_messages_receiving_proofs: Vec<TestMessagesReceivingProof>,
+		is_target_fails: bool,
+		is_target_reconnected: bool,
+		target_state: SourceClientState<TestMessageLane>,
+		target_latest_received_nonce: MessageNonce,
+		target_latest_confirmed_received_nonce: MessageNonce,
+		submitted_messages_proofs: Vec<TestMessagesProof>,
+		target_to_source_header_required: Option<TestTargetHeaderId>,
+		target_to_source_header_requirements: Vec<TestTargetHeaderId>,
+		source_to_target_header_required: Option<TestSourceHeaderId>,
+		source_to_target_header_requirements: Vec<TestSourceHeaderId>,
+	}
+
+	#[derive(Clone)]
+	pub struct TestSourceClient {
+		data: Arc<Mutex<TestClientData>>,
+		tick: Arc<dyn Fn(&mut TestClientData) + Send + Sync>,
+	}
+
+	#[async_trait]
+	impl RelayClient for TestSourceClient {
+		type Error = TestError;
+
+		async fn reconnect(&mut self) -> Result<(), TestError> {
+			{
+				let mut data = self.data.lock();
+				(self.tick)(&mut *data);
+				data.is_source_reconnected = true;
+			}
+			Ok(())
+		}
+	}
+
+	#[async_trait]
+	impl SourceClient<TestMessageLane> for TestSourceClient {
+		async fn state(&self) -> Result<SourceClientState<TestMessageLane>, TestError> {
+			let mut data = self.data.lock();
+			(self.tick)(&mut *data);
+			if data.is_source_fails {
+				return Err(TestError);
+			}
+			Ok(data.source_state.clone())
+		}
+
+		async fn latest_generated_nonce(
+			&self,
+			id: SourceHeaderIdOf<TestMessageLane>,
+		) -> Result<(SourceHeaderIdOf<TestMessageLane>, MessageNonce), TestError> {
+			let mut data = self.data.lock();
+			(self.tick)(&mut *data);
+			if data.is_source_fails {
+				return Err(TestError);
+			}
+			Ok((id, data.source_latest_generated_nonce))
+		}
+
+		async fn latest_confirmed_received_nonce(
+			&self,
+			id: SourceHeaderIdOf<TestMessageLane>,
+		) -> Result<(SourceHeaderIdOf<TestMessageLane>, MessageNonce), TestError> {
+			let mut data = self.data.lock();
+			(self.tick)(&mut *data);
+			Ok((id, data.source_latest_confirmed_received_nonce))
+		}
+
+		async fn generated_messages_weights(
+			&self,
+			_id: SourceHeaderIdOf<TestMessageLane>,
+			nonces: RangeInclusive<MessageNonce>,
+		) -> Result<MessageWeightsMap, TestError> {
+			Ok(nonces
+				.map(|nonce| (nonce, MessageWeights { weight: 1, size: 1 }))
+				.collect())
+		}
+
+		async fn prove_messages(
+			&self,
+			id: SourceHeaderIdOf<TestMessageLane>,
+			nonces: RangeInclusive<MessageNonce>,
+			proof_parameters: MessageProofParameters,
+		) -> Result<
+			(
+				SourceHeaderIdOf<TestMessageLane>,
+				RangeInclusive<MessageNonce>,
+				TestMessagesProof,
+			),
+			TestError,
+		> {
+			let mut data = self.data.lock();
+			(self.tick)(&mut *data);
+			Ok((
+				id,
+				nonces.clone(),
+				(
+					nonces,
+					if proof_parameters.outbound_state_proof_required {
+						Some(data.source_latest_confirmed_received_nonce)
+					} else {
+						None
+					},
+				),
+			))
+		}
+
+		async fn submit_messages_receiving_proof(
+			&self,
+			_generated_at_block: TargetHeaderIdOf<TestMessageLane>,
+			proof: TestMessagesReceivingProof,
+		) -> Result<(), TestError> {
+			let mut data = self.data.lock();
+			(self.tick)(&mut *data);
+			data.submitted_messages_receiving_proofs.push(proof);
+			data.source_latest_confirmed_received_nonce = proof;
+			Ok(())
+		}
+
+		async fn require_target_header_on_source(&self, id: TargetHeaderIdOf<TestMessageLane>) {
+			let mut data = self.data.lock();
+			data.target_to_source_header_required = Some(id);
+			data.target_to_source_header_requirements.push(id);
+			(self.tick)(&mut *data);
+		}
+	}
+
+	#[derive(Clone)]
+	pub struct TestTargetClient {
+		data: Arc<Mutex<TestClientData>>,
+		tick: Arc<dyn Fn(&mut TestClientData) + Send + Sync>,
+	}
+
+	#[async_trait]
+	impl RelayClient for TestTargetClient {
+		type Error = TestError;
+
+		async fn reconnect(&mut self) -> Result<(), TestError> {
+			{
+				let mut data = self.data.lock();
+				(self.tick)(&mut *data);
+				data.is_target_reconnected = true;
+			}
+			Ok(())
+		}
+	}
+
+	#[async_trait]
+	impl TargetClient<TestMessageLane> for TestTargetClient {
+		async fn state(&self) -> Result<TargetClientState<TestMessageLane>, TestError> {
+			let mut data = self.data.lock();
+			(self.tick)(&mut *data);
+			if data.is_target_fails {
+				return Err(TestError);
+			}
+			Ok(data.target_state.clone())
+		}
+
+		async fn latest_received_nonce(
+			&self,
+			id: TargetHeaderIdOf<TestMessageLane>,
+		) -> Result<(TargetHeaderIdOf<TestMessageLane>, MessageNonce), TestError> {
+			let mut data = self.data.lock();
+			(self.tick)(&mut *data);
+			if data.is_target_fails {
+				return Err(TestError);
+			}
+			Ok((id, data.target_latest_received_nonce))
+		}
+
+		async fn unrewarded_relayers_state(
+			&self,
+			id: TargetHeaderIdOf<TestMessageLane>,
+		) -> Result<(TargetHeaderIdOf<TestMessageLane>, UnrewardedRelayersState), TestError> {
+			Ok((
+				id,
+				UnrewardedRelayersState {
+					unrewarded_relayer_entries: 0,
+					messages_in_oldest_entry: 0,
+					total_messages: 0,
+				},
+			))
+		}
+
+		async fn latest_confirmed_received_nonce(
+			&self,
+			id: TargetHeaderIdOf<TestMessageLane>,
+		) -> Result<(TargetHeaderIdOf<TestMessageLane>, MessageNonce), TestError> {
+			let mut data = self.data.lock();
+			(self.tick)(&mut *data);
+			if data.is_target_fails {
+				return Err(TestError);
+			}
+			Ok((id, data.target_latest_confirmed_received_nonce))
+		}
+
+		async fn prove_messages_receiving(
+			&self,
+			id: TargetHeaderIdOf<TestMessageLane>,
+		) -> Result<(TargetHeaderIdOf<TestMessageLane>, TestMessagesReceivingProof), TestError> {
+			Ok((id, self.data.lock().target_latest_received_nonce))
+		}
+
+		async fn submit_messages_proof(
+			&self,
+			_generated_at_header: SourceHeaderIdOf<TestMessageLane>,
+			nonces: RangeInclusive<MessageNonce>,
+			proof: TestMessagesProof,
+		) -> Result<RangeInclusive<MessageNonce>, TestError> {
+			let mut data = self.data.lock();
+			(self.tick)(&mut *data);
+			if data.is_target_fails {
+				return Err(TestError);
+			}
+			data.target_state.best_self =
+				HeaderId(data.target_state.best_self.0 + 1, data.target_state.best_self.1 + 1);
+			data.target_latest_received_nonce = *proof.0.end();
+			if let Some(target_latest_confirmed_received_nonce) = proof.1 {
+				data.target_latest_confirmed_received_nonce = target_latest_confirmed_received_nonce;
+			}
+			data.submitted_messages_proofs.push(proof);
+			Ok(nonces)
+		}
+
+		async fn require_source_header_on_target(&self, id: SourceHeaderIdOf<TestMessageLane>) {
+			let mut data = self.data.lock();
+			data.source_to_target_header_required = Some(id);
+			data.source_to_target_header_requirements.push(id);
+			(self.tick)(&mut *data);
+		}
+	}
+
+	fn run_loop_test(
+		data: TestClientData,
+		source_tick: Arc<dyn Fn(&mut TestClientData) + Send + Sync>,
+		target_tick: Arc<dyn Fn(&mut TestClientData) + Send + Sync>,
+		exit_signal: impl Future<Output = ()>,
+	) -> TestClientData {
+		async_std::task::block_on(async {
+			let data = Arc::new(Mutex::new(data));
+
+			let source_client = TestSourceClient {
+				data: data.clone(),
+				tick: source_tick,
+			};
+			let target_client = TestTargetClient {
+				data: data.clone(),
+				tick: target_tick,
+			};
+			let _ = run(
+				Params {
+					lane: [0, 0, 0, 0],
+					source_tick: Duration::from_millis(100),
+					target_tick: Duration::from_millis(100),
+					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,
+						max_messages_size_in_single_batch: 4,
+					},
+				},
+				source_client,
+				target_client,
+				MetricsParams::disabled(),
+				exit_signal,
+			)
+			.await;
+			let result = data.lock().clone();
+			result
+		})
+	}
+
+	#[test]
+	fn message_lane_loop_is_able_to_recover_from_connection_errors() {
+		// with this configuration, source client will return Err, making source client
+		// reconnect. Then the target client will fail with Err + reconnect. Then we finally
+		// able to deliver messages.
+		let (exit_sender, exit_receiver) = unbounded();
+		let result = run_loop_test(
+			TestClientData {
+				is_source_fails: true,
+				source_state: ClientState {
+					best_self: HeaderId(0, 0),
+					best_finalized_self: HeaderId(0, 0),
+					best_finalized_peer_at_best_self: HeaderId(0, 0),
+				},
+				source_latest_generated_nonce: 1,
+				target_state: ClientState {
+					best_self: HeaderId(0, 0),
+					best_finalized_self: HeaderId(0, 0),
+					best_finalized_peer_at_best_self: HeaderId(0, 0),
+				},
+				target_latest_received_nonce: 0,
+				..Default::default()
+			},
+			Arc::new(|data: &mut TestClientData| {
+				if data.is_source_reconnected {
+					data.is_source_fails = false;
+					data.is_target_fails = true;
+				}
+			}),
+			Arc::new(move |data: &mut TestClientData| {
+				if data.is_target_reconnected {
+					data.is_target_fails = false;
+				}
+				if data.target_state.best_finalized_peer_at_best_self.0 < 10 {
+					data.target_state.best_finalized_peer_at_best_self = HeaderId(
+						data.target_state.best_finalized_peer_at_best_self.0 + 1,
+						data.target_state.best_finalized_peer_at_best_self.0 + 1,
+					);
+				}
+				if !data.submitted_messages_proofs.is_empty() {
+					exit_sender.unbounded_send(()).unwrap();
+				}
+			}),
+			exit_receiver.into_future().map(|(_, _)| ()),
+		);
+
+		assert_eq!(result.submitted_messages_proofs, vec![(1..=1, None)],);
+	}
+
+	#[test]
+	fn message_lane_loop_works() {
+		let (exit_sender, exit_receiver) = unbounded();
+		let result = run_loop_test(
+			TestClientData {
+				source_state: ClientState {
+					best_self: HeaderId(10, 10),
+					best_finalized_self: HeaderId(10, 10),
+					best_finalized_peer_at_best_self: HeaderId(0, 0),
+				},
+				source_latest_generated_nonce: 10,
+				target_state: ClientState {
+					best_self: HeaderId(0, 0),
+					best_finalized_self: HeaderId(0, 0),
+					best_finalized_peer_at_best_self: HeaderId(0, 0),
+				},
+				target_latest_received_nonce: 0,
+				..Default::default()
+			},
+			Arc::new(|data: &mut TestClientData| {
+				// headers relay must only be started when we need new target headers at source node
+				if data.target_to_source_header_required.is_some() {
+					assert!(data.source_state.best_finalized_peer_at_best_self.0 < data.target_state.best_self.0);
+					data.target_to_source_header_required = None;
+				}
+			}),
+			Arc::new(move |data: &mut TestClientData| {
+				// headers relay must only be started when we need new source headers at target node
+				if data.source_to_target_header_required.is_some() {
+					assert!(data.target_state.best_finalized_peer_at_best_self.0 < data.source_state.best_self.0);
+					data.source_to_target_header_required = None;
+				}
+				// syncing source headers -> target chain (all at once)
+				if data.target_state.best_finalized_peer_at_best_self.0 < data.source_state.best_finalized_self.0 {
+					data.target_state.best_finalized_peer_at_best_self = data.source_state.best_finalized_self;
+				}
+				// syncing source headers -> target chain (all at once)
+				if data.source_state.best_finalized_peer_at_best_self.0 < data.target_state.best_finalized_self.0 {
+					data.source_state.best_finalized_peer_at_best_self = data.target_state.best_finalized_self;
+				}
+				// if target has received messages batch => increase blocks so that confirmations may be sent
+				if data.target_latest_received_nonce == 4
+					|| data.target_latest_received_nonce == 8
+					|| data.target_latest_received_nonce == 10
+				{
+					data.target_state.best_self =
+						HeaderId(data.target_state.best_self.0 + 1, data.target_state.best_self.0 + 1);
+					data.target_state.best_finalized_self = data.target_state.best_self;
+					data.source_state.best_self =
+						HeaderId(data.source_state.best_self.0 + 1, data.source_state.best_self.0 + 1);
+					data.source_state.best_finalized_self = data.source_state.best_self;
+				}
+				// if source has received all messages receiving confirmations => stop
+				if data.source_latest_confirmed_received_nonce == 10 {
+					exit_sender.unbounded_send(()).unwrap();
+				}
+			}),
+			exit_receiver.into_future().map(|(_, _)| ()),
+		);
+
+		// there are no strict restrictions on when reward confirmation should come
+		// (because `max_unconfirmed_nonces_at_target` is `100` in tests and this confirmation
+		// depends on the state of both clients)
+		// => we do not check it here
+		assert_eq!(result.submitted_messages_proofs[0].0, 1..=4);
+		assert_eq!(result.submitted_messages_proofs[1].0, 5..=8);
+		assert_eq!(result.submitted_messages_proofs[2].0, 9..=10);
+		assert!(!result.submitted_messages_receiving_proofs.is_empty());
+
+		// check that we have at least once required new source->target or target->source headers
+		assert!(!result.target_to_source_header_requirements.is_empty());
+		assert!(!result.source_to_target_header_requirements.is_empty());
+	}
+}

polkadot/bridges/relays/messages/src/message_race_delivery.rs

@@ -0,0 +1,879 @@
+// Copyright 2019-2021 Parity Technologies (UK) Ltd.
+// This file is part of Parity Bridges Common.
+
+// Parity Bridges Common is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity Bridges Common is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+//! Message delivery race delivers proof-of-messages from lane.source to lane.target.
+
+use crate::message_lane::{MessageLane, SourceHeaderIdOf, TargetHeaderIdOf};
+use crate::message_lane_loop::{
+	MessageDeliveryParams, MessageProofParameters, MessageWeightsMap, SourceClient as MessageLaneSourceClient,
+	SourceClientState, TargetClient as MessageLaneTargetClient, TargetClientState,
+};
+use crate::message_race_loop::{
+	MessageRace, NoncesRange, RaceState, RaceStrategy, SourceClient, SourceClientNonces, TargetClient,
+	TargetClientNonces,
+};
+use crate::message_race_strategy::BasicStrategy;
+use crate::metrics::MessageLaneLoopMetrics;
+
+use async_trait::async_trait;
+use bp_messages::{MessageNonce, UnrewardedRelayersState, Weight};
+use futures::stream::FusedStream;
+use relay_utils::FailedClient;
+use std::{
+	collections::{BTreeMap, VecDeque},
+	marker::PhantomData,
+	ops::RangeInclusive,
+	time::Duration,
+};
+
+/// Run message delivery race.
+pub async fn run<P: MessageLane>(
+	source_client: impl MessageLaneSourceClient<P>,
+	source_state_updates: impl FusedStream<Item = SourceClientState<P>>,
+	target_client: impl MessageLaneTargetClient<P>,
+	target_state_updates: impl FusedStream<Item = TargetClientState<P>>,
+	stall_timeout: Duration,
+	metrics_msg: Option<MessageLaneLoopMetrics>,
+	params: MessageDeliveryParams,
+) -> Result<(), FailedClient> {
+	crate::message_race_loop::run(
+		MessageDeliveryRaceSource {
+			client: source_client,
+			metrics_msg: metrics_msg.clone(),
+			_phantom: Default::default(),
+		},
+		source_state_updates,
+		MessageDeliveryRaceTarget {
+			client: target_client,
+			metrics_msg,
+			_phantom: Default::default(),
+		},
+		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,
+			max_messages_size_in_single_batch: params.max_messages_size_in_single_batch,
+			latest_confirmed_nonces_at_source: VecDeque::new(),
+			target_nonces: None,
+			strategy: BasicStrategy::new(),
+		},
+	)
+	.await
+}
+
+/// Message delivery race.
+struct MessageDeliveryRace<P>(std::marker::PhantomData<P>);
+
+impl<P: MessageLane> MessageRace for MessageDeliveryRace<P> {
+	type SourceHeaderId = SourceHeaderIdOf<P>;
+	type TargetHeaderId = TargetHeaderIdOf<P>;
+
+	type MessageNonce = MessageNonce;
+	type Proof = P::MessagesProof;
+
+	fn source_name() -> String {
+		format!("{}::MessagesDelivery", P::SOURCE_NAME)
+	}
+
+	fn target_name() -> String {
+		format!("{}::MessagesDelivery", P::TARGET_NAME)
+	}
+}
+
+/// Message delivery race source, which is a source of the lane.
+struct MessageDeliveryRaceSource<P: MessageLane, C> {
+	client: C,
+	metrics_msg: Option<MessageLaneLoopMetrics>,
+	_phantom: PhantomData<P>,
+}
+
+#[async_trait]
+impl<P, C> SourceClient<MessageDeliveryRace<P>> for MessageDeliveryRaceSource<P, C>
+where
+	P: MessageLane,
+	C: MessageLaneSourceClient<P>,
+{
+	type Error = C::Error;
+	type NoncesRange = MessageWeightsMap;
+	type ProofParameters = MessageProofParameters;
+
+	async fn nonces(
+		&self,
+		at_block: SourceHeaderIdOf<P>,
+		prev_latest_nonce: MessageNonce,
+	) -> Result<(SourceHeaderIdOf<P>, SourceClientNonces<Self::NoncesRange>), Self::Error> {
+		let (at_block, latest_generated_nonce) = self.client.latest_generated_nonce(at_block).await?;
+		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_generated_nonce::<P>(latest_generated_nonce);
+			metrics_msg.update_source_latest_confirmed_nonce::<P>(latest_confirmed_nonce);
+		}
+
+		let new_nonces = if latest_generated_nonce > prev_latest_nonce {
+			self.client
+				.generated_messages_weights(at_block.clone(), prev_latest_nonce + 1..=latest_generated_nonce)
+				.await?
+		} else {
+			MessageWeightsMap::new()
+		};
+
+		Ok((
+			at_block,
+			SourceClientNonces {
+				new_nonces,
+				confirmed_nonce: Some(latest_confirmed_nonce),
+			},
+		))
+	}
+
+	async fn generate_proof(
+		&self,
+		at_block: SourceHeaderIdOf<P>,
+		nonces: RangeInclusive<MessageNonce>,
+		proof_parameters: Self::ProofParameters,
+	) -> Result<(SourceHeaderIdOf<P>, RangeInclusive<MessageNonce>, P::MessagesProof), Self::Error> {
+		self.client.prove_messages(at_block, nonces, proof_parameters).await
+	}
+}
+
+/// Message delivery race target, which is a target of the lane.
+struct MessageDeliveryRaceTarget<P: MessageLane, C> {
+	client: C,
+	metrics_msg: Option<MessageLaneLoopMetrics>,
+	_phantom: PhantomData<P>,
+}
+
+#[async_trait]
+impl<P, C> TargetClient<MessageDeliveryRace<P>> for MessageDeliveryRaceTarget<P, C>
+where
+	P: MessageLane,
+	C: MessageLaneTargetClient<P>,
+{
+	type Error = C::Error;
+	type TargetNoncesData = DeliveryRaceTargetNoncesData;
+
+	async fn require_source_header(&self, id: SourceHeaderIdOf<P>) {
+		self.client.require_source_header_on_target(id).await
+	}
+
+	async fn nonces(
+		&self,
+		at_block: TargetHeaderIdOf<P>,
+		update_metrics: bool,
+	) -> 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 update_metrics {
+			if let Some(metrics_msg) = self.metrics_msg.as_ref() {
+				metrics_msg.update_target_latest_received_nonce::<P>(latest_received_nonce);
+				metrics_msg.update_target_latest_confirmed_nonce::<P>(latest_confirmed_nonce);
+			}
+		}
+
+		Ok((
+			at_block,
+			TargetClientNonces {
+				latest_nonce: latest_received_nonce,
+				nonces_data: DeliveryRaceTargetNoncesData {
+					confirmed_nonce: latest_confirmed_nonce,
+					unrewarded_relayers,
+				},
+			},
+		))
+	}
+
+	async fn submit_proof(
+		&self,
+		generated_at_block: SourceHeaderIdOf<P>,
+		nonces: RangeInclusive<MessageNonce>,
+		proof: P::MessagesProof,
+	) -> Result<RangeInclusive<MessageNonce>, Self::Error> {
+		self.client
+			.submit_messages_proof(generated_at_block, nonces, proof)
+			.await
+	}
+}
+
+/// 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.
+	max_messages_in_single_batch: MessageNonce,
+	/// Maximal cumulative messages weight in the single delivery transaction.
+	max_messages_weight_in_single_batch: Weight,
+	/// Maximal messages size in the single delivery transaction.
+	max_messages_size_in_single_batch: usize,
+	/// Latest confirmed nonces at the source client + the header id where we have first met this nonce.
+	latest_confirmed_nonces_at_source: VecDeque<(SourceHeaderIdOf<P>, MessageNonce)>,
+	/// Target nonces from the source client.
+	target_nonces: Option<TargetClientNonces<DeliveryRaceTargetNoncesData>>,
+	/// Basic delivery strategy.
+	strategy: MessageDeliveryStrategyBase<P>,
+}
+
+type MessageDeliveryStrategyBase<P> = BasicStrategy<
+	<P as MessageLane>::SourceHeaderNumber,
+	<P as MessageLane>::SourceHeaderHash,
+	<P as MessageLane>::TargetHeaderNumber,
+	<P as MessageLane>::TargetHeaderHash,
+	MessageWeightsMap,
+	<P as MessageLane>::MessagesProof,
+>;
+
+impl<P: MessageLane> std::fmt::Debug for MessageDeliveryStrategy<P> {
+	fn fmt(&self, fmt: &mut std::fmt::Formatter) -> std::fmt::Result {
+		fmt.debug_struct("MessageDeliveryStrategy")
+			.field(
+				"max_unrewarded_relayer_entries_at_target",
+				&self.max_unrewarded_relayer_entries_at_target,
+			)
+			.field(
+				"max_unconfirmed_nonces_at_target",
+				&self.max_unconfirmed_nonces_at_target,
+			)
+			.field("max_messages_in_single_batch", &self.max_messages_in_single_batch)
+			.field(
+				"max_messages_weight_in_single_batch",
+				&self.max_messages_weight_in_single_batch,
+			)
+			.field(
+				"max_messages_size_in_single_batch",
+				&self.max_messages_size_in_single_batch,
+			)
+			.field(
+				"latest_confirmed_nonces_at_source",
+				&self.latest_confirmed_nonces_at_source,
+			)
+			.field("target_nonces", &self.target_nonces)
+			.field("strategy", &self.strategy)
+			.finish()
+	}
+}
+
+impl<P: MessageLane> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::MessagesProof>
+	for MessageDeliveryStrategy<P>
+{
+	type SourceNoncesRange = MessageWeightsMap;
+	type ProofParameters = MessageProofParameters;
+	type TargetNoncesData = DeliveryRaceTargetNoncesData;
+
+	fn is_empty(&self) -> bool {
+		self.strategy.is_empty()
+	}
+
+	fn required_source_header_at_target(&self, current_best: &SourceHeaderIdOf<P>) -> Option<SourceHeaderIdOf<P>> {
+		self.strategy.required_source_header_at_target(current_best)
+	}
+
+	fn best_at_source(&self) -> Option<MessageNonce> {
+		self.strategy.best_at_source()
+	}
+
+	fn best_at_target(&self) -> Option<MessageNonce> {
+		self.strategy.best_at_target()
+	}
+
+	fn source_nonces_updated(
+		&mut self,
+		at_block: SourceHeaderIdOf<P>,
+		nonces: SourceClientNonces<Self::SourceNoncesRange>,
+	) {
+		if let Some(confirmed_nonce) = nonces.confirmed_nonce {
+			let is_confirmed_nonce_updated = self
+				.latest_confirmed_nonces_at_source
+				.back()
+				.map(|(_, prev_nonce)| *prev_nonce != confirmed_nonce)
+				.unwrap_or(true);
+			if is_confirmed_nonce_updated {
+				self.latest_confirmed_nonces_at_source
+					.push_back((at_block.clone(), confirmed_nonce));
+			}
+		}
+		self.strategy.source_nonces_updated(at_block, nonces)
+	}
+
+	fn best_target_nonces_updated(
+		&mut self,
+		nonces: TargetClientNonces<DeliveryRaceTargetNoncesData>,
+		race_state: &mut RaceState<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::MessagesProof>,
+	) {
+		// best target nonces must always be ge than finalized target nonces
+		let mut target_nonces = self.target_nonces.take().unwrap_or_else(|| nonces.clone());
+		target_nonces.nonces_data = nonces.nonces_data.clone();
+		target_nonces.latest_nonce = std::cmp::max(target_nonces.latest_nonce, nonces.latest_nonce);
+		self.target_nonces = Some(target_nonces);
+
+		self.strategy.best_target_nonces_updated(
+			TargetClientNonces {
+				latest_nonce: nonces.latest_nonce,
+				nonces_data: (),
+			},
+			race_state,
+		)
+	}
+
+	fn finalized_target_nonces_updated(
+		&mut self,
+		nonces: TargetClientNonces<DeliveryRaceTargetNoncesData>,
+		race_state: &mut RaceState<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::MessagesProof>,
+	) {
+		if let Some(ref best_finalized_source_header_id_at_best_target) =
+			race_state.best_finalized_source_header_id_at_best_target
+		{
+			let oldest_header_number_to_keep = best_finalized_source_header_id_at_best_target.0;
+			while self
+				.latest_confirmed_nonces_at_source
+				.front()
+				.map(|(id, _)| id.0 < oldest_header_number_to_keep)
+				.unwrap_or(false)
+			{
+				self.latest_confirmed_nonces_at_source.pop_front();
+			}
+		}
+
+		if let Some(ref mut target_nonces) = self.target_nonces {
+			target_nonces.latest_nonce = std::cmp::max(target_nonces.latest_nonce, nonces.latest_nonce);
+		}
+
+		self.strategy.finalized_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)> {
+		let best_finalized_source_header_id_at_best_target =
+			race_state.best_finalized_source_header_id_at_best_target.clone()?;
+		let latest_confirmed_nonce_at_source = self
+			.latest_confirmed_nonces_at_source
+			.iter()
+			.take_while(|(id, _)| id.0 <= best_finalized_source_header_id_at_best_target.0)
+			.last()
+			.map(|(_, nonce)| *nonce)?;
+		let target_nonces = self.target_nonces.as_ref()?;
+
+		// There's additional condition in the message delivery race: target would reject messages
+		// if there are too much unconfirmed messages at the inbound lane.
+
+		// The receiving race is responsible to deliver confirmations back to the source chain. So if
+		// there's a lot of unconfirmed messages, let's wait until it'll be able to do its job.
+		let latest_received_nonce_at_target = target_nonces.latest_nonce;
+		let confirmations_missing = latest_received_nonce_at_target.checked_sub(latest_confirmed_nonce_at_source);
+		match confirmations_missing {
+			Some(confirmations_missing) if confirmations_missing >= self.max_unconfirmed_nonces_at_target => {
+				log::debug!(
+					target: "bridge",
+					"Cannot deliver any more messages from {} to {}. Too many unconfirmed nonces \
+					at target: target.latest_received={:?}, source.latest_confirmed={:?}, max={:?}",
+					MessageDeliveryRace::<P>::source_name(),
+					MessageDeliveryRace::<P>::target_name(),
+					latest_received_nonce_at_target,
+					latest_confirmed_nonce_at_source,
+					self.max_unconfirmed_nonces_at_target,
+				);
+
+				return None;
+			}
+			_ => (),
+		}
+
+		// Ok - we may have new nonces to deliver. But target may still reject new messages, because we haven't
+		// notified it that (some) messages have been confirmed. So we may want to include updated
+		// `source.latest_confirmed` in the proof.
+		//
+		// 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.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.
+		//
+		// We may deliver at most:
+		//
+		// max_unconfirmed_nonces_at_target - (latest_received_nonce_at_target - latest_confirmed_nonce_at_target)
+		//
+		// messages in the batch. But since we're including outbound state proof in the batch, then it
+		// may be increased to:
+		//
+		// max_unconfirmed_nonces_at_target - (latest_received_nonce_at_target - latest_confirmed_nonce_at_source)
+		let future_confirmed_nonce_at_target = if outbound_state_proof_required {
+			latest_confirmed_nonce_at_source
+		} else {
+			latest_confirmed_nonce_at_target
+		};
+		let max_nonces = latest_received_nonce_at_target
+			.checked_sub(future_confirmed_nonce_at_target)
+			.and_then(|diff| self.max_unconfirmed_nonces_at_target.checked_sub(diff))
+			.unwrap_or_default();
+		let max_nonces = std::cmp::min(max_nonces, self.max_messages_in_single_batch);
+		let max_messages_weight_in_single_batch = self.max_messages_weight_in_single_batch;
+		let max_messages_size_in_single_batch = self.max_messages_size_in_single_batch;
+		let mut selected_weight: Weight = 0;
+		let mut selected_size: usize = 0;
+		let mut selected_count: MessageNonce = 0;
+
+		let selected_nonces = self
+			.strategy
+			.select_nonces_to_deliver_with_selector(race_state, |range| {
+				let to_requeue = range
+					.into_iter()
+					.skip_while(|(_, weight)| {
+						// Since we (hopefully) have some reserves in `max_messages_weight_in_single_batch`
+						// and `max_messages_size_in_single_batch`, we may still try to submit transaction
+						// with single message if message overflows these limits. The worst case would be if
+						// transaction will be rejected by the target runtime, but at least we have tried.
+
+						// limit messages in the batch by weight
+						let new_selected_weight = match selected_weight.checked_add(weight.weight) {
+							Some(new_selected_weight) if new_selected_weight <= max_messages_weight_in_single_batch => {
+								new_selected_weight
+							}
+							new_selected_weight if selected_count == 0 => {
+								log::warn!(
+									target: "bridge",
+									"Going to submit message delivery transaction with declared dispatch \
+									weight {:?} that overflows maximal configured weight {}",
+									new_selected_weight,
+									max_messages_weight_in_single_batch,
+								);
+								new_selected_weight.unwrap_or(Weight::MAX)
+							}
+							_ => return false,
+						};
+
+						// limit messages in the batch by size
+						let new_selected_size = match selected_size.checked_add(weight.size) {
+							Some(new_selected_size) if new_selected_size <= max_messages_size_in_single_batch => {
+								new_selected_size
+							}
+							new_selected_size if selected_count == 0 => {
+								log::warn!(
+									target: "bridge",
+									"Going to submit message delivery transaction with message \
+									size {:?} that overflows maximal configured size {}",
+									new_selected_size,
+									max_messages_size_in_single_batch,
+								);
+								new_selected_size.unwrap_or(usize::MAX)
+							}
+							_ => return false,
+						};
+
+						// limit number of messages in the batch
+						let new_selected_count = selected_count + 1;
+						if new_selected_count > max_nonces {
+							return false;
+						}
+
+						selected_weight = new_selected_weight;
+						selected_size = new_selected_size;
+						selected_count = new_selected_count;
+						true
+					})
+					.collect::<BTreeMap<_, _>>();
+				if to_requeue.is_empty() {
+					None
+				} else {
+					Some(to_requeue)
+				}
+			})?;
+
+		Some((
+			selected_nonces,
+			MessageProofParameters {
+				outbound_state_proof_required,
+				dispatch_weight: selected_weight,
+			},
+		))
+	}
+}
+
+impl NoncesRange for MessageWeightsMap {
+	fn begin(&self) -> MessageNonce {
+		self.keys().next().cloned().unwrap_or_default()
+	}
+
+	fn end(&self) -> MessageNonce {
+		self.keys().next_back().cloned().unwrap_or_default()
+	}
+
+	fn greater_than(mut self, nonce: MessageNonce) -> Option<Self> {
+		let gte = self.split_off(&(nonce + 1));
+		if gte.is_empty() {
+			None
+		} else {
+			Some(gte)
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::message_lane_loop::{
+		tests::{header_id, TestMessageLane, TestMessagesProof, TestSourceHeaderId, TestTargetHeaderId},
+		MessageWeights,
+	};
+
+	type TestRaceState = RaceState<TestSourceHeaderId, TestTargetHeaderId, TestMessagesProof>;
+	type TestStrategy = MessageDeliveryStrategy<TestMessageLane>;
+
+	fn prepare_strategy() -> (TestRaceState, TestStrategy) {
+		let mut race_state = RaceState {
+			best_finalized_source_header_id_at_source: Some(header_id(1)),
+			best_finalized_source_header_id_at_best_target: Some(header_id(1)),
+			best_target_header_id: Some(header_id(1)),
+			best_finalized_target_header_id: Some(header_id(1)),
+			nonces_to_submit: None,
+			nonces_submitted: None,
+		};
+
+		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,
+			max_messages_size_in_single_batch: 4,
+			latest_confirmed_nonces_at_source: vec![(header_id(1), 19)].into_iter().collect(),
+			target_nonces: Some(TargetClientNonces {
+				latest_nonce: 19,
+				nonces_data: DeliveryRaceTargetNoncesData {
+					confirmed_nonce: 19,
+					unrewarded_relayers: UnrewardedRelayersState {
+						unrewarded_relayer_entries: 0,
+						messages_in_oldest_entry: 0,
+						total_messages: 0,
+					},
+				},
+			}),
+			strategy: BasicStrategy::new(),
+		};
+
+		race_strategy.strategy.source_nonces_updated(
+			header_id(1),
+			SourceClientNonces {
+				new_nonces: vec![
+					(20, MessageWeights { weight: 1, size: 1 }),
+					(21, MessageWeights { weight: 1, size: 1 }),
+					(22, MessageWeights { weight: 1, size: 1 }),
+					(23, MessageWeights { weight: 1, size: 1 }),
+				]
+				.into_iter()
+				.collect(),
+				confirmed_nonce: Some(19),
+			},
+		);
+
+		let target_nonces = TargetClientNonces {
+			latest_nonce: 19,
+			nonces_data: (),
+		};
+		race_strategy
+			.strategy
+			.best_target_nonces_updated(target_nonces.clone(), &mut race_state);
+		race_strategy
+			.strategy
+			.finalized_target_nonces_updated(target_nonces, &mut race_state);
+
+		(race_state, race_strategy)
+	}
+
+	fn proof_parameters(state_required: bool, weight: Weight) -> MessageProofParameters {
+		MessageProofParameters {
+			outbound_state_proof_required: state_required,
+			dispatch_weight: weight,
+		}
+	}
+
+	#[test]
+	fn weights_map_works_as_nonces_range() {
+		fn build_map(range: RangeInclusive<MessageNonce>) -> MessageWeightsMap {
+			range
+				.map(|idx| {
+					(
+						idx,
+						MessageWeights {
+							weight: idx,
+							size: idx as _,
+						},
+					)
+				})
+				.collect()
+		}
+
+		let map = build_map(20..=30);
+
+		assert_eq!(map.begin(), 20);
+		assert_eq!(map.end(), 30);
+		assert_eq!(map.clone().greater_than(10), Some(build_map(20..=30)));
+		assert_eq!(map.clone().greater_than(19), Some(build_map(20..=30)));
+		assert_eq!(map.clone().greater_than(20), Some(build_map(21..=30)));
+		assert_eq!(map.clone().greater_than(25), Some(build_map(26..=30)));
+		assert_eq!(map.clone().greater_than(29), Some(build_map(30..=30)));
+		assert_eq!(map.greater_than(30), None);
+	}
+
+	#[test]
+	fn message_delivery_strategy_selects_messages_to_deliver() {
+		let (state, mut strategy) = prepare_strategy();
+
+		// both sides are ready to relay new messages
+		assert_eq!(
+			strategy.select_nonces_to_deliver(&state),
+			Some(((20..=23), proof_parameters(false, 4)))
+		);
+	}
+
+	#[test]
+	fn message_delivery_strategy_selects_nothing_if_too_many_confirmations_missing() {
+		let (state, mut strategy) = prepare_strategy();
+
+		// if there are already `max_unconfirmed_nonces_at_target` messages on target,
+		// we need to wait until confirmations will be delivered by receiving race
+		strategy.latest_confirmed_nonces_at_source = vec![(
+			header_id(1),
+			strategy.target_nonces.as_ref().unwrap().latest_nonce - strategy.max_unconfirmed_nonces_at_target,
+		)]
+		.into_iter()
+		.collect();
+		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
+	}
+
+	#[test]
+	fn message_delivery_strategy_includes_outbound_state_proof_when_new_nonces_are_available() {
+		let (state, mut strategy) = prepare_strategy();
+
+		// 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_nonces_at_source.back().unwrap().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_nonces_at_source.back().unwrap().1;
+		{
+			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_nonces_at_source.back().unwrap().1;
+		{
+			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)))
+		);
+	}
+
+	#[test]
+	fn message_delivery_strategy_limits_batch_by_messages_weight() {
+		let (state, mut strategy) = prepare_strategy();
+
+		// not all queued messages may fit in the batch, because batch has max weight
+		strategy.max_messages_weight_in_single_batch = 3;
+		assert_eq!(
+			strategy.select_nonces_to_deliver(&state),
+			Some(((20..=22), proof_parameters(false, 3)))
+		);
+	}
+
+	#[test]
+	fn message_delivery_strategy_accepts_single_message_even_if_its_weight_overflows_maximal_weight() {
+		let (state, mut strategy) = prepare_strategy();
+
+		// first message doesn't fit in the batch, because it has weight (10) that overflows max weight (4)
+		strategy.strategy.source_queue_mut()[0].1.get_mut(&20).unwrap().weight = 10;
+		assert_eq!(
+			strategy.select_nonces_to_deliver(&state),
+			Some(((20..=20), proof_parameters(false, 10)))
+		);
+	}
+
+	#[test]
+	fn message_delivery_strategy_limits_batch_by_messages_size() {
+		let (state, mut strategy) = prepare_strategy();
+
+		// not all queued messages may fit in the batch, because batch has max weight
+		strategy.max_messages_size_in_single_batch = 3;
+		assert_eq!(
+			strategy.select_nonces_to_deliver(&state),
+			Some(((20..=22), proof_parameters(false, 3)))
+		);
+	}
+
+	#[test]
+	fn message_delivery_strategy_accepts_single_message_even_if_its_weight_overflows_maximal_size() {
+		let (state, mut strategy) = prepare_strategy();
+
+		// first message doesn't fit in the batch, because it has weight (10) that overflows max weight (4)
+		strategy.strategy.source_queue_mut()[0].1.get_mut(&20).unwrap().size = 10;
+		assert_eq!(
+			strategy.select_nonces_to_deliver(&state),
+			Some(((20..=20), proof_parameters(false, 1)))
+		);
+	}
+
+	#[test]
+	fn message_delivery_strategy_limits_batch_by_messages_count_when_there_is_upper_limit() {
+		let (state, mut strategy) = prepare_strategy();
+
+		// not all queued messages may fit in the batch, because batch has max number of messages limit
+		strategy.max_messages_in_single_batch = 3;
+		assert_eq!(
+			strategy.select_nonces_to_deliver(&state),
+			Some(((20..=22), proof_parameters(false, 3)))
+		);
+	}
+
+	#[test]
+	fn message_delivery_strategy_limits_batch_by_messages_count_when_there_are_unconfirmed_nonces() {
+		let (state, mut strategy) = prepare_strategy();
+
+		// 1 delivery confirmation from target to source is still missing, so we may only
+		// relay 3 new messages
+		let prev_confirmed_nonce_at_source = strategy.latest_confirmed_nonces_at_source.back().unwrap().1;
+		strategy.latest_confirmed_nonces_at_source = vec![(header_id(1), prev_confirmed_nonce_at_source - 1)]
+			.into_iter()
+			.collect();
+		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)))
+		);
+	}
+
+	#[test]
+	fn message_delivery_strategy_waits_for_confirmed_nonce_header_to_appear_on_target() {
+		// 1 delivery confirmation from target to source is still missing, so we may deliver
+		// reward confirmation with our message delivery transaction. But the problem is that
+		// the reward has been paid at header 2 && this header is still unknown to target node.
+		//
+		// => so we can't deliver more than 3 messages
+		let (mut state, mut strategy) = prepare_strategy();
+		let prev_confirmed_nonce_at_source = strategy.latest_confirmed_nonces_at_source.back().unwrap().1;
+		strategy.latest_confirmed_nonces_at_source = vec![
+			(header_id(1), prev_confirmed_nonce_at_source - 1),
+			(header_id(2), prev_confirmed_nonce_at_source),
+		]
+		.into_iter()
+		.collect();
+		strategy.target_nonces.as_mut().unwrap().nonces_data.confirmed_nonce = prev_confirmed_nonce_at_source - 1;
+		state.best_finalized_source_header_id_at_best_target = Some(header_id(1));
+		assert_eq!(
+			strategy.select_nonces_to_deliver(&state),
+			Some(((20..=22), proof_parameters(false, 3)))
+		);
+
+		// the same situation, but the header 2 is known to the target node, so we may deliver reward confirmation
+		let (mut state, mut strategy) = prepare_strategy();
+		let prev_confirmed_nonce_at_source = strategy.latest_confirmed_nonces_at_source.back().unwrap().1;
+		strategy.latest_confirmed_nonces_at_source = vec![
+			(header_id(1), prev_confirmed_nonce_at_source - 1),
+			(header_id(2), prev_confirmed_nonce_at_source),
+		]
+		.into_iter()
+		.collect();
+		strategy.target_nonces.as_mut().unwrap().nonces_data.confirmed_nonce = prev_confirmed_nonce_at_source - 1;
+		state.best_finalized_source_header_id_at_source = Some(header_id(2));
+		state.best_finalized_source_header_id_at_best_target = Some(header_id(2));
+		assert_eq!(
+			strategy.select_nonces_to_deliver(&state),
+			Some(((20..=23), proof_parameters(true, 4)))
+		);
+	}
+}

polkadot/bridges/relays/messages/src/message_race_loop.rs

@@ -0,0 +1,627 @@
+// Copyright 2019-2021 Parity Technologies (UK) Ltd.
+// This file is part of Parity Bridges Common.
+
+// Parity Bridges Common is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity Bridges Common is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+//! Loop that is serving single race within message lane. This could be
+//! message delivery race, receiving confirmations race or processing
+//! confirmations race.
+//!
+//! The idea of the race is simple - we have `nonce`-s on source and target
+//! nodes. We're trying to prove that the source node has this nonce (and
+//! associated data - like messages, lane state, etc) to the target node by
+//! generating and submitting proof.
+
+use crate::message_lane_loop::ClientState;
+
+use async_trait::async_trait;
+use bp_messages::MessageNonce;
+use futures::{
+	future::FutureExt,
+	stream::{FusedStream, StreamExt},
+};
+use relay_utils::{process_future_result, retry_backoff, FailedClient, MaybeConnectionError};
+use std::{
+	fmt::Debug,
+	ops::RangeInclusive,
+	time::{Duration, Instant},
+};
+
+/// One of races within lane.
+pub trait MessageRace {
+	/// Header id of the race source.
+	type SourceHeaderId: Debug + Clone + PartialEq;
+	/// Header id of the race source.
+	type TargetHeaderId: Debug + Clone + PartialEq;
+
+	/// Message nonce used in the race.
+	type MessageNonce: Debug + Clone;
+	/// Proof that is generated and delivered in this race.
+	type Proof: Debug + Clone;
+
+	/// Name of the race source.
+	fn source_name() -> String;
+	/// Name of the race target.
+	fn target_name() -> String;
+}
+
+/// State of race source client.
+type SourceClientState<P> = ClientState<<P as MessageRace>::SourceHeaderId, <P as MessageRace>::TargetHeaderId>;
+
+/// State of race target client.
+type TargetClientState<P> = ClientState<<P as MessageRace>::TargetHeaderId, <P as MessageRace>::SourceHeaderId>;
+
+/// Inclusive nonces range.
+pub trait NoncesRange: Debug + Sized {
+	/// Get begin of the range.
+	fn begin(&self) -> MessageNonce;
+	/// Get end of the range.
+	fn end(&self) -> MessageNonce;
+	/// Returns new range with current range nonces that are greater than the passed `nonce`.
+	/// If there are no such nonces, `None` is returned.
+	fn greater_than(self, nonce: MessageNonce) -> Option<Self>;
+}
+
+/// Nonces on the race source client.
+#[derive(Debug, Clone)]
+pub struct SourceClientNonces<NoncesRange> {
+	/// New nonces range known to the client. `New` here means all nonces generated after
+	/// `prev_latest_nonce` passed to the `SourceClient::nonces` method.
+	pub new_nonces: NoncesRange,
+	/// 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>,
+}
+
+/// Nonces on the race target client.
+#[derive(Debug, Clone)]
+pub struct TargetClientNonces<TargetNoncesData> {
+	/// Latest nonce that is known to the target client.
+	pub latest_nonce: 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.
+#[async_trait]
+pub trait SourceClient<P: MessageRace> {
+	/// Type of error this clients returns.
+	type Error: std::fmt::Debug + MaybeConnectionError;
+	/// Type of nonces range returned by the source client.
+	type NoncesRange: NoncesRange;
+	/// Additional proof parameters required to generate proof.
+	type ProofParameters;
+
+	/// Return nonces that are known to the source client.
+	async fn nonces(
+		&self,
+		at_block: P::SourceHeaderId,
+		prev_latest_nonce: MessageNonce,
+	) -> Result<(P::SourceHeaderId, SourceClientNonces<Self::NoncesRange>), Self::Error>;
+	/// Generate proof for delivering to the target client.
+	async fn generate_proof(
+		&self,
+		at_block: P::SourceHeaderId,
+		nonces: RangeInclusive<MessageNonce>,
+		proof_parameters: Self::ProofParameters,
+	) -> Result<(P::SourceHeaderId, RangeInclusive<MessageNonce>, P::Proof), Self::Error>;
+}
+
+/// One of message lane clients, which is target client for the race.
+#[async_trait]
+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;
+
+	/// Ask headers relay to relay finalized headers up to (and including) given header
+	/// from race source to race target.
+	async fn require_source_header(&self, id: P::SourceHeaderId);
+
+	/// Return nonces that are known to the target client.
+	async fn nonces(
+		&self,
+		at_block: P::TargetHeaderId,
+		update_metrics: bool,
+	) -> Result<(P::TargetHeaderId, TargetClientNonces<Self::TargetNoncesData>), Self::Error>;
+	/// Submit proof to the target client.
+	async fn submit_proof(
+		&self,
+		generated_at_block: P::SourceHeaderId,
+		nonces: RangeInclusive<MessageNonce>,
+		proof: P::Proof,
+	) -> Result<RangeInclusive<MessageNonce>, Self::Error>;
+}
+
+/// Race strategy.
+pub trait RaceStrategy<SourceHeaderId, TargetHeaderId, Proof>: Debug {
+	/// Type of nonces range expected from the source client.
+	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;
+	/// Return id of source header that is required to be on target to continue synchronization.
+	fn required_source_header_at_target(&self, current_best: &SourceHeaderId) -> Option<SourceHeaderId>;
+	/// Return best nonce at source node.
+	///
+	/// `Some` is returned only if we are sure that the value is greater or equal
+	/// than the result of `best_at_target`.
+	fn best_at_source(&self) -> Option<MessageNonce>;
+	/// Return best nonce at target node.
+	///
+	/// May return `None` if value is yet unknown.
+	fn best_at_target(&self) -> Option<MessageNonce>;
+
+	/// Called when nonces are updated at source node of the race.
+	fn source_nonces_updated(&mut self, at_block: SourceHeaderId, nonces: SourceClientNonces<Self::SourceNoncesRange>);
+	/// Called when best nonces are updated at target node of the race.
+	fn best_target_nonces_updated(
+		&mut self,
+		nonces: TargetClientNonces<Self::TargetNoncesData>,
+		race_state: &mut RaceState<SourceHeaderId, TargetHeaderId, Proof>,
+	);
+	/// Called when finalized nonces are updated at target node of the race.
+	fn finalized_target_nonces_updated(
+		&mut self,
+		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
+	/// data) from source to target node.
+	/// Additionally, parameters required to generate proof are returned.
+	fn select_nonces_to_deliver(
+		&mut self,
+		race_state: &RaceState<SourceHeaderId, TargetHeaderId, Proof>,
+	) -> Option<(RangeInclusive<MessageNonce>, Self::ProofParameters)>;
+}
+
+/// State of the race.
+#[derive(Debug)]
+pub struct RaceState<SourceHeaderId, TargetHeaderId, Proof> {
+	/// Best finalized source header id at the source client.
+	pub best_finalized_source_header_id_at_source: Option<SourceHeaderId>,
+	/// Best finalized source header id at the best block on the target
+	/// client (at the `best_finalized_source_header_id_at_best_target`).
+	pub best_finalized_source_header_id_at_best_target: Option<SourceHeaderId>,
+	/// Best header id at the target client.
+	pub best_target_header_id: Option<TargetHeaderId>,
+	/// Best finalized header id at the target client.
+	pub best_finalized_target_header_id: Option<TargetHeaderId>,
+	/// Range of nonces that we have selected to submit.
+	pub nonces_to_submit: Option<(SourceHeaderId, RangeInclusive<MessageNonce>, Proof)>,
+	/// Range of nonces that is currently submitted.
+	pub nonces_submitted: Option<RangeInclusive<MessageNonce>>,
+}
+
+/// Run race loop until connection with target or source node is lost.
+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: TC,
+	race_target_updated: impl FusedStream<Item = TargetClientState<P>>,
+	stall_timeout: Duration,
+	mut strategy: impl RaceStrategy<
+		P::SourceHeaderId,
+		P::TargetHeaderId,
+		P::Proof,
+		SourceNoncesRange = SC::NoncesRange,
+		ProofParameters = SC::ProofParameters,
+		TargetNoncesData = TC::TargetNoncesData,
+	>,
+) -> Result<(), FailedClient> {
+	let mut progress_context = Instant::now();
+	let mut race_state = RaceState::default();
+	let mut stall_countdown = Instant::now();
+
+	let mut source_retry_backoff = retry_backoff();
+	let mut source_client_is_online = true;
+	let mut source_nonces_required = false;
+	let source_nonces = futures::future::Fuse::terminated();
+	let source_generate_proof = futures::future::Fuse::terminated();
+	let source_go_offline_future = futures::future::Fuse::terminated();
+
+	let mut target_retry_backoff = retry_backoff();
+	let mut target_client_is_online = true;
+	let mut target_best_nonces_required = false;
+	let mut target_finalized_nonces_required = false;
+	let target_best_nonces = futures::future::Fuse::terminated();
+	let target_finalized_nonces = futures::future::Fuse::terminated();
+	let target_submit_proof = futures::future::Fuse::terminated();
+	let target_go_offline_future = futures::future::Fuse::terminated();
+
+	futures::pin_mut!(
+		race_source_updated,
+		source_nonces,
+		source_generate_proof,
+		source_go_offline_future,
+		race_target_updated,
+		target_best_nonces,
+		target_finalized_nonces,
+		target_submit_proof,
+		target_go_offline_future,
+	);
+
+	loop {
+		futures::select! {
+			// when headers ids are updated
+			source_state = race_source_updated.next() => {
+				if let Some(source_state) = source_state {
+					let is_source_state_updated = race_state.best_finalized_source_header_id_at_source.as_ref()
+						!= Some(&source_state.best_finalized_self);
+					if is_source_state_updated {
+						source_nonces_required = true;
+						race_state.best_finalized_source_header_id_at_source = Some(source_state.best_finalized_self);
+					}
+				}
+			},
+			target_state = race_target_updated.next() => {
+				if let Some(target_state) = target_state {
+					let is_target_best_state_updated = race_state.best_target_header_id.as_ref()
+						!= Some(&target_state.best_self);
+
+					if is_target_best_state_updated {
+						target_best_nonces_required = true;
+						race_state.best_target_header_id = Some(target_state.best_self);
+						race_state.best_finalized_source_header_id_at_best_target
+							= Some(target_state.best_finalized_peer_at_best_self);
+					}
+
+					let is_target_finalized_state_updated = race_state.best_finalized_target_header_id.as_ref()
+						!= Some(&target_state.best_finalized_self);
+					if  is_target_finalized_state_updated {
+						target_finalized_nonces_required = true;
+						race_state.best_finalized_target_header_id = Some(target_state.best_finalized_self);
+					}
+				}
+			},
+
+			// when nonces are updated
+			nonces = source_nonces => {
+				source_nonces_required = false;
+
+				source_client_is_online = process_future_result(
+					nonces,
+					&mut source_retry_backoff,
+					|(at_block, nonces)| {
+						log::debug!(
+							target: "bridge",
+							"Received nonces from {}: {:?}",
+							P::source_name(),
+							nonces,
+						);
+
+						strategy.source_nonces_updated(at_block, nonces);
+					},
+					&mut source_go_offline_future,
+					async_std::task::sleep,
+					|| format!("Error retrieving nonces from {}", P::source_name()),
+				).fail_if_connection_error(FailedClient::Source)?;
+
+				// ask for more headers if we have nonces to deliver and required headers are missing
+				let required_source_header_id = race_state
+					.best_finalized_source_header_id_at_best_target
+					.as_ref()
+					.and_then(|best|strategy.required_source_header_at_target(best));
+				if let Some(required_source_header_id) = required_source_header_id {
+					race_target.require_source_header(required_source_header_id).await;
+				}
+			},
+			nonces = target_best_nonces => {
+				target_best_nonces_required = false;
+
+				target_client_is_online = process_future_result(
+					nonces,
+					&mut target_retry_backoff,
+					|(_, nonces)| {
+						log::debug!(
+							target: "bridge",
+							"Received best nonces from {}: {:?}",
+							P::target_name(),
+							nonces,
+						);
+
+						let prev_best_at_target = strategy.best_at_target();
+						strategy.best_target_nonces_updated(nonces, &mut race_state);
+						if strategy.best_at_target() != prev_best_at_target {
+							stall_countdown = Instant::now();
+						}
+					},
+					&mut target_go_offline_future,
+					async_std::task::sleep,
+					|| format!("Error retrieving best nonces from {}", P::target_name()),
+				).fail_if_connection_error(FailedClient::Target)?;
+			},
+			nonces = target_finalized_nonces => {
+				target_finalized_nonces_required = false;
+
+				target_client_is_online = process_future_result(
+					nonces,
+					&mut target_retry_backoff,
+					|(_, nonces)| {
+						log::debug!(
+							target: "bridge",
+							"Received finalized nonces from {}: {:?}",
+							P::target_name(),
+							nonces,
+						);
+
+						strategy.finalized_target_nonces_updated(nonces, &mut race_state);
+					},
+					&mut target_go_offline_future,
+					async_std::task::sleep,
+					|| format!("Error retrieving finalized nonces from {}", P::target_name()),
+				).fail_if_connection_error(FailedClient::Target)?;
+			},
+
+			// proof generation and submission
+			proof = source_generate_proof => {
+				source_client_is_online = process_future_result(
+					proof,
+					&mut source_retry_backoff,
+					|(at_block, nonces_range, proof)| {
+						log::debug!(
+							target: "bridge",
+							"Received proof for nonces in range {:?} from {}",
+							nonces_range,
+							P::source_name(),
+						);
+
+						race_state.nonces_to_submit = Some((at_block, nonces_range, proof));
+					},
+					&mut source_go_offline_future,
+					async_std::task::sleep,
+					|| format!("Error generating proof at {}", P::source_name()),
+				).fail_if_connection_error(FailedClient::Source)?;
+			},
+			proof_submit_result = target_submit_proof => {
+				target_client_is_online = process_future_result(
+					proof_submit_result,
+					&mut target_retry_backoff,
+					|nonces_range| {
+						log::debug!(
+							target: "bridge",
+							"Successfully submitted proof of nonces {:?} to {}",
+							nonces_range,
+							P::target_name(),
+						);
+
+						race_state.nonces_to_submit = None;
+						race_state.nonces_submitted = Some(nonces_range);
+						stall_countdown = Instant::now();
+					},
+					&mut target_go_offline_future,
+					async_std::task::sleep,
+					|| format!("Error submitting proof {}", P::target_name()),
+				).fail_if_connection_error(FailedClient::Target)?;
+			},
+
+			// when we're ready to retry request
+			_ = source_go_offline_future => {
+				source_client_is_online = true;
+			},
+			_ = target_go_offline_future => {
+				target_client_is_online = true;
+			},
+		}
+
+		progress_context = print_race_progress::<P, _>(progress_context, &strategy);
+
+		if stall_countdown.elapsed() > stall_timeout {
+			log::warn!(
+				target: "bridge",
+				"{} -> {} race has stalled. State: {:?}. Strategy: {:?}",
+				P::source_name(),
+				P::target_name(),
+				race_state,
+				strategy,
+			);
+
+			return Err(FailedClient::Both);
+		} else if race_state.nonces_to_submit.is_none() && race_state.nonces_submitted.is_none() && strategy.is_empty()
+		{
+			stall_countdown = Instant::now();
+		}
+
+		if source_client_is_online {
+			source_client_is_online = false;
+
+			let nonces_to_deliver = select_nonces_to_deliver(&race_state, &mut strategy);
+			let best_at_source = strategy.best_at_source();
+
+			if let Some((at_block, nonces_range, proof_parameters)) = nonces_to_deliver {
+				log::debug!(
+					target: "bridge",
+					"Asking {} to prove nonces in range {:?} at block {:?}",
+					P::source_name(),
+					nonces_range,
+					at_block,
+				);
+				source_generate_proof.set(
+					race_source
+						.generate_proof(at_block, nonces_range, proof_parameters)
+						.fuse(),
+				);
+			} else if source_nonces_required && best_at_source.is_some() {
+				log::debug!(target: "bridge", "Asking {} about message nonces", P::source_name());
+				let at_block = race_state
+					.best_finalized_source_header_id_at_source
+					.as_ref()
+					.expect(
+						"source_nonces_required is only true when\
+						best_finalized_source_header_id_at_source is Some; qed",
+					)
+					.clone();
+				source_nonces.set(
+					race_source
+						.nonces(at_block, best_at_source.expect("guaranteed by if condition; qed"))
+						.fuse(),
+				);
+			} else {
+				source_client_is_online = true;
+			}
+		}
+
+		if target_client_is_online {
+			target_client_is_online = false;
+
+			if let Some((at_block, nonces_range, proof)) = race_state.nonces_to_submit.as_ref() {
+				log::debug!(
+					target: "bridge",
+					"Going to submit proof of messages in range {:?} to {} node",
+					nonces_range,
+					P::target_name(),
+				);
+				target_submit_proof.set(
+					race_target
+						.submit_proof(at_block.clone(), nonces_range.clone(), proof.clone())
+						.fuse(),
+				);
+			} else if target_best_nonces_required {
+				log::debug!(target: "bridge", "Asking {} about best message nonces", P::target_name());
+				let at_block = race_state
+					.best_target_header_id
+					.as_ref()
+					.expect("target_best_nonces_required is only true when best_target_header_id is Some; qed")
+					.clone();
+				target_best_nonces.set(race_target.nonces(at_block, false).fuse());
+			} else if target_finalized_nonces_required {
+				log::debug!(target: "bridge", "Asking {} about finalized message nonces", P::target_name());
+				let at_block = race_state
+					.best_finalized_target_header_id
+					.as_ref()
+					.expect(
+						"target_finalized_nonces_required is only true when\
+						best_finalized_target_header_id is Some; qed",
+					)
+					.clone();
+				target_finalized_nonces.set(race_target.nonces(at_block, true).fuse());
+			} else {
+				target_client_is_online = true;
+			}
+		}
+	}
+}
+
+impl<SourceHeaderId, TargetHeaderId, Proof> Default for RaceState<SourceHeaderId, TargetHeaderId, Proof> {
+	fn default() -> Self {
+		RaceState {
+			best_finalized_source_header_id_at_source: None,
+			best_finalized_source_header_id_at_best_target: None,
+			best_target_header_id: None,
+			best_finalized_target_header_id: None,
+			nonces_to_submit: None,
+			nonces_submitted: None,
+		}
+	}
+}
+
+/// Print race progress.
+fn print_race_progress<P, S>(prev_time: Instant, strategy: &S) -> Instant
+where
+	P: MessageRace,
+	S: RaceStrategy<P::SourceHeaderId, P::TargetHeaderId, P::Proof>,
+{
+	let now_time = Instant::now();
+
+	let need_update = now_time.saturating_duration_since(prev_time) > Duration::from_secs(10);
+	if !need_update {
+		return prev_time;
+	}
+
+	let now_best_nonce_at_source = strategy.best_at_source();
+	let now_best_nonce_at_target = strategy.best_at_target();
+	log::info!(
+		target: "bridge",
+		"Synced {:?} of {:?} nonces in {} -> {} race",
+		now_best_nonce_at_target,
+		now_best_nonce_at_source,
+		P::source_name(),
+		P::target_name(),
+	);
+	now_time
+}
+
+fn select_nonces_to_deliver<SourceHeaderId, TargetHeaderId, Proof, Strategy>(
+	race_state: &RaceState<SourceHeaderId, TargetHeaderId, Proof>,
+	strategy: &mut Strategy,
+) -> Option<(SourceHeaderId, RangeInclusive<MessageNonce>, Strategy::ProofParameters)>
+where
+	SourceHeaderId: Clone,
+	Strategy: RaceStrategy<SourceHeaderId, TargetHeaderId, Proof>,
+{
+	race_state
+		.best_finalized_source_header_id_at_best_target
+		.as_ref()
+		.and_then(|best_finalized_source_header_id_at_best_target| {
+			strategy
+				.select_nonces_to_deliver(&race_state)
+				.map(|(nonces_range, proof_parameters)| {
+					(
+						best_finalized_source_header_id_at_best_target.clone(),
+						nonces_range,
+						proof_parameters,
+					)
+				})
+		})
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::message_race_strategy::BasicStrategy;
+	use relay_utils::HeaderId;
+
+	#[test]
+	fn proof_is_generated_at_best_block_known_to_target_node() {
+		const GENERATED_AT: u64 = 6;
+		const BEST_AT_SOURCE: u64 = 10;
+		const BEST_AT_TARGET: u64 = 8;
+
+		// target node only knows about source' BEST_AT_TARGET block
+		// source node has BEST_AT_SOURCE > BEST_AT_TARGET block
+		let mut race_state = RaceState::<_, _, ()> {
+			best_finalized_source_header_id_at_source: Some(HeaderId(BEST_AT_SOURCE, BEST_AT_SOURCE)),
+			best_finalized_source_header_id_at_best_target: Some(HeaderId(BEST_AT_TARGET, BEST_AT_TARGET)),
+			best_target_header_id: Some(HeaderId(0, 0)),
+			best_finalized_target_header_id: Some(HeaderId(0, 0)),
+			nonces_to_submit: None,
+			nonces_submitted: None,
+		};
+
+		// we have some nonces to deliver and they're generated at GENERATED_AT < BEST_AT_SOURCE
+		let mut strategy = BasicStrategy::new();
+		strategy.source_nonces_updated(
+			HeaderId(GENERATED_AT, GENERATED_AT),
+			SourceClientNonces {
+				new_nonces: 0..=10,
+				confirmed_nonce: None,
+			},
+		);
+		strategy.best_target_nonces_updated(
+			TargetClientNonces {
+				latest_nonce: 5u64,
+				nonces_data: (),
+			},
+			&mut race_state,
+		);
+
+		// the proof will be generated on source, but using BEST_AT_TARGET block
+		assert_eq!(
+			select_nonces_to_deliver(&race_state, &mut strategy),
+			Some((HeaderId(BEST_AT_TARGET, BEST_AT_TARGET), 6..=10, (),))
+		);
+	}
+}

polkadot/bridges/relays/messages/src/message_race_receiving.rs

@@ -0,0 +1,236 @@
+// Copyright 2019-2021 Parity Technologies (UK) Ltd.
+// This file is part of Parity Bridges Common.
+
+// Parity Bridges Common is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity Bridges Common is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+//! Message receiving race delivers proof-of-messages-delivery from lane.target to lane.source.
+
+use crate::message_lane::{MessageLane, SourceHeaderIdOf, TargetHeaderIdOf};
+use crate::message_lane_loop::{
+	SourceClient as MessageLaneSourceClient, SourceClientState, TargetClient as MessageLaneTargetClient,
+	TargetClientState,
+};
+use crate::message_race_loop::{
+	MessageRace, NoncesRange, SourceClient, SourceClientNonces, TargetClient, TargetClientNonces,
+};
+use crate::message_race_strategy::BasicStrategy;
+use crate::metrics::MessageLaneLoopMetrics;
+
+use async_trait::async_trait;
+use bp_messages::MessageNonce;
+use futures::stream::FusedStream;
+use relay_utils::FailedClient;
+use std::{marker::PhantomData, ops::RangeInclusive, time::Duration};
+
+/// Message receiving confirmations delivery strategy.
+type ReceivingConfirmationsBasicStrategy<P> = BasicStrategy<
+	<P as MessageLane>::TargetHeaderNumber,
+	<P as MessageLane>::TargetHeaderHash,
+	<P as MessageLane>::SourceHeaderNumber,
+	<P as MessageLane>::SourceHeaderHash,
+	RangeInclusive<MessageNonce>,
+	<P as MessageLane>::MessagesReceivingProof,
+>;
+
+/// Run receiving confirmations race.
+pub async fn run<P: MessageLane>(
+	source_client: impl MessageLaneSourceClient<P>,
+	source_state_updates: impl FusedStream<Item = SourceClientState<P>>,
+	target_client: impl MessageLaneTargetClient<P>,
+	target_state_updates: impl FusedStream<Item = TargetClientState<P>>,
+	stall_timeout: Duration,
+	metrics_msg: Option<MessageLaneLoopMetrics>,
+) -> Result<(), FailedClient> {
+	crate::message_race_loop::run(
+		ReceivingConfirmationsRaceSource {
+			client: target_client,
+			metrics_msg: metrics_msg.clone(),
+			_phantom: Default::default(),
+		},
+		target_state_updates,
+		ReceivingConfirmationsRaceTarget {
+			client: source_client,
+			metrics_msg,
+			_phantom: Default::default(),
+		},
+		source_state_updates,
+		stall_timeout,
+		ReceivingConfirmationsBasicStrategy::<P>::new(),
+	)
+	.await
+}
+
+/// Messages receiving confirmations race.
+struct ReceivingConfirmationsRace<P>(std::marker::PhantomData<P>);
+
+impl<P: MessageLane> MessageRace for ReceivingConfirmationsRace<P> {
+	type SourceHeaderId = TargetHeaderIdOf<P>;
+	type TargetHeaderId = SourceHeaderIdOf<P>;
+
+	type MessageNonce = MessageNonce;
+	type Proof = P::MessagesReceivingProof;
+
+	fn source_name() -> String {
+		format!("{}::ReceivingConfirmationsDelivery", P::TARGET_NAME)
+	}
+
+	fn target_name() -> String {
+		format!("{}::ReceivingConfirmationsDelivery", P::SOURCE_NAME)
+	}
+}
+
+/// Message receiving confirmations race source, which is a target of the lane.
+struct ReceivingConfirmationsRaceSource<P: MessageLane, C> {
+	client: C,
+	metrics_msg: Option<MessageLaneLoopMetrics>,
+	_phantom: PhantomData<P>,
+}
+
+#[async_trait]
+impl<P, C> SourceClient<ReceivingConfirmationsRace<P>> for ReceivingConfirmationsRaceSource<P, C>
+where
+	P: MessageLane,
+	C: MessageLaneTargetClient<P>,
+{
+	type Error = C::Error;
+	type NoncesRange = RangeInclusive<MessageNonce>;
+	type ProofParameters = ();
+
+	async fn nonces(
+		&self,
+		at_block: TargetHeaderIdOf<P>,
+		prev_latest_nonce: MessageNonce,
+	) -> Result<(TargetHeaderIdOf<P>, SourceClientNonces<Self::NoncesRange>), Self::Error> {
+		let (at_block, latest_received_nonce) = self.client.latest_received_nonce(at_block).await?;
+		if let Some(metrics_msg) = self.metrics_msg.as_ref() {
+			metrics_msg.update_target_latest_received_nonce::<P>(latest_received_nonce);
+		}
+		Ok((
+			at_block,
+			SourceClientNonces {
+				new_nonces: prev_latest_nonce + 1..=latest_received_nonce,
+				confirmed_nonce: None,
+			},
+		))
+	}
+
+	#[allow(clippy::unit_arg)]
+	async fn generate_proof(
+		&self,
+		at_block: TargetHeaderIdOf<P>,
+		nonces: RangeInclusive<MessageNonce>,
+		_proof_parameters: Self::ProofParameters,
+	) -> Result<
+		(
+			TargetHeaderIdOf<P>,
+			RangeInclusive<MessageNonce>,
+			P::MessagesReceivingProof,
+		),
+		Self::Error,
+	> {
+		self.client
+			.prove_messages_receiving(at_block)
+			.await
+			.map(|(at_block, proof)| (at_block, nonces, proof))
+	}
+}
+
+/// Message receiving confirmations race target, which is a source of the lane.
+struct ReceivingConfirmationsRaceTarget<P: MessageLane, C> {
+	client: C,
+	metrics_msg: Option<MessageLaneLoopMetrics>,
+	_phantom: PhantomData<P>,
+}
+
+#[async_trait]
+impl<P, C> TargetClient<ReceivingConfirmationsRace<P>> for ReceivingConfirmationsRaceTarget<P, C>
+where
+	P: MessageLane,
+	C: MessageLaneSourceClient<P>,
+{
+	type Error = C::Error;
+	type TargetNoncesData = ();
+
+	async fn require_source_header(&self, id: TargetHeaderIdOf<P>) {
+		self.client.require_target_header_on_source(id).await
+	}
+
+	async fn nonces(
+		&self,
+		at_block: SourceHeaderIdOf<P>,
+		update_metrics: bool,
+	) -> Result<(SourceHeaderIdOf<P>, TargetClientNonces<()>), Self::Error> {
+		let (at_block, latest_confirmed_nonce) = self.client.latest_confirmed_received_nonce(at_block).await?;
+		if update_metrics {
+			if let Some(metrics_msg) = self.metrics_msg.as_ref() {
+				metrics_msg.update_source_latest_confirmed_nonce::<P>(latest_confirmed_nonce);
+			}
+		}
+		Ok((
+			at_block,
+			TargetClientNonces {
+				latest_nonce: latest_confirmed_nonce,
+				nonces_data: (),
+			},
+		))
+	}
+
+	async fn submit_proof(
+		&self,
+		generated_at_block: TargetHeaderIdOf<P>,
+		nonces: RangeInclusive<MessageNonce>,
+		proof: P::MessagesReceivingProof,
+	) -> Result<RangeInclusive<MessageNonce>, Self::Error> {
+		self.client
+			.submit_messages_receiving_proof(generated_at_block, proof)
+			.await?;
+		Ok(nonces)
+	}
+}
+
+impl NoncesRange for RangeInclusive<MessageNonce> {
+	fn begin(&self) -> MessageNonce {
+		*RangeInclusive::<MessageNonce>::start(self)
+	}
+
+	fn end(&self) -> MessageNonce {
+		*RangeInclusive::<MessageNonce>::end(self)
+	}
+
+	fn greater_than(self, nonce: MessageNonce) -> Option<Self> {
+		let next_nonce = nonce + 1;
+		let end = *self.end();
+		if next_nonce > end {
+			None
+		} else {
+			Some(std::cmp::max(self.begin(), next_nonce)..=end)
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn range_inclusive_works_as_nonces_range() {
+		let range = 20..=30;
+
+		assert_eq!(NoncesRange::begin(&range), 20);
+		assert_eq!(NoncesRange::end(&range), 30);
+		assert_eq!(range.clone().greater_than(10), Some(20..=30));
+		assert_eq!(range.clone().greater_than(19), Some(20..=30));
+		assert_eq!(range.clone().greater_than(20), Some(21..=30));
+		assert_eq!(range.clone().greater_than(25), Some(26..=30));
+		assert_eq!(range.clone().greater_than(29), Some(30..=30));
+		assert_eq!(range.greater_than(30), None);
+	}
+}

polkadot/bridges/relays/messages/src/message_race_strategy.rs

@@ -0,0 +1,488 @@
+// Copyright 2019-2021 Parity Technologies (UK) Ltd.
+// This file is part of Parity Bridges Common.
+
+// Parity Bridges Common is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity Bridges Common is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+//! Basic delivery strategy. The strategy selects nonces if:
+//!
+//! 1) there are more nonces on the source side than on the target side;
+//! 2) new nonces may be proved to target node (i.e. they have appeared at the
+//!    block, which is known to the target node).
+
+use crate::message_race_loop::{NoncesRange, RaceState, RaceStrategy, SourceClientNonces, TargetClientNonces};
+
+use bp_messages::MessageNonce;
+use relay_utils::HeaderId;
+use std::{collections::VecDeque, fmt::Debug, marker::PhantomData, ops::RangeInclusive};
+
+/// Nonces delivery strategy.
+#[derive(Debug)]
+pub struct BasicStrategy<
+	SourceHeaderNumber,
+	SourceHeaderHash,
+	TargetHeaderNumber,
+	TargetHeaderHash,
+	SourceNoncesRange,
+	Proof,
+> {
+	/// All queued nonces.
+	source_queue: VecDeque<(HeaderId<SourceHeaderHash, SourceHeaderNumber>, SourceNoncesRange)>,
+	/// Best nonce known to target node (at its best block). `None` if it has not been received yet.
+	best_target_nonce: Option<MessageNonce>,
+	/// Unused generic types dump.
+	_phantom: PhantomData<(TargetHeaderNumber, TargetHeaderHash, Proof)>,
+}
+
+impl<SourceHeaderNumber, SourceHeaderHash, TargetHeaderNumber, TargetHeaderHash, SourceNoncesRange, Proof>
+	BasicStrategy<SourceHeaderNumber, SourceHeaderHash, TargetHeaderNumber, TargetHeaderHash, SourceNoncesRange, Proof>
+where
+	SourceHeaderHash: Clone,
+	SourceHeaderNumber: Clone + Ord,
+	SourceNoncesRange: NoncesRange,
+{
+	/// Create new delivery strategy.
+	pub fn new() -> Self {
+		BasicStrategy {
+			source_queue: VecDeque::new(),
+			best_target_nonce: None,
+			_phantom: Default::default(),
+		}
+	}
+
+	/// Mutable reference to source queue to use in tests.
+	#[cfg(test)]
+	pub(crate) fn source_queue_mut(
+		&mut self,
+	) -> &mut VecDeque<(HeaderId<SourceHeaderHash, SourceHeaderNumber>, SourceNoncesRange)> {
+		&mut self.source_queue
+	}
+
+	/// Should return `Some(nonces)` if we need to deliver proof of `nonces` (and associated
+	/// data) from source to target node.
+	///
+	/// The `selector` function receives range of nonces and should return `None` if the whole
+	/// range needs to be delivered. If there are some nonces in the range that can't be delivered
+	/// right now, it should return `Some` with 'undeliverable' nonces. Please keep in mind that
+	/// this should be the sub-range that the passed range ends with, because nonces are always
+	/// delivered in-order. Otherwise the function will panic.
+	pub fn select_nonces_to_deliver_with_selector(
+		&mut self,
+		race_state: &RaceState<
+			HeaderId<SourceHeaderHash, SourceHeaderNumber>,
+			HeaderId<TargetHeaderHash, TargetHeaderNumber>,
+			Proof,
+		>,
+		mut selector: impl FnMut(SourceNoncesRange) -> Option<SourceNoncesRange>,
+	) -> Option<RangeInclusive<MessageNonce>> {
+		// if we do not know best nonce at target node, we can't select anything
+		let target_nonce = self.best_target_nonce?;
+
+		// if we have already selected nonces that we want to submit, do nothing
+		if race_state.nonces_to_submit.is_some() {
+			return None;
+		}
+
+		// if we already submitted some nonces, do nothing
+		if race_state.nonces_submitted.is_some() {
+			return None;
+		}
+
+		// 1) we want to deliver all nonces, starting from `target_nonce + 1`
+		// 2) we can't deliver new nonce until header, that has emitted this nonce, is finalized
+		// by target client
+		// 3) selector is used for more complicated logic
+		let best_header_at_target = &race_state.best_finalized_source_header_id_at_best_target.as_ref()?;
+		let mut nonces_end = None;
+		while let Some((queued_at, queued_range)) = self.source_queue.pop_front() {
+			// select (sub) range to deliver
+			let queued_range_begin = queued_range.begin();
+			let queued_range_end = queued_range.end();
+			let range_to_requeue = if queued_at.0 > best_header_at_target.0 {
+				// if header that has queued the range is not yet finalized at bridged chain,
+				// we can't prove anything
+				Some(queued_range)
+			} else {
+				// selector returns `Some(range)` if this `range` needs to be requeued
+				selector(queued_range)
+			};
+
+			// requeue (sub) range and update range to deliver
+			match range_to_requeue {
+				Some(range_to_requeue) => {
+					assert!(
+						range_to_requeue.begin() <= range_to_requeue.end()
+							&& range_to_requeue.begin() >= queued_range_begin
+							&& range_to_requeue.end() == queued_range_end,
+						"Incorrect implementation of internal `selector` function. Expected original\
+						range {:?} to end with returned range {:?}",
+						queued_range_begin..=queued_range_end,
+						range_to_requeue,
+					);
+
+					if range_to_requeue.begin() != queued_range_begin {
+						nonces_end = Some(range_to_requeue.begin() - 1);
+					}
+					self.source_queue.push_front((queued_at, range_to_requeue));
+					break;
+				}
+				None => {
+					nonces_end = Some(queued_range_end);
+				}
+			}
+		}
+
+		nonces_end.map(|nonces_end| RangeInclusive::new(target_nonce + 1, nonces_end))
+	}
+}
+
+impl<SourceHeaderNumber, SourceHeaderHash, TargetHeaderNumber, TargetHeaderHash, SourceNoncesRange, Proof>
+	RaceStrategy<HeaderId<SourceHeaderHash, SourceHeaderNumber>, HeaderId<TargetHeaderHash, TargetHeaderNumber>, Proof>
+	for BasicStrategy<SourceHeaderNumber, SourceHeaderHash, TargetHeaderNumber, TargetHeaderHash, SourceNoncesRange, Proof>
+where
+	SourceHeaderHash: Clone + Debug,
+	SourceHeaderNumber: Clone + Ord + Debug,
+	SourceNoncesRange: NoncesRange + Debug,
+	TargetHeaderHash: Debug,
+	TargetHeaderNumber: Debug,
+	Proof: Debug,
+{
+	type SourceNoncesRange = SourceNoncesRange;
+	type ProofParameters = ();
+	type TargetNoncesData = ();
+
+	fn is_empty(&self) -> bool {
+		self.source_queue.is_empty()
+	}
+
+	fn required_source_header_at_target(
+		&self,
+		current_best: &HeaderId<SourceHeaderHash, SourceHeaderNumber>,
+	) -> Option<HeaderId<SourceHeaderHash, SourceHeaderNumber>> {
+		self.source_queue
+			.back()
+			.and_then(|(h, _)| if h.0 > current_best.0 { Some(h.clone()) } else { None })
+	}
+
+	fn best_at_source(&self) -> Option<MessageNonce> {
+		let best_in_queue = self.source_queue.back().map(|(_, range)| range.end());
+		match (best_in_queue, self.best_target_nonce) {
+			(Some(best_in_queue), Some(best_target_nonce)) if best_in_queue > best_target_nonce => Some(best_in_queue),
+			(_, Some(best_target_nonce)) => Some(best_target_nonce),
+			(_, None) => None,
+		}
+	}
+
+	fn best_at_target(&self) -> Option<MessageNonce> {
+		self.best_target_nonce
+	}
+
+	fn source_nonces_updated(
+		&mut self,
+		at_block: HeaderId<SourceHeaderHash, SourceHeaderNumber>,
+		nonces: SourceClientNonces<SourceNoncesRange>,
+	) {
+		let best_in_queue = self
+			.source_queue
+			.back()
+			.map(|(_, range)| range.end())
+			.or(self.best_target_nonce)
+			.unwrap_or_default();
+		self.source_queue.extend(
+			nonces
+				.new_nonces
+				.greater_than(best_in_queue)
+				.into_iter()
+				.map(move |range| (at_block.clone(), range)),
+		)
+	}
+
+	fn best_target_nonces_updated(
+		&mut self,
+		nonces: TargetClientNonces<()>,
+		race_state: &mut RaceState<
+			HeaderId<SourceHeaderHash, SourceHeaderNumber>,
+			HeaderId<TargetHeaderHash, TargetHeaderNumber>,
+			Proof,
+		>,
+	) {
+		let nonce = nonces.latest_nonce;
+
+		if let Some(best_target_nonce) = self.best_target_nonce {
+			if nonce < best_target_nonce {
+				return;
+			}
+		}
+
+		while let Some(true) = self.source_queue.front().map(|(_, range)| range.begin() <= nonce) {
+			let maybe_subrange = self
+				.source_queue
+				.pop_front()
+				.and_then(|(at_block, range)| range.greater_than(nonce).map(|subrange| (at_block, subrange)));
+			if let Some((at_block, subrange)) = maybe_subrange {
+				self.source_queue.push_front((at_block, subrange));
+				break;
+			}
+		}
+
+		let need_to_select_new_nonces = race_state
+			.nonces_to_submit
+			.as_ref()
+			.map(|(_, nonces, _)| *nonces.end() <= nonce)
+			.unwrap_or(false);
+		if need_to_select_new_nonces {
+			race_state.nonces_to_submit = None;
+		}
+
+		let need_new_nonces_to_submit = race_state
+			.nonces_submitted
+			.as_ref()
+			.map(|nonces| *nonces.end() <= nonce)
+			.unwrap_or(false);
+		if need_new_nonces_to_submit {
+			race_state.nonces_submitted = None;
+		}
+
+		self.best_target_nonce = Some(std::cmp::max(
+			self.best_target_nonce.unwrap_or(nonces.latest_nonce),
+			nonce,
+		));
+	}
+
+	fn finalized_target_nonces_updated(
+		&mut self,
+		nonces: TargetClientNonces<()>,
+		_race_state: &mut RaceState<
+			HeaderId<SourceHeaderHash, SourceHeaderNumber>,
+			HeaderId<TargetHeaderHash, TargetHeaderNumber>,
+			Proof,
+		>,
+	) {
+		self.best_target_nonce = Some(std::cmp::max(
+			self.best_target_nonce.unwrap_or(nonces.latest_nonce),
+			nonces.latest_nonce,
+		));
+	}
+
+	fn select_nonces_to_deliver(
+		&mut self,
+		race_state: &RaceState<
+			HeaderId<SourceHeaderHash, SourceHeaderNumber>,
+			HeaderId<TargetHeaderHash, TargetHeaderNumber>,
+			Proof,
+		>,
+	) -> Option<(RangeInclusive<MessageNonce>, Self::ProofParameters)> {
+		self.select_nonces_to_deliver_with_selector(race_state, |_| None)
+			.map(|range| (range, ()))
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use crate::message_lane::MessageLane;
+	use crate::message_lane_loop::tests::{header_id, TestMessageLane, TestMessagesProof};
+
+	type SourceNoncesRange = RangeInclusive<MessageNonce>;
+
+	type BasicStrategy<P> = super::BasicStrategy<
+		<P as MessageLane>::SourceHeaderNumber,
+		<P as MessageLane>::SourceHeaderHash,
+		<P as MessageLane>::TargetHeaderNumber,
+		<P as MessageLane>::TargetHeaderHash,
+		SourceNoncesRange,
+		<P as MessageLane>::MessagesProof,
+	>;
+
+	fn source_nonces(new_nonces: SourceNoncesRange) -> SourceClientNonces<SourceNoncesRange> {
+		SourceClientNonces {
+			new_nonces,
+			confirmed_nonce: None,
+		}
+	}
+
+	fn target_nonces(latest_nonce: MessageNonce) -> TargetClientNonces<()> {
+		TargetClientNonces {
+			latest_nonce,
+			nonces_data: (),
+		}
+	}
+
+	#[test]
+	fn strategy_is_empty_works() {
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
+		assert_eq!(strategy.is_empty(), true);
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=1));
+		assert_eq!(strategy.is_empty(), false);
+	}
+
+	#[test]
+	fn best_at_source_is_never_lower_than_target_nonce() {
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
+		assert_eq!(strategy.best_at_source(), None);
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=5));
+		assert_eq!(strategy.best_at_source(), None);
+		strategy.best_target_nonces_updated(target_nonces(10), &mut Default::default());
+		assert_eq!(strategy.source_queue, vec![]);
+		assert_eq!(strategy.best_at_source(), Some(10));
+	}
+
+	#[test]
+	fn source_nonce_is_never_lower_than_known_target_nonce() {
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
+		strategy.best_target_nonces_updated(target_nonces(10), &mut Default::default());
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=5));
+		assert_eq!(strategy.source_queue, vec![]);
+	}
+
+	#[test]
+	fn source_nonce_is_never_lower_than_latest_known_source_nonce() {
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=5));
+		strategy.source_nonces_updated(header_id(2), source_nonces(1..=3));
+		strategy.source_nonces_updated(header_id(2), source_nonces(1..=5));
+		assert_eq!(strategy.source_queue, vec![(header_id(1), 1..=5)]);
+	}
+
+	#[test]
+	fn target_nonce_is_never_lower_than_latest_known_target_nonce() {
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
+		assert_eq!(strategy.best_target_nonce, None);
+		strategy.best_target_nonces_updated(target_nonces(10), &mut Default::default());
+		assert_eq!(strategy.best_target_nonce, Some(10));
+		strategy.best_target_nonces_updated(target_nonces(5), &mut Default::default());
+		assert_eq!(strategy.best_target_nonce, Some(10));
+	}
+
+	#[test]
+	fn updated_target_nonce_removes_queued_entries() {
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=5));
+		strategy.source_nonces_updated(header_id(2), source_nonces(6..=10));
+		strategy.source_nonces_updated(header_id(3), source_nonces(11..=15));
+		strategy.source_nonces_updated(header_id(4), source_nonces(16..=20));
+		strategy.best_target_nonces_updated(target_nonces(15), &mut Default::default());
+		assert_eq!(strategy.source_queue, vec![(header_id(4), 16..=20)]);
+		strategy.best_target_nonces_updated(target_nonces(17), &mut Default::default());
+		assert_eq!(strategy.source_queue, vec![(header_id(4), 18..=20)]);
+	}
+
+	#[test]
+	fn selected_nonces_are_dropped_on_target_nonce_update() {
+		let mut state = RaceState::default();
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
+		state.nonces_to_submit = Some((header_id(1), 5..=10, (5..=10, None)));
+		strategy.best_target_nonces_updated(target_nonces(7), &mut state);
+		assert!(state.nonces_to_submit.is_some());
+		strategy.best_target_nonces_updated(target_nonces(10), &mut state);
+		assert!(state.nonces_to_submit.is_none());
+	}
+
+	#[test]
+	fn submitted_nonces_are_dropped_on_target_nonce_update() {
+		let mut state = RaceState::default();
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
+		state.nonces_submitted = Some(5..=10);
+		strategy.best_target_nonces_updated(target_nonces(7), &mut state);
+		assert!(state.nonces_submitted.is_some());
+		strategy.best_target_nonces_updated(target_nonces(10), &mut state);
+		assert!(state.nonces_submitted.is_none());
+	}
+
+	#[test]
+	fn nothing_is_selected_if_something_is_already_selected() {
+		let mut state = RaceState::default();
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
+		state.nonces_to_submit = Some((header_id(1), 1..=10, (1..=10, None)));
+		strategy.best_target_nonces_updated(target_nonces(0), &mut state);
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=10));
+		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
+	}
+
+	#[test]
+	fn nothing_is_selected_if_something_is_already_submitted() {
+		let mut state = RaceState::default();
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
+		state.nonces_submitted = Some(1..=10);
+		strategy.best_target_nonces_updated(target_nonces(0), &mut state);
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=10));
+		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
+	}
+
+	#[test]
+	fn select_nonces_to_deliver_works() {
+		let mut state = RaceState::<_, _, TestMessagesProof>::default();
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
+		strategy.best_target_nonces_updated(target_nonces(0), &mut state);
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=1));
+		strategy.source_nonces_updated(header_id(2), source_nonces(2..=2));
+		strategy.source_nonces_updated(header_id(3), source_nonces(3..=6));
+		strategy.source_nonces_updated(header_id(5), source_nonces(7..=8));
+
+		state.best_finalized_source_header_id_at_best_target = Some(header_id(4));
+		assert_eq!(strategy.select_nonces_to_deliver(&state), Some((1..=6, ())));
+		strategy.best_target_nonces_updated(target_nonces(6), &mut state);
+		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
+
+		state.best_finalized_source_header_id_at_best_target = Some(header_id(5));
+		assert_eq!(strategy.select_nonces_to_deliver(&state), Some((7..=8, ())));
+		strategy.best_target_nonces_updated(target_nonces(8), &mut state);
+		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
+	}
+
+	#[test]
+	fn select_nonces_to_deliver_able_to_split_ranges_with_selector() {
+		let mut state = RaceState::<_, _, TestMessagesProof>::default();
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
+		strategy.best_target_nonces_updated(target_nonces(0), &mut state);
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=100));
+
+		state.best_finalized_source_header_id_at_source = Some(header_id(1));
+		state.best_finalized_source_header_id_at_best_target = Some(header_id(1));
+		state.best_target_header_id = Some(header_id(1));
+
+		assert_eq!(
+			strategy.select_nonces_to_deliver_with_selector(&state, |_| Some(50..=100)),
+			Some(1..=49),
+		);
+	}
+
+	fn run_panic_test_for_incorrect_selector(
+		invalid_selector: impl Fn(SourceNoncesRange) -> Option<SourceNoncesRange>,
+	) {
+		let mut state = RaceState::<_, _, TestMessagesProof>::default();
+		let mut strategy = BasicStrategy::<TestMessageLane>::new();
+		strategy.source_nonces_updated(header_id(1), source_nonces(1..=100));
+		strategy.best_target_nonces_updated(target_nonces(50), &mut state);
+		state.best_finalized_source_header_id_at_source = Some(header_id(1));
+		state.best_finalized_source_header_id_at_best_target = Some(header_id(1));
+		state.best_target_header_id = Some(header_id(1));
+		strategy.select_nonces_to_deliver_with_selector(&state, invalid_selector);
+	}
+
+	#[test]
+	#[should_panic]
+	fn select_nonces_to_deliver_panics_if_selector_returns_empty_range() {
+		#[allow(clippy::reversed_empty_ranges)]
+		run_panic_test_for_incorrect_selector(|_| Some(2..=1))
+	}
+
+	#[test]
+	#[should_panic]
+	fn select_nonces_to_deliver_panics_if_selector_returns_range_that_starts_before_passed_range() {
+		run_panic_test_for_incorrect_selector(|range| Some(range.begin() - 1..=*range.end()))
+	}
+
+	#[test]
+	#[should_panic]
+	fn select_nonces_to_deliver_panics_if_selector_returns_range_with_mismatched_end() {
+		run_panic_test_for_incorrect_selector(|range| Some(range.begin()..=*range.end() + 1))
+	}
+}

polkadot/bridges/relays/messages/src/metrics.rs

@@ -0,0 +1,110 @@
+// Copyright 2019-2021 Parity Technologies (UK) Ltd.
+// This file is part of Parity Bridges Common.
+
+// Parity Bridges Common is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity Bridges Common is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Metrics for message lane relay loop.
+
+use crate::message_lane::MessageLane;
+use crate::message_lane_loop::{SourceClientState, TargetClientState};
+
+use bp_messages::MessageNonce;
+use relay_utils::metrics::{metric_name, register, GaugeVec, Opts, PrometheusError, Registry, U64};
+
+/// Message lane relay metrics.
+///
+/// Cloning only clones references.
+#[derive(Clone)]
+pub struct MessageLaneLoopMetrics {
+	/// Best finalized block numbers - "source", "target", "source_at_target", "target_at_source".
+	best_block_numbers: GaugeVec<U64>,
+	/// Lane state nonces: "source_latest_generated", "source_latest_confirmed",
+	/// "target_latest_received", "target_latest_confirmed".
+	lane_state_nonces: GaugeVec<U64>,
+}
+
+impl MessageLaneLoopMetrics {
+	/// Create and register messages loop metrics.
+	pub fn new(registry: &Registry, prefix: Option<&str>) -> Result<Self, PrometheusError> {
+		Ok(MessageLaneLoopMetrics {
+			best_block_numbers: register(
+				GaugeVec::new(
+					Opts::new(
+						metric_name(prefix, "best_block_numbers"),
+						"Best finalized block numbers",
+					),
+					&["type"],
+				)?,
+				registry,
+			)?,
+			lane_state_nonces: register(
+				GaugeVec::new(
+					Opts::new(metric_name(prefix, "lane_state_nonces"), "Nonces of the lane state"),
+					&["type"],
+				)?,
+				registry,
+			)?,
+		})
+	}
+}
+
+impl MessageLaneLoopMetrics {
+	/// Update source client state metrics.
+	pub fn update_source_state<P: MessageLane>(&self, source_client_state: SourceClientState<P>) {
+		self.best_block_numbers
+			.with_label_values(&["source"])
+			.set(source_client_state.best_self.0.into());
+		self.best_block_numbers
+			.with_label_values(&["target_at_source"])
+			.set(source_client_state.best_finalized_peer_at_best_self.0.into());
+	}
+
+	/// Update target client state metrics.
+	pub fn update_target_state<P: MessageLane>(&self, target_client_state: TargetClientState<P>) {
+		self.best_block_numbers
+			.with_label_values(&["target"])
+			.set(target_client_state.best_self.0.into());
+		self.best_block_numbers
+			.with_label_values(&["source_at_target"])
+			.set(target_client_state.best_finalized_peer_at_best_self.0.into());
+	}
+
+	/// Update latest generated nonce at source.
+	pub fn update_source_latest_generated_nonce<P: MessageLane>(&self, source_latest_generated_nonce: MessageNonce) {
+		self.lane_state_nonces
+			.with_label_values(&["source_latest_generated"])
+			.set(source_latest_generated_nonce);
+	}
+
+	/// Update latest confirmed nonce at source.
+	pub fn update_source_latest_confirmed_nonce<P: MessageLane>(&self, source_latest_confirmed_nonce: MessageNonce) {
+		self.lane_state_nonces
+			.with_label_values(&["source_latest_confirmed"])
+			.set(source_latest_confirmed_nonce);
+	}
+
+	/// Update latest received nonce at target.
+	pub fn update_target_latest_received_nonce<P: MessageLane>(&self, target_latest_generated_nonce: MessageNonce) {
+		self.lane_state_nonces
+			.with_label_values(&["target_latest_received"])
+			.set(target_latest_generated_nonce);
+	}
+
+	/// Update latest confirmed nonce at target.
+	pub fn update_target_latest_confirmed_nonce<P: MessageLane>(&self, target_latest_confirmed_nonce: MessageNonce) {
+		self.lane_state_nonces
+			.with_label_values(&["target_latest_confirmed"])
+			.set(target_latest_confirmed_nonce);
+	}
+}