use transaction tracker in messages relay (#1581)

relays/client-substrate/src/lib.rs

@@ -88,30 +88,3 @@ pub fn transaction_stall_timeout(
 		.map(|mortality_period| average_block_interval.saturating_mul(mortality_period + 1 + 1))
 		.unwrap_or(default_stall_timeout)
 }
-
-/// Returns stall timeout for relay loop that submit transactions to two chains.
-///
-/// Bidirectional relay may have two active transactions. Even if one of them has been spoiled, we
-/// can't just restart the loop - the other transaction may still be alive and we'll be submitting
-/// duplicate transaction, which may result in funds loss. So we'll be selecting maximal mortality
-/// for choosing loop stall timeout.
-pub fn bidirectional_transaction_stall_timeout(
-	left_mortality_period: Option<u32>,
-	right_mortality_period: Option<u32>,
-	left_average_block_interval: Duration,
-	right_average_block_interval: Duration,
-	default_stall_timeout: Duration,
-) -> Duration {
-	std::cmp::max(
-		transaction_stall_timeout(
-			left_mortality_period,
-			left_average_block_interval,
-			default_stall_timeout,
-		),
-		transaction_stall_timeout(
-			right_mortality_period,
-			right_average_block_interval,
-			default_stall_timeout,
-		),
-	)
-}

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

@@ -164,13 +164,6 @@ where
 {
 	let source_client = params.source_client;
 	let target_client = params.target_client;
-	let stall_timeout = relay_substrate_client::bidirectional_transaction_stall_timeout(
-		params.source_transaction_params.mortality,
-		params.target_transaction_params.mortality,
-		P::SourceChain::AVERAGE_BLOCK_INTERVAL,
-		P::TargetChain::AVERAGE_BLOCK_INTERVAL,
-		STALL_TIMEOUT,
-	);
 	let relayer_id_at_source: AccountIdOf<P::SourceChain> =
 		params.source_transaction_params.signer.public().into();
 
@@ -202,8 +195,7 @@ where
 			Max messages in single transaction: {}\n\t\
 			Max messages size in single transaction: {}\n\t\
 			Max messages weight in single transaction: {}\n\t\
-			Tx mortality: {:?} (~{}m)/{:?} (~{}m)\n\t\
-			Stall timeout: {:?}",
+			Tx mortality: {:?} (~{}m)/{:?} (~{}m)",
 		P::SourceChain::NAME,
 		P::TargetChain::NAME,
 		P::SourceChain::NAME,
@@ -223,7 +215,6 @@ where
 			P::TargetChain::AVERAGE_BLOCK_INTERVAL,
 			STALL_TIMEOUT,
 		).as_secs_f64() / 60.0f64,
-		stall_timeout,
 	);
 
 	messages_relay::message_lane_loop::run(
@@ -232,7 +223,6 @@ where
 			source_tick: P::SourceChain::AVERAGE_BLOCK_INTERVAL,
 			target_tick: P::TargetChain::AVERAGE_BLOCK_INTERVAL,
 			reconnect_delay: relay_utils::relay_loop::RECONNECT_DELAY,
-			stall_timeout,
 			delivery_params: messages_relay::message_lane_loop::MessageDeliveryParams {
 				max_unrewarded_relayer_entries_at_target:
 					P::SourceChain::MAX_UNREWARDED_RELAYERS_IN_CONFIRMATION_TX,

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

@@ -51,7 +51,7 @@ use num_traits::{Bounded, Zero};
 use relay_substrate_client::{
 	AccountIdOf, AccountKeyPairOf, BalanceOf, BlockNumberOf, Chain, ChainWithMessages, Client,
 	Error as SubstrateError, HashOf, HeaderIdOf, IndexOf, SignParam, TransactionEra,
-	TransactionSignScheme, UnsignedTransaction,
+	TransactionSignScheme, TransactionTracker, UnsignedTransaction,
 };
 use relay_utils::{relay_loop::Client as RelayClient, HeaderId};
 use sp_core::{Bytes, Pair};
@@ -144,6 +144,8 @@ where
 		From<<AccountKeyPairOf<P::SourceTransactionSignScheme> as Pair>::Public>,
 	P::SourceTransactionSignScheme: TransactionSignScheme<Chain = P::SourceChain>,
 {
+	type TransactionTracker = TransactionTracker<P::SourceChain>;
+
 	async fn state(&self) -> Result<SourceClientState<MessageLaneAdapter<P>>, SubstrateError> {
 		// we can't continue to deliver confirmations if source node is out of sync, because
 		// it may have already received confirmations that we're going to deliver
@@ -338,13 +340,13 @@ where
 		&self,
 		_generated_at_block: TargetHeaderIdOf<MessageLaneAdapter<P>>,
 		proof: <MessageLaneAdapter<P> as MessageLane>::MessagesReceivingProof,
-	) -> Result<(), SubstrateError> {
+	) -> Result<Self::TransactionTracker, SubstrateError> {
 		let genesis_hash = *self.source_client.genesis_hash();
 		let transaction_params = self.transaction_params.clone();
 		let (spec_version, transaction_version) =
 			self.source_client.simple_runtime_version().await?;
 		self.source_client
-			.submit_signed_extrinsic(
+			.submit_and_watch_signed_extrinsic(
 				self.transaction_params.signer.public().into(),
 				SignParam::<P::SourceTransactionSignScheme> {
 					spec_version,
@@ -362,8 +364,7 @@ where
 					)
 				},
 			)
-			.await?;
-		Ok(())
+			.await
 	}
 
 	async fn require_target_header_on_source(&self, id: TargetHeaderIdOf<MessageLaneAdapter<P>>) {

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

@@ -39,13 +39,13 @@ use codec::Encode;
 use frame_support::weights::{Weight, WeightToFee};
 use messages_relay::{
 	message_lane::{MessageLane, SourceHeaderIdOf, TargetHeaderIdOf},
-	message_lane_loop::{TargetClient, TargetClientState},
+	message_lane_loop::{NoncesSubmitArtifacts, TargetClient, TargetClientState},
 };
 use num_traits::{Bounded, Zero};
 use relay_substrate_client::{
 	AccountIdOf, AccountKeyPairOf, BalanceOf, BlockNumberOf, Chain, ChainWithMessages, Client,
 	Error as SubstrateError, HashOf, HeaderIdOf, IndexOf, SignParam, TransactionEra,
-	TransactionSignScheme, UnsignedTransaction, WeightToFeeOf,
+	TransactionSignScheme, TransactionTracker, UnsignedTransaction, WeightToFeeOf,
 };
 use relay_utils::{relay_loop::Client as RelayClient, HeaderId};
 use sp_core::{Bytes, Pair};
@@ -145,6 +145,8 @@ where
 	P::TargetTransactionSignScheme: TransactionSignScheme<Chain = P::TargetChain>,
 	BalanceOf<P::SourceChain>: TryFrom<BalanceOf<P::TargetChain>>,
 {
+	type TransactionTracker = TransactionTracker<P::TargetChain>;
+
 	async fn state(&self) -> Result<TargetClientState<MessageLaneAdapter<P>>, SubstrateError> {
 		// we can't continue to deliver confirmations if source node is out of sync, because
 		// it may have already received confirmations that we're going to deliver
@@ -245,15 +247,16 @@ where
 		_generated_at_header: SourceHeaderIdOf<MessageLaneAdapter<P>>,
 		nonces: RangeInclusive<MessageNonce>,
 		proof: <MessageLaneAdapter<P> as MessageLane>::MessagesProof,
-	) -> Result<RangeInclusive<MessageNonce>, SubstrateError> {
+	) -> Result<NoncesSubmitArtifacts<Self::TransactionTracker>, SubstrateError> {
 		let genesis_hash = *self.target_client.genesis_hash();
 		let transaction_params = self.transaction_params.clone();
 		let relayer_id_at_source = self.relayer_id_at_source.clone();
 		let nonces_clone = nonces.clone();
 		let (spec_version, transaction_version) =
 			self.target_client.simple_runtime_version().await?;
-		self.target_client
-			.submit_signed_extrinsic(
+		let tx_tracker = self
+			.target_client
+			.submit_and_watch_signed_extrinsic(
 				self.transaction_params.signer.public().into(),
 				SignParam::<P::TargetTransactionSignScheme> {
 					spec_version,
@@ -274,7 +277,7 @@ where
 				},
 			)
 			.await?;
-		Ok(nonces)
+		Ok(NoncesSubmitArtifacts { nonces, tx_tracker })
 	}
 
 	async fn require_source_header_on_target(&self, id: SourceHeaderIdOf<MessageLaneAdapter<P>>) {