substrate_relay_helper/parachains/
source.rs

1// Copyright 2019-2021 Parity Technologies (UK) Ltd.
2// This file is part of Parity Bridges Common.
3
4// Parity Bridges Common is free software: you can redistribute it and/or modify
5// it under the terms of the GNU General Public License as published by
6// the Free Software Foundation, either version 3 of the License, or
7// (at your option) any later version.
8
9// Parity Bridges Common is distributed in the hope that it will be useful,
10// but WITHOUT ANY WARRANTY; without even the implied warranty of
11// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12// GNU General Public License for more details.
13
14// You should have received a copy of the GNU General Public License
15// along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
16
17//! Parachain heads source.
18
19use crate::{
20	parachains::{ParachainsPipelineAdapter, SubstrateParachainsPipeline},
21	proofs::to_raw_storage_proof,
22};
23use async_std::sync::{Arc, Mutex};
24use async_trait::async_trait;
25use bp_parachains::parachain_head_storage_key_at_source;
26use bp_polkadot_core::parachains::{ParaHash, ParaHead, ParaHeadsProof, ParaId};
27use bp_runtime::HeaderIdProvider;
28use codec::Decode;
29use parachains_relay::parachains_loop::{AvailableHeader, SourceClient};
30use relay_substrate_client::{
31	is_ancient_block, Chain, Client, Error as SubstrateError, HeaderIdOf, HeaderOf, ParachainBase,
32	RelayChain,
33};
34use relay_utils::relay_loop::Client as RelayClient;
35
36/// Shared updatable reference to the maximal parachain header id that we want to sync from the
37/// source.
38pub type RequiredHeaderIdRef<C> = Arc<Mutex<AvailableHeader<HeaderIdOf<C>>>>;
39
40/// Substrate client as parachain heads source.
41#[derive(Clone)]
42pub struct ParachainsSource<P: SubstrateParachainsPipeline, SourceRelayClnt> {
43	client: SourceRelayClnt,
44	max_head_id: RequiredHeaderIdRef<P::SourceParachain>,
45}
46
47impl<P: SubstrateParachainsPipeline, SourceRelayClnt: Client<P::SourceRelayChain>>
48	ParachainsSource<P, SourceRelayClnt>
49{
50	/// Creates new parachains source client.
51	pub fn new(
52		client: SourceRelayClnt,
53		max_head_id: RequiredHeaderIdRef<P::SourceParachain>,
54	) -> Self {
55		ParachainsSource { client, max_head_id }
56	}
57
58	/// Returns reference to the underlying RPC client.
59	pub fn client(&self) -> &SourceRelayClnt {
60		&self.client
61	}
62
63	/// Return decoded head of given parachain.
64	pub async fn on_chain_para_head_id(
65		&self,
66		at_block: HeaderIdOf<P::SourceRelayChain>,
67	) -> Result<Option<HeaderIdOf<P::SourceParachain>>, SubstrateError> {
68		let para_id = ParaId(P::SourceParachain::PARACHAIN_ID);
69		let storage_key =
70			parachain_head_storage_key_at_source(P::SourceRelayChain::PARAS_PALLET_NAME, para_id);
71		let para_head: Option<ParaHead> =
72			self.client.storage_value(at_block.hash(), storage_key).await?;
73		let para_head = match para_head {
74			Some(para_head) => para_head,
75			None => return Ok(None),
76		};
77		let para_head: HeaderOf<P::SourceParachain> = Decode::decode(&mut &para_head.0[..])?;
78		Ok(Some(para_head.id()))
79	}
80}
81
82#[async_trait]
83impl<P: SubstrateParachainsPipeline, SourceRelayClnt: Client<P::SourceRelayChain>> RelayClient
84	for ParachainsSource<P, SourceRelayClnt>
85{
86	type Error = SubstrateError;
87
88	async fn reconnect(&mut self) -> Result<(), SubstrateError> {
89		self.client.reconnect().await
90	}
91}
92
93#[async_trait]
94impl<P: SubstrateParachainsPipeline, SourceRelayClnt: Client<P::SourceRelayChain>>
95	SourceClient<ParachainsPipelineAdapter<P>> for ParachainsSource<P, SourceRelayClnt>
96where
97	P::SourceParachain: Chain<Hash = ParaHash>,
98{
99	async fn ensure_synced(&self) -> Result<bool, Self::Error> {
100		match self.client.ensure_synced().await {
101			Ok(_) => Ok(true),
102			Err(SubstrateError::ClientNotSynced(_)) => Ok(false),
103			Err(e) => Err(e),
104		}
105	}
106
107	async fn parachain_head(
108		&self,
109		at_block: HeaderIdOf<P::SourceRelayChain>,
110	) -> Result<AvailableHeader<HeaderIdOf<P::SourceParachain>>, Self::Error> {
111		// if requested relay header is ancient, then we don't even want to try to read the
112		// parachain head - we simply return `Unavailable`
113		let best_block_number = self.client.best_finalized_header_number().await?;
114		if is_ancient_block(at_block.number(), best_block_number) {
115			log::trace!(
116				target: "bridge",
117				"{} block {:?} is ancient. Cannot prove the {} header there",
118				P::SourceRelayChain::NAME,
119				at_block,
120				P::SourceParachain::NAME,
121			);
122			return Ok(AvailableHeader::Unavailable)
123		}
124
125		// else - try to read head from the source client
126		let mut para_head_id = AvailableHeader::Missing;
127		if let Some(on_chain_para_head_id) = self.on_chain_para_head_id(at_block).await? {
128			// Never return head that is larger than requested. This way we'll never sync
129			// headers past `max_header_id`.
130			para_head_id = match *self.max_head_id.lock().await {
131				AvailableHeader::Unavailable => AvailableHeader::Unavailable,
132				AvailableHeader::Missing => {
133					// `max_header_id` is not set. There is no limit.
134					AvailableHeader::Available(on_chain_para_head_id)
135				},
136				AvailableHeader::Available(max_head_id) if on_chain_para_head_id >= max_head_id => {
137					// We report at most `max_header_id`.
138					AvailableHeader::Available(std::cmp::min(on_chain_para_head_id, max_head_id))
139				},
140				AvailableHeader::Available(_) => {
141					// the `max_head_id` is not yet available at the source chain => wait and avoid
142					// syncing extra headers
143					AvailableHeader::Unavailable
144				},
145			}
146		}
147
148		Ok(para_head_id)
149	}
150
151	async fn prove_parachain_head(
152		&self,
153		at_block: HeaderIdOf<P::SourceRelayChain>,
154	) -> Result<(ParaHeadsProof, ParaHash), Self::Error> {
155		let parachain = ParaId(P::SourceParachain::PARACHAIN_ID);
156		let storage_key =
157			parachain_head_storage_key_at_source(P::SourceRelayChain::PARAS_PALLET_NAME, parachain);
158
159		let storage_proof =
160			self.client.prove_storage(at_block.hash(), vec![storage_key.clone()]).await?;
161
162		// why we're reading parachain head here once again (it has already been read at the
163		// `parachain_head`)? that's because `parachain_head` sometimes returns obsolete parachain
164		// head and loop sometimes asks to prove this obsolete head and gets other (actual) head
165		// instead
166		//
167		// => since we want to provide proper hashes in our `submit_parachain_heads` call, we're
168		// rereading actual value here
169		let parachain_head = self
170			.client
171			.storage_value::<ParaHead>(at_block.hash(), storage_key)
172			.await?
173			.ok_or_else(|| {
174				SubstrateError::Custom(format!(
175					"Failed to read expected parachain {parachain:?} head at {at_block:?}"
176				))
177			})?;
178		let parachain_head_hash = parachain_head.hash();
179
180		Ok((
181			ParaHeadsProof {
182				storage_proof: to_raw_storage_proof::<P::SourceRelayChain>(storage_proof),
183			},
184			parachain_head_hash,
185		))
186	}
187}
substrate_relay_helper/parachains/source.rs

substrate_relay_helper/parachains/
source.rs
