Common crate for substrate-relay (#1082)

* lib-substrate-relay

* pass to build

* remove unused file

* clippy && tests

* Changed follow suggestions

* License

* Make changes follow suggested

* docs

* fix fmt

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

@@ -0,0 +1,51 @@
+[package]
+name = "substrate-relay-helper"
+version = "0.1.0"
+authors = ["Parity Technologies <admin@parity.io>"]
+edition = "2018"
+license = "GPL-3.0-or-later WITH Classpath-exception-2.0"
+
+[dependencies]
+anyhow = "1.0"
+async-std = "1.9.0"
+async-trait = "0.1.42"
+codec = { package = "parity-scale-codec", version = "2.2.0" }
+futures = "0.3.12"
+num-format = "0.4"
+num-traits = "0.2"
+log = "0.4.14"
+
+
+# Bridge dependencies
+
+bp-header-chain = { path = "../../primitives/header-chain" }
+bridge-runtime-common = { path = "../../bin/runtime-common" }
+
+finality-grandpa = { version = "0.14.0" }
+finality-relay = { path = "../finality" }
+relay-utils = { path = "../utils" }
+headers-relay = { path = "../headers" }
+messages-relay = { path = "../messages" }
+relay-substrate-client = { path = "../client-substrate" }
+
+pallet-bridge-messages = { path = "../../modules/messages" }
+
+bp-runtime = { path = "../../primitives/runtime" }
+bp-messages = { path = "../../primitives/messages" }
+
+
+# Substrate Dependencies
+
+frame-support = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-finality-grandpa = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-runtime = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-trie = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-version = { git = "https://github.com/paritytech/substrate", branch = "master" }
+
+[dev-dependencies]
+relay-millau-client = { path = "../client-millau" }
+relay-rialto-client = { path = "../client-rialto" }
+bp-rialto = { path = "../../primitives/chain-rialto" }
+bp-millau = { path = "../../primitives/chain-millau" }
+rialto-runtime = { path = "../../bin/rialto/runtime" }

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

@@ -0,0 +1,160 @@
+// 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/>.
+
+//! Substrate-to-Substrate headers sync entrypoint.
+
+use crate::finality_target::SubstrateFinalityTarget;
+
+use bp_header_chain::justification::GrandpaJustification;
+use finality_relay::{FinalitySyncParams, FinalitySyncPipeline};
+use relay_substrate_client::{finality_source::FinalitySource, BlockNumberOf, Chain, Client, HashOf, SyncHeader};
+use relay_utils::{metrics::MetricsParams, BlockNumberBase};
+use sp_core::Bytes;
+use std::{fmt::Debug, marker::PhantomData, time::Duration};
+
+/// Default synchronization loop timeout.
+pub(crate) const STALL_TIMEOUT: Duration = Duration::from_secs(120);
+/// Default limit of recent finality proofs.
+///
+/// Finality delay of 4096 blocks is unlikely to happen in practice in
+/// Substrate+GRANDPA based chains (good to know).
+pub(crate) const RECENT_FINALITY_PROOFS_LIMIT: usize = 4096;
+
+/// Headers sync pipeline for Substrate <-> Substrate relays.
+pub trait SubstrateFinalitySyncPipeline: 'static + Clone + Debug + Send + Sync {
+	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.
+	type TargetChain: Chain;
+
+	/// Customize metrics exposed by headers sync loop.
+	fn customize_metrics(params: MetricsParams) -> anyhow::Result<MetricsParams> {
+		Ok(params)
+	}
+
+	/// 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) -> <Self::TargetChain as Chain>::AccountId;
+
+	/// Make submit header transaction.
+	fn make_submit_finality_proof_transaction(
+		&self,
+		transaction_nonce: <Self::TargetChain as Chain>::Index,
+		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()
+	}
+}
+
+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(),
+		}
+	}
+}
+
+impl<SourceChain, TargetChain, TargetSign> FinalitySyncPipeline
+	for SubstrateFinalityToSubstrate<SourceChain, TargetChain, TargetSign>
+where
+	SourceChain: Clone + Chain + Debug,
+	BlockNumberOf<SourceChain>: BlockNumberBase,
+	TargetChain: Clone + Chain + Debug,
+	TargetSign: 'static + Clone + Send + Sync,
+{
+	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>;
+}
+
+/// Run Substrate-to-Substrate finality sync.
+pub async fn run<SourceChain, TargetChain, P>(
+	pipeline: P,
+	source_client: Client<SourceChain>,
+	target_client: Client<TargetChain>,
+	only_mandatory_headers: bool,
+	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,
+{
+	log::info!(
+		target: "bridge",
+		"Starting {} -> {} finality proof relay",
+		SourceChain::NAME,
+		TargetChain::NAME,
+	);
+
+	finality_relay::run(
+		FinalitySource::new(source_client, None),
+		SubstrateFinalityTarget::new(target_client, pipeline),
+		FinalitySyncParams {
+			tick: std::cmp::max(SourceChain::AVERAGE_BLOCK_INTERVAL, TargetChain::AVERAGE_BLOCK_INTERVAL),
+			recent_finality_proofs_limit: RECENT_FINALITY_PROOFS_LIMIT,
+			stall_timeout: STALL_TIMEOUT,
+			only_mandatory_headers,
+		},
+		metrics_params,
+		futures::future::pending(),
+	)
+	.await
+	.map_err(|e| anyhow::format_err!("{}", e))
+}

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

@@ -0,0 +1,99 @@
+// 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/>.
+
+//! 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.
+
+use crate::finality_pipeline::SubstrateFinalitySyncPipeline;
+
+use async_trait::async_trait;
+use codec::Decode;
+use finality_relay::{FinalitySyncPipeline, TargetClient};
+use relay_substrate_client::{Chain, Client, Error as SubstrateError};
+use relay_utils::relay_loop::Client as RelayClient;
+
+/// Substrate client as Substrate finality target.
+pub struct SubstrateFinalityTarget<C: Chain, P> {
+	client: Client<C>,
+	pipeline: P,
+}
+
+impl<C: Chain, P> SubstrateFinalityTarget<C, P> {
+	/// Create new Substrate headers target.
+	pub fn new(client: Client<C>, pipeline: P) -> Self {
+		SubstrateFinalityTarget { client, pipeline }
+	}
+}
+
+impl<C: Chain, P: SubstrateFinalitySyncPipeline> Clone for SubstrateFinalityTarget<C, P> {
+	fn clone(&self) -> Self {
+		SubstrateFinalityTarget {
+			client: self.client.clone(),
+			pipeline: self.pipeline.clone(),
+		}
+	}
+}
+
+#[async_trait]
+impl<C: Chain, P: SubstrateFinalitySyncPipeline> RelayClient for SubstrateFinalityTarget<C, P> {
+	type Error = SubstrateError;
+
+	async fn reconnect(&mut self) -> Result<(), SubstrateError> {
+		self.client.reconnect().await
+	}
+}
+
+#[async_trait]
+impl<C, P> TargetClient<P::FinalitySyncPipeline> for SubstrateFinalityTarget<C, P>
+where
+	C: Chain,
+	P: SubstrateFinalitySyncPipeline<TargetChain = C>,
+	<P::FinalitySyncPipeline as FinalitySyncPipeline>::Number: Decode,
+	<P::FinalitySyncPipeline as FinalitySyncPipeline>::Hash: Decode,
+{
+	async fn best_finalized_source_block_number(
+		&self,
+	) -> Result<<P::FinalitySyncPipeline as FinalitySyncPipeline>::Number, SubstrateError> {
+		// 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)
+		.await?
+		.best_finalized_peer_at_best_self
+		.0)
+	}
+
+	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();
+		self.client
+			.submit_signed_extrinsic(transactions_author, move |transaction_nonce| {
+				pipeline.make_submit_finality_proof_transaction(transaction_nonce, header, proof)
+			})
+			.await
+			.map(drop)
+	}
+}

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

@@ -0,0 +1,266 @@
+// 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/>.
+
+//! Initialize Substrate -> Substrate headers bridge.
+//!
+//! Initialization is a transaction that calls `initialize()` function of the
+//! `pallet-bridge-grandpa` pallet. This transaction brings initial header
+//! and authorities set from source to target chain. The headers sync starts
+//! with this header.
+
+use bp_header_chain::InitializationData;
+use bp_header_chain::{
+	find_grandpa_authorities_scheduled_change,
+	justification::{verify_justification, GrandpaJustification},
+};
+use codec::Decode;
+use finality_grandpa::voter_set::VoterSet;
+use num_traits::{One, Zero};
+use relay_substrate_client::{Chain, Client};
+use sp_core::Bytes;
+use sp_finality_grandpa::AuthorityList as GrandpaAuthoritiesSet;
+use sp_runtime::traits::Header as HeaderT;
+
+/// Submit headers-bridge initialization transaction.
+pub async fn initialize<SourceChain: Chain, TargetChain: Chain>(
+	source_client: Client<SourceChain>,
+	target_client: Client<TargetChain>,
+	target_transactions_signer: TargetChain::AccountId,
+	prepare_initialize_transaction: impl FnOnce(TargetChain::Index, InitializationData<SourceChain::Header>) -> Bytes
+		+ Send
+		+ 'static,
+) {
+	let result = do_initialize(
+		source_client,
+		target_client,
+		target_transactions_signer,
+		prepare_initialize_transaction,
+	)
+	.await;
+
+	match result {
+		Ok(tx_hash) => log::info!(
+			target: "bridge",
+			"Successfully submitted {}-headers bridge initialization transaction to {}: {:?}",
+			SourceChain::NAME,
+			TargetChain::NAME,
+			tx_hash,
+		),
+		Err(err) => log::error!(
+			target: "bridge",
+			"Failed to submit {}-headers bridge initialization transaction to {}: {:?}",
+			SourceChain::NAME,
+			TargetChain::NAME,
+			err,
+		),
+	}
+}
+
+/// Craft and submit initialization transaction, returning any error that may occur.
+async fn do_initialize<SourceChain: Chain, TargetChain: Chain>(
+	source_client: Client<SourceChain>,
+	target_client: Client<TargetChain>,
+	target_transactions_signer: TargetChain::AccountId,
+	prepare_initialize_transaction: impl FnOnce(TargetChain::Index, InitializationData<SourceChain::Header>) -> Bytes
+		+ Send
+		+ 'static,
+) -> Result<TargetChain::Hash, String> {
+	let initialization_data = prepare_initialization_data(source_client).await?;
+	log::info!(
+		target: "bridge",
+		"Prepared initialization data for {}-headers bridge at {}: {:?}",
+		SourceChain::NAME,
+		TargetChain::NAME,
+		initialization_data,
+	);
+
+	let initialization_tx_hash = target_client
+		.submit_signed_extrinsic(target_transactions_signer, move |transaction_nonce| {
+			prepare_initialize_transaction(transaction_nonce, initialization_data)
+		})
+		.await
+		.map_err(|err| format!("Failed to submit {} transaction: {:?}", TargetChain::NAME, err))?;
+	Ok(initialization_tx_hash)
+}
+
+/// Prepare initialization data for the GRANDPA verifier pallet.
+async fn prepare_initialization_data<SourceChain: Chain>(
+	source_client: Client<SourceChain>,
+) -> Result<InitializationData<SourceChain::Header>, String> {
+	// In ideal world we just need to get best finalized header and then to read GRANDPA authorities
+	// set (`pallet_grandpa::CurrentSetId` + `GrandpaApi::grandpa_authorities()`) at this header.
+	//
+	// But now there are problems with this approach - `CurrentSetId` may return invalid value. So here
+	// we're waiting for the next justification, read the authorities set and then try to figure out
+	// the set id with bruteforce.
+	let justifications = source_client
+		.subscribe_justifications()
+		.await
+		.map_err(|err| format!("Failed to subscribe to {} justifications: {:?}", SourceChain::NAME, err))?;
+
+	// Read next justification - the header that it finalizes will be used as initial header.
+	let justification = justifications
+		.next()
+		.await
+		.map_err(|err| err.to_string())
+		.and_then(|justification| justification.ok_or_else(|| "stream has ended unexpectedly".into()))
+		.map_err(|err| {
+			format!(
+				"Failed to read {} justification from the stream: {}",
+				SourceChain::NAME,
+				err,
+			)
+		})?;
+
+	// Read initial header.
+	let justification: GrandpaJustification<SourceChain::Header> = Decode::decode(&mut &justification.0[..])
+		.map_err(|err| format!("Failed to decode {} justification: {:?}", SourceChain::NAME, err))?;
+
+	let (initial_header_hash, initial_header_number) =
+		(justification.commit.target_hash, justification.commit.target_number);
+
+	let initial_header = source_header(&source_client, initial_header_hash).await?;
+	log::trace!(target: "bridge", "Selected {} initial header: {}/{}",
+		SourceChain::NAME,
+		initial_header_number,
+		initial_header_hash,
+	);
+
+	// Read GRANDPA authorities set at initial header.
+	let initial_authorities_set = source_authorities_set(&source_client, initial_header_hash).await?;
+	log::trace!(target: "bridge", "Selected {} initial authorities set: {:?}",
+		SourceChain::NAME,
+		initial_authorities_set,
+	);
+
+	// If initial header changes the GRANDPA authorities set, then we need previous authorities
+	// to verify justification.
+	let mut authorities_for_verification = initial_authorities_set.clone();
+	let scheduled_change = find_grandpa_authorities_scheduled_change(&initial_header);
+	assert!(
+		scheduled_change.as_ref().map(|c| c.delay.is_zero()).unwrap_or(true),
+		"GRANDPA authorities change at {} scheduled to happen in {:?} blocks. We expect\
+		regular hange to have zero delay",
+		initial_header_hash,
+		scheduled_change.as_ref().map(|c| c.delay),
+	);
+	let schedules_change = scheduled_change.is_some();
+	if schedules_change {
+		authorities_for_verification = source_authorities_set(&source_client, *initial_header.parent_hash()).await?;
+		log::trace!(
+			target: "bridge",
+			"Selected {} header is scheduling GRANDPA authorities set changes. Using previous set: {:?}",
+			SourceChain::NAME,
+			authorities_for_verification,
+		);
+	}
+
+	// Now let's try to guess authorities set id by verifying justification.
+	let mut initial_authorities_set_id = 0;
+	let mut min_possible_block_number = SourceChain::BlockNumber::zero();
+	let authorities_for_verification = VoterSet::new(authorities_for_verification.clone()).ok_or_else(|| {
+		format!(
+			"Read invalid {} authorities set: {:?}",
+			SourceChain::NAME,
+			authorities_for_verification,
+		)
+	})?;
+	loop {
+		log::trace!(
+			target: "bridge", "Trying {} GRANDPA authorities set id: {}",
+			SourceChain::NAME,
+			initial_authorities_set_id,
+		);
+
+		let is_valid_set_id = verify_justification::<SourceChain::Header>(
+			(initial_header_hash, initial_header_number),
+			initial_authorities_set_id,
+			&authorities_for_verification,
+			&justification,
+		)
+		.is_ok();
+
+		if is_valid_set_id {
+			break;
+		}
+
+		initial_authorities_set_id += 1;
+		min_possible_block_number += One::one();
+		if min_possible_block_number > initial_header_number {
+			// there can't be more authorities set changes than headers => if we have reached `initial_block_number`
+			// and still have not found correct value of `initial_authorities_set_id`, then something
+			// else is broken => fail
+			return Err(format!(
+				"Failed to guess initial {} GRANDPA authorities set id: checked all\
+			possible ids in range [0; {}]",
+				SourceChain::NAME,
+				initial_header_number
+			));
+		}
+	}
+
+	Ok(InitializationData {
+		header: initial_header,
+		authority_list: initial_authorities_set,
+		set_id: if schedules_change {
+			initial_authorities_set_id + 1
+		} else {
+			initial_authorities_set_id
+		},
+		is_halted: false,
+	})
+}
+
+/// Read header by hash from the source client.
+async fn source_header<SourceChain: Chain>(
+	source_client: &Client<SourceChain>,
+	header_hash: SourceChain::Hash,
+) -> Result<SourceChain::Header, String> {
+	source_client.header_by_hash(header_hash).await.map_err(|err| {
+		format!(
+			"Failed to retrive {} header with hash {}: {:?}",
+			SourceChain::NAME,
+			header_hash,
+			err,
+		)
+	})
+}
+
+/// Read GRANDPA authorities set at given header.
+async fn source_authorities_set<SourceChain: Chain>(
+	source_client: &Client<SourceChain>,
+	header_hash: SourceChain::Hash,
+) -> Result<GrandpaAuthoritiesSet, String> {
+	let raw_authorities_set = source_client
+		.grandpa_authorities_set(header_hash)
+		.await
+		.map_err(|err| {
+			format!(
+				"Failed to retrive {} GRANDPA authorities set at header {}: {:?}",
+				SourceChain::NAME,
+				header_hash,
+				err,
+			)
+		})?;
+	GrandpaAuthoritiesSet::decode(&mut &raw_authorities_set[..]).map_err(|err| {
+		format!(
+			"Failed to decode {} GRANDPA authorities set at header {}: {:?}",
+			SourceChain::NAME,
+			header_hash,
+			err,
+		)
+	})
+}

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

@@ -0,0 +1,41 @@
+// 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/>.
+
+//! Substrate relay helpers
+
+use relay_utils::metrics::{FloatJsonValueMetric, PrometheusError, Registry};
+
+/// Creates standalone token price metric.
+pub fn token_price_metric(
+	registry: &Registry,
+	prefix: Option<&str>,
+	token_id: &str,
+) -> Result<FloatJsonValueMetric, PrometheusError> {
+	FloatJsonValueMetric::new(
+		registry,
+		prefix,
+		format!(
+			"https://api.coingecko.com/api/v3/simple/price?ids={}&vs_currencies=btc",
+			token_id
+		),
+		format!("$.{}.btc", token_id),
+		format!("{}_to_base_conversion_rate", token_id.replace("-", "_")),
+		format!(
+			"Rate used to convert from {} to some BASE tokens",
+			token_id.to_uppercase()
+		),
+	)
+}

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

@@ -0,0 +1,28 @@
+// 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/>.
+
+//! The library of substrate relay. contains some public codes to provide to substrate relay.
+
+#![warn(missing_docs)]
+
+pub mod finality_pipeline;
+pub mod finality_target;
+pub mod headers_initialize;
+pub mod helpers;
+pub mod messages_lane;
+pub mod messages_source;
+pub mod messages_target;
+pub mod on_demand_headers;

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

@@ -0,0 +1,309 @@
+// 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/>.
+
+use crate::messages_source::SubstrateMessagesProof;
+use crate::messages_target::SubstrateMessagesReceivingProof;
+use crate::on_demand_headers::OnDemandHeadersRelay;
+
+use bp_messages::{LaneId, MessageNonce};
+use frame_support::weights::Weight;
+use messages_relay::message_lane::{MessageLane, SourceHeaderIdOf, TargetHeaderIdOf};
+use relay_substrate_client::{
+	metrics::{FloatStorageValueMetric, StorageProofOverheadMetric},
+	BlockNumberOf, Chain, Client, HashOf,
+};
+use relay_utils::{
+	metrics::{F64SharedRef, MetricsParams},
+	BlockNumberBase,
+};
+use sp_core::{storage::StorageKey, Bytes};
+use sp_runtime::FixedU128;
+use std::ops::RangeInclusive;
+
+/// Substrate <-> Substrate messages relay parameters.
+pub struct MessagesRelayParams<SC: Chain, SS, TC: Chain, TS> {
+	/// Messages source client.
+	pub source_client: Client<SC>,
+	/// Sign parameters for messages source chain.
+	pub source_sign: SS,
+	/// Messages target client.
+	pub target_client: Client<TC>,
+	/// Sign parameters for messages target chain.
+	pub target_sign: TS,
+	/// Optional on-demand source to target headers relay.
+	pub source_to_target_headers_relay: Option<OnDemandHeadersRelay<SC>>,
+	/// Optional on-demand target to source headers relay.
+	pub target_to_source_headers_relay: Option<OnDemandHeadersRelay<TC>>,
+	/// Identifier of lane that needs to be served.
+	pub lane_id: LaneId,
+	/// Relayer operating mode.
+	pub relayer_mode: messages_relay::message_lane_loop::RelayerMode,
+	/// Metrics parameters.
+	pub metrics_params: MetricsParams,
+}
+
+/// Message sync pipeline for Substrate <-> Substrate relays.
+pub trait SubstrateMessageLane: 'static + Clone + Send + Sync {
+	type MessageLane: MessageLane;
+
+	/// Name of the runtime method that returns dispatch weight of outbound messages at the source chain.
+	const OUTBOUND_LANE_MESSAGE_DETAILS_METHOD: &'static str;
+	/// Name of the runtime method that returns latest generated nonce at the source chain.
+	const OUTBOUND_LANE_LATEST_GENERATED_NONCE_METHOD: &'static str;
+	/// Name of the runtime method that returns latest received (confirmed) nonce at the the source chain.
+	const OUTBOUND_LANE_LATEST_RECEIVED_NONCE_METHOD: &'static str;
+
+	/// Name of the runtime method that returns latest received nonce at the target chain.
+	const INBOUND_LANE_LATEST_RECEIVED_NONCE_METHOD: &'static str;
+	/// Name of the runtime method that returns the latest confirmed (reward-paid) nonce at the target chain.
+	const INBOUND_LANE_LATEST_CONFIRMED_NONCE_METHOD: &'static str;
+	/// Number of the runtime method that returns state of "unrewarded relayers" set at the target chain.
+	const INBOUND_LANE_UNREWARDED_RELAYERS_STATE: &'static str;
+
+	/// 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;
+	/// Name of the runtime method that returns id of best finalized target header at source chain.
+	const BEST_FINALIZED_TARGET_HEADER_ID_AT_SOURCE: &'static str;
+
+	/// Source chain.
+	type SourceChain: Chain;
+	/// Target chain.
+	type TargetChain: Chain;
+
+	/// Returns id of account that we're using to sign transactions at target chain (messages proof).
+	fn target_transactions_author(&self) -> <Self::TargetChain as Chain>::AccountId;
+
+	/// Make messages delivery transaction.
+	fn make_messages_delivery_transaction(
+		&self,
+		transaction_nonce: <Self::TargetChain as Chain>::Index,
+		generated_at_header: SourceHeaderIdOf<Self::MessageLane>,
+		nonces: RangeInclusive<MessageNonce>,
+		proof: <Self::MessageLane as MessageLane>::MessagesProof,
+	) -> Bytes;
+
+	/// Returns id of account that we're using to sign transactions at source chain (delivery proof).
+	fn source_transactions_author(&self) -> <Self::SourceChain as Chain>::AccountId;
+
+	/// Make messages receiving proof transaction.
+	fn make_messages_receiving_proof_transaction(
+		&self,
+		transaction_nonce: <Self::SourceChain as Chain>::Index,
+		generated_at_header: TargetHeaderIdOf<Self::MessageLane>,
+		proof: <Self::MessageLane as MessageLane>::MessagesReceivingProof,
+	) -> Bytes;
+}
+
+/// Substrate-to-Substrate message lane.
+#[derive(Debug)]
+pub struct SubstrateMessageLaneToSubstrate<Source: Chain, SourceSignParams, Target: Chain, TargetSignParams> {
+	/// Client for the source Substrate chain.
+	pub source_client: Client<Source>,
+	/// Parameters required to sign transactions for source chain.
+	pub source_sign: SourceSignParams,
+	/// Client for the target Substrate chain.
+	pub target_client: Client<Target>,
+	/// Parameters required to sign transactions for target chain.
+	pub target_sign: TargetSignParams,
+	/// Account id of relayer at the source chain.
+	pub relayer_id_at_source: Source::AccountId,
+}
+
+impl<Source: Chain, SourceSignParams: Clone, Target: Chain, TargetSignParams: Clone> Clone
+	for SubstrateMessageLaneToSubstrate<Source, SourceSignParams, Target, TargetSignParams>
+{
+	fn clone(&self) -> Self {
+		Self {
+			source_client: self.source_client.clone(),
+			source_sign: self.source_sign.clone(),
+			target_client: self.target_client.clone(),
+			target_sign: self.target_sign.clone(),
+			relayer_id_at_source: self.relayer_id_at_source.clone(),
+		}
+	}
+}
+
+impl<Source: Chain, SourceSignParams, Target: Chain, TargetSignParams> MessageLane
+	for SubstrateMessageLaneToSubstrate<Source, SourceSignParams, Target, TargetSignParams>
+where
+	SourceSignParams: Clone + Send + Sync + 'static,
+	TargetSignParams: Clone + Send + Sync + 'static,
+	BlockNumberOf<Source>: BlockNumberBase,
+	BlockNumberOf<Target>: BlockNumberBase,
+{
+	const SOURCE_NAME: &'static str = Source::NAME;
+	const TARGET_NAME: &'static str = Target::NAME;
+
+	type MessagesProof = SubstrateMessagesProof<Source>;
+	type MessagesReceivingProof = SubstrateMessagesReceivingProof<Target>;
+
+	type SourceChainBalance = Source::Balance;
+	type SourceHeaderNumber = BlockNumberOf<Source>;
+	type SourceHeaderHash = HashOf<Source>;
+
+	type TargetHeaderNumber = BlockNumberOf<Target>;
+	type TargetHeaderHash = HashOf<Target>;
+}
+
+/// Returns maximal number of messages and their maximal cumulative dispatch weight, based
+/// on given chain parameters.
+pub fn select_delivery_transaction_limits<W: pallet_bridge_messages::WeightInfoExt>(
+	max_extrinsic_weight: Weight,
+	max_unconfirmed_messages_at_inbound_lane: MessageNonce,
+) -> (MessageNonce, Weight) {
+	// We may try to guess accurate value, based on maximal number of messages and per-message
+	// weight overhead, but the relay loop isn't using this info in a super-accurate way anyway.
+	// So just a rough guess: let's say 1/3 of max tx weight is for tx itself and the rest is
+	// for messages dispatch.
+
+	// Another thing to keep in mind is that our runtimes (when this code was written) accept
+	// messages with dispatch weight <= max_extrinsic_weight/2. So we can't reserve less than
+	// that for dispatch.
+
+	let weight_for_delivery_tx = max_extrinsic_weight / 3;
+	let weight_for_messages_dispatch = max_extrinsic_weight - weight_for_delivery_tx;
+
+	let delivery_tx_base_weight =
+		W::receive_messages_proof_overhead() + W::receive_messages_proof_outbound_lane_state_overhead();
+	let delivery_tx_weight_rest = weight_for_delivery_tx - delivery_tx_base_weight;
+	let max_number_of_messages = std::cmp::min(
+		delivery_tx_weight_rest / W::receive_messages_proof_messages_overhead(1),
+		max_unconfirmed_messages_at_inbound_lane,
+	);
+
+	assert!(
+		max_number_of_messages > 0,
+		"Relay should fit at least one message in every delivery transaction",
+	);
+	assert!(
+		weight_for_messages_dispatch >= max_extrinsic_weight / 2,
+		"Relay shall be able to deliver messages with dispatch weight = max_extrinsic_weight / 2",
+	);
+
+	(max_number_of_messages, weight_for_messages_dispatch)
+}
+
+/// Shared references to the values of standalone metrics of the message lane relay loop.
+#[derive(Debug, Clone)]
+pub struct StandaloneMessagesMetrics {
+	/// Shared reference to the actual target -> <base> chain token conversion rate.
+	pub target_to_base_conversion_rate: Option<F64SharedRef>,
+	/// Shared reference to the actual source -> <base> chain token conversion rate.
+	pub source_to_base_conversion_rate: Option<F64SharedRef>,
+}
+
+impl StandaloneMessagesMetrics {
+	/// Return conversion rate from target to source tokens.
+	pub async fn target_to_source_conversion_rate(&self) -> Option<f64> {
+		let target_to_base_conversion_rate = (*self.target_to_base_conversion_rate.as_ref()?.read().await)?;
+		let source_to_base_conversion_rate = (*self.source_to_base_conversion_rate.as_ref()?.read().await)?;
+		Some(target_to_base_conversion_rate / source_to_base_conversion_rate)
+	}
+}
+
+/// Add general standalone metrics for the message lane relay loop.
+pub fn add_standalone_metrics<P: SubstrateMessageLane>(
+	metrics_prefix: Option<String>,
+	metrics_params: MetricsParams,
+	source_client: Client<P::SourceChain>,
+	source_chain_token_id: Option<&str>,
+	target_chain_token_id: Option<&str>,
+	target_to_source_conversion_rate_params: Option<(StorageKey, FixedU128)>,
+) -> anyhow::Result<(MetricsParams, StandaloneMessagesMetrics)> {
+	let mut source_to_base_conversion_rate = None;
+	let mut target_to_base_conversion_rate = None;
+	let mut metrics_params =
+		relay_utils::relay_metrics(metrics_prefix, metrics_params).standalone_metric(|registry, prefix| {
+			StorageProofOverheadMetric::new(
+				registry,
+				prefix,
+				source_client.clone(),
+				format!("{}_storage_proof_overhead", P::SourceChain::NAME.to_lowercase()),
+				format!("{} storage proof overhead", P::SourceChain::NAME),
+			)
+		})?;
+	if let Some((target_to_source_conversion_rate_storage_key, initial_target_to_source_conversion_rate)) =
+		target_to_source_conversion_rate_params
+	{
+		metrics_params = metrics_params.standalone_metric(|registry, prefix| {
+			let metric = FloatStorageValueMetric::<_, sp_runtime::FixedU128>::new(
+				registry,
+				prefix,
+				source_client,
+				target_to_source_conversion_rate_storage_key,
+				Some(initial_target_to_source_conversion_rate),
+				format!(
+					"{}_{}_to_{}_conversion_rate",
+					P::SourceChain::NAME,
+					P::TargetChain::NAME,
+					P::SourceChain::NAME
+				),
+				format!(
+					"{} to {} tokens conversion rate (used by {})",
+					P::TargetChain::NAME,
+					P::SourceChain::NAME,
+					P::SourceChain::NAME
+				),
+			)?;
+			Ok(metric)
+		})?;
+	}
+	if let Some(source_chain_token_id) = source_chain_token_id {
+		metrics_params = metrics_params.standalone_metric(|registry, prefix| {
+			let metric = crate::helpers::token_price_metric(registry, prefix, source_chain_token_id)?;
+			source_to_base_conversion_rate = Some(metric.shared_value_ref());
+			Ok(metric)
+		})?;
+	}
+	if let Some(target_chain_token_id) = target_chain_token_id {
+		metrics_params = metrics_params.standalone_metric(|registry, prefix| {
+			let metric = crate::helpers::token_price_metric(registry, prefix, target_chain_token_id)?;
+			target_to_base_conversion_rate = Some(metric.shared_value_ref());
+			Ok(metric)
+		})?;
+	}
+	Ok((
+		metrics_params.into_params(),
+		StandaloneMessagesMetrics {
+			target_to_base_conversion_rate,
+			source_to_base_conversion_rate,
+		},
+	))
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	type RialtoToMillauMessagesWeights = pallet_bridge_messages::weights::RialtoWeight<rialto_runtime::Runtime>;
+
+	#[test]
+	fn select_delivery_transaction_limits_works() {
+		let (max_count, max_weight) = select_delivery_transaction_limits::<RialtoToMillauMessagesWeights>(
+			bp_millau::max_extrinsic_weight(),
+			bp_millau::MAX_UNREWARDED_RELAYER_ENTRIES_AT_INBOUND_LANE,
+		);
+		assert_eq!(
+			(max_count, max_weight),
+			// We don't actually care about these values, so feel free to update them whenever test
+			// fails. The only thing to do before that is to ensure that new values looks sane: i.e. weight
+			// reserved for messages dispatch allows dispatch of non-trivial messages.
+			//
+			// Any significant change in this values should attract additional attention.
+			(782, 216_583_333_334),
+		);
+	}
+}

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

@@ -0,0 +1,538 @@
+// 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/>.
+
+//! Substrate client as Substrate messages source. The chain we connect to should have
+//! runtime that implements `<BridgedChainName>HeaderApi` to allow bridging with
+//! <BridgedName> chain.
+
+use crate::messages_lane::SubstrateMessageLane;
+use crate::messages_target::SubstrateMessagesReceivingProof;
+use crate::on_demand_headers::OnDemandHeadersRelay;
+
+use async_trait::async_trait;
+use bp_messages::{LaneId, MessageNonce, UnrewardedRelayersState};
+use bp_runtime::{messages::DispatchFeePayment, ChainId};
+use bridge_runtime_common::messages::{
+	source::FromBridgedChainMessagesDeliveryProof, target::FromBridgedChainMessagesProof,
+};
+use codec::{Decode, Encode};
+use frame_support::{traits::Instance, weights::Weight};
+use messages_relay::message_lane::MessageLane;
+use messages_relay::{
+	message_lane::{SourceHeaderIdOf, TargetHeaderIdOf},
+	message_lane_loop::{
+		ClientState, MessageDetails, MessageDetailsMap, MessageProofParameters, SourceClient, SourceClientState,
+	},
+};
+use num_traits::{Bounded, Zero};
+use relay_substrate_client::{Chain, Client, Error as SubstrateError, HashOf, HeaderIdOf};
+use relay_utils::{relay_loop::Client as RelayClient, BlockNumberBase, HeaderId};
+use sp_core::Bytes;
+use sp_runtime::{traits::Header as HeaderT, DeserializeOwned};
+use std::{marker::PhantomData, ops::RangeInclusive};
+
+/// Intermediate message proof returned by the source Substrate node. Includes everything
+/// required to submit to the target node: cumulative dispatch weight of bundled messages and
+/// the proof itself.
+pub type SubstrateMessagesProof<C> = (Weight, FromBridgedChainMessagesProof<HashOf<C>>);
+
+/// Substrate client as Substrate messages source.
+pub struct SubstrateMessagesSource<SC: Chain, TC: Chain, P: SubstrateMessageLane, I> {
+	client: Client<SC>,
+	lane: P,
+	lane_id: LaneId,
+	instance: ChainId,
+	target_to_source_headers_relay: Option<OnDemandHeadersRelay<TC>>,
+	_phantom: PhantomData<I>,
+}
+
+impl<SC: Chain, TC: Chain, P: SubstrateMessageLane, I> SubstrateMessagesSource<SC, TC, P, I> {
+	/// Create new Substrate headers source.
+	pub fn new(
+		client: Client<SC>,
+		lane: P,
+		lane_id: LaneId,
+		instance: ChainId,
+		target_to_source_headers_relay: Option<OnDemandHeadersRelay<TC>>,
+	) -> Self {
+		SubstrateMessagesSource {
+			client,
+			lane,
+			lane_id,
+			instance,
+			target_to_source_headers_relay,
+			_phantom: Default::default(),
+		}
+	}
+}
+
+impl<SC: Chain, TC: Chain, P: SubstrateMessageLane, I> Clone for SubstrateMessagesSource<SC, TC, P, I> {
+	fn clone(&self) -> Self {
+		Self {
+			client: self.client.clone(),
+			lane: self.lane.clone(),
+			lane_id: self.lane_id,
+			instance: self.instance,
+			target_to_source_headers_relay: self.target_to_source_headers_relay.clone(),
+			_phantom: Default::default(),
+		}
+	}
+}
+
+#[async_trait]
+impl<SC, TC, P, I> RelayClient for SubstrateMessagesSource<SC, TC, P, I>
+where
+	SC: Chain,
+	TC: Chain,
+	P: SubstrateMessageLane,
+	I: Send + Sync + Instance,
+{
+	type Error = SubstrateError;
+
+	async fn reconnect(&mut self) -> Result<(), SubstrateError> {
+		self.client.reconnect().await
+	}
+}
+
+#[async_trait]
+impl<SC, TC, P, I> SourceClient<P::MessageLane> for SubstrateMessagesSource<SC, TC, P, I>
+where
+	SC: Chain<
+		Hash = <P::MessageLane as MessageLane>::SourceHeaderHash,
+		BlockNumber = <P::MessageLane as MessageLane>::SourceHeaderNumber,
+		Balance = <P::MessageLane as MessageLane>::SourceChainBalance,
+	>,
+	SC::Hash: Copy,
+	SC::BlockNumber: Copy,
+	SC::Balance: Decode + Bounded,
+	SC::Header: DeserializeOwned,
+	SC::Index: DeserializeOwned,
+	SC::BlockNumber: BlockNumberBase,
+	TC: Chain<
+		Hash = <P::MessageLane as MessageLane>::TargetHeaderHash,
+		BlockNumber = <P::MessageLane as MessageLane>::TargetHeaderNumber,
+	>,
+	P: SubstrateMessageLane<SourceChain = SC, TargetChain = TC>,
+	P::MessageLane: MessageLane<
+		MessagesProof = SubstrateMessagesProof<SC>,
+		MessagesReceivingProof = SubstrateMessagesReceivingProof<TC>,
+	>,
+	<P::MessageLane as MessageLane>::TargetHeaderNumber: Decode,
+	<P::MessageLane as MessageLane>::TargetHeaderHash: Decode,
+	I: Send + Sync + Instance,
+{
+	async fn state(&self) -> Result<SourceClientState<P::MessageLane>, 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
+		self.client.ensure_synced().await?;
+
+		read_client_state::<
+			_,
+			<P::MessageLane as MessageLane>::TargetHeaderHash,
+			<P::MessageLane as MessageLane>::TargetHeaderNumber,
+		>(&self.client, P::BEST_FINALIZED_TARGET_HEADER_ID_AT_SOURCE)
+		.await
+	}
+
+	async fn latest_generated_nonce(
+		&self,
+		id: SourceHeaderIdOf<P::MessageLane>,
+	) -> Result<(SourceHeaderIdOf<P::MessageLane>, MessageNonce), SubstrateError> {
+		let encoded_response = self
+			.client
+			.state_call(
+				P::OUTBOUND_LANE_LATEST_GENERATED_NONCE_METHOD.into(),
+				Bytes(self.lane_id.encode()),
+				Some(id.1),
+			)
+			.await?;
+		let latest_generated_nonce: MessageNonce =
+			Decode::decode(&mut &encoded_response.0[..]).map_err(SubstrateError::ResponseParseFailed)?;
+		Ok((id, latest_generated_nonce))
+	}
+
+	async fn latest_confirmed_received_nonce(
+		&self,
+		id: SourceHeaderIdOf<P::MessageLane>,
+	) -> Result<(SourceHeaderIdOf<P::MessageLane>, MessageNonce), SubstrateError> {
+		let encoded_response = self
+			.client
+			.state_call(
+				P::OUTBOUND_LANE_LATEST_RECEIVED_NONCE_METHOD.into(),
+				Bytes(self.lane_id.encode()),
+				Some(id.1),
+			)
+			.await?;
+		let latest_received_nonce: MessageNonce =
+			Decode::decode(&mut &encoded_response.0[..]).map_err(SubstrateError::ResponseParseFailed)?;
+		Ok((id, latest_received_nonce))
+	}
+
+	async fn generated_message_details(
+		&self,
+		id: SourceHeaderIdOf<P::MessageLane>,
+		nonces: RangeInclusive<MessageNonce>,
+	) -> Result<MessageDetailsMap<<P::MessageLane as MessageLane>::SourceChainBalance>, SubstrateError> {
+		let encoded_response = self
+			.client
+			.state_call(
+				P::OUTBOUND_LANE_MESSAGE_DETAILS_METHOD.into(),
+				Bytes((self.lane_id, nonces.start(), nonces.end()).encode()),
+				Some(id.1),
+			)
+			.await?;
+
+		make_message_details_map::<SC>(
+			Decode::decode(&mut &encoded_response.0[..]).map_err(SubstrateError::ResponseParseFailed)?,
+			nonces,
+		)
+	}
+
+	async fn prove_messages(
+		&self,
+		id: SourceHeaderIdOf<P::MessageLane>,
+		nonces: RangeInclusive<MessageNonce>,
+		proof_parameters: MessageProofParameters,
+	) -> Result<
+		(
+			SourceHeaderIdOf<P::MessageLane>,
+			RangeInclusive<MessageNonce>,
+			<P::MessageLane as MessageLane>::MessagesProof,
+		),
+		SubstrateError,
+	> {
+		let mut storage_keys = Vec::with_capacity(nonces.end().saturating_sub(*nonces.start()) as usize + 1);
+		let mut message_nonce = *nonces.start();
+		while message_nonce <= *nonces.end() {
+			let message_key = pallet_bridge_messages::storage_keys::message_key::<I>(&self.lane_id, message_nonce);
+			storage_keys.push(message_key);
+			message_nonce += 1;
+		}
+		if proof_parameters.outbound_state_proof_required {
+			storage_keys.push(pallet_bridge_messages::storage_keys::outbound_lane_data_key::<I>(
+				&self.lane_id,
+			));
+		}
+
+		let proof = self
+			.client
+			.prove_storage(storage_keys, id.1)
+			.await?
+			.iter_nodes()
+			.collect();
+		let proof = FromBridgedChainMessagesProof {
+			bridged_header_hash: id.1,
+			storage_proof: proof,
+			lane: self.lane_id,
+			nonces_start: *nonces.start(),
+			nonces_end: *nonces.end(),
+		};
+		Ok((id, nonces, (proof_parameters.dispatch_weight, proof)))
+	}
+
+	async fn submit_messages_receiving_proof(
+		&self,
+		generated_at_block: TargetHeaderIdOf<P::MessageLane>,
+		proof: <P::MessageLane as MessageLane>::MessagesReceivingProof,
+	) -> Result<(), SubstrateError> {
+		let lane = self.lane.clone();
+		self.client
+			.submit_signed_extrinsic(self.lane.source_transactions_author(), move |transaction_nonce| {
+				lane.make_messages_receiving_proof_transaction(transaction_nonce, generated_at_block, proof)
+			})
+			.await?;
+		Ok(())
+	}
+
+	async fn require_target_header_on_source(&self, id: TargetHeaderIdOf<P::MessageLane>) {
+		if let Some(ref target_to_source_headers_relay) = self.target_to_source_headers_relay {
+			target_to_source_headers_relay.require_finalized_header(id).await;
+		}
+	}
+
+	async fn estimate_confirmation_transaction(&self) -> <P::MessageLane as MessageLane>::SourceChainBalance {
+		self.client
+			.estimate_extrinsic_fee(self.lane.make_messages_receiving_proof_transaction(
+				Zero::zero(),
+				HeaderId(Default::default(), Default::default()),
+				prepare_dummy_messages_delivery_proof::<SC, TC>(),
+			))
+			.await
+			.unwrap_or_else(|_| SC::Balance::max_value())
+	}
+}
+
+/// Prepare 'dummy' messages delivery proof that will compose the delivery confirmation transaction.
+///
+/// We don't care about proof actually being the valid proof, because its validity doesn't
+/// affect the call weight - we only care about its size.
+fn prepare_dummy_messages_delivery_proof<SC: Chain, TC: Chain>() -> SubstrateMessagesReceivingProof<TC> {
+	let single_message_confirmation_size =
+		bp_messages::InboundLaneData::<()>::encoded_size_hint(SC::MAXIMAL_ENCODED_ACCOUNT_ID_SIZE, 1, 1)
+			.unwrap_or(u32::MAX);
+	let proof_size = TC::STORAGE_PROOF_OVERHEAD.saturating_add(single_message_confirmation_size);
+	(
+		UnrewardedRelayersState {
+			unrewarded_relayer_entries: 1,
+			messages_in_oldest_entry: 1,
+			total_messages: 1,
+		},
+		FromBridgedChainMessagesDeliveryProof {
+			bridged_header_hash: Default::default(),
+			storage_proof: vec![vec![0; proof_size as usize]],
+			lane: Default::default(),
+		},
+	)
+}
+
+pub async fn read_client_state<SelfChain, BridgedHeaderHash, BridgedHeaderNumber>(
+	self_client: &Client<SelfChain>,
+	best_finalized_header_id_method_name: &str,
+) -> Result<ClientState<HeaderIdOf<SelfChain>, HeaderId<BridgedHeaderHash, BridgedHeaderNumber>>, SubstrateError>
+where
+	SelfChain: Chain,
+	SelfChain::Header: DeserializeOwned,
+	SelfChain::Index: DeserializeOwned,
+	BridgedHeaderHash: Decode,
+	BridgedHeaderNumber: Decode,
+{
+	// let's read our state first: we need best finalized header hash on **this** chain
+	let self_best_finalized_header_hash = self_client.best_finalized_header_hash().await?;
+	let self_best_finalized_header = self_client.header_by_hash(self_best_finalized_header_hash).await?;
+	let self_best_finalized_id = HeaderId(*self_best_finalized_header.number(), self_best_finalized_header_hash);
+
+	// now let's read our best header on **this** chain
+	let self_best_header = self_client.best_header().await?;
+	let self_best_hash = self_best_header.hash();
+	let self_best_id = HeaderId(*self_best_header.number(), self_best_hash);
+
+	// now let's read id of best finalized peer header at our best finalized block
+	let encoded_best_finalized_peer_on_self = self_client
+		.state_call(
+			best_finalized_header_id_method_name.into(),
+			Bytes(Vec::new()),
+			Some(self_best_hash),
+		)
+		.await?;
+	let decoded_best_finalized_peer_on_self: (BridgedHeaderNumber, BridgedHeaderHash) =
+		Decode::decode(&mut &encoded_best_finalized_peer_on_self.0[..]).map_err(SubstrateError::ResponseParseFailed)?;
+	let peer_on_self_best_finalized_id = HeaderId(
+		decoded_best_finalized_peer_on_self.0,
+		decoded_best_finalized_peer_on_self.1,
+	);
+
+	Ok(ClientState {
+		best_self: self_best_id,
+		best_finalized_self: self_best_finalized_id,
+		best_finalized_peer_at_best_self: peer_on_self_best_finalized_id,
+	})
+}
+
+fn make_message_details_map<C: Chain>(
+	weights: Vec<bp_messages::MessageDetails<C::Balance>>,
+	nonces: RangeInclusive<MessageNonce>,
+) -> Result<MessageDetailsMap<C::Balance>, SubstrateError> {
+	let make_missing_nonce_error = |expected_nonce| {
+		Err(SubstrateError::Custom(format!(
+			"Missing nonce {} in message_details call result. Expected all nonces from {:?}",
+			expected_nonce, nonces,
+		)))
+	};
+
+	let mut weights_map = MessageDetailsMap::new();
+
+	// this is actually prevented by external logic
+	if nonces.is_empty() {
+		return Ok(weights_map);
+	}
+
+	// check if last nonce is missing - loop below is not checking this
+	let last_nonce_is_missing = weights
+		.last()
+		.map(|details| details.nonce != *nonces.end())
+		.unwrap_or(true);
+	if last_nonce_is_missing {
+		return make_missing_nonce_error(*nonces.end());
+	}
+
+	let mut expected_nonce = *nonces.start();
+	let mut is_at_head = true;
+
+	for details in weights {
+		match (details.nonce == expected_nonce, is_at_head) {
+			(true, _) => (),
+			(false, true) => {
+				// this may happen if some messages were already pruned from the source node
+				//
+				// this is not critical error and will be auto-resolved by messages lane (and target node)
+				log::info!(
+					target: "bridge",
+					"Some messages are missing from the {} node: {:?}. Target node may be out of sync?",
+					C::NAME,
+					expected_nonce..details.nonce,
+				);
+			}
+			(false, false) => {
+				// some nonces are missing from the middle/tail of the range
+				//
+				// this is critical error, because we can't miss any nonces
+				return make_missing_nonce_error(expected_nonce);
+			}
+		}
+
+		weights_map.insert(
+			details.nonce,
+			MessageDetails {
+				dispatch_weight: details.dispatch_weight,
+				size: details.size as _,
+				reward: details.delivery_and_dispatch_fee,
+				dispatch_fee_payment: DispatchFeePayment::AtSourceChain,
+			},
+		);
+		expected_nonce = details.nonce + 1;
+		is_at_head = false;
+	}
+
+	Ok(weights_map)
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use bp_runtime::messages::DispatchFeePayment;
+	use relay_millau_client::Millau;
+	use relay_rialto_client::Rialto;
+
+	fn message_details_from_rpc(
+		nonces: RangeInclusive<MessageNonce>,
+	) -> Vec<bp_messages::MessageDetails<bp_rialto::Balance>> {
+		nonces
+			.into_iter()
+			.map(|nonce| bp_messages::MessageDetails {
+				nonce,
+				dispatch_weight: 0,
+				size: 0,
+				delivery_and_dispatch_fee: 0,
+				dispatch_fee_payment: DispatchFeePayment::AtSourceChain,
+			})
+			.collect()
+	}
+
+	#[test]
+	fn make_message_details_map_succeeds_if_no_messages_are_missing() {
+		assert_eq!(
+			make_message_details_map::<relay_rialto_client::Rialto>(message_details_from_rpc(1..=3), 1..=3,).unwrap(),
+			vec![
+				(
+					1,
+					MessageDetails {
+						dispatch_weight: 0,
+						size: 0,
+						reward: 0,
+						dispatch_fee_payment: DispatchFeePayment::AtSourceChain,
+					}
+				),
+				(
+					2,
+					MessageDetails {
+						dispatch_weight: 0,
+						size: 0,
+						reward: 0,
+						dispatch_fee_payment: DispatchFeePayment::AtSourceChain,
+					}
+				),
+				(
+					3,
+					MessageDetails {
+						dispatch_weight: 0,
+						size: 0,
+						reward: 0,
+						dispatch_fee_payment: DispatchFeePayment::AtSourceChain,
+					}
+				),
+			]
+			.into_iter()
+			.collect(),
+		);
+	}
+
+	#[test]
+	fn make_message_details_map_succeeds_if_head_messages_are_missing() {
+		assert_eq!(
+			make_message_details_map::<relay_rialto_client::Rialto>(message_details_from_rpc(2..=3), 1..=3,).unwrap(),
+			vec![
+				(
+					2,
+					MessageDetails {
+						dispatch_weight: 0,
+						size: 0,
+						reward: 0,
+						dispatch_fee_payment: DispatchFeePayment::AtSourceChain,
+					}
+				),
+				(
+					3,
+					MessageDetails {
+						dispatch_weight: 0,
+						size: 0,
+						reward: 0,
+						dispatch_fee_payment: DispatchFeePayment::AtSourceChain,
+					}
+				),
+			]
+			.into_iter()
+			.collect(),
+		);
+	}
+
+	#[test]
+	fn make_message_details_map_fails_if_mid_messages_are_missing() {
+		let mut message_details_from_rpc = message_details_from_rpc(1..=3);
+		message_details_from_rpc.remove(1);
+		assert!(matches!(
+			make_message_details_map::<relay_rialto_client::Rialto>(message_details_from_rpc, 1..=3,),
+			Err(SubstrateError::Custom(_))
+		));
+	}
+
+	#[test]
+	fn make_message_details_map_fails_if_tail_messages_are_missing() {
+		assert!(matches!(
+			make_message_details_map::<relay_rialto_client::Rialto>(message_details_from_rpc(1..=2), 1..=3,),
+			Err(SubstrateError::Custom(_))
+		));
+	}
+
+	#[test]
+	fn make_message_details_map_fails_if_all_messages_are_missing() {
+		assert!(matches!(
+			make_message_details_map::<relay_rialto_client::Rialto>(vec![], 1..=3),
+			Err(SubstrateError::Custom(_))
+		));
+	}
+
+	#[test]
+	fn prepare_dummy_messages_delivery_proof_works() {
+		let expected_minimal_size = Rialto::MAXIMAL_ENCODED_ACCOUNT_ID_SIZE + Millau::STORAGE_PROOF_OVERHEAD;
+		let dummy_proof = prepare_dummy_messages_delivery_proof::<Rialto, Millau>();
+		assert!(
+			dummy_proof.1.encode().len() as u32 > expected_minimal_size,
+			"Expected proof size at least {}. Got: {}",
+			expected_minimal_size,
+			dummy_proof.1.encode().len(),
+		);
+	}
+}

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

@@ -0,0 +1,360 @@
+// 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/>.
+
+//! Substrate client as Substrate messages target. The chain we connect to should have
+//! runtime that implements `<BridgedChainName>HeaderApi` to allow bridging with
+//! <BridgedName> chain.
+
+use crate::messages_lane::{StandaloneMessagesMetrics, SubstrateMessageLane};
+use crate::messages_source::{read_client_state, SubstrateMessagesProof};
+use crate::on_demand_headers::OnDemandHeadersRelay;
+
+use async_trait::async_trait;
+use bp_messages::{LaneId, MessageNonce, UnrewardedRelayersState};
+use bp_runtime::ChainId;
+use bridge_runtime_common::messages::{
+	source::FromBridgedChainMessagesDeliveryProof, target::FromBridgedChainMessagesProof,
+};
+use codec::{Decode, Encode};
+use frame_support::{traits::Instance, weights::Weight};
+use messages_relay::message_lane::MessageLane;
+use messages_relay::{
+	message_lane::{SourceHeaderIdOf, TargetHeaderIdOf},
+	message_lane_loop::{TargetClient, TargetClientState},
+};
+use num_traits::{Bounded, Zero};
+use relay_substrate_client::{Chain, Client, Error as SubstrateError, HashOf};
+use relay_utils::{relay_loop::Client as RelayClient, BlockNumberBase, HeaderId};
+use sp_core::Bytes;
+use sp_runtime::{traits::Header as HeaderT, DeserializeOwned, FixedPointNumber, FixedU128};
+use std::{convert::TryFrom, marker::PhantomData, ops::RangeInclusive};
+
+/// Message receiving proof returned by the target Substrate node.
+pub type SubstrateMessagesReceivingProof<C> = (
+	UnrewardedRelayersState,
+	FromBridgedChainMessagesDeliveryProof<HashOf<C>>,
+);
+
+/// Substrate client as Substrate messages target.
+pub struct SubstrateMessagesTarget<SC: Chain, TC: Chain, P: SubstrateMessageLane, I> {
+	client: Client<TC>,
+	lane: P,
+	lane_id: LaneId,
+	instance: ChainId,
+	metric_values: StandaloneMessagesMetrics,
+	source_to_target_headers_relay: Option<OnDemandHeadersRelay<SC>>,
+	_phantom: PhantomData<I>,
+}
+
+impl<SC: Chain, TC: Chain, P: SubstrateMessageLane, I> SubstrateMessagesTarget<SC, TC, P, I> {
+	/// Create new Substrate headers target.
+	pub fn new(
+		client: Client<TC>,
+		lane: P,
+		lane_id: LaneId,
+		instance: ChainId,
+		metric_values: StandaloneMessagesMetrics,
+		source_to_target_headers_relay: Option<OnDemandHeadersRelay<SC>>,
+	) -> Self {
+		SubstrateMessagesTarget {
+			client,
+			lane,
+			lane_id,
+			instance,
+			metric_values,
+			source_to_target_headers_relay,
+			_phantom: Default::default(),
+		}
+	}
+}
+
+impl<SC: Chain, TC: Chain, P: SubstrateMessageLane, I> Clone for SubstrateMessagesTarget<SC, TC, P, I> {
+	fn clone(&self) -> Self {
+		Self {
+			client: self.client.clone(),
+			lane: self.lane.clone(),
+			lane_id: self.lane_id,
+			instance: self.instance,
+			metric_values: self.metric_values.clone(),
+			source_to_target_headers_relay: self.source_to_target_headers_relay.clone(),
+			_phantom: Default::default(),
+		}
+	}
+}
+
+#[async_trait]
+impl<SC, TC, P, I> RelayClient for SubstrateMessagesTarget<SC, TC, P, I>
+where
+	SC: Chain,
+	TC: Chain,
+	P: SubstrateMessageLane,
+	I: Send + Sync + Instance,
+{
+	type Error = SubstrateError;
+
+	async fn reconnect(&mut self) -> Result<(), SubstrateError> {
+		self.client.reconnect().await
+	}
+}
+
+#[async_trait]
+impl<SC, TC, P, I> TargetClient<P::MessageLane> for SubstrateMessagesTarget<SC, TC, P, I>
+where
+	SC: Chain<
+		Hash = <P::MessageLane as MessageLane>::SourceHeaderHash,
+		BlockNumber = <P::MessageLane as MessageLane>::SourceHeaderNumber,
+		Balance = <P::MessageLane as MessageLane>::SourceChainBalance,
+	>,
+	SC::Balance: TryFrom<TC::Balance> + Bounded,
+	TC: Chain<
+		Hash = <P::MessageLane as MessageLane>::TargetHeaderHash,
+		BlockNumber = <P::MessageLane as MessageLane>::TargetHeaderNumber,
+	>,
+	TC::Hash: Copy,
+	TC::BlockNumber: Copy,
+	TC::Header: DeserializeOwned,
+	TC::Index: DeserializeOwned,
+	<TC::Header as HeaderT>::Number: BlockNumberBase,
+	P: SubstrateMessageLane<SourceChain = SC, TargetChain = TC>,
+	P::MessageLane: MessageLane<
+		MessagesProof = SubstrateMessagesProof<SC>,
+		MessagesReceivingProof = SubstrateMessagesReceivingProof<TC>,
+	>,
+	<P::MessageLane as MessageLane>::SourceHeaderNumber: Decode,
+	<P::MessageLane as MessageLane>::SourceHeaderHash: Decode,
+	I: Send + Sync + Instance,
+{
+	async fn state(&self) -> Result<TargetClientState<P::MessageLane>, SubstrateError> {
+		// we can't continue to deliver messages if target node is out of sync, because
+		// it may have already received (some of) messages that we're going to deliver
+		self.client.ensure_synced().await?;
+
+		read_client_state::<
+			_,
+			<P::MessageLane as MessageLane>::SourceHeaderHash,
+			<P::MessageLane as MessageLane>::SourceHeaderNumber,
+		>(&self.client, P::BEST_FINALIZED_SOURCE_HEADER_ID_AT_TARGET)
+		.await
+	}
+
+	async fn latest_received_nonce(
+		&self,
+		id: TargetHeaderIdOf<P::MessageLane>,
+	) -> Result<(TargetHeaderIdOf<P::MessageLane>, MessageNonce), SubstrateError> {
+		let encoded_response = self
+			.client
+			.state_call(
+				P::INBOUND_LANE_LATEST_RECEIVED_NONCE_METHOD.into(),
+				Bytes(self.lane_id.encode()),
+				Some(id.1),
+			)
+			.await?;
+		let latest_received_nonce: MessageNonce =
+			Decode::decode(&mut &encoded_response.0[..]).map_err(SubstrateError::ResponseParseFailed)?;
+		Ok((id, latest_received_nonce))
+	}
+
+	async fn latest_confirmed_received_nonce(
+		&self,
+		id: TargetHeaderIdOf<P::MessageLane>,
+	) -> Result<(TargetHeaderIdOf<P::MessageLane>, MessageNonce), SubstrateError> {
+		let encoded_response = self
+			.client
+			.state_call(
+				P::INBOUND_LANE_LATEST_CONFIRMED_NONCE_METHOD.into(),
+				Bytes(self.lane_id.encode()),
+				Some(id.1),
+			)
+			.await?;
+		let latest_received_nonce: MessageNonce =
+			Decode::decode(&mut &encoded_response.0[..]).map_err(SubstrateError::ResponseParseFailed)?;
+		Ok((id, latest_received_nonce))
+	}
+
+	async fn unrewarded_relayers_state(
+		&self,
+		id: TargetHeaderIdOf<P::MessageLane>,
+	) -> Result<(TargetHeaderIdOf<P::MessageLane>, UnrewardedRelayersState), SubstrateError> {
+		let encoded_response = self
+			.client
+			.state_call(
+				P::INBOUND_LANE_UNREWARDED_RELAYERS_STATE.into(),
+				Bytes(self.lane_id.encode()),
+				Some(id.1),
+			)
+			.await?;
+		let unrewarded_relayers_state: UnrewardedRelayersState =
+			Decode::decode(&mut &encoded_response.0[..]).map_err(SubstrateError::ResponseParseFailed)?;
+		Ok((id, unrewarded_relayers_state))
+	}
+
+	async fn prove_messages_receiving(
+		&self,
+		id: TargetHeaderIdOf<P::MessageLane>,
+	) -> Result<
+		(
+			TargetHeaderIdOf<P::MessageLane>,
+			<P::MessageLane as MessageLane>::MessagesReceivingProof,
+		),
+		SubstrateError,
+	> {
+		let (id, relayers_state) = self.unrewarded_relayers_state(id).await?;
+		let inbound_data_key = pallet_bridge_messages::storage_keys::inbound_lane_data_key::<I>(&self.lane_id);
+		let proof = self
+			.client
+			.prove_storage(vec![inbound_data_key], id.1)
+			.await?
+			.iter_nodes()
+			.collect();
+		let proof = FromBridgedChainMessagesDeliveryProof {
+			bridged_header_hash: id.1,
+			storage_proof: proof,
+			lane: self.lane_id,
+		};
+		Ok((id, (relayers_state, proof)))
+	}
+
+	async fn submit_messages_proof(
+		&self,
+		generated_at_header: SourceHeaderIdOf<P::MessageLane>,
+		nonces: RangeInclusive<MessageNonce>,
+		proof: <P::MessageLane as MessageLane>::MessagesProof,
+	) -> Result<RangeInclusive<MessageNonce>, SubstrateError> {
+		let lane = self.lane.clone();
+		let nonces_clone = nonces.clone();
+		self.client
+			.submit_signed_extrinsic(self.lane.target_transactions_author(), move |transaction_nonce| {
+				lane.make_messages_delivery_transaction(transaction_nonce, generated_at_header, nonces_clone, proof)
+			})
+			.await?;
+		Ok(nonces)
+	}
+
+	async fn require_source_header_on_target(&self, id: SourceHeaderIdOf<P::MessageLane>) {
+		if let Some(ref source_to_target_headers_relay) = self.source_to_target_headers_relay {
+			source_to_target_headers_relay.require_finalized_header(id).await;
+		}
+	}
+
+	async fn estimate_delivery_transaction_in_source_tokens(
+		&self,
+		nonces: RangeInclusive<MessageNonce>,
+		total_dispatch_weight: Weight,
+		total_size: u32,
+	) -> Result<<P::MessageLane as MessageLane>::SourceChainBalance, SubstrateError> {
+		let conversion_rate = self
+			.metric_values
+			.target_to_source_conversion_rate()
+			.await
+			.ok_or_else(|| {
+				SubstrateError::Custom(format!(
+					"Failed to compute conversion rate from {} to {}",
+					TC::NAME,
+					SC::NAME,
+				))
+			})?;
+		log::trace!(
+			target: "bridge",
+			"Using conversion rate {} when converting from {} tokens to {} tokens",
+			conversion_rate,
+			TC::NAME,
+			SC::NAME
+		);
+		Ok(convert_target_tokens_to_source_tokens::<SC, TC>(
+			FixedU128::from_float(conversion_rate),
+			self.client
+				.estimate_extrinsic_fee(self.lane.make_messages_delivery_transaction(
+					Zero::zero(),
+					HeaderId(Default::default(), Default::default()),
+					nonces.clone(),
+					prepare_dummy_messages_proof::<SC>(nonces, total_dispatch_weight, total_size),
+				))
+				.await
+				.unwrap_or_else(|_| TC::Balance::max_value()),
+		))
+	}
+}
+
+/// Prepare 'dummy' messages proof that will compose the delivery transaction.
+///
+/// We don't care about proof actually being the valid proof, because its validity doesn't
+/// affect the call weight - we only care about its size.
+fn prepare_dummy_messages_proof<SC: Chain>(
+	nonces: RangeInclusive<MessageNonce>,
+	total_dispatch_weight: Weight,
+	total_size: u32,
+) -> SubstrateMessagesProof<SC> {
+	(
+		total_dispatch_weight,
+		FromBridgedChainMessagesProof {
+			bridged_header_hash: Default::default(),
+			storage_proof: vec![vec![0; SC::STORAGE_PROOF_OVERHEAD.saturating_add(total_size) as usize]],
+			lane: Default::default(),
+			nonces_start: *nonces.start(),
+			nonces_end: *nonces.end(),
+		},
+	)
+}
+
+/// Given delivery transaction fee in target chain tokens and conversion rate to the source
+/// chain tokens, compute transaction cost in source chain tokens.
+fn convert_target_tokens_to_source_tokens<SC: Chain, TC: Chain>(
+	target_to_source_conversion_rate: FixedU128,
+	target_transaction_fee: TC::Balance,
+) -> SC::Balance
+where
+	SC::Balance: TryFrom<TC::Balance>,
+{
+	SC::Balance::try_from(target_to_source_conversion_rate.saturating_mul_int(target_transaction_fee))
+		.unwrap_or_else(|_| SC::Balance::max_value())
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use relay_millau_client::Millau;
+	use relay_rialto_client::Rialto;
+
+	#[test]
+	fn prepare_dummy_messages_proof_works() {
+		const DISPATCH_WEIGHT: Weight = 1_000_000;
+		const SIZE: u32 = 1_000;
+		let dummy_proof = prepare_dummy_messages_proof::<Rialto>(1..=10, DISPATCH_WEIGHT, SIZE);
+		assert_eq!(dummy_proof.0, DISPATCH_WEIGHT);
+		assert!(
+			dummy_proof.1.encode().len() as u32 > SIZE,
+			"Expected proof size at least {}. Got: {}",
+			SIZE,
+			dummy_proof.1.encode().len(),
+		);
+	}
+
+	#[test]
+	fn convert_target_tokens_to_source_tokens_works() {
+		assert_eq!(
+			convert_target_tokens_to_source_tokens::<Rialto, Millau>((150, 100).into(), 1_000),
+			1_500
+		);
+		assert_eq!(
+			convert_target_tokens_to_source_tokens::<Rialto, Millau>((50, 100).into(), 1_000),
+			500
+		);
+		assert_eq!(
+			convert_target_tokens_to_source_tokens::<Rialto, Millau>((100, 100).into(), 1_000),
+			1_000
+		);
+	}
+}

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

@@ -0,0 +1,440 @@
+// 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/>.
+
+//! 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 relay_substrate_client::{
+	finality_source::{FinalitySource as SubstrateFinalitySource, RequiredHeaderNumberRef},
+	Chain, Client, HeaderIdOf, SyncHeader,
+};
+use relay_utils::{
+	metrics::MetricsParams, relay_loop::Client as RelayClient, BlockNumberBase, FailedClient, MaybeConnectionError,
+};
+
+use crate::finality_pipeline::{
+	SubstrateFinalitySyncPipeline, SubstrateFinalityToSubstrate, RECENT_FINALITY_PROOFS_LIMIT, STALL_TIMEOUT,
+};
+use crate::finality_target::SubstrateFinalityTarget;
+
+/// On-demand Substrate <-> Substrate headers relay.
+///
+/// This relay may be requested to sync more headers, whenever some other relay (e.g. messages relay) needs
+/// it to continue its regular work. When enough headers are relayed, on-demand stops syncing headers.
+#[derive(Clone)]
+pub struct OnDemandHeadersRelay<SourceChain: Chain> {
+	/// Relay task name.
+	relay_task_name: String,
+	/// Shared reference to maximal required finalized header number.
+	required_header_number: RequiredHeaderNumberRef<SourceChain>,
+}
+
+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>,
+		pipeline: P,
+		maximal_headers_difference: SourceChain::BlockNumber,
+	) -> 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,
+		>,
+	{
+		let required_header_number = Arc::new(Mutex::new(Zero::zero()));
+		let this = OnDemandHeadersRelay {
+			relay_task_name: on_demand_headers_relay_name::<SourceChain, TargetChain>(),
+			required_header_number: required_header_number.clone(),
+		};
+		async_std::task::spawn(async move {
+			background_task(
+				source_client,
+				target_client,
+				pipeline,
+				maximal_headers_difference,
+				required_header_number,
+			)
+			.await;
+		});
+
+		this
+	}
+
+	/// Someone is asking us to relay given finalized header.
+	pub async fn require_finalized_header(&self, header_id: HeaderIdOf<SourceChain>) {
+		let mut required_header_number = self.required_header_number.lock().await;
+		if header_id.0 > *required_header_number {
+			log::trace!(
+				target: "bridge",
+				"More {} headers required in {} relay. Going to sync up to the {}",
+				SourceChain::NAME,
+				self.relay_task_name,
+				header_id.0,
+			);
+
+			*required_header_number = header_id.0;
+		}
+	}
+}
+
+/// 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>,
+	// pipeline: SubstrateFinalityToSubstrate<SourceChain, TargetChain, TargetSign>,
+	pipeline: P,
+	maximal_headers_difference: SourceChain::BlockNumber,
+	required_header_number: RequiredHeaderNumberRef<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,
+	>,
+{
+	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());
+	let mut latest_non_mandatory_at_source = Zero::zero();
+
+	let mut restart_relay = true;
+	let finality_relay_task = futures::future::Fuse::terminated();
+	futures::pin_mut!(finality_relay_task);
+
+	loop {
+		select! {
+			_ = async_std::task::sleep(TargetChain::AVERAGE_BLOCK_INTERVAL).fuse() => {},
+			_ = finality_relay_task => {
+				// this should never happen in practice given the current code
+				restart_relay = true;
+			},
+		}
+
+		// read best finalized source header number from source
+		let best_finalized_source_header_at_source =
+			best_finalized_source_header_at_source(&finality_source, &relay_task_name).await;
+		if matches!(best_finalized_source_header_at_source, Err(ref e) if e.is_connection_error()) {
+			relay_utils::relay_loop::reconnect_failed_client(
+				FailedClient::Source,
+				relay_utils::relay_loop::RECONNECT_DELAY,
+				&mut finality_source,
+				&mut finality_target,
+			)
+			.await;
+			continue;
+		}
+
+		// 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;
+		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,
+				relay_utils::relay_loop::RECONNECT_DELAY,
+				&mut finality_source,
+				&mut finality_target,
+			)
+			.await;
+			continue;
+		}
+
+		// 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>(
+			best_finalized_source_header_at_source.ok(),
+			best_finalized_source_header_at_target.ok(),
+			maximal_headers_difference,
+			&required_header_number,
+		)
+		.await;
+		if let Some(mandatory_scan_range) = mandatory_scan_range {
+			let relay_mandatory_header_result = relay_mandatory_header_from_range(
+				&finality_source,
+				&required_header_number,
+				best_finalized_source_header_at_target_fmt,
+				(
+					std::cmp::max(mandatory_scan_range.0, latest_non_mandatory_at_source),
+					mandatory_scan_range.1,
+				),
+				&relay_task_name,
+			)
+			.await;
+			match relay_mandatory_header_result {
+				Ok(true) => (),
+				Ok(false) => {
+					// there are no (or we don't need to relay them) mandatory headers in the range
+					// => to avoid scanning the same headers over and over again, remember that
+					latest_non_mandatory_at_source = mandatory_scan_range.1;
+				}
+				Err(e) => {
+					if e.is_connection_error() {
+						relay_utils::relay_loop::reconnect_failed_client(
+							FailedClient::Source,
+							relay_utils::relay_loop::RECONNECT_DELAY,
+							&mut finality_source,
+							&mut finality_target,
+						)
+						.await;
+						continue;
+					}
+				}
+			}
+		}
+
+		// start/restart relay
+		if restart_relay {
+			finality_relay_task.set(
+				finality_relay::run(
+					finality_source.clone(),
+					finality_target.clone(),
+					FinalitySyncParams {
+						tick: std::cmp::max(SourceChain::AVERAGE_BLOCK_INTERVAL, TargetChain::AVERAGE_BLOCK_INTERVAL),
+						recent_finality_proofs_limit: RECENT_FINALITY_PROOFS_LIMIT,
+						stall_timeout: STALL_TIMEOUT,
+						only_mandatory_headers: false,
+					},
+					MetricsParams::disabled(),
+					futures::future::pending(),
+				)
+				.fuse(),
+			);
+
+			restart_relay = false;
+		}
+	}
+}
+
+/// Returns `Some()` with inclusive range of headers which must be scanned for mandatory headers
+/// and the first of such headers must be submitted to the target node.
+async fn mandatory_headers_scan_range<C: Chain>(
+	best_finalized_source_header_at_source: Option<C::BlockNumber>,
+	best_finalized_source_header_at_target: Option<C::BlockNumber>,
+	maximal_headers_difference: C::BlockNumber,
+	required_header_number: &RequiredHeaderNumberRef<C>,
+) -> Option<(C::BlockNumber, C::BlockNumber)> {
+	let required_header_number = *required_header_number.lock().await;
+
+	// if we have been unable to read header number from the target, then let's assume
+	// that it is the same as required header number. Otherwise we risk submitting
+	// unneeded transactions
+	let best_finalized_source_header_at_target =
+		best_finalized_source_header_at_target.unwrap_or(required_header_number);
+
+	// if we have been unable to read header number from the source, then let's assume
+	// that it is the same as at the target
+	let best_finalized_source_header_at_source =
+		best_finalized_source_header_at_source.unwrap_or(best_finalized_source_header_at_target);
+
+	// if there are too many source headers missing from the target node, sync mandatory
+	// headers to target
+	//
+	// why do we need that? When complex headers+messages relay is used, it'll normally only relay
+	// headers when there are undelivered messages/confirmations. But security model of the
+	// `pallet-bridge-grandpa` module relies on the fact that headers are synced in real-time and
+	// that it'll see authorities-change header before unbonding period will end for previous
+	// authorities set.
+	let current_headers_difference = best_finalized_source_header_at_source
+		.checked_sub(&best_finalized_source_header_at_target)
+		.unwrap_or_else(Zero::zero);
+	if current_headers_difference <= maximal_headers_difference {
+		return None;
+	}
+
+	// if relay is already asked to sync headers, don't do anything yet
+	if required_header_number > best_finalized_source_header_at_target {
+		return None;
+	}
+
+	Some((
+		best_finalized_source_header_at_target + One::one(),
+		best_finalized_source_header_at_source,
+	))
+}
+
+/// Try to find mandatory header in the inclusive headers range and, if one is found, ask to relay 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>,
+	best_finalized_source_header_at_target: String,
+	range: (SourceChain::BlockNumber, SourceChain::BlockNumber),
+	relay_task_name: &str,
+) -> Result<bool, relay_substrate_client::Error>
+where
+	SubstrateFinalitySource<SourceChain, P>: FinalitySourceClient<P>,
+	P: FinalitySyncPipeline<Number = SourceChain::BlockNumber>,
+{
+	// search for mandatory header first
+	let mandatory_source_header_number = find_mandatory_header_in_range(finality_source, range).await?;
+
+	// if there are no mandatory headers - we have nothing to do
+	let mandatory_source_header_number = match mandatory_source_header_number {
+		Some(mandatory_source_header_number) => mandatory_source_header_number,
+		None => return Ok(false),
+	};
+
+	// `find_mandatory_header` call may take a while => check if `required_header_number` is still
+	// less than our `mandatory_source_header_number` before logging anything
+	let mut required_header_number = required_header_number.lock().await;
+	if *required_header_number >= mandatory_source_header_number {
+		return Ok(false);
+	}
+
+	log::trace!(
+		target: "bridge",
+		"Too many {} headers missing at target in {} relay ({} vs {}). Going to sync up to the mandatory {}",
+		SourceChain::NAME,
+		relay_task_name,
+		best_finalized_source_header_at_target,
+		range.1,
+		mandatory_source_header_number,
+	);
+
+	*required_header_number = mandatory_source_header_number;
+	Ok(true)
+}
+
+/// 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>,
+	relay_task_name: &str,
+) -> Result<SourceChain::BlockNumber, relay_substrate_client::Error>
+where
+	SubstrateFinalitySource<SourceChain, P>: FinalitySourceClient<P>,
+	P: FinalitySyncPipeline<Number = SourceChain::BlockNumber>,
+{
+	finality_source
+		.on_chain_best_finalized_block_number()
+		.await
+		.map_err(|error| {
+			log::error!(
+				target: "bridge",
+				"Failed to read best finalized source header from source in {} relay: {:?}",
+				relay_task_name,
+				error,
+			);
+
+			error
+		})
+}
+
+/// 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>,
+	relay_task_name: &str,
+) -> Result<SourceChain::BlockNumber, <SubstrateFinalityTarget<TargetChain, P> as RelayClient>::Error>
+where
+	SubstrateFinalityTarget<TargetChain, P>: FinalityTargetClient<P::FinalitySyncPipeline>,
+	P: SubstrateFinalitySyncPipeline,
+	P::FinalitySyncPipeline: FinalitySyncPipeline<Number = SourceChain::BlockNumber>,
+{
+	finality_target
+		.best_finalized_source_block_number()
+		.await
+		.map_err(|error| {
+			log::error!(
+				target: "bridge",
+				"Failed to read best finalized source header from target in {} relay: {:?}",
+				relay_task_name,
+				error,
+			);
+
+			error
+		})
+}
+
+/// 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>,
+{
+	let mut current = range.0;
+	while current <= range.1 {
+		let header: SyncHeader<SourceChain::Header> = finality_source.client().header_by_number(current).await?.into();
+		if header.is_mandatory() {
+			return Ok(Some(current));
+		}
+
+		current += One::one();
+	}
+
+	Ok(None)
+}
+
+/// On-demand headers relay task name.
+fn on_demand_headers_relay_name<SourceChain: Chain, TargetChain: Chain>() -> String {
+	format!("on-demand-{}-to-{}", SourceChain::NAME, TargetChain::NAME)
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	type TestChain = relay_millau_client::Millau;
+
+	const AT_SOURCE: Option<bp_millau::BlockNumber> = Some(10);
+	const AT_TARGET: Option<bp_millau::BlockNumber> = Some(1);
+
+	#[async_std::test]
+	async fn mandatory_headers_scan_range_selects_range_if_too_many_headers_are_missing() {
+		assert_eq!(
+			mandatory_headers_scan_range::<TestChain>(AT_SOURCE, AT_TARGET, 5, &Arc::new(Mutex::new(0))).await,
+			Some((AT_TARGET.unwrap() + 1, AT_SOURCE.unwrap())),
+		);
+	}
+
+	#[async_std::test]
+	async fn mandatory_headers_scan_range_selects_nothing_if_enough_headers_are_relayed() {
+		assert_eq!(
+			mandatory_headers_scan_range::<TestChain>(AT_SOURCE, AT_TARGET, 10, &Arc::new(Mutex::new(0))).await,
+			None,
+		);
+	}
+}