Finality loop: get block justification and authorities change by consensus engine ID (#1619)

* SignedBlock: get justification by consensus engine id

* Define ConsensusLogReader

Making the check for authority changes more generic

* cod review changes

bridges/primitives/header-chain/src/lib.rs

@@ -26,7 +26,7 @@ use scale_info::TypeInfo;
 #[cfg(feature = "std")]
 use serde::{Deserialize, Serialize};
 use sp_finality_grandpa::{AuthorityList, ConsensusLog, SetId, GRANDPA_ENGINE_ID};
-use sp_runtime::{generic::OpaqueDigestItemId, traits::Header as HeaderT, RuntimeDebug};
+use sp_runtime::{traits::Header as HeaderT, Digest, RuntimeDebug};
 use sp_std::boxed::Box;
 
 pub mod justification;
@@ -77,18 +77,31 @@ pub trait FinalityProof<Number>: Clone + Send + Sync + Debug {
 	fn target_header_number(&self) -> Number;
 }
 
-/// Find header digest that schedules next GRANDPA authorities set.
+/// A trait that provides helper methods for querying the consensus log.
-pub fn find_grandpa_authorities_scheduled_change<H: HeaderT>(
+pub trait ConsensusLogReader {
-	header: &H,
+	fn schedules_authorities_change(digest: &Digest) -> bool;
-) -> Option<sp_finality_grandpa::ScheduledChange<H::Number>> {
+}
-	let id = OpaqueDigestItemId::Consensus(&GRANDPA_ENGINE_ID);
 
-	let filter_log = |log: ConsensusLog<H::Number>| match log {
+/// A struct that provides helper methods for querying the GRANDPA consensus log.
-		ConsensusLog::ScheduledChange(change) => Some(change),
+pub struct GrandpaConsensusLogReader<Number>(sp_std::marker::PhantomData<Number>);
-		_ => None,
-	};
 
+impl<Number: Codec> GrandpaConsensusLogReader<Number> {
+	pub fn find_authorities_change(
+		digest: &Digest,
+	) -> Option<sp_finality_grandpa::ScheduledChange<Number>> {
 		// find the first consensus digest with the right ID which converts to
 		// the right kind of consensus log.
-	header.digest().convert_first(|l| l.try_to(id).and_then(filter_log))
+		digest
+			.convert_first(|log| log.consensus_try_to(&GRANDPA_ENGINE_ID))
+			.and_then(|log| match log {
+				ConsensusLog::ScheduledChange(change) => Some(change),
+				_ => None,
+			})
+	}
+}
+
+impl<Number: Codec> ConsensusLogReader for GrandpaConsensusLogReader<Number> {
+	fn schedules_authorities_change(digest: &Digest) -> bool {
+		GrandpaConsensusLogReader::<Number>::find_authorities_change(digest).is_some()
+	}
 }

bridges/relays/client-substrate/src/chain.rs

@@ -27,7 +27,7 @@ use sp_core::{storage::StorageKey, Pair};
 use sp_runtime::{
 	generic::SignedBlock,
 	traits::{Block as BlockT, Dispatchable, Member},
-	EncodedJustification,
+	ConsensusEngineId, EncodedJustification,
 };
 use std::{fmt::Debug, time::Duration};
 
@@ -147,7 +147,7 @@ pub trait BlockWithJustification<Header> {
 	/// Return encoded block extrinsics.
 	fn extrinsics(&self) -> Vec<EncodedExtrinsic>;
 	/// Return block justification, if known.
-	fn justification(&self) -> Option<&EncodedJustification>;
+	fn justification(&self, engine_id: ConsensusEngineId) -> Option<&EncodedJustification>;
 }
 
 /// Transaction before it is signed.
@@ -237,9 +237,7 @@ impl<Block: BlockT> BlockWithJustification<Block::Header> for SignedBlock<Block>
 		self.block.extrinsics().iter().map(Encode::encode).collect()
 	}
 
-	fn justification(&self) -> Option<&EncodedJustification> {
+	fn justification(&self, engine_id: ConsensusEngineId) -> Option<&EncodedJustification> {
-		self.justifications
+		self.justifications.as_ref().and_then(|j| j.get(engine_id))
-			.as_ref()
-			.and_then(|j| j.get(sp_finality_grandpa::GRANDPA_ENGINE_ID))
 	}
 }

bridges/relays/client-substrate/src/sync_header.rs

@@ -14,7 +14,7 @@
 // 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 bp_header_chain::find_grandpa_authorities_scheduled_change;
+use bp_header_chain::ConsensusLogReader;
 use finality_relay::SourceHeader as FinalitySourceHeader;
 use sp_runtime::traits::Header as HeaderT;
 
@@ -44,7 +44,9 @@ impl<Header> From<Header> for SyncHeader<Header> {
 	}
 }
 
-impl<Header: HeaderT> FinalitySourceHeader<Header::Hash, Header::Number> for SyncHeader<Header> {
+impl<Header: HeaderT, R: ConsensusLogReader> FinalitySourceHeader<Header::Hash, Header::Number, R>
+	for SyncHeader<Header>
+{
 	fn hash(&self) -> Header::Hash {
 		self.0.hash()
 	}
@@ -54,6 +56,6 @@ impl<Header: HeaderT> FinalitySourceHeader<Header::Hash, Header::Number> for Syn
 	}
 
 	fn is_mandatory(&self) -> bool {
-		find_grandpa_authorities_scheduled_change(&self.0).is_some()
+		R::schedules_authorities_change(self.digest())
 	}
 }

bridges/relays/finality/src/finality_loop_tests.rs

@@ -30,6 +30,7 @@ use crate::{
 };
 
 use async_trait::async_trait;
+use bp_header_chain::GrandpaConsensusLogReader;
 use futures::{FutureExt, Stream, StreamExt};
 use parking_lot::Mutex;
 use relay_utils::{
@@ -85,6 +86,7 @@ impl FinalitySyncPipeline for TestFinalitySyncPipeline {
 
 	type Hash = TestHash;
 	type Number = TestNumber;
+	type ConsensusLogReader = GrandpaConsensusLogReader<TestNumber>;
 	type Header = TestSourceHeader;
 	type FinalityProof = TestFinalityProof;
 }
@@ -92,7 +94,9 @@ impl FinalitySyncPipeline for TestFinalitySyncPipeline {
 #[derive(Debug, Clone, PartialEq, Eq)]
 struct TestSourceHeader(IsMandatory, TestNumber, TestHash);
 
-impl SourceHeader<TestHash, TestNumber> for TestSourceHeader {
+impl SourceHeader<TestHash, TestNumber, GrandpaConsensusLogReader<TestNumber>>
+	for TestSourceHeader
+{
 	fn hash(&self) -> TestHash {
 		self.2
 	}

bridges/relays/finality/src/lib.rs

@@ -24,7 +24,7 @@ pub use crate::{
 	sync_loop_metrics::SyncLoopMetrics,
 };
 
-use bp_header_chain::FinalityProof;
+use bp_header_chain::{ConsensusLogReader, FinalityProof};
 use std::fmt::Debug;
 
 mod finality_loop;
@@ -42,14 +42,16 @@ pub trait FinalitySyncPipeline: 'static + Clone + Debug + Send + Sync {
 	type Hash: Eq + Clone + Copy + Send + Sync + Debug;
 	/// Headers we're syncing are identified by this number.
 	type Number: relay_utils::BlockNumberBase;
+	/// A reader that can extract the consensus log from the header digest and interpret it.
+	type ConsensusLogReader: ConsensusLogReader;
 	/// Type of header that we're syncing.
-	type Header: SourceHeader<Self::Hash, Self::Number>;
+	type Header: SourceHeader<Self::Hash, Self::Number, Self::ConsensusLogReader>;
 	/// Finality proof type.
 	type FinalityProof: FinalityProof<Self::Number>;
 }
 
 /// Header that we're receiving from source node.
-pub trait SourceHeader<Hash, Number>: Clone + Debug + PartialEq + Send + Sync {
+pub trait SourceHeader<Hash, Number, Reader>: Clone + Debug + PartialEq + Send + Sync {
 	/// Returns hash of header.
 	fn hash(&self) -> Hash;
 	/// Returns number of header.

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

@@ -19,9 +19,8 @@
 use crate::error::Error;
 use async_trait::async_trait;
 use bp_header_chain::{
-	find_grandpa_authorities_scheduled_change,
 	justification::{verify_justification, GrandpaJustification},
-	FinalityProof,
+	ConsensusLogReader, FinalityProof, GrandpaConsensusLogReader,
 };
 use bp_runtime::{BasicOperatingMode, OperatingMode};
 use codec::{Decode, Encode};
@@ -32,7 +31,7 @@ use relay_substrate_client::{
 	Subscription, SubstrateFinalityClient, SubstrateGrandpaFinalityClient,
 };
 use sp_core::{storage::StorageKey, Bytes};
-use sp_finality_grandpa::AuthorityList as GrandpaAuthoritiesSet;
+use sp_finality_grandpa::{AuthorityList as GrandpaAuthoritiesSet, GRANDPA_ENGINE_ID};
 use sp_runtime::{traits::Header, ConsensusEngineId};
 use std::marker::PhantomData;
 
@@ -41,6 +40,8 @@ use std::marker::PhantomData;
 pub trait Engine<C: Chain>: Send {
 	/// Unique consensus engine identifier.
 	const ID: ConsensusEngineId;
+	/// A reader that can extract the consensus log from the header digest and interpret it.
+	type ConsensusLogReader: ConsensusLogReader;
 	/// Type of Finality RPC client used by this engine.
 	type FinalityClient: SubstrateFinalityClient<C>;
 	/// Type of finality proofs, used by consensus engine.
@@ -50,22 +51,11 @@ pub trait Engine<C: Chain>: Send {
 	/// Type of bridge pallet operating mode.
 	type OperatingMode: OperatingMode + 'static;
 
-	/// Returns storage key at the bridged (target) chain that corresponds to the variable
-	/// that holds the operating mode of the pallet.
-	fn pallet_operating_mode_key() -> StorageKey;
 	/// Returns storage at the bridged (target) chain that corresponds to some value that is
 	/// missing from the storage until bridge pallet is initialized.
 	///
 	/// Note that we don't care about type of the value - just if it present or not.
 	fn is_initialized_key() -> StorageKey;
-	/// A method to subscribe to encoded finality proofs, given source client.
-	async fn finality_proofs(client: &Client<C>) -> Result<Subscription<Bytes>, SubstrateError> {
-		client.subscribe_finality_justifications::<Self::FinalityClient>().await
-	}
-	/// Prepare initialization data for the finality bridge pallet.
-	async fn prepare_initialization_data(
-		client: Client<C>,
-	) -> Result<Self::InitializationData, Error<HashOf<C>, BlockNumberOf<C>>>;
 
 	/// Returns `Ok(true)` if finality pallet at the bridged chain has already been initialized.
 	async fn is_initialized<TargetChain: Chain>(
@@ -77,6 +67,10 @@ pub trait Engine<C: Chain>: Send {
 			.is_some())
 	}
 
+	/// Returns storage key at the bridged (target) chain that corresponds to the variable
+	/// that holds the operating mode of the pallet.
+	fn pallet_operating_mode_key() -> StorageKey;
+
 	/// Returns `Ok(true)` if finality pallet at the bridged chain is halted.
 	async fn is_halted<TargetChain: Chain>(
 		target_client: &Client<TargetChain>,
@@ -87,6 +81,16 @@ pub trait Engine<C: Chain>: Send {
 			.map(|operating_mode| operating_mode.is_halted())
 			.unwrap_or(false))
 	}
+
+	/// A method to subscribe to encoded finality proofs, given source client.
+	async fn finality_proofs(client: &Client<C>) -> Result<Subscription<Bytes>, SubstrateError> {
+		client.subscribe_finality_justifications::<Self::FinalityClient>().await
+	}
+
+	/// Prepare initialization data for the finality bridge pallet.
+	async fn prepare_initialization_data(
+		client: Client<C>,
+	) -> Result<Self::InitializationData, Error<HashOf<C>, BlockNumberOf<C>>>;
 }
 
 /// GRANDPA finality engine.
@@ -120,20 +124,21 @@ impl<C: ChainWithGrandpa> Grandpa<C> {
 
 #[async_trait]
 impl<C: ChainWithGrandpa> Engine<C> for Grandpa<C> {
-	const ID: ConsensusEngineId = sp_finality_grandpa::GRANDPA_ENGINE_ID;
+	const ID: ConsensusEngineId = GRANDPA_ENGINE_ID;
+	type ConsensusLogReader = GrandpaConsensusLogReader<<C::Header as Header>::Number>;
 	type FinalityClient = SubstrateGrandpaFinalityClient;
 	type FinalityProof = GrandpaJustification<HeaderOf<C>>;
 	type InitializationData = bp_header_chain::InitializationData<C::Header>;
 	type OperatingMode = BasicOperatingMode;
 
-	fn pallet_operating_mode_key() -> StorageKey {
-		bp_header_chain::storage_keys::pallet_operating_mode_key(C::WITH_CHAIN_GRANDPA_PALLET_NAME)
-	}
-
 	fn is_initialized_key() -> StorageKey {
 		bp_header_chain::storage_keys::best_finalized_key(C::WITH_CHAIN_GRANDPA_PALLET_NAME)
 	}
 
+	fn pallet_operating_mode_key() -> StorageKey {
+		bp_header_chain::storage_keys::pallet_operating_mode_key(C::WITH_CHAIN_GRANDPA_PALLET_NAME)
+	}
+
 	/// Prepare initialization data for the GRANDPA verifier pallet.
 	async fn prepare_initialization_data(
 		source_client: Client<C>,
@@ -183,11 +188,14 @@ impl<C: ChainWithGrandpa> Engine<C> for Grandpa<C> {
 		// 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);
+		let scheduled_change =
+			GrandpaConsensusLogReader::<BlockNumberOf<C>>::find_authorities_change(
+				initial_header.digest(),
+			);
 		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",
+			regular change to have zero delay",
 			initial_header_hash,
 			scheduled_change.as_ref().map(|c| c.delay),
 		);

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

@@ -87,7 +87,8 @@ impl<P: SubstrateFinalitySyncPipeline> FinalitySyncPipeline for FinalitySyncPipe
 
 	type Hash = HashOf<P::SourceChain>;
 	type Number = BlockNumberOf<P::SourceChain>;
-	type Header = relay_substrate_client::SyncHeader<HeaderOf<P::SourceChain>>;
+	type ConsensusLogReader = <P::FinalityEngine as Engine<P::SourceChain>>::ConsensusLogReader;
+	type Header = SyncHeader<HeaderOf<P::SourceChain>>;
 	type FinalityProof = SubstrateFinalityProof<P>;
 }
 

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

@@ -33,15 +33,8 @@ use std::pin::Pin;
 pub type RequiredHeaderNumberRef<C> = Arc<Mutex<<C as bp_runtime::Chain>::BlockNumber>>;
 
 /// Substrate finality proofs stream.
-pub type SubstrateFinalityProofsStream<P> = Pin<
+pub type SubstrateFinalityProofsStream<P> =
-	Box<
+	Pin<Box<dyn Stream<Item = SubstrateFinalityProof<P>> + Send>>;
-		dyn Stream<
-				Item = <<P as SubstrateFinalitySyncPipeline>::FinalityEngine as Engine<
-					<P as SubstrateFinalitySyncPipeline>::SourceChain,
-				>>::FinalityProof,
-			> + Send,
-	>,
->;
 
 /// Substrate finality proof. Specific to the used `FinalityEngine`.
 pub type SubstrateFinalityProof<P> =
@@ -130,7 +123,7 @@ impl<P: SubstrateFinalitySyncPipeline> SourceClient<FinalitySyncPipelineAdapter<
 		let signed_block = self.client.get_block(Some(header_hash)).await?;
 
 		let justification = signed_block
-			.justification()
+			.justification(P::FinalityEngine::ID)
 			.map(|raw_justification| {
 				SubstrateFinalityProof::<P>::decode(&mut raw_justification.as_slice())
 			})

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

@@ -18,13 +18,14 @@
 
 use async_std::sync::{Arc, Mutex};
 use async_trait::async_trait;
+use bp_header_chain::ConsensusLogReader;
 use futures::{select, FutureExt};
 use num_traits::{One, Zero};
+use sp_runtime::traits::Header;
 
-use finality_relay::{FinalitySyncParams, SourceHeader, TargetClient as FinalityTargetClient};
+use finality_relay::{FinalitySyncParams, TargetClient as FinalityTargetClient};
 use relay_substrate_client::{
-	AccountIdOf, AccountKeyPairOf, BlockNumberOf, Chain, Client, HeaderOf, SyncHeader,
+	AccountIdOf, AccountKeyPairOf, BlockNumberOf, Chain, Client, TransactionSignScheme,
-	TransactionSignScheme,
 };
 use relay_utils::{
 	metrics::MetricsParams, relay_loop::Client as RelayClient, FailedClient, MaybeConnectionError,
@@ -33,6 +34,7 @@ use relay_utils::{
 
 use crate::{
 	finality::{
+		engine::Engine,
 		source::{RequiredHeaderNumberRef, SubstrateFinalitySource},
 		target::SubstrateFinalityTarget,
 		SubstrateFinalitySyncPipeline, RECENT_FINALITY_PROOFS_LIMIT,
@@ -416,9 +418,10 @@ async fn find_mandatory_header_in_range<P: SubstrateFinalitySyncPipeline>(
 ) -> Result<Option<BlockNumberOf<P::SourceChain>>, relay_substrate_client::Error> {
 	let mut current = range.0;
 	while current <= range.1 {
-		let header: SyncHeader<HeaderOf<P::SourceChain>> =
+		let header = finality_source.client().header_by_number(current).await?;
-			finality_source.client().header_by_number(current).await?.into();
+		if <P::FinalityEngine as Engine<P::SourceChain>>::ConsensusLogReader::schedules_authorities_change(
-		if header.is_mandatory() {
+			header.digest(),
+		) {
 			return Ok(Some(current))
 		}