submit lane unblock transactions from relay (paritytech#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,22 +622,38 @@ 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;
 			}
 			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,