submit lane unblock transactions from relay (#2030)

* submit lane unblock transactions from relay

* moved body of select_nonces_to_deliver to the separate select_race_action

* extracted latest_confirmed_nonce_at_source method

* return Option<RaceAction> from select_race_action

* make required_source_header_at_target async

* remove extra argument from required_source_header_at_target

* small fixes in tests

* Revert "return Option<RaceAction> from select_race_action"

This reverts commit 9f13dbfae39a5a45564550e8c89b10a524a68729.

* implement required_source_header_at_target using what-if approach

* fix compilation

* fmt

* clippy

* moved some code to the can_submit_transaction_with

bridges/relays/client-substrate/src/chain.rs

@@ -55,7 +55,7 @@ pub trait Chain: ChainBase + Clone {
 	/// Block type.
 	type SignedBlock: Member + Serialize + DeserializeOwned + BlockWithJustification<Self::Header>;
 	/// The aggregated `Call` type.
-	type Call: Clone + Codec + Debug + Send;
+	type Call: Clone + Codec + Debug + Send + Sync;
 }
 
 /// Substrate-based relay chain that supports parachains.

bridges/relays/lib-substrate-relay/src/lib.rs

@@ -91,7 +91,7 @@ impl<AccountId> TaggedAccount<AccountId> {
 }
 
 /// Batch call builder.
-pub trait BatchCallBuilder<Call>: Clone + Send {
+pub trait BatchCallBuilder<Call>: Clone + Send + Sync {
 	/// Create batch call from given calls vector.
 	fn build_batch_call(&self, _calls: Vec<Call>) -> Call;
 }

bridges/relays/messages/Cargo.toml

@@ -8,6 +8,7 @@ license = "GPL-3.0-or-later WITH Classpath-exception-2.0"
 [dependencies]
 async-std = { version = "1.6.5", features = ["attributes"] }
 async-trait = "0.1"
+env_logger = "0.10"
 futures = "0.3.28"
 hex = "0.4"
 log = "0.4.17"

bridges/relays/messages/src/message_lane_loop.rs

@@ -111,7 +111,7 @@ pub struct NoncesSubmitArtifacts<T> {
 
 /// Batch transaction that already submit some headers and needs to be extended with
 /// messages/delivery proof before sending.
-pub trait BatchTransaction<HeaderId>: Debug + Send {
+pub trait BatchTransaction<HeaderId>: Debug + Send + Sync {
 	/// Header that was required in the original call and which is bundled within this
 	/// batch transaction.
 	fn required_header_id(&self) -> HeaderId;
@@ -622,11 +622,19 @@ pub(crate) mod tests {
 	}
 
 	impl TestClientData {
-		fn receive_messages(&mut self, proof: TestMessagesProof) {
+		fn receive_messages(
+			&mut self,
+			maybe_batch_tx: Option<TestMessagesBatchTransaction>,
+			proof: TestMessagesProof,
+		) {
 			self.target_state.best_self =
 				HeaderId(self.target_state.best_self.0 + 1, self.target_state.best_self.1 + 1);
 			self.target_state.best_finalized_self = self.target_state.best_self;
 			self.target_latest_received_nonce = *proof.0.end();
+			if let Some(maybe_batch_tx) = maybe_batch_tx {
+				self.target_state.best_finalized_peer_at_best_self =
+					Some(maybe_batch_tx.required_header_id());
+			}
 			if let Some(target_latest_confirmed_received_nonce) = proof.1 {
 				self.target_latest_confirmed_received_nonce =
 					target_latest_confirmed_received_nonce;
@@ -634,10 +642,18 @@ pub(crate) mod tests {
 			self.submitted_messages_proofs.push(proof);
 		}
 
-		fn receive_messages_delivery_proof(&mut self, proof: TestMessagesReceivingProof) {
+		fn receive_messages_delivery_proof(
+			&mut self,
+			maybe_batch_tx: Option<TestConfirmationBatchTransaction>,
+			proof: TestMessagesReceivingProof,
+		) {
 			self.source_state.best_self =
 				HeaderId(self.source_state.best_self.0 + 1, self.source_state.best_self.1 + 1);
 			self.source_state.best_finalized_self = self.source_state.best_self;
+			if let Some(maybe_batch_tx) = maybe_batch_tx {
+				self.source_state.best_finalized_peer_at_best_self =
+					Some(maybe_batch_tx.required_header_id());
+			}
 			self.submitted_messages_receiving_proofs.push(proof);
 			self.source_latest_confirmed_received_nonce = proof;
 		}
@@ -760,13 +776,13 @@ pub(crate) mod tests {
 
 		async fn submit_messages_receiving_proof(
 			&self,
-			_maybe_batch_tx: Option<Self::BatchTransaction>,
+			maybe_batch_tx: Option<Self::BatchTransaction>,
 			_generated_at_block: TargetHeaderIdOf<TestMessageLane>,
 			proof: TestMessagesReceivingProof,
 		) -> Result<Self::TransactionTracker, TestError> {
 			let mut data = self.data.lock();
 			(self.tick)(&mut data);
-			data.receive_messages_delivery_proof(proof);
+			data.receive_messages_delivery_proof(maybe_batch_tx, proof);
 			(self.post_tick)(&mut data);
 			Ok(TestTransactionTracker(data.source_tracked_transaction_status))
 		}
@@ -885,7 +901,7 @@ pub(crate) mod tests {
 
 		async fn submit_messages_proof(
 			&self,
-			_maybe_batch_tx: Option<Self::BatchTransaction>,
+			maybe_batch_tx: Option<Self::BatchTransaction>,
 			_generated_at_header: SourceHeaderIdOf<TestMessageLane>,
 			nonces: RangeInclusive<MessageNonce>,
 			proof: TestMessagesProof,
@@ -895,7 +911,7 @@ pub(crate) mod tests {
 			if data.is_target_fails {
 				return Err(TestError)
 			}
-			data.receive_messages(proof);
+			data.receive_messages(maybe_batch_tx, proof);
 			(self.post_tick)(&mut data);
 			Ok(NoncesSubmitArtifacts {
 				nonces,

bridges/relays/messages/src/message_race_loop.rs

@@ -41,14 +41,14 @@ use std::{
 /// One of races within lane.
 pub trait MessageRace {
 	/// Header id of the race source.
-	type SourceHeaderId: Debug + Clone + PartialEq + Send;
+	type SourceHeaderId: Debug + Clone + PartialEq + Send + Sync;
 	/// Header id of the race source.
-	type TargetHeaderId: Debug + Clone + PartialEq + Send;
+	type TargetHeaderId: Debug + Clone + PartialEq + Send + Sync;
 
 	/// Message nonce used in the race.
 	type MessageNonce: Debug + Clone;
 	/// Proof that is generated and delivered in this race.
-	type Proof: Debug + Clone + Send;
+	type Proof: Debug + Clone + Send + Sync;
 
 	/// Name of the race source.
 	fn source_name() -> String;
@@ -175,9 +175,8 @@ pub trait RaceStrategy<SourceHeaderId, TargetHeaderId, Proof>: Debug {
 	/// 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<RS: RaceState<SourceHeaderId, TargetHeaderId>>(
+	async fn required_source_header_at_target<RS: RaceState<SourceHeaderId, TargetHeaderId>>(
 		&self,
-		current_best: &SourceHeaderId,
 		race_state: RS,
 	) -> Option<SourceHeaderId>;
 	/// Return the best nonce at source node.
@@ -218,7 +217,11 @@ pub trait RaceStrategy<SourceHeaderId, TargetHeaderId, Proof>: Debug {
 }
 
 /// State of the race.
-pub trait RaceState<SourceHeaderId, TargetHeaderId>: Send {
+pub trait RaceState<SourceHeaderId, TargetHeaderId>: Clone + Send + Sync {
+	/// Set best finalized source header id at the best block on the target
+	/// client (at the `best_finalized_source_header_id_at_best_target`).
+	fn set_best_finalized_source_header_id_at_best_target(&mut self, id: SourceHeaderId);
+
 	/// Best finalized source header id at the source client.
 	fn best_finalized_source_header_id_at_source(&self) -> Option<SourceHeaderId>;
 	/// Best finalized source header id at the best block on the target
@@ -281,11 +284,15 @@ impl<SourceHeaderId, TargetHeaderId, Proof, BatchTx> Default
 impl<SourceHeaderId, TargetHeaderId, Proof, BatchTx> RaceState<SourceHeaderId, TargetHeaderId>
 	for RaceStateImpl<SourceHeaderId, TargetHeaderId, Proof, BatchTx>
 where
-	SourceHeaderId: Clone + Send,
-	TargetHeaderId: Clone + Send,
-	Proof: Clone + Send,
-	BatchTx: Clone + Send,
+	SourceHeaderId: Clone + Send + Sync,
+	TargetHeaderId: Clone + Send + Sync,
+	Proof: Clone + Send + Sync,
+	BatchTx: Clone + Send + Sync,
 {
+	fn set_best_finalized_source_header_id_at_best_target(&mut self, id: SourceHeaderId) {
+		self.best_finalized_source_header_id_at_best_target = Some(id);
+	}
+
 	fn best_finalized_source_header_id_at_source(&self) -> Option<SourceHeaderId> {
 		self.best_finalized_source_header_id_at_source.clone()
 	}
@@ -430,10 +437,9 @@ pub async fn run<P: MessageRace, SC: SourceClient<P>, TC: TargetClient<P>>(
 				).fail_if_connection_error(FailedClient::Source)?;
 
 				// ask for more headers if we have nonces to deliver and required headers are missing
-				source_required_header = race_state
-					.best_finalized_source_header_id_at_best_target
-					.as_ref()
-					.and_then(|best| strategy.required_source_header_at_target(best, race_state.clone()));
+				source_required_header = strategy
+					.required_source_header_at_target(race_state.clone())
+					.await;
 			},
 			nonces = target_best_nonces => {
 				target_best_nonces_required = false;

bridges/relays/messages/src/message_race_strategy.rs

@@ -205,16 +205,16 @@ impl<
 		self.source_queue.is_empty()
 	}
 
-	fn required_source_header_at_target<
+	async fn required_source_header_at_target<
 		RS: RaceState<
 			HeaderId<SourceHeaderHash, SourceHeaderNumber>,
 			HeaderId<TargetHeaderHash, TargetHeaderNumber>,
 		>,
 	>(
 		&self,
-		current_best: &HeaderId<SourceHeaderHash, SourceHeaderNumber>,
-		_race_state: RS,
+		race_state: RS,
 	) -> Option<HeaderId<SourceHeaderHash, SourceHeaderNumber>> {
+		let current_best = race_state.best_finalized_source_header_id_at_best_target()?;
 		self.source_queue
 			.back()
 			.and_then(|(h, _)| if h.0 > current_best.0 { Some(h.clone()) } else { None })