Refactor finality relay helpers (#1220)

* refactor finality relay helper definitions

* add missing doc

* removed commented code

* fmt

* disable rustfmt for macro

* move best_finalized method const to relay chain def

bridges/relays/lib-substrate-relay/Cargo.toml

@@ -27,6 +27,7 @@ relay-utils = { path = "../utils" }
 messages-relay = { path = "../messages" }
 relay-substrate-client = { path = "../client-substrate" }
 
+pallet-bridge-grandpa = { path = "../../modules/grandpa" }
 pallet-bridge-messages = { path = "../../modules/messages" }
 
 bp-runtime = { path = "../../primitives/runtime" }

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

@@ -14,18 +14,20 @@
 // You should have received a copy of the GNU General Public License
 // along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
 
-//! Substrate-to-Substrate headers sync entrypoint.
+//! Types and functions intended to ease adding of new Substrate -> Substrate
+//! finality proofs synchronization pipelines.
 
-use crate::{finality_target::SubstrateFinalityTarget, STALL_TIMEOUT};
+use crate::{finality_source::SubstrateFinalitySource, finality_target::SubstrateFinalityTarget};
 
 use bp_header_chain::justification::GrandpaJustification;
-use bp_runtime::AccountIdOf;
-use finality_relay::{FinalitySyncParams, FinalitySyncPipeline};
+use finality_relay::FinalitySyncPipeline;
+use pallet_bridge_grandpa::{Call as BridgeGrandpaCall, Config as BridgeGrandpaConfig};
 use relay_substrate_client::{
-	finality_source::FinalitySource, BlockNumberOf, Chain, Client, HashOf, SyncHeader,
+	transaction_stall_timeout, AccountIdOf, AccountKeyPairOf, BlockNumberOf, CallOf, Chain, Client,
+	HashOf, HeaderOf, SyncHeader, TransactionSignScheme,
 };
-use relay_utils::{metrics::MetricsParams, BlockNumberBase};
-use sp_core::Bytes;
+use relay_utils::metrics::MetricsParams;
+use sp_core::Pair;
 use std::{fmt::Debug, marker::PhantomData};
 
 /// Default limit of recent finality proofs.
@@ -34,130 +36,152 @@ use std::{fmt::Debug, marker::PhantomData};
 /// Substrate+GRANDPA based chains (good to know).
 pub(crate) const RECENT_FINALITY_PROOFS_LIMIT: usize = 4096;
 
-/// Headers sync pipeline for Substrate <-> Substrate relays.
+/// Submit-finality-proofs transaction creation parameters.
+#[derive(Clone, Debug)]
+pub struct TransactionParams<TS> {
+	/// Transactions author.
+	pub transactions_signer: TS,
+	/// Transactions mortality.
+	pub transactions_mortality: Option<u32>,
+}
+
+/// Substrate -> Substrate finality proofs synchronization pipeline.
 pub trait SubstrateFinalitySyncPipeline: 'static + Clone + Debug + Send + Sync {
-	/// Pipeline for syncing finalized Source chain headers to Target chain.
-	type FinalitySyncPipeline: FinalitySyncPipeline;
-
-	/// Name of the runtime method that returns id of best finalized source header at target chain.
-	const BEST_FINALIZED_SOURCE_HEADER_ID_AT_TARGET: &'static str;
-
-	/// Chain with GRANDPA bridge pallet.
+	/// Headers of this chain are submitted to the `TargetChain`.
+	type SourceChain: Chain;
+	/// Headers of the `SourceChain` are submitted to this chain.
 	type TargetChain: Chain;
 
-	/// Customize metrics exposed by headers sync loop.
-	fn customize_metrics(params: MetricsParams) -> anyhow::Result<MetricsParams> {
-		Ok(params)
-	}
+	/// How submit finality proof call is built?
+	type SubmitFinalityProofCallBuilder: SubmitFinalityProofCallBuilder<Self>;
+	/// Scheme used to sign target chain transactions.
+	type TransactionSignScheme: TransactionSignScheme;
 
-	/// Start finality relay guards.
-	///
-	/// Different finality bridges may have different set of guards - e.g. on ephemeral chains we
-	/// don't need a version guards, on test chains we don't care that much about relayer account
-	/// balance, ... So the implementation is left to the specific bridges.
-	fn start_relay_guards(&self) {}
-
-	/// Returns id of account that we're using to sign transactions at target chain.
-	fn transactions_author(&self) -> AccountIdOf<Self::TargetChain>;
-
-	/// Make submit header transaction.
-	fn make_submit_finality_proof_transaction(
-		&self,
-		era: bp_runtime::TransactionEraOf<Self::TargetChain>,
-		transaction_nonce: bp_runtime::IndexOf<Self::TargetChain>,
-		header: <Self::FinalitySyncPipeline as FinalitySyncPipeline>::Header,
-		proof: <Self::FinalitySyncPipeline as FinalitySyncPipeline>::FinalityProof,
-	) -> Bytes;
-}
-
-/// Substrate-to-Substrate finality proof pipeline.
-#[derive(Clone)]
-pub struct SubstrateFinalityToSubstrate<SourceChain, TargetChain: Chain, TargetSign> {
-	/// Client for the target chain.
-	pub target_client: Client<TargetChain>,
-	/// Data required to sign target chain transactions.
-	pub target_sign: TargetSign,
-	/// Unused generic arguments dump.
-	_marker: PhantomData<SourceChain>,
-}
-
-impl<SourceChain, TargetChain: Chain, TargetSign> Debug
-	for SubstrateFinalityToSubstrate<SourceChain, TargetChain, TargetSign>
-{
-	fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
-		f.debug_struct("SubstrateFinalityToSubstrate")
-			.field("target_client", &self.target_client)
-			.finish()
+	/// Add relay guards if required.
+	fn start_relay_guards(
+		_target_client: &Client<Self::TargetChain>,
+		_transaction_params: &TransactionParams<AccountKeyPairOf<Self::TransactionSignScheme>>,
+	) {
 	}
 }
 
-impl<SourceChain, TargetChain: Chain, TargetSign>
-	SubstrateFinalityToSubstrate<SourceChain, TargetChain, TargetSign>
-{
-	/// Create new Substrate-to-Substrate headers pipeline.
-	pub fn new(target_client: Client<TargetChain>, target_sign: TargetSign) -> Self {
-		SubstrateFinalityToSubstrate { target_client, target_sign, _marker: Default::default() }
-	}
+/// Adapter that allows all `SubstrateFinalitySyncPipeline` to act as `FinalitySyncPipeline`.
+#[derive(Clone, Debug)]
+pub struct FinalitySyncPipelineAdapter<P: SubstrateFinalitySyncPipeline> {
+	_phantom: PhantomData<P>,
 }
 
-impl<SourceChain, TargetChain, TargetSign> FinalitySyncPipeline
-	for SubstrateFinalityToSubstrate<SourceChain, TargetChain, TargetSign>
+impl<P: SubstrateFinalitySyncPipeline> FinalitySyncPipeline for FinalitySyncPipelineAdapter<P> {
+	const SOURCE_NAME: &'static str = P::SourceChain::NAME;
+	const TARGET_NAME: &'static str = P::TargetChain::NAME;
+
+	type Hash = HashOf<P::SourceChain>;
+	type Number = BlockNumberOf<P::SourceChain>;
+	type Header = relay_substrate_client::SyncHeader<HeaderOf<P::SourceChain>>;
+	type FinalityProof = GrandpaJustification<HeaderOf<P::SourceChain>>;
+}
+
+/// Different ways of building `submit_finality_proof` calls.
+pub trait SubmitFinalityProofCallBuilder<P: SubstrateFinalitySyncPipeline> {
+	/// Given source chain header and its finality proofs, build call of `submit_finality_proof`
+	/// function of bridge GRANDPA module at the target chain.
+	fn build_submit_finality_proof_call(
+		header: SyncHeader<HeaderOf<P::SourceChain>>,
+		proof: GrandpaJustification<HeaderOf<P::SourceChain>>,
+	) -> CallOf<P::TargetChain>;
+}
+
+/// Building `submit_finality_proof` call when you have direct access to the target
+/// chain runtime.
+pub struct DirectSubmitFinalityProofCallBuilder<P, R, I> {
+	_phantom: PhantomData<(P, R, I)>,
+}
+
+impl<P, R, I> SubmitFinalityProofCallBuilder<P> for DirectSubmitFinalityProofCallBuilder<P, R, I>
 where
-	SourceChain: Clone + Chain + Debug,
-	BlockNumberOf<SourceChain>: BlockNumberBase,
-	TargetChain: Clone + Chain + Debug,
-	TargetSign: 'static + Clone + Send + Sync,
+	P: SubstrateFinalitySyncPipeline,
+	R: BridgeGrandpaConfig<I>,
+	I: 'static,
+	R::BridgedChain: bp_runtime::Chain<Header = HeaderOf<P::SourceChain>>,
+	CallOf<P::TargetChain>: From<BridgeGrandpaCall<R, I>>,
 {
-	const SOURCE_NAME: &'static str = SourceChain::NAME;
-	const TARGET_NAME: &'static str = TargetChain::NAME;
-
-	type Hash = HashOf<SourceChain>;
-	type Number = BlockNumberOf<SourceChain>;
-	type Header = SyncHeader<SourceChain::Header>;
-	type FinalityProof = GrandpaJustification<SourceChain::Header>;
+	fn build_submit_finality_proof_call(
+		header: SyncHeader<HeaderOf<P::SourceChain>>,
+		proof: GrandpaJustification<HeaderOf<P::SourceChain>>,
+	) -> CallOf<P::TargetChain> {
+		BridgeGrandpaCall::<R, I>::submit_finality_proof {
+			finality_target: Box::new(header.into_inner()),
+			justification: proof,
+		}
+		.into()
+	}
 }
 
-/// Run Substrate-to-Substrate finality sync.
-pub async fn run<SourceChain, TargetChain, P>(
-	pipeline: P,
-	source_client: Client<SourceChain>,
-	target_client: Client<TargetChain>,
+/// Macro that generates `SubmitFinalityProofCallBuilder` implementation for the case when
+/// you only have an access to the mocked version of target chain runtime. In this case you
+/// should provide "name" of the call variant for the bridge GRANDPA calls and the "name" of
+/// the variant for the `submit_finality_proof` call within that first option.
+#[rustfmt::skip]
+#[macro_export]
+macro_rules! generate_mocked_submit_finality_proof_call_builder {
+	($pipeline:ident, $mocked_builder:ident, $bridge_grandpa:path, $submit_finality_proof:path) => {
+		pub struct $mocked_builder;
+
+		impl $crate::finality_pipeline::SubmitFinalityProofCallBuilder<$pipeline>
+			for $mocked_builder
+		{
+			fn build_submit_finality_proof_call(
+				header: relay_substrate_client::SyncHeader<
+					relay_substrate_client::HeaderOf<
+						<$pipeline as $crate::finality_pipeline::SubstrateFinalitySyncPipeline>::SourceChain
+					>
+				>,
+				proof: bp_header_chain::justification::GrandpaJustification<
+					relay_substrate_client::HeaderOf<
+						<$pipeline as $crate::finality_pipeline::SubstrateFinalitySyncPipeline>::SourceChain
+					>
+				>,
+			) -> relay_substrate_client::CallOf<
+				<$pipeline as $crate::finality_pipeline::SubstrateFinalitySyncPipeline>::TargetChain
+			> {
+				$bridge_grandpa($submit_finality_proof(Box::new(header.into_inner()), proof))
+			}
+		}
+	};
+}
+
+/// Run Substrate-to-Substrate finality sync loop.
+pub async fn run<P: SubstrateFinalitySyncPipeline>(
+	source_client: Client<P::SourceChain>,
+	target_client: Client<P::TargetChain>,
 	only_mandatory_headers: bool,
-	transactions_mortality: Option<u32>,
+	transaction_params: TransactionParams<AccountKeyPairOf<P::TransactionSignScheme>>,
 	metrics_params: MetricsParams,
 ) -> anyhow::Result<()>
 where
-	P: SubstrateFinalitySyncPipeline<TargetChain = TargetChain>,
-	P::FinalitySyncPipeline: FinalitySyncPipeline<
-		Hash = HashOf<SourceChain>,
-		Number = BlockNumberOf<SourceChain>,
-		Header = SyncHeader<SourceChain::Header>,
-		FinalityProof = GrandpaJustification<SourceChain::Header>,
-	>,
-	SourceChain: Clone + Chain,
-	BlockNumberOf<SourceChain>: BlockNumberBase,
-	TargetChain: Clone + Chain,
+	AccountIdOf<P::TargetChain>: From<<AccountKeyPairOf<P::TransactionSignScheme> as Pair>::Public>,
+	P::TransactionSignScheme: TransactionSignScheme<Chain = P::TargetChain>,
 {
 	log::info!(
 		target: "bridge",
 		"Starting {} -> {} finality proof relay",
-		SourceChain::NAME,
-		TargetChain::NAME,
+		P::SourceChain::NAME,
+		P::TargetChain::NAME,
 	);
 
 	finality_relay::run(
-		FinalitySource::new(source_client, None),
-		SubstrateFinalityTarget::new(target_client, pipeline, transactions_mortality),
-		FinalitySyncParams {
+		SubstrateFinalitySource::<P>::new(source_client, None),
+		SubstrateFinalityTarget::<P>::new(target_client, transaction_params.clone()),
+		finality_relay::FinalitySyncParams {
 			tick: std::cmp::max(
-				SourceChain::AVERAGE_BLOCK_INTERVAL,
-				TargetChain::AVERAGE_BLOCK_INTERVAL,
+				P::SourceChain::AVERAGE_BLOCK_INTERVAL,
+				P::TargetChain::AVERAGE_BLOCK_INTERVAL,
 			),
 			recent_finality_proofs_limit: RECENT_FINALITY_PROOFS_LIMIT,
-			stall_timeout: relay_substrate_client::transaction_stall_timeout(
-				transactions_mortality,
-				TargetChain::AVERAGE_BLOCK_INTERVAL,
-				STALL_TIMEOUT,
+			stall_timeout: transaction_stall_timeout(
+				transaction_params.transactions_mortality,
+				P::TargetChain::AVERAGE_BLOCK_INTERVAL,
+				crate::STALL_TIMEOUT,
 			),
 			only_mandatory_headers,
 		},

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

@@ -0,0 +1,182 @@
+// Copyright 2019-2021 Parity Technologies (UK) Ltd.
+// This file is part of Parity Bridges Common.
+
+// Parity Bridges Common is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity Bridges Common is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Default generic implementation of finality source for basic Substrate client.
+
+use crate::finality_pipeline::{FinalitySyncPipelineAdapter, SubstrateFinalitySyncPipeline};
+
+use async_std::sync::{Arc, Mutex};
+use async_trait::async_trait;
+use bp_header_chain::justification::GrandpaJustification;
+use codec::Decode;
+use finality_relay::SourceClient;
+use futures::stream::{unfold, Stream, StreamExt};
+use relay_substrate_client::{
+	BlockNumberOf, BlockWithJustification, Chain, Client, Error, HeaderOf,
+};
+use relay_utils::relay_loop::Client as RelayClient;
+use sp_runtime::traits::Header as HeaderT;
+use std::pin::Pin;
+
+/// Shared updatable reference to the maximal header number that we want to sync from the source.
+pub type RequiredHeaderNumberRef<C> = Arc<Mutex<<C as bp_runtime::Chain>::BlockNumber>>;
+
+/// Substrate finality proofs stream.
+pub type SubstrateFinalityProofsStream<P> = Pin<
+	Box<
+		dyn Stream<
+				Item = GrandpaJustification<
+					HeaderOf<<P as SubstrateFinalitySyncPipeline>::SourceChain>,
+				>,
+			> + Send,
+	>,
+>;
+
+/// Substrate node as finality source.
+pub struct SubstrateFinalitySource<P: SubstrateFinalitySyncPipeline> {
+	client: Client<P::SourceChain>,
+	maximal_header_number: Option<RequiredHeaderNumberRef<P::SourceChain>>,
+}
+
+impl<P: SubstrateFinalitySyncPipeline> SubstrateFinalitySource<P> {
+	/// Create new headers source using given client.
+	pub fn new(
+		client: Client<P::SourceChain>,
+		maximal_header_number: Option<RequiredHeaderNumberRef<P::SourceChain>>,
+	) -> Self {
+		SubstrateFinalitySource { client, maximal_header_number }
+	}
+
+	/// Returns reference to the underlying RPC client.
+	pub fn client(&self) -> &Client<P::SourceChain> {
+		&self.client
+	}
+
+	/// Returns best finalized block number.
+	pub async fn on_chain_best_finalized_block_number(
+		&self,
+	) -> Result<BlockNumberOf<P::SourceChain>, Error> {
+		// we **CAN** continue to relay finality proofs if source node is out of sync, because
+		// target node may be missing proofs that are already available at the source
+		let finalized_header_hash = self.client.best_finalized_header_hash().await?;
+		let finalized_header = self.client.header_by_hash(finalized_header_hash).await?;
+		Ok(*finalized_header.number())
+	}
+}
+
+impl<P: SubstrateFinalitySyncPipeline> Clone for SubstrateFinalitySource<P> {
+	fn clone(&self) -> Self {
+		SubstrateFinalitySource {
+			client: self.client.clone(),
+			maximal_header_number: self.maximal_header_number.clone(),
+		}
+	}
+}
+
+#[async_trait]
+impl<P: SubstrateFinalitySyncPipeline> RelayClient for SubstrateFinalitySource<P> {
+	type Error = Error;
+
+	async fn reconnect(&mut self) -> Result<(), Error> {
+		self.client.reconnect().await
+	}
+}
+
+#[async_trait]
+impl<P: SubstrateFinalitySyncPipeline> SourceClient<FinalitySyncPipelineAdapter<P>>
+	for SubstrateFinalitySource<P>
+{
+	type FinalityProofsStream = SubstrateFinalityProofsStream<P>;
+
+	async fn best_finalized_block_number(&self) -> Result<BlockNumberOf<P::SourceChain>, Error> {
+		let mut finalized_header_number = self.on_chain_best_finalized_block_number().await?;
+		// never return block number larger than requested. This way we'll never sync headers
+		// past `maximal_header_number`
+		if let Some(ref maximal_header_number) = self.maximal_header_number {
+			let maximal_header_number = *maximal_header_number.lock().await;
+			if finalized_header_number > maximal_header_number {
+				finalized_header_number = maximal_header_number;
+			}
+		}
+		Ok(finalized_header_number)
+	}
+
+	async fn header_and_finality_proof(
+		&self,
+		number: BlockNumberOf<P::SourceChain>,
+	) -> Result<
+		(
+			relay_substrate_client::SyncHeader<HeaderOf<P::SourceChain>>,
+			Option<GrandpaJustification<HeaderOf<P::SourceChain>>>,
+		),
+		Error,
+	> {
+		let header_hash = self.client.block_hash_by_number(number).await?;
+		let signed_block = self.client.get_block(Some(header_hash)).await?;
+
+		let justification = signed_block
+			.justification()
+			.map(|raw_justification| {
+				GrandpaJustification::<HeaderOf<P::SourceChain>>::decode(
+					&mut raw_justification.as_slice(),
+				)
+			})
+			.transpose()
+			.map_err(Error::ResponseParseFailed)?;
+
+		Ok((signed_block.header().into(), justification))
+	}
+
+	async fn finality_proofs(&self) -> Result<Self::FinalityProofsStream, Error> {
+		Ok(unfold(
+			self.client.clone().subscribe_justifications().await?,
+			move |subscription| async move {
+				loop {
+					let log_error = |err| {
+						log::error!(
+							target: "bridge",
+							"Failed to read justification target from the {} justifications stream: {:?}",
+							P::SourceChain::NAME,
+							err,
+						);
+					};
+
+					let next_justification = subscription
+						.next()
+						.await
+						.map_err(|err| log_error(err.to_string()))
+						.ok()??;
+
+					let decoded_justification =
+						GrandpaJustification::<HeaderOf<P::SourceChain>>::decode(
+							&mut &next_justification[..],
+						);
+
+					let justification = match decoded_justification {
+						Ok(j) => j,
+						Err(err) => {
+							log_error(format!("decode failed with error {:?}", err));
+							continue
+						},
+					};
+
+					return Some((justification, subscription))
+				}
+			},
+		)
+		.boxed())
+	}
+}

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

@@ -15,70 +15,78 @@
 // along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
 
 //! Substrate client as Substrate finality proof target. The chain we connect to should have
-//! runtime that implements `<BridgedChainName>FinalityApi` to allow bridging with
-//! <BridgedName> chain.
+//! bridge GRANDPA pallet deployed and provide `<BridgedChainName>FinalityApi` to allow bridging
+//! with <BridgedName> chain.
 
-use crate::finality_pipeline::SubstrateFinalitySyncPipeline;
+use crate::finality_pipeline::{
+	FinalitySyncPipelineAdapter, SubmitFinalityProofCallBuilder, SubstrateFinalitySyncPipeline,
+	TransactionParams,
+};
 
 use async_trait::async_trait;
-use codec::Decode;
-use finality_relay::{FinalitySyncPipeline, TargetClient};
-use relay_substrate_client::{Chain, Client, Error as SubstrateError};
+use bp_header_chain::justification::GrandpaJustification;
+use codec::Encode;
+use finality_relay::TargetClient;
+use relay_substrate_client::{
+	AccountIdOf, AccountKeyPairOf, BlockNumberOf, Chain, Client, Error, HashOf, HeaderOf,
+	SyncHeader, TransactionEra, TransactionSignScheme, UnsignedTransaction,
+};
 use relay_utils::relay_loop::Client as RelayClient;
+use sp_core::{Bytes, Pair};
 
 /// Substrate client as Substrate finality target.
-pub struct SubstrateFinalityTarget<C: Chain, P> {
-	client: Client<C>,
-	pipeline: P,
-	transactions_mortality: Option<u32>,
+pub struct SubstrateFinalityTarget<P: SubstrateFinalitySyncPipeline> {
+	client: Client<P::TargetChain>,
+	transaction_params: TransactionParams<AccountKeyPairOf<P::TransactionSignScheme>>,
 }
 
-impl<C: Chain, P> SubstrateFinalityTarget<C, P> {
+impl<P: SubstrateFinalitySyncPipeline> SubstrateFinalityTarget<P> {
 	/// Create new Substrate headers target.
-	pub fn new(client: Client<C>, pipeline: P, transactions_mortality: Option<u32>) -> Self {
-		SubstrateFinalityTarget { client, pipeline, transactions_mortality }
+	pub fn new(
+		client: Client<P::TargetChain>,
+		transaction_params: TransactionParams<AccountKeyPairOf<P::TransactionSignScheme>>,
+	) -> Self {
+		SubstrateFinalityTarget { client, transaction_params }
 	}
 }
 
-impl<C: Chain, P: SubstrateFinalitySyncPipeline> Clone for SubstrateFinalityTarget<C, P> {
+impl<P: SubstrateFinalitySyncPipeline> Clone for SubstrateFinalityTarget<P> {
 	fn clone(&self) -> Self {
 		SubstrateFinalityTarget {
 			client: self.client.clone(),
-			pipeline: self.pipeline.clone(),
-			transactions_mortality: self.transactions_mortality,
+			transaction_params: self.transaction_params.clone(),
 		}
 	}
 }
 
 #[async_trait]
-impl<C: Chain, P: SubstrateFinalitySyncPipeline> RelayClient for SubstrateFinalityTarget<C, P> {
-	type Error = SubstrateError;
+impl<P: SubstrateFinalitySyncPipeline> RelayClient for SubstrateFinalityTarget<P> {
+	type Error = Error;
 
-	async fn reconnect(&mut self) -> Result<(), SubstrateError> {
+	async fn reconnect(&mut self) -> Result<(), Error> {
 		self.client.reconnect().await
 	}
 }
 
 #[async_trait]
-impl<C, P> TargetClient<P::FinalitySyncPipeline> for SubstrateFinalityTarget<C, P>
+impl<P: SubstrateFinalitySyncPipeline> TargetClient<FinalitySyncPipelineAdapter<P>>
+	for SubstrateFinalityTarget<P>
 where
-	C: Chain,
-	P: SubstrateFinalitySyncPipeline<TargetChain = C>,
-	<P::FinalitySyncPipeline as FinalitySyncPipeline>::Number: Decode,
-	<P::FinalitySyncPipeline as FinalitySyncPipeline>::Hash: Decode,
+	AccountIdOf<P::TargetChain>: From<<AccountKeyPairOf<P::TransactionSignScheme> as Pair>::Public>,
+	P::TransactionSignScheme: TransactionSignScheme<Chain = P::TargetChain>,
 {
 	async fn best_finalized_source_block_number(
 		&self,
-	) -> Result<<P::FinalitySyncPipeline as FinalitySyncPipeline>::Number, SubstrateError> {
+	) -> Result<BlockNumberOf<P::SourceChain>, Error> {
 		// we can't continue to relay finality if target node is out of sync, because
 		// it may have already received (some of) headers that we're going to relay
 		self.client.ensure_synced().await?;
 
 		Ok(crate::messages_source::read_client_state::<
-			C,
-			<P::FinalitySyncPipeline as FinalitySyncPipeline>::Hash,
-			<P::FinalitySyncPipeline as FinalitySyncPipeline>::Number,
-		>(&self.client, P::BEST_FINALIZED_SOURCE_HEADER_ID_AT_TARGET)
+			P::TargetChain,
+			HashOf<P::SourceChain>,
+			BlockNumberOf<P::SourceChain>,
+		>(&self.client, P::SourceChain::BEST_FINALIZED_HEADER_ID_METHOD)
 		.await?
 		.best_finalized_peer_at_best_self
 		.0)
@@ -86,24 +94,28 @@ where
 
 	async fn submit_finality_proof(
 		&self,
-		header: <P::FinalitySyncPipeline as FinalitySyncPipeline>::Header,
-		proof: <P::FinalitySyncPipeline as FinalitySyncPipeline>::FinalityProof,
-	) -> Result<(), SubstrateError> {
-		let transactions_author = self.pipeline.transactions_author();
-		let pipeline = self.pipeline.clone();
-		let transactions_mortality = self.transactions_mortality;
+		header: SyncHeader<HeaderOf<P::SourceChain>>,
+		proof: GrandpaJustification<HeaderOf<P::SourceChain>>,
+	) -> Result<(), Error> {
+		let genesis_hash = *self.client.genesis_hash();
+		let transaction_params = self.transaction_params.clone();
+		let call =
+			P::SubmitFinalityProofCallBuilder::build_submit_finality_proof_call(header, proof);
 		self.client
 			.submit_signed_extrinsic(
-				transactions_author,
+				self.transaction_params.transactions_signer.public().into(),
 				move |best_block_id, transaction_nonce| {
-					pipeline.make_submit_finality_proof_transaction(
-						relay_substrate_client::TransactionEra::new(
-							best_block_id,
-							transactions_mortality,
-						),
-						transaction_nonce,
-						header,
-						proof,
+					Bytes(
+						P::TransactionSignScheme::sign_transaction(
+							genesis_hash,
+							&transaction_params.transactions_signer,
+							TransactionEra::new(
+								best_block_id,
+								transaction_params.transactions_mortality,
+							),
+							UnsignedTransaction::new(call, transaction_nonce),
+						)
+						.encode(),
 					)
 				},
 			)

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

@@ -23,6 +23,7 @@ use std::time::Duration;
 pub mod conversion_rate_update;
 pub mod error;
 pub mod finality_pipeline;
+pub mod finality_source;
 pub mod finality_target;
 pub mod headers_initialize;
 pub mod helpers;

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

@@ -16,29 +16,24 @@
 
 //! On-demand Substrate -> Substrate headers relay.
 
-use std::fmt::Debug;
-
 use async_std::sync::{Arc, Mutex};
 use futures::{select, FutureExt};
 use num_traits::{CheckedSub, One, Zero};
 
-use finality_relay::{
-	FinalitySyncParams, FinalitySyncPipeline, SourceClient as FinalitySourceClient, SourceHeader,
-	TargetClient as FinalityTargetClient,
-};
+use finality_relay::{FinalitySyncParams, SourceHeader, TargetClient as FinalityTargetClient};
 use relay_substrate_client::{
-	finality_source::{FinalitySource as SubstrateFinalitySource, RequiredHeaderNumberRef},
-	Chain, Client, HeaderIdOf, SyncHeader,
+	AccountIdOf, AccountKeyPairOf, BlockNumberOf, Chain, Client, HeaderIdOf, HeaderOf, SyncHeader,
+	TransactionSignScheme,
 };
 use relay_utils::{
-	metrics::MetricsParams, relay_loop::Client as RelayClient, BlockNumberBase, FailedClient,
-	MaybeConnectionError,
+	metrics::MetricsParams, relay_loop::Client as RelayClient, FailedClient, MaybeConnectionError,
 };
 
 use crate::{
 	finality_pipeline::{
-		SubstrateFinalitySyncPipeline, SubstrateFinalityToSubstrate, RECENT_FINALITY_PROOFS_LIMIT,
+		SubstrateFinalitySyncPipeline, TransactionParams, RECENT_FINALITY_PROOFS_LIMIT,
 	},
+	finality_source::{RequiredHeaderNumberRef, SubstrateFinalitySource},
 	finality_target::SubstrateFinalityTarget,
 	STALL_TIMEOUT,
 };
@@ -58,40 +53,28 @@ pub struct OnDemandHeadersRelay<SourceChain: Chain> {
 
 impl<SourceChain: Chain> OnDemandHeadersRelay<SourceChain> {
 	/// Create new on-demand headers relay.
-	pub fn new<TargetChain: Chain, TargetSign, P>(
-		source_client: Client<SourceChain>,
-		target_client: Client<TargetChain>,
-		target_transactions_mortality: Option<u32>,
-		pipeline: P,
-		maximal_headers_difference: SourceChain::BlockNumber,
+	pub fn new<P: SubstrateFinalitySyncPipeline<SourceChain = SourceChain>>(
+		source_client: Client<P::SourceChain>,
+		target_client: Client<P::TargetChain>,
+		target_transaction_params: TransactionParams<AccountKeyPairOf<P::TransactionSignScheme>>,
+		maximal_headers_difference: BlockNumberOf<P::SourceChain>,
 		only_mandatory_headers: bool,
 	) -> Self
 	where
-		SourceChain: Chain + Debug,
-		SourceChain::BlockNumber: BlockNumberBase,
-		TargetChain: Chain + Debug,
-		TargetChain::BlockNumber: BlockNumberBase,
-		TargetSign: Clone + Send + Sync + 'static,
-		P: SubstrateFinalitySyncPipeline<
-			FinalitySyncPipeline = SubstrateFinalityToSubstrate<
-				SourceChain,
-				TargetChain,
-				TargetSign,
-			>,
-			TargetChain = TargetChain,
-		>,
+		AccountIdOf<P::TargetChain>:
+			From<<AccountKeyPairOf<P::TransactionSignScheme> as sp_core::Pair>::Public>,
+		P::TransactionSignScheme: TransactionSignScheme<Chain = P::TargetChain>,
 	{
 		let required_header_number = Arc::new(Mutex::new(Zero::zero()));
 		let this = OnDemandHeadersRelay {
-			relay_task_name: on_demand_headers_relay_name::<SourceChain, TargetChain>(),
+			relay_task_name: on_demand_headers_relay_name::<P::SourceChain, P::TargetChain>(),
 			required_header_number: required_header_number.clone(),
 		};
 		async_std::task::spawn(async move {
-			background_task(
+			background_task::<P>(
 				source_client,
 				target_client,
-				target_transactions_mortality,
-				pipeline,
+				target_transaction_params,
 				maximal_headers_difference,
 				only_mandatory_headers,
 				required_header_number,
@@ -120,35 +103,25 @@ impl<SourceChain: Chain> OnDemandHeadersRelay<SourceChain> {
 }
 
 /// Background task that is responsible for starting headers relay.
-async fn background_task<SourceChain, TargetChain, TargetSign, P>(
-	source_client: Client<SourceChain>,
-	target_client: Client<TargetChain>,
-	target_transactions_mortality: Option<u32>,
-	pipeline: P,
-	maximal_headers_difference: SourceChain::BlockNumber,
+async fn background_task<P: SubstrateFinalitySyncPipeline>(
+	source_client: Client<P::SourceChain>,
+	target_client: Client<P::TargetChain>,
+	target_transaction_params: TransactionParams<AccountKeyPairOf<P::TransactionSignScheme>>,
+	maximal_headers_difference: BlockNumberOf<P::SourceChain>,
 	only_mandatory_headers: bool,
-	required_header_number: RequiredHeaderNumberRef<SourceChain>,
+	required_header_number: RequiredHeaderNumberRef<P::SourceChain>,
 ) where
-	SourceChain: Chain + Debug,
-	SourceChain::BlockNumber: BlockNumberBase,
-	TargetChain: Chain + Debug,
-	TargetChain::BlockNumber: BlockNumberBase,
-	TargetSign: Clone + Send + Sync + 'static,
-	P: SubstrateFinalitySyncPipeline<
-		FinalitySyncPipeline = SubstrateFinalityToSubstrate<SourceChain, TargetChain, TargetSign>,
-		TargetChain = TargetChain,
-	>,
+	AccountIdOf<P::TargetChain>:
+		From<<AccountKeyPairOf<P::TransactionSignScheme> as sp_core::Pair>::Public>,
+	P::TransactionSignScheme: TransactionSignScheme<Chain = P::TargetChain>,
 {
-	let relay_task_name = on_demand_headers_relay_name::<SourceChain, TargetChain>();
-	let mut finality_source = SubstrateFinalitySource::<
-		_,
-		SubstrateFinalityToSubstrate<SourceChain, TargetChain, TargetSign>,
-	>::new(source_client.clone(), Some(required_header_number.clone()));
-	let mut finality_target = SubstrateFinalityTarget::new(
-		target_client.clone(),
-		pipeline.clone(),
-		target_transactions_mortality,
+	let relay_task_name = on_demand_headers_relay_name::<P::SourceChain, P::TargetChain>();
+	let mut finality_source = SubstrateFinalitySource::<P>::new(
+		source_client.clone(),
+		Some(required_header_number.clone()),
 	);
+	let mut finality_target =
+		SubstrateFinalityTarget::new(target_client.clone(), target_transaction_params);
 	let mut latest_non_mandatory_at_source = Zero::zero();
 
 	let mut restart_relay = true;
@@ -157,7 +130,7 @@ async fn background_task<SourceChain, TargetChain, TargetSign, P>(
 
 	loop {
 		select! {
-			_ = async_std::task::sleep(TargetChain::AVERAGE_BLOCK_INTERVAL).fuse() => {},
+			_ = async_std::task::sleep(P::TargetChain::AVERAGE_BLOCK_INTERVAL).fuse() => {},
 			_ = finality_relay_task => {
 				// this should never happen in practice given the current code
 				restart_relay = true;
@@ -179,12 +152,8 @@ async fn background_task<SourceChain, TargetChain, TargetSign, P>(
 		}
 
 		// read best finalized source header number from target
-		let best_finalized_source_header_at_target = best_finalized_source_header_at_target::<
-			SourceChain,
-			_,
-			_,
-		>(&finality_target, &relay_task_name)
-		.await;
+		let best_finalized_source_header_at_target =
+			best_finalized_source_header_at_target::<P>(&finality_target, &relay_task_name).await;
 		if matches!(best_finalized_source_header_at_target, Err(ref e) if e.is_connection_error()) {
 			relay_utils::relay_loop::reconnect_failed_client(
 				FailedClient::Target,
@@ -199,7 +168,7 @@ async fn background_task<SourceChain, TargetChain, TargetSign, P>(
 		// submit mandatory header if some headers are missing
 		let best_finalized_source_header_at_target_fmt =
 			format!("{:?}", best_finalized_source_header_at_target);
-		let mandatory_scan_range = mandatory_headers_scan_range::<SourceChain>(
+		let mandatory_scan_range = mandatory_headers_scan_range::<P::SourceChain>(
 			best_finalized_source_header_at_source.ok(),
 			best_finalized_source_header_at_target.ok(),
 			maximal_headers_difference,
@@ -247,8 +216,8 @@ async fn background_task<SourceChain, TargetChain, TargetSign, P>(
 					finality_target.clone(),
 					FinalitySyncParams {
 						tick: std::cmp::max(
-							SourceChain::AVERAGE_BLOCK_INTERVAL,
-							TargetChain::AVERAGE_BLOCK_INTERVAL,
+							P::SourceChain::AVERAGE_BLOCK_INTERVAL,
+							P::TargetChain::AVERAGE_BLOCK_INTERVAL,
 						),
 						recent_finality_proofs_limit: RECENT_FINALITY_PROOFS_LIMIT,
 						stall_timeout: STALL_TIMEOUT,
@@ -316,17 +285,13 @@ async fn mandatory_headers_scan_range<C: Chain>(
 /// it.
 ///
 /// Returns `true` if header was found and (asked to be) relayed and `false` otherwise.
-async fn relay_mandatory_header_from_range<SourceChain: Chain, P>(
-	finality_source: &SubstrateFinalitySource<SourceChain, P>,
-	required_header_number: &RequiredHeaderNumberRef<SourceChain>,
+async fn relay_mandatory_header_from_range<P: SubstrateFinalitySyncPipeline>(
+	finality_source: &SubstrateFinalitySource<P>,
+	required_header_number: &RequiredHeaderNumberRef<P::SourceChain>,
 	best_finalized_source_header_at_target: String,
-	range: (SourceChain::BlockNumber, SourceChain::BlockNumber),
+	range: (BlockNumberOf<P::SourceChain>, BlockNumberOf<P::SourceChain>),
 	relay_task_name: &str,
-) -> Result<bool, relay_substrate_client::Error>
-where
-	SubstrateFinalitySource<SourceChain, P>: FinalitySourceClient<P>,
-	P: FinalitySyncPipeline<Number = SourceChain::BlockNumber>,
-{
+) -> Result<bool, relay_substrate_client::Error> {
 	// search for mandatory header first
 	let mandatory_source_header_number =
 		find_mandatory_header_in_range(finality_source, range).await?;
@@ -347,7 +312,7 @@ where
 	log::trace!(
 		target: "bridge",
 		"Too many {} headers missing at target in {} relay ({} vs {}). Going to sync up to the mandatory {}",
-		SourceChain::NAME,
+		P::SourceChain::NAME,
 		relay_task_name,
 		best_finalized_source_header_at_target,
 		range.1,
@@ -361,14 +326,10 @@ where
 /// Read best finalized source block number from source client.
 ///
 /// Returns `None` if we have failed to read the number.
-async fn best_finalized_source_header_at_source<SourceChain: Chain, P>(
-	finality_source: &SubstrateFinalitySource<SourceChain, P>,
+async fn best_finalized_source_header_at_source<P: SubstrateFinalitySyncPipeline>(
+	finality_source: &SubstrateFinalitySource<P>,
 	relay_task_name: &str,
-) -> Result<SourceChain::BlockNumber, relay_substrate_client::Error>
-where
-	SubstrateFinalitySource<SourceChain, P>: FinalitySourceClient<P>,
-	P: FinalitySyncPipeline<Number = SourceChain::BlockNumber>,
-{
+) -> Result<BlockNumberOf<P::SourceChain>, relay_substrate_client::Error> {
 	finality_source.on_chain_best_finalized_block_number().await.map_err(|error| {
 		log::error!(
 			target: "bridge",
@@ -384,14 +345,14 @@ where
 /// Read best finalized source block number from target client.
 ///
 /// Returns `None` if we have failed to read the number.
-async fn best_finalized_source_header_at_target<SourceChain: Chain, TargetChain: Chain, P>(
-	finality_target: &SubstrateFinalityTarget<TargetChain, P>,
+async fn best_finalized_source_header_at_target<P: SubstrateFinalitySyncPipeline>(
+	finality_target: &SubstrateFinalityTarget<P>,
 	relay_task_name: &str,
-) -> Result<SourceChain::BlockNumber, <SubstrateFinalityTarget<TargetChain, P> as RelayClient>::Error>
+) -> Result<BlockNumberOf<P::SourceChain>, <SubstrateFinalityTarget<P> as RelayClient>::Error>
 where
-	SubstrateFinalityTarget<TargetChain, P>: FinalityTargetClient<P::FinalitySyncPipeline>,
-	P: SubstrateFinalitySyncPipeline,
-	P::FinalitySyncPipeline: FinalitySyncPipeline<Number = SourceChain::BlockNumber>,
+	AccountIdOf<P::TargetChain>:
+		From<<AccountKeyPairOf<P::TransactionSignScheme> as sp_core::Pair>::Public>,
+	P::TransactionSignScheme: TransactionSignScheme<Chain = P::TargetChain>,
 {
 	finality_target.best_finalized_source_block_number().await.map_err(|error| {
 		log::error!(
@@ -408,17 +369,13 @@ where
 /// Read first mandatory header in given inclusive range.
 ///
 /// Returns `Ok(None)` if there were no mandatory headers in the range.
-async fn find_mandatory_header_in_range<SourceChain: Chain, P>(
-	finality_source: &SubstrateFinalitySource<SourceChain, P>,
-	range: (SourceChain::BlockNumber, SourceChain::BlockNumber),
-) -> Result<Option<SourceChain::BlockNumber>, relay_substrate_client::Error>
-where
-	SubstrateFinalitySource<SourceChain, P>: FinalitySourceClient<P>,
-	P: FinalitySyncPipeline<Number = SourceChain::BlockNumber>,
-{
+async fn find_mandatory_header_in_range<P: SubstrateFinalitySyncPipeline>(
+	finality_source: &SubstrateFinalitySource<P>,
+	range: (BlockNumberOf<P::SourceChain>, BlockNumberOf<P::SourceChain>),
+) -> Result<Option<BlockNumberOf<P::SourceChain>>, relay_substrate_client::Error> {
 	let mut current = range.0;
 	while current <= range.1 {
-		let header: SyncHeader<SourceChain::Header> =
+		let header: SyncHeader<HeaderOf<P::SourceChain>> =
 			finality_source.client().header_by_number(current).await?.into();
 		if header.is_mandatory() {
 			return Ok(Some(current))