Relay receiving + processing confirmations (paritytech#351)

* relay receiving + processing confirmations

* fmt && clippy

* removed message processing race

* remove more traces

* generic args names

relays/ethereum/src/main.rs

@@ -31,6 +31,7 @@ mod message_lane;
 mod message_lane_loop;
 mod message_race_delivery;
 mod message_race_loop;
+mod message_race_receiving;
 mod metrics;
 mod rpc;
 mod rpc_errors;

relays/ethereum/src/message_lane.rs

@@ -17,11 +17,11 @@
 //! 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-receiving from target to source node;
-//! 3) relay proof-of-processing from target no source node.
+//! 2) relay proof-of-receiving from target to source node.
 
 use crate::utils::HeaderId;
 
+use num_traits::{One, Zero};
 use std::fmt::Debug;
 
 /// One-way message lane.
@@ -32,10 +32,20 @@ pub trait MessageLane {
 	const TARGET_NAME: &'static str;
 
 	/// Message nonce type.
-	type MessageNonce: Clone + Copy + Debug + Default + From<u32> + Ord + std::ops::Add<Output = Self::MessageNonce>;
+	type MessageNonce: Clone
+		+ Copy
+		+ Debug
+		+ Default
+		+ From<u32>
+		+ Ord
+		+ std::ops::Add<Output = Self::MessageNonce>
+		+ One
+		+ Zero;
 
 	/// Messages proof.
 	type MessagesProof: Clone;
+	/// Messages receiving proof.
+	type MessagesReceivingProof: Clone;
 
 	/// Number of the source header.
 	type SourceHeaderNumber: Clone + Debug + Default + Ord + PartialEq;

relays/ethereum/src/message_lane_loop.rs

@@ -29,6 +29,7 @@
 
 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::utils::{interval, process_future_result, retry_backoff, FailedClient, MaybeConnectionError};
 
 use async_trait::async_trait;
@@ -52,13 +53,25 @@ pub trait SourceClient<P: MessageLane>: Clone {
 		&self,
 		id: SourceHeaderIdOf<P>,
 	) -> Result<(SourceHeaderIdOf<P>, 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>, P::MessageNonce), Self::Error>;
 
 	/// Prove messages in inclusive range [begin; end].
 	async fn prove_messages(
 		&self,
 		id: SourceHeaderIdOf<P>,
 		nonces: RangeInclusive<P::MessageNonce>,
 	) -> Result<(SourceHeaderIdOf<P>, RangeInclusive<P::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<RangeInclusive<P::MessageNonce>, Self::Error>;
 }
 
 /// Target client trait.
@@ -73,12 +86,18 @@ pub trait TargetClient<P: MessageLane>: Clone {
 	/// Returns state of the client.
 	async fn state(&self) -> Result<TargetClientState<P>, Self::Error>;
 
-	/// Get nonce of latest message, which receival has been confirmed.
+	/// Get nonce of latest received message.
 	async fn latest_received_nonce(
 		&self,
 		id: TargetHeaderIdOf<P>,
 	) -> Result<(TargetHeaderIdOf<P>, P::MessageNonce), 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,
@@ -196,6 +215,19 @@ async fn run_until_connection_lost<P: MessageLane, SC: SourceClient<P>, TC: Targ
 	)
 	.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,
+		stall_timeout,
+	)
+	.fuse();
+
 	let exit_signal = exit_signal.fuse();
 
 	futures::pin_mut!(
@@ -206,6 +238,7 @@ async fn run_until_connection_lost<P: MessageLane, SC: SourceClient<P>, TC: Targ
 		target_go_offline_future,
 		target_tick_stream,
 		delivery_race_loop,
+		receiving_race_loop,
 		exit_signal
 	);
 
@@ -224,7 +257,8 @@ async fn run_until_connection_lost<P: MessageLane, SC: SourceClient<P>, TC: Targ
 							P::SOURCE_NAME,
 							new_source_state,
 						);
-						let _ = delivery_source_state_sender.unbounded_send(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());
 					},
 					&mut source_go_offline_future,
 					|delay| async_std::task::sleep(delay),
@@ -250,7 +284,8 @@ async fn run_until_connection_lost<P: MessageLane, SC: SourceClient<P>, TC: Targ
 							P::TARGET_NAME,
 							new_target_state,
 						);
-						let _ = delivery_target_state_sender.unbounded_send(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());
 					},
 					&mut target_go_offline_future,
 					|delay| async_std::task::sleep(delay),
@@ -270,6 +305,12 @@ async fn run_until_connection_lost<P: MessageLane, SC: SourceClient<P>, TC: Targ
 					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(());
@@ -304,6 +345,7 @@ pub(crate) mod tests {
 
 	pub type TestMessageNonce = u64;
 	pub type TestMessagesProof = RangeInclusive<TestMessageNonce>;
+	pub type TestMessagesReceivingProof = TestMessageNonce;
 
 	pub type TestSourceHeaderNumber = u64;
 	pub type TestSourceHeaderHash = u64;
@@ -333,7 +375,9 @@ pub(crate) mod tests {
 		const TARGET_NAME: &'static str = "TestTarget";
 
 		type MessageNonce = TestMessageNonce;
+
 		type MessagesProof = TestMessagesProof;
+		type MessagesReceivingProof = TestMessagesReceivingProof;
 
 		type SourceHeaderNumber = TestSourceHeaderNumber;
 		type SourceHeaderHash = TestSourceHeaderHash;
@@ -348,6 +392,8 @@ pub(crate) mod tests {
 		is_source_reconnected: bool,
 		source_state: SourceClientState<TestMessageLane>,
 		source_latest_generated_nonce: TestMessageNonce,
+		source_latest_confirmed_received_nonce: TestMessageNonce,
+		submitted_messages_receiving_proofs: Vec<TestMessagesReceivingProof>,
 		is_target_fails: bool,
 		is_target_reconnected: bool,
 		target_state: SourceClientState<TestMessageLane>,
@@ -383,7 +429,6 @@ pub(crate) mod tests {
 			Ok(data.source_state.clone())
 		}
 
-		/// Get nonce of instance of latest generated message.
 		async fn latest_generated_nonce(
 			&self,
 			id: SourceHeaderIdOf<TestMessageLane>,
@@ -396,6 +441,15 @@ pub(crate) mod tests {
 			Ok((id, data.source_latest_generated_nonce))
 		}
 
+		async fn latest_confirmed_received_nonce(
+			&self,
+			id: SourceHeaderIdOf<TestMessageLane>,
+		) -> Result<(SourceHeaderIdOf<TestMessageLane>, TestMessageNonce), Self::Error> {
+			let mut data = self.data.lock();
+			(self.tick)(&mut *data);
+			Ok((id, data.source_latest_confirmed_received_nonce))
+		}
+
 		async fn prove_messages(
 			&self,
 			id: SourceHeaderIdOf<TestMessageLane>,
@@ -410,6 +464,18 @@ pub(crate) mod tests {
 		> {
 			Ok((id, nonces.clone(), nonces))
 		}
+
+		async fn submit_messages_receiving_proof(
+			&self,
+			_generated_at_block: TargetHeaderIdOf<TestMessageLane>,
+			proof: TestMessagesReceivingProof,
+		) -> Result<RangeInclusive<TestMessageNonce>, Self::Error> {
+			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(proof..=proof)
+		}
 	}
 
 	#[derive(Clone)]
@@ -452,7 +518,13 @@ pub(crate) mod tests {
 			Ok((id, data.target_latest_received_nonce))
 		}
 
-		/// Submit messages proof.
+		async fn prove_messages_receiving(
+			&self,
+			id: TargetHeaderIdOf<TestMessageLane>,
+		) -> Result<(TargetHeaderIdOf<TestMessageLane>, TestMessagesReceivingProof), Self::Error> {
+			Ok((id, self.data.lock().target_latest_received_nonce))
+		}
+
 		async fn submit_messages_proof(
 			&self,
 			_generated_at_header: SourceHeaderIdOf<TestMessageLane>,
@@ -570,22 +642,29 @@ pub(crate) mod tests {
 			},
 			Arc::new(|_: &mut TestClientData| {}),
 			Arc::new(move |data: &mut TestClientData| {
-				if data.target_state.best_peer.0 < 10 {
+				// syncing source headers -> target chain (by one)
+				if data.target_state.best_peer.0 < data.source_state.best_self.0 {
 					data.target_state.best_peer =
 						HeaderId(data.target_state.best_peer.0 + 1, data.target_state.best_peer.0 + 1);
 				}
-				if data
-					.submitted_messages_proofs
-					.last()
-					.map(|last| *last.end() == 10)
-					.unwrap_or(false)
-				{
+				// syncing source headers -> target chain (all at once)
+				if data.source_state.best_peer.0 < data.target_state.best_self.0 {
+					data.source_state.best_peer = data.target_state.best_self;
+				}
+				// if target has received all messages => increase target block so that confirmations may be sent
+				if data.target_latest_received_nonce == 10 {
+					data.target_state.best_self =
+						HeaderId(data.source_state.best_self.0 + 1, data.source_state.best_self.0 + 1);
+				}
+				// if source has received all messages receiving confirmations => increase source block so that confirmations may be sent
+				if data.source_latest_confirmed_received_nonce == 10 {
 					exit_sender.unbounded_send(()).unwrap();
 				}
 			}),
 			exit_receiver.into_future().map(|(_, _)| ()),
 		);
 
 		assert_eq!(result.submitted_messages_proofs, vec![1..=4, 5..=8, 9..=10],);
+		assert!(!result.submitted_messages_receiving_proofs.is_empty());
 	}
 }