Bridge: add subcommand to relay messages delivery confirmation

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

@@ -26,9 +26,12 @@ use async_trait::async_trait;
 use sp_core::Pair;
 use structopt::StructOpt;
 
+use bp_messages::MessageNonce;
+use bp_runtime::HeaderIdProvider;
 use relay_substrate_client::{
-	AccountIdOf, AccountKeyPairOf, BalanceOf, ChainWithRuntimeVersion, ChainWithTransactions,
+	AccountIdOf, AccountKeyPairOf, BalanceOf, Chain, ChainWithRuntimeVersion, ChainWithTransactions,
 };
+use relay_utils::UniqueSaturatedInto;
 
 /// Messages relaying params.
 #[derive(StructOpt)]
@@ -48,6 +51,53 @@ pub struct RelayMessagesParams {
 	prometheus_params: PrometheusParams,
 }
 
+/// Messages range relaying params.
+#[derive(StructOpt)]
+pub struct RelayMessagesRangeParams {
+	/// Number of the source chain header that we will use to prepare a messages proof.
+	/// This header must be previously proved to the target chain.
+	#[structopt(long)]
+	at_source_block: u128,
+	/// Hex-encoded lane id that should be served by the relay. Defaults to `00000000`.
+	#[structopt(long, default_value = "00000000")]
+	lane: HexLaneId,
+	/// Nonce (inclusive) of the first message to relay.
+	#[structopt(long)]
+	messages_start: MessageNonce,
+	/// Nonce (inclusive) of the last message to relay.
+	#[structopt(long)]
+	messages_end: MessageNonce,
+	/// Whether the outbound lane state proof should be included into transaction.
+	#[structopt(long)]
+	outbound_state_proof_required: bool,
+	#[structopt(flatten)]
+	source: SourceConnectionParams,
+	#[structopt(flatten)]
+	source_sign: SourceSigningParams,
+	#[structopt(flatten)]
+	target: TargetConnectionParams,
+	#[structopt(flatten)]
+	target_sign: TargetSigningParams,
+}
+
+/// Messages delivery confirmation relaying params.
+#[derive(StructOpt)]
+pub struct RelayMessagesDeliveryConfirmationParams {
+	/// Number of the target chain header that we will use to prepare a messages
+	/// delivery proof. This header must be previously proved to the source chain.
+	#[structopt(long)]
+	at_target_block: u128,
+	/// Hex-encoded lane id that should be served by the relay. Defaults to `00000000`.
+	#[structopt(long, default_value = "00000000")]
+	lane: HexLaneId,
+	#[structopt(flatten)]
+	source: SourceConnectionParams,
+	#[structopt(flatten)]
+	source_sign: SourceSigningParams,
+	#[structopt(flatten)]
+	target: TargetConnectionParams,
+}
+
 /// Trait used for relaying messages between 2 chains.
 #[async_trait]
 pub trait MessagesRelayer: MessagesCliBridge
@@ -86,4 +136,73 @@ where
 		.await
 		.map_err(|e| anyhow::format_err!("{}", e))
 	}
+
+	/// Relay a consequitive range of messages.
+	async fn relay_messages_range(data: RelayMessagesRangeParams) -> anyhow::Result<()> {
+		let source_client = data.source.into_client::<Self::Source>().await?;
+		let target_client = data.target.into_client::<Self::Target>().await?;
+		let source_sign = data.source_sign.to_keypair::<Self::Source>()?;
+		let source_transactions_mortality = data.source_sign.transactions_mortality()?;
+		let target_sign = data.target_sign.to_keypair::<Self::Target>()?;
+		let target_transactions_mortality = data.target_sign.transactions_mortality()?;
+
+		let at_source_block = source_client
+			.header_by_number(data.at_source_block.unique_saturated_into())
+			.await
+			.map_err(|e| {
+				log::trace!(
+					target: "bridge",
+					"Failed to read {} header with number {}: {e:?}",
+					Self::Source::NAME,
+					data.at_source_block,
+				);
+				anyhow::format_err!("The command has failed")
+			})?
+			.id();
+
+		crate::messages_lane::relay_messages_range::<Self::MessagesLane>(
+			source_client,
+			target_client,
+			TransactionParams { signer: source_sign, mortality: source_transactions_mortality },
+			TransactionParams { signer: target_sign, mortality: target_transactions_mortality },
+			at_source_block,
+			data.lane.into(),
+			data.messages_start..=data.messages_end,
+			data.outbound_state_proof_required,
+		)
+		.await
+	}
+
+	/// Relay a messages delivery confirmation.
+	async fn relay_messages_delivery_confirmation(
+		data: RelayMessagesDeliveryConfirmationParams,
+	) -> anyhow::Result<()> {
+		let source_client = data.source.into_client::<Self::Source>().await?;
+		let target_client = data.target.into_client::<Self::Target>().await?;
+		let source_sign = data.source_sign.to_keypair::<Self::Source>()?;
+		let source_transactions_mortality = data.source_sign.transactions_mortality()?;
+
+		let at_target_block = target_client
+			.header_by_number(data.at_target_block.unique_saturated_into())
+			.await
+			.map_err(|e| {
+				log::trace!(
+					target: "bridge",
+					"Failed to read {} header with number {}: {e:?}",
+					Self::Target::NAME,
+					data.at_target_block,
+				);
+				anyhow::format_err!("The command has failed")
+			})?
+			.id();
+
+		crate::messages_lane::relay_messages_delivery_confirmation::<Self::MessagesLane>(
+			source_client,
+			target_client,
+			TransactionParams { signer: source_sign, mortality: source_transactions_mortality },
+			at_target_block,
+			data.lane.into(),
+		)
+		.await
+	}
 }

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

@@ -46,7 +46,7 @@ use relay_utils::{
 };
 use sp_core::Pair;
 use sp_runtime::traits::Zero;
-use std::{fmt::Debug, marker::PhantomData};
+use std::{fmt::Debug, marker::PhantomData, ops::RangeInclusive};
 
 /// Substrate -> Substrate messages synchronization pipeline.
 pub trait SubstrateMessageLane: 'static + Clone + Debug + Send + Sync {
@@ -262,7 +262,7 @@ where
 			source_client,
 			params.lane_id,
 			relayer_id_at_source,
-			params.target_transaction_params,
+			Some(params.target_transaction_params),
 			params.source_to_target_headers_relay,
 		),
 		{
@@ -275,6 +275,87 @@ where
 	.map_err(Into::into)
 }
 
+/// Deliver range of Substrate-to-Substrate messages. No checks are made to ensure that transaction
+/// will succeed.
+pub async fn relay_messages_range<P: SubstrateMessageLane>(
+	source_client: Client<P::SourceChain>,
+	target_client: Client<P::TargetChain>,
+	source_transaction_params: TransactionParams<AccountKeyPairOf<P::SourceChain>>,
+	target_transaction_params: TransactionParams<AccountKeyPairOf<P::TargetChain>>,
+	at_source_block: HeaderIdOf<P::SourceChain>,
+	lane_id: LaneId,
+	range: RangeInclusive<MessageNonce>,
+	outbound_state_proof_required: bool,
+) -> anyhow::Result<()>
+where
+	AccountIdOf<P::SourceChain>: From<<AccountKeyPairOf<P::SourceChain> as Pair>::Public>,
+	AccountIdOf<P::TargetChain>: From<<AccountKeyPairOf<P::TargetChain> as Pair>::Public>,
+	BalanceOf<P::SourceChain>: TryFrom<BalanceOf<P::TargetChain>>,
+{
+	let relayer_id_at_source: AccountIdOf<P::SourceChain> =
+		source_transaction_params.signer.public().into();
+	messages_relay::relay_messages_range(
+		SubstrateMessagesSource::<P>::new(
+			source_client.clone(),
+			target_client.clone(),
+			lane_id,
+			source_transaction_params,
+			None,
+		),
+		SubstrateMessagesTarget::<P>::new(
+			target_client,
+			source_client,
+			lane_id,
+			relayer_id_at_source,
+			Some(target_transaction_params),
+			None,
+		),
+		at_source_block,
+		range,
+		outbound_state_proof_required,
+	)
+	.await
+	.map_err(|_| anyhow::format_err!("The command has failed"))
+}
+
+/// Relay messages delivery confirmation of Substrate-to-Substrate messages.
+/// No checks are made to ensure that transaction will succeed.
+pub async fn relay_messages_delivery_confirmation<P: SubstrateMessageLane>(
+	source_client: Client<P::SourceChain>,
+	target_client: Client<P::TargetChain>,
+	source_transaction_params: TransactionParams<AccountKeyPairOf<P::SourceChain>>,
+	at_target_block: HeaderIdOf<P::TargetChain>,
+	lane_id: LaneId,
+) -> anyhow::Result<()>
+where
+	AccountIdOf<P::SourceChain>: From<<AccountKeyPairOf<P::SourceChain> as Pair>::Public>,
+	AccountIdOf<P::TargetChain>: From<<AccountKeyPairOf<P::TargetChain> as Pair>::Public>,
+	BalanceOf<P::SourceChain>: TryFrom<BalanceOf<P::TargetChain>>,
+{
+	let relayer_id_at_source: AccountIdOf<P::SourceChain> =
+		source_transaction_params.signer.public().into();
+	messages_relay::relay_messages_delivery_confirmation(
+		SubstrateMessagesSource::<P>::new(
+			source_client.clone(),
+			target_client.clone(),
+			lane_id,
+			source_transaction_params,
+			None,
+		),
+		SubstrateMessagesTarget::<P>::new(
+			target_client,
+			source_client,
+			lane_id,
+			relayer_id_at_source,
+			None,
+			None,
+		),
+		at_target_block,
+	)
+	.await
+	.map_err(|_| anyhow::format_err!("The command has failed"))
+}
+
 /// Different ways of building `receive_messages_proof` calls.
 pub trait ReceiveMessagesProofCallBuilder<P: SubstrateMessageLane> {
 	/// Given messages proof, build call of `receive_messages_proof` function of bridge

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

@@ -40,8 +40,8 @@ use messages_relay::{
 	message_lane_loop::{NoncesSubmitArtifacts, TargetClient, TargetClientState},
 };
 use relay_substrate_client::{
-	AccountIdOf, AccountKeyPairOf, BalanceOf, CallOf, Client, Error as SubstrateError, HashOf,
-	TransactionEra, TransactionTracker, UnsignedTransaction,
+	AccountIdOf, AccountKeyPairOf, BalanceOf, CallOf, Chain, Client, Error as SubstrateError,
+	HashOf, TransactionEra, TransactionTracker, UnsignedTransaction,
 };
 use relay_utils::relay_loop::Client as RelayClient;
 use sp_core::Pair;
@@ -57,7 +57,7 @@ pub struct SubstrateMessagesTarget<P: SubstrateMessageLane> {
 	source_client: Client<P::SourceChain>,
 	lane_id: LaneId,
 	relayer_id_at_source: AccountIdOf<P::SourceChain>,
-	transaction_params: TransactionParams<AccountKeyPairOf<P::TargetChain>>,
+	transaction_params: Option<TransactionParams<AccountKeyPairOf<P::TargetChain>>>,
 	source_to_target_headers_relay: Option<Arc<dyn OnDemandRelay<P::SourceChain, P::TargetChain>>>,
 }
 
@@ -68,7 +68,7 @@ impl<P: SubstrateMessageLane> SubstrateMessagesTarget<P> {
 		source_client: Client<P::SourceChain>,
 		lane_id: LaneId,
 		relayer_id_at_source: AccountIdOf<P::SourceChain>,
-		transaction_params: TransactionParams<AccountKeyPairOf<P::TargetChain>>,
+		transaction_params: Option<TransactionParams<AccountKeyPairOf<P::TargetChain>>>,
 		source_to_target_headers_relay: Option<
 			Arc<dyn OnDemandRelay<P::SourceChain, P::TargetChain>>,
 		>,
@@ -249,11 +249,18 @@ where
 			None => messages_proof_call,
 		};
 
-		let transaction_params = self.transaction_params.clone();
+		let transaction_params = self.transaction_params.clone().map(Ok).unwrap_or_else(|| {
+			// this error shall never happen in practice, so it not deserves
+			// a separate error variant
+			Err(SubstrateError::Custom(format!(
+				"Cannot sign transaction of {} chain",
+				P::TargetChain::NAME,
+			)))
+		})?;
 		let tx_tracker = self
 			.target_client
 			.submit_and_watch_signed_extrinsic(
-				&self.transaction_params.signer,
+				&transaction_params.signer,
 				move |best_block_id, transaction_nonce| {
 					Ok(UnsignedTransaction::new(final_call.into(), transaction_nonce)
 						.era(TransactionEra::new(best_block_id, transaction_params.mortality)))

bridges/relays/messages/src/lib.rs

@@ -35,3 +35,6 @@ mod message_race_limits;
 mod message_race_loop;
 mod message_race_receiving;
 mod message_race_strategy;
+
+pub use message_race_delivery::relay_messages_range;
+pub use message_race_receiving::relay_messages_delivery_confirmation;

bridges/relays/messages/src/message_race_delivery.rs

@@ -19,7 +19,7 @@ use async_trait::async_trait;
 use futures::stream::FusedStream;
 
 use bp_messages::{MessageNonce, UnrewardedRelayersState, Weight};
-use relay_utils::FailedClient;
+use relay_utils::{FailedClient, TrackedTransactionStatus, TransactionTracker};
 
 use crate::{
 	message_lane::{MessageLane, SourceHeaderIdOf, TargetHeaderIdOf},
@@ -77,6 +77,69 @@ pub async fn run<P: MessageLane>(
 	.await
 }
 
+/// Relay range of messages.
+pub async fn relay_messages_range<P: MessageLane>(
+	source_client: impl MessageLaneSourceClient<P>,
+	target_client: impl MessageLaneTargetClient<P>,
+	at: SourceHeaderIdOf<P>,
+	range: RangeInclusive<MessageNonce>,
+	outbound_state_proof_required: bool,
+) -> Result<(), ()> {
+	// compute cumulative dispatch weight of all messages in given range
+	let dispatch_weight = source_client
+		.generated_message_details(at.clone(), range.clone())
+		.await
+		.map_err(|e| {
+			log::error!(
+				target: "bridge",
+				"Failed to get generated message details at {:?} for messages {:?}: {:?}",
+				at,
+				range,
+				e,
+			);
+		})?
+		.values()
+		.fold(Weight::zero(), |total, details| total.saturating_add(details.dispatch_weight));
+	// prepare messages proof
+	let (at, range, proof) = source_client
+		.prove_messages(
+			at.clone(),
+			range.clone(),
+			MessageProofParameters { outbound_state_proof_required, dispatch_weight },
+		)
+		.await
+		.map_err(|e| {
+			log::error!(
+				target: "bridge",
+				"Failed to generate messages proof at {:?} for messages {:?}: {:?}",
+				at,
+				range,
+				e,
+			);
+		})?;
+	// submit messages proof to the target node
+	let tx_tracker = target_client
+		.submit_messages_proof(None, at, range.clone(), proof)
+		.await
+		.map_err(|e| {
+			log::error!(
+				target: "bridge",
+				"Failed to submit messages proof for messages {:?}: {:?}",
+				range,
+				e,
+			);
+		})?
+		.tx_tracker;
+
+	match tx_tracker.wait().await {
+		TrackedTransactionStatus::Finalized(_) => Ok(()),
+		TrackedTransactionStatus::Lost => {
+			log::error!("Transaction with messages {:?} is considered lost", range,);
+			Err(())
+		},
+	}
+}
+
 /// Message delivery race.
 struct MessageDeliveryRace<P>(std::marker::PhantomData<P>);