parachains.rs

// 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 parachain finality relay.

use crate::{
	messages_source::best_finalized_peer_header_at_self,
	on_demand::OnDemandRelay,
	parachains::{
		source::ParachainsSource, target::ParachainsTarget, ParachainsPipelineAdapter,
		SubmitParachainHeadsCallBuilder, SubstrateParachainsPipeline,
	},
	TransactionParams,
};

use async_std::{
	channel::{unbounded, Receiver, Sender},
	sync::{Arc, Mutex},
};
use async_trait::async_trait;
use bp_polkadot_core::parachains::{ParaHash, ParaId};
use bp_runtime::HeaderIdProvider;
use futures::{select, FutureExt};
use num_traits::Zero;
use pallet_bridge_parachains::{RelayBlockHash, RelayBlockHasher, RelayBlockNumber};
use parachains_relay::parachains_loop::{AvailableHeader, SourceClient, TargetClient};
use relay_substrate_client::{
	is_ancient_block, AccountIdOf, AccountKeyPairOf, BlockNumberOf, CallOf, Chain, Client,
	Error as SubstrateError, HashOf, HeaderIdOf, ParachainBase,
};
use relay_utils::{
	metrics::MetricsParams, relay_loop::Client as RelayClient, BlockNumberBase, FailedClient,
	HeaderId, UniqueSaturatedInto,
};
use std::fmt::Debug;

/// On-demand Substrate <-> Substrate parachain finality relay.
///
/// This relay may be requested to sync more parachain headers, whenever some other relay
/// (e.g. messages relay) needs it to continue its regular work. When enough parachain headers
/// are relayed, on-demand stops syncing headers.
#[derive(Clone)]
pub struct OnDemandParachainsRelay<P: SubstrateParachainsPipeline> {
	/// Relay task name.
	relay_task_name: String,
	/// Channel used to communicate with background task and ask for relay of parachain heads.
	required_header_number_sender: Sender<BlockNumberOf<P::SourceParachain>>,
	/// Source relay chain client.
	source_relay_client: Client<P::SourceRelayChain>,
	/// Target chain client.
	target_client: Client<P::TargetChain>,
	/// On-demand relay chain relay.
	on_demand_source_relay_to_target_headers:
		Arc<dyn OnDemandRelay<P::SourceRelayChain, P::TargetChain>>,
}

impl<P: SubstrateParachainsPipeline> OnDemandParachainsRelay<P> {
	/// Create new on-demand parachains relay.
	///
	/// Note that the argument is the source relay chain client, not the parachain client.
	/// That's because parachain finality is determined by the relay chain and we don't
	/// need to connect to the parachain itself here.
	pub fn new(
		source_relay_client: Client<P::SourceRelayChain>,
		target_client: Client<P::TargetChain>,
		target_transaction_params: TransactionParams<AccountKeyPairOf<P::TargetChain>>,
		on_demand_source_relay_to_target_headers: Arc<
			dyn OnDemandRelay<P::SourceRelayChain, P::TargetChain>,
		>,
	) -> Self
	where
		P::SourceParachain: Chain<Hash = ParaHash>,
		P::SourceRelayChain:
			Chain<BlockNumber = RelayBlockNumber, Hash = RelayBlockHash, Hasher = RelayBlockHasher>,
		AccountIdOf<P::TargetChain>:
			From<<AccountKeyPairOf<P::TargetChain> as sp_core::Pair>::Public>,
	{
		let (required_header_number_sender, required_header_number_receiver) = unbounded();
		let this = OnDemandParachainsRelay {
			relay_task_name: on_demand_parachains_relay_name::<P::SourceParachain, P::TargetChain>(
			),
			required_header_number_sender,
			source_relay_client: source_relay_client.clone(),
			target_client: target_client.clone(),
			on_demand_source_relay_to_target_headers: on_demand_source_relay_to_target_headers
				.clone(),
		};
		async_std::task::spawn(async move {
			background_task::<P>(
				source_relay_client,
				target_client,
				target_transaction_params,
				on_demand_source_relay_to_target_headers,
				required_header_number_receiver,
			)
			.await;
		});

		this
	}
}

#[async_trait]
impl<P: SubstrateParachainsPipeline> OnDemandRelay<P::SourceParachain, P::TargetChain>
	for OnDemandParachainsRelay<P>
where
	P::SourceParachain: Chain<Hash = ParaHash>,
{
	async fn reconnect(&self) -> Result<(), SubstrateError> {
		// using clone is fine here (to avoid mut requirement), because clone on Client clones
		// internal references
		self.source_relay_client.clone().reconnect().await?;
		self.target_client.clone().reconnect().await?;
		// we'll probably need to reconnect relay chain relayer clients also
		self.on_demand_source_relay_to_target_headers.reconnect().await
	}

	async fn require_more_headers(&self, required_header: BlockNumberOf<P::SourceParachain>) {
		if let Err(e) = self.required_header_number_sender.send(required_header).await {
			log::trace!(
				target: "bridge",
				"[{}] Failed to request {} header {:?}: {:?}",
				self.relay_task_name,
				P::SourceParachain::NAME,
				required_header,
				e,
			);
		}
	}

	/// Ask relay to prove source `required_header` to the `TargetChain`.
	async fn prove_header(
		&self,
		required_parachain_header: BlockNumberOf<P::SourceParachain>,
	) -> Result<(HeaderIdOf<P::SourceParachain>, Vec<CallOf<P::TargetChain>>), SubstrateError> {
		// select headers to prove
		let parachains_source = ParachainsSource::<P>::new(
			self.source_relay_client.clone(),
			Arc::new(Mutex::new(AvailableHeader::Missing)),
		);
		let env = (self, &parachains_source);
		let (need_to_prove_relay_block, selected_relay_block, selected_parachain_block) =
			select_headers_to_prove(env, required_parachain_header).await?;

		log::debug!(
			target: "bridge",
			"[{}] Requested to prove {} head {:?}. Selected to prove {} head {:?} and {} head {:?}",
			self.relay_task_name,
			P::SourceParachain::NAME,
			required_parachain_header,
			P::SourceParachain::NAME,
			selected_parachain_block,
			P::SourceRelayChain::NAME,
			if need_to_prove_relay_block {
				Some(selected_relay_block)
			} else {
				None
			},
		);

		// now let's prove relay chain block (if needed)
		let mut calls = Vec::new();
		let mut proved_relay_block = selected_relay_block;
		if need_to_prove_relay_block {
			let (relay_block, relay_prove_call) = self
				.on_demand_source_relay_to_target_headers
				.prove_header(selected_relay_block.number())
				.await?;
			proved_relay_block = relay_block;
			calls.extend(relay_prove_call);
		}

		// despite what we've selected before (in `select_headers_to_prove` call), if headers relay
		// have chose the different header (e.g. because there's no GRANDPA jusstification for it),
		// we need to prove parachain head available at this header
		let para_id = ParaId(P::SourceParachain::PARACHAIN_ID);
		let mut proved_parachain_block = selected_parachain_block;
		if proved_relay_block != selected_relay_block {
			proved_parachain_block = parachains_source
				.on_chain_para_head_id(proved_relay_block)
				.await?
				// this could happen e.g. if parachain has been offboarded?
				.ok_or_else(|| {
					SubstrateError::MissingRequiredParachainHead(
						para_id,
						proved_relay_block.number().unique_saturated_into(),
					)