From 63e84708259f51cb90a8d6eb5e3ba1af744ac10f Mon Sep 17 00:00:00 2001 From: Serban Iorga Date: Tue, 2 Aug 2022 12:49:08 +0300 Subject: [PATCH] Follow-up on #1419 (#1531) * Parachains source cosmetic changes - Make `ParaHashAtSource` more generic - Modify `on_chain_parachain_header` to return `HeaderId` - Shortening variable names Signed-off-by: Serban Iorga * Change ParachainsSource::max_head_id type Change ParachainsSource::max_head_id to Arc> Signed-off-by: Serban Iorga * code review changes --- bridges/primitives/runtime/src/lib.rs | 4 +- .../bin-substrate/src/cli/relay_parachains.rs | 11 ++- .../src/finality/source.rs | 5 +- .../src/on_demand/parachains.rs | 22 +++--- .../src/parachains/source.rs | 76 ++++++++----------- .../relays/parachains/src/parachains_loop.rs | 74 +++++++++--------- 6 files changed, 91 insertions(+), 101 deletions(-) diff --git a/bridges/primitives/runtime/src/lib.rs b/bridges/primitives/runtime/src/lib.rs index a51327e89b802..a7e3e44626ad0 100644 --- a/bridges/primitives/runtime/src/lib.rs +++ b/bridges/primitives/runtime/src/lib.rs @@ -82,7 +82,9 @@ pub const ACCOUNT_DERIVATION_PREFIX: &[u8] = b"pallet-bridge/account-derivation/ pub const ROOT_ACCOUNT_DERIVATION_PREFIX: &[u8] = b"pallet-bridge/account-derivation/root"; /// Generic header Id. -#[derive(RuntimeDebug, Default, Clone, Encode, Decode, Copy, Eq, Hash, PartialEq)] +#[derive( + RuntimeDebug, Default, Clone, Encode, Decode, Copy, Eq, Hash, PartialEq, PartialOrd, Ord, +)] pub struct HeaderId(pub Number, pub Hash); /// Generic header id provider. diff --git a/bridges/relays/bin-substrate/src/cli/relay_parachains.rs b/bridges/relays/bin-substrate/src/cli/relay_parachains.rs index 2fa6dcaf9eb4b..09251972587da 100644 --- a/bridges/relays/bin-substrate/src/cli/relay_parachains.rs +++ b/bridges/relays/bin-substrate/src/cli/relay_parachains.rs @@ -18,10 +18,14 @@ use crate::chains::{ rialto_parachains_to_millau::RialtoParachainToMillauCliBridge, westend_parachains_to_millau::WestmintToMillauCliBridge, }; +use async_std::sync::Mutex; use async_trait::async_trait; use bp_polkadot_core::parachains::ParaId; -use parachains_relay::parachains_loop::{ParachainSyncParams, SourceClient, TargetClient}; +use parachains_relay::parachains_loop::{ + AvailableHeader, ParachainSyncParams, SourceClient, TargetClient, +}; use relay_utils::metrics::{GlobalMetrics, StandaloneMetric}; +use std::sync::Arc; use structopt::StructOpt; use strum::{EnumString, EnumVariantNames, VariantNames}; use substrate_relay_helper::{ @@ -65,7 +69,10 @@ where { async fn relay_headers(data: RelayParachains) -> anyhow::Result<()> { let source_client = data.source.into_client::().await?; - let source_client = ParachainsSource::::new(source_client, None); + let source_client = ParachainsSource::::new( + source_client, + Arc::new(Mutex::new(AvailableHeader::Missing)), + ); let target_transaction_params = TransactionParams { signer: data.target_sign.to_keypair::()?, diff --git a/bridges/relays/lib-substrate-relay/src/finality/source.rs b/bridges/relays/lib-substrate-relay/src/finality/source.rs index d0acb7717682f..c8360bbddbc57 100644 --- a/bridges/relays/lib-substrate-relay/src/finality/source.rs +++ b/bridges/relays/lib-substrate-relay/src/finality/source.rs @@ -27,7 +27,6 @@ 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. @@ -76,9 +75,7 @@ impl SubstrateFinalitySource

{ ) -> Result, 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()) + self.client.best_finalized_header_number().await } } diff --git a/bridges/relays/lib-substrate-relay/src/on_demand/parachains.rs b/bridges/relays/lib-substrate-relay/src/on_demand/parachains.rs index 8300bf74cd6c2..dc400115502eb 100644 --- a/bridges/relays/lib-substrate-relay/src/on_demand/parachains.rs +++ b/bridges/relays/lib-substrate-relay/src/on_demand/parachains.rs @@ -36,7 +36,7 @@ use bp_runtime::HeaderIdProvider; use futures::{select, FutureExt}; use num_traits::Zero; use pallet_bridge_parachains::{RelayBlockHash, RelayBlockHasher, RelayBlockNumber}; -use parachains_relay::parachains_loop::{ParachainSyncParams, TargetClient}; +use parachains_relay::parachains_loop::{AvailableHeader, ParachainSyncParams, TargetClient}; use relay_substrate_client::{ AccountIdOf, AccountKeyPairOf, BlockNumberOf, Chain, Client, Error as SubstrateError, HashOf, TransactionSignScheme, @@ -143,7 +143,7 @@ async fn background_task( let mut relay_state = RelayState::Idle; let mut required_parachain_header_number = Zero::zero(); - let required_para_header_number_ref = Arc::new(Mutex::new(None)); + let required_para_header_number_ref = Arc::new(Mutex::new(AvailableHeader::Unavailable)); let mut restart_relay = true; let parachains_relay_task = futures::future::Fuse::terminated(); @@ -151,7 +151,7 @@ async fn background_task( let mut parachains_source = ParachainsSource::

::new( source_relay_client.clone(), - Some(required_para_header_number_ref.clone()), + required_para_header_number_ref.clone(), ); let mut parachains_target = ParachainsTarget::

::new(target_client.clone(), target_transaction_params.clone()); @@ -253,7 +253,8 @@ async fn background_task( .await; }, RelayState::RelayingParaHeader(required_para_header) => { - *required_para_header_number_ref.lock().await = Some(required_para_header); + *required_para_header_number_ref.lock().await = + AvailableHeader::Available(required_para_header); }, } @@ -389,13 +390,9 @@ where source.client().best_finalized_header().await.map_err(map_source_err)?; let best_finalized_relay_block_id = best_finalized_relay_header.id(); let para_header_at_source = source - .on_chain_parachain_header( - best_finalized_relay_block_id, - P::SOURCE_PARACHAIN_PARA_ID.into(), - ) + .on_chain_para_head_id(best_finalized_relay_block_id, P::SOURCE_PARACHAIN_PARA_ID.into()) .await - .map_err(map_source_err)? - .map(|h| h.id()); + .map_err(map_source_err)?; let relay_header_at_source = best_finalized_relay_block_id.0; let relay_header_at_target = @@ -408,10 +405,9 @@ where .map_err(map_target_err)?; let para_header_at_relay_header_at_target = source - .on_chain_parachain_header(relay_header_at_target, P::SOURCE_PARACHAIN_PARA_ID.into()) + .on_chain_para_head_id(relay_header_at_target, P::SOURCE_PARACHAIN_PARA_ID.into()) .await - .map_err(map_source_err)? - .map(|h| h.id()); + .map_err(map_source_err)?; Ok(RelayData { required_para_header: required_header_number, diff --git a/bridges/relays/lib-substrate-relay/src/parachains/source.rs b/bridges/relays/lib-substrate-relay/src/parachains/source.rs index f763b744f192e..c23ac87947632 100644 --- a/bridges/relays/lib-substrate-relay/src/parachains/source.rs +++ b/bridges/relays/lib-substrate-relay/src/parachains/source.rs @@ -22,35 +22,35 @@ use async_std::sync::{Arc, Mutex}; use async_trait::async_trait; use bp_parachains::parachain_head_storage_key_at_source; use bp_polkadot_core::parachains::{ParaHash, ParaHead, ParaHeadsProof, ParaId}; +use bp_runtime::HeaderIdProvider; use codec::Decode; use parachains_relay::{ - parachains_loop::{ParaHashAtSource, SourceClient}, + parachains_loop::{AvailableHeader, SourceClient}, parachains_loop_metrics::ParachainsLoopMetrics, }; use relay_substrate_client::{ Chain, Client, Error as SubstrateError, HeaderIdOf, HeaderOf, RelayChain, }; use relay_utils::relay_loop::Client as RelayClient; -use sp_runtime::traits::Header as HeaderT; /// Shared updatable reference to the maximal parachain header id that we want to sync from the /// source. -pub type RequiredHeaderIdRef = Arc>>>; +pub type RequiredHeaderIdRef = Arc>>>; /// Substrate client as parachain heads source. #[derive(Clone)] pub struct ParachainsSource { client: Client, - maximal_header_id: Option>, + max_head_id: RequiredHeaderIdRef, } impl ParachainsSource

{ /// Creates new parachains source client. pub fn new( client: Client, - maximal_header_id: Option>, + max_head_id: RequiredHeaderIdRef, ) -> Self { - ParachainsSource { client, maximal_header_id } + ParachainsSource { client, max_head_id } } /// Returns reference to the underlying RPC client. @@ -59,11 +59,11 @@ impl ParachainsSource

{ } /// Return decoded head of given parachain. - pub async fn on_chain_parachain_header( + pub async fn on_chain_para_head_id( &self, at_block: HeaderIdOf, para_id: ParaId, - ) -> Result>, SubstrateError> { + ) -> Result>, SubstrateError> { let storage_key = parachain_head_storage_key_at_source(P::SourceRelayChain::PARAS_PALLET_NAME, para_id); let para_head = self.client.raw_storage_value(storage_key, Some(at_block.1)).await?; @@ -72,8 +72,8 @@ impl ParachainsSource

{ Some(para_head) => para_head, None => return Ok(None), }; - - Ok(Some(Decode::decode(&mut ¶_head.0[..])?)) + let para_head: HeaderOf = Decode::decode(&mut ¶_head.0[..])?; + Ok(Some(para_head.id())) } } @@ -105,7 +105,7 @@ where at_block: HeaderIdOf, metrics: Option<&ParachainsLoopMetrics>, para_id: ParaId, - ) -> Result { + ) -> Result, Self::Error> { // we don't need to support many parachains now if para_id.0 != P::SOURCE_PARACHAIN_PARA_ID { return Err(SubstrateError::Custom(format!( @@ -115,44 +115,28 @@ where ))) } - let mut para_hash_at_source = ParaHashAtSource::None; - let mut para_header_number_at_source = None; - match self.on_chain_parachain_header(at_block, para_id).await? { - Some(parachain_header) => { - para_hash_at_source = ParaHashAtSource::Some(parachain_header.hash()); - para_header_number_at_source = Some(*parachain_header.number()); - // never return head that is larger than requested. This way we'll never sync - // headers past `maximal_header_id` - if let Some(ref maximal_header_id) = self.maximal_header_id { - let maximal_header_id = *maximal_header_id.lock().await; - match maximal_header_id { - Some(maximal_header_id) - if *parachain_header.number() > maximal_header_id.0 => - { - // we don't want this header yet => let's report previously requested - // header - para_hash_at_source = ParaHashAtSource::Some(maximal_header_id.1); - para_header_number_at_source = Some(maximal_header_id.0); - }, - Some(_) => (), - None => { - // on-demand relay has not yet asked us to sync anything let's do that - para_hash_at_source = ParaHashAtSource::Unavailable; - para_header_number_at_source = None; - }, - } - } - }, - None => {}, - }; + let mut para_head_id = AvailableHeader::Missing; + if let Some(on_chain_para_head_id) = self.on_chain_para_head_id(at_block, para_id).await? { + // Never return head that is larger than requested. This way we'll never sync + // headers past `max_header_id`. + para_head_id = match *self.max_head_id.lock().await { + AvailableHeader::Unavailable => AvailableHeader::Unavailable, + AvailableHeader::Missing => { + // `max_header_id` is not set. There is no limit. + AvailableHeader::Available(on_chain_para_head_id) + }, + AvailableHeader::Available(max_head_id) => { + // We report at most `max_header_id`. + AvailableHeader::Available(std::cmp::min(on_chain_para_head_id, max_head_id)) + }, + } + } - if let (Some(metrics), Some(para_header_number_at_source)) = - (metrics, para_header_number_at_source) - { - metrics.update_best_parachain_block_at_source(para_id, para_header_number_at_source); + if let (Some(metrics), AvailableHeader::Available(para_head_id)) = (metrics, para_head_id) { + metrics.update_best_parachain_block_at_source(para_id, para_head_id.0); } - Ok(para_hash_at_source) + Ok(para_head_id.map(|para_head_id| para_head_id.1)) } async fn prove_parachain_heads( diff --git a/bridges/relays/parachains/src/parachains_loop.rs b/bridges/relays/parachains/src/parachains_loop.rs index 3c7ac85913947..f44f986d63231 100644 --- a/bridges/relays/parachains/src/parachains_loop.rs +++ b/bridges/relays/parachains/src/parachains_loop.rs @@ -52,29 +52,33 @@ pub enum ParachainSyncStrategy { All, } -/// Parachain head hash, available at the source (relay) chain. +/// Parachain header availability at a certain chain. #[derive(Clone, Copy, Debug)] -pub enum ParaHashAtSource { - /// There's no parachain head at the source chain. - /// - /// Normally it means that the parachain is not registered there. - None, - /// Parachain head with given hash is available at the source chain. - Some(ParaHash), - /// The source client refuses to report parachain head hash at this moment. +pub enum AvailableHeader { + /// The client refuses to report parachain head at this moment. /// /// It is a "mild" error, which may appear when e.g. on-demand parachains relay is used. /// This variant must be treated as "we don't want to update parachain head value at the /// target chain at this moment". Unavailable, + /// There's no parachain header at the relay chain. + /// + /// Normally it means that the parachain is not registered there. + Missing, + /// Parachain head with given hash is available at the source chain. + Available(T), } -impl ParaHashAtSource { - /// Return parachain head hash, if available. - pub fn hash(&self) -> Option<&ParaHash> { - match *self { - ParaHashAtSource::Some(ref para_hash) => Some(para_hash), - _ => None, +impl AvailableHeader { + /// Transform contained value. + pub fn map(self, f: F) -> AvailableHeader + where + F: FnOnce(T) -> U, + { + match self { + AvailableHeader::Unavailable => AvailableHeader::Unavailable, + AvailableHeader::Missing => AvailableHeader::Missing, + AvailableHeader::Available(val) => AvailableHeader::Available(f(val)), } } } @@ -94,7 +98,7 @@ pub trait SourceClient: RelayClient { at_block: HeaderIdOf, metrics: Option<&ParachainsLoopMetrics>, para_id: ParaId, - ) -> Result; + ) -> Result, Self::Error>; /// Get parachain heads proof. /// @@ -342,7 +346,7 @@ where /// Given heads at source and target clients, returns set of heads that are out of sync. fn select_parachains_to_update( - heads_at_source: BTreeMap, + heads_at_source: BTreeMap>, heads_at_target: BTreeMap>, best_finalized_relay_block: HeaderIdOf, ) -> Vec @@ -368,12 +372,12 @@ where .zip(heads_at_target.into_iter()) .filter(|((para, head_at_source), (_, head_at_target))| { let needs_update = match (head_at_source, head_at_target) { - (ParaHashAtSource::Unavailable, _) => { + (AvailableHeader::Unavailable, _) => { // source client has politely asked us not to update current parachain head // at the target chain false }, - (ParaHashAtSource::Some(head_at_source), Some(head_at_target)) + (AvailableHeader::Available(head_at_source), Some(head_at_target)) if head_at_target.at_relay_block_number < best_finalized_relay_block.0 && head_at_target.head_hash != *head_at_source => { @@ -381,22 +385,22 @@ where // client true }, - (ParaHashAtSource::Some(_), Some(_)) => { + (AvailableHeader::Available(_), Some(_)) => { // this is normal case when relay has recently updated heads, when parachain is // not progressing, or when our source client is still syncing false }, - (ParaHashAtSource::Some(_), None) => { + (AvailableHeader::Available(_), None) => { // parachain is not yet known to the target client. This is true when parachain // or bridge has been just onboarded/started true }, - (ParaHashAtSource::None, Some(_)) => { + (AvailableHeader::Missing, Some(_)) => { // parachain/parathread has been offboarded removed from the system. It needs to // be propageted to the target client true }, - (ParaHashAtSource::None, None) => { + (AvailableHeader::Missing, None) => { // all's good - parachain is unknown to both clients false }, @@ -435,7 +439,7 @@ async fn read_heads_at_source( metrics: Option<&ParachainsLoopMetrics>, at_relay_block: &HeaderIdOf, parachains: &[ParaId], -) -> Result, FailedClient> { +) -> Result>, FailedClient> { let mut para_head_hashes = BTreeMap::new(); for para in parachains { let para_head = source_client.parachain_head(*at_relay_block, metrics, *para).await; @@ -612,7 +616,7 @@ mod tests { #[derive(Clone, Debug)] struct TestClientData { source_sync_status: Result, - source_heads: BTreeMap>, + source_heads: BTreeMap, TestError>>, source_proofs: BTreeMap, TestError>>, target_best_block: Result, TestError>, @@ -627,7 +631,7 @@ mod tests { pub fn minimal() -> Self { TestClientData { source_sync_status: Ok(true), - source_heads: vec![(PARA_ID, Ok(ParaHashAtSource::Some(PARA_0_HASH)))] + source_heads: vec![(PARA_ID, Ok(AvailableHeader::Available(PARA_0_HASH)))] .into_iter() .collect(), source_proofs: vec![(PARA_ID, Ok(PARA_0_HASH.encode()))].into_iter().collect(), @@ -676,10 +680,10 @@ mod tests { _at_block: HeaderIdOf, _metrics: Option<&ParachainsLoopMetrics>, para_id: ParaId, - ) -> Result { + ) -> Result, TestError> { match self.data.lock().await.source_heads.get(¶_id.0).cloned() { Some(result) => result, - None => Ok(ParaHashAtSource::None), + None => Ok(AvailableHeader::Missing), } } @@ -988,7 +992,7 @@ mod tests { fn parachain_is_not_updated_if_it_is_unknown_to_both_clients() { assert_eq!( select_parachains_to_update::( - vec![(ParaId(PARA_ID), ParaHashAtSource::None)].into_iter().collect(), + vec![(ParaId(PARA_ID), AvailableHeader::Missing)].into_iter().collect(), vec![(ParaId(PARA_ID), None)].into_iter().collect(), HeaderId(10, Default::default()), ), @@ -1000,7 +1004,7 @@ mod tests { fn parachain_is_not_updated_if_it_has_been_updated_at_better_relay_block() { assert_eq!( select_parachains_to_update::( - vec![(ParaId(PARA_ID), ParaHashAtSource::Some(PARA_0_HASH))] + vec![(ParaId(PARA_ID), AvailableHeader::Available(PARA_0_HASH))] .into_iter() .collect(), vec![( @@ -1019,7 +1023,7 @@ mod tests { fn parachain_is_not_updated_if_hash_is_the_same_at_next_relay_block() { assert_eq!( select_parachains_to_update::( - vec![(ParaId(PARA_ID), ParaHashAtSource::Some(PARA_0_HASH))] + vec![(ParaId(PARA_ID), AvailableHeader::Available(PARA_0_HASH))] .into_iter() .collect(), vec![( @@ -1038,7 +1042,7 @@ mod tests { fn parachain_is_updated_after_offboarding() { assert_eq!( select_parachains_to_update::( - vec![(ParaId(PARA_ID), ParaHashAtSource::None)].into_iter().collect(), + vec![(ParaId(PARA_ID), AvailableHeader::Missing)].into_iter().collect(), vec![( ParaId(PARA_ID), Some(BestParaHeadHash { @@ -1058,7 +1062,7 @@ mod tests { fn parachain_is_updated_after_onboarding() { assert_eq!( select_parachains_to_update::( - vec![(ParaId(PARA_ID), ParaHashAtSource::Some(PARA_0_HASH))] + vec![(ParaId(PARA_ID), AvailableHeader::Available(PARA_0_HASH))] .into_iter() .collect(), vec![(ParaId(PARA_ID), None)].into_iter().collect(), @@ -1072,7 +1076,7 @@ mod tests { fn parachain_is_updated_if_newer_head_is_known() { assert_eq!( select_parachains_to_update::( - vec![(ParaId(PARA_ID), ParaHashAtSource::Some(PARA_1_HASH))] + vec![(ParaId(PARA_ID), AvailableHeader::Available(PARA_1_HASH))] .into_iter() .collect(), vec![( @@ -1091,7 +1095,7 @@ mod tests { fn parachain_is_not_updated_if_source_head_is_unavailable() { assert_eq!( select_parachains_to_update::( - vec![(ParaId(PARA_ID), ParaHashAtSource::Unavailable)].into_iter().collect(), + vec![(ParaId(PARA_ID), AvailableHeader::Unavailable)].into_iter().collect(), vec![( ParaId(PARA_ID), Some(BestParaHeadHash { at_relay_block_number: 0, head_hash: PARA_0_HASH })