pezkuwi-omni-node-lib 0.1.0

// Copyright (C) Parity Technologies (UK) Ltd. and Dijital Kurdistan Tech Institute
// This file is part of Pezcumulus.
// SPDX-License-Identifier: Apache-2.0

// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// 	http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use crate::{
	cli::{AuthoringPolicy, DevSealMode},
	common::{
		aura::{AuraIdT, AuraRuntimeApi},
		rpc::{BuildRpcExtensions, BuildTeyrchainRpcExtensions},
		spec::{
			BaseNodeSpec, BuildImportQueue, ClientBlockImport, DynNodeSpec, InitBlockImport,
			NodeSpec, StartConsensus,
		},
		types::{
			AccountId, Balance, Hash, Nonce, TeyrchainBackend, TeyrchainBlockImport,
			TeyrchainClient,
		},
		ConstructNodeRuntimeApi, NodeBlock, NodeExtraArgs,
	},
};
use codec::Encode;
use futures::{prelude::*, FutureExt};
use pezcumulus_client_collator::service::{
	CollatorService, ServiceInterface as CollatorServiceInterface,
};
#[docify::export(slot_based_colator_import)]
use pezcumulus_client_consensus_aura::collators::slot_based::{
	self as slot_based, Params as SlotBasedParams,
};
use pezcumulus_client_consensus_aura::{
	collators::{
		lookahead::{self as aura, Params as AuraParams},
		slot_based::{SlotBasedBlockImport, SlotBasedBlockImportHandle},
	},
	equivocation_import_queue::Verifier as EquivocationVerifier,
};
use pezcumulus_client_consensus_proposer::ProposerInterface;
use pezcumulus_client_consensus_relay_chain::Verifier as RelayChainVerifier;
use pezcumulus_client_service::CollatorSybilResistance;
use pezcumulus_client_teyrchain_inherent::MockValidationDataInherentDataProvider;
use pezcumulus_primitives_core::{
	relay_chain::ValidationCode, CollectCollationInfo, GetTeyrchainInfo, ParaId,
};
use pezcumulus_relay_chain_interface::{OverseerHandle, RelayChainInterface};
use pezkuwi_primitives::{CollatorPair, UpgradeGoAhead};
use pezsc_client_api::{Backend, BlockchainEvents};
use pezsc_client_db::DbHash;
use pezsc_consensus::{
	import_queue::{BasicQueue, Verifier as VerifierT},
	BlockImportParams, DefaultImportQueue, LongestChain,
};
use pezsc_consensus_manual_seal::consensus::aura::AuraConsensusDataProvider;
use pezsc_network::{config::FullNetworkConfiguration, NotificationMetrics, PeerId};
use pezsc_service::{Configuration, Error, PartialComponents, TaskManager};
use pezsc_telemetry::TelemetryHandle;
use pezsc_transaction_pool::TransactionPoolHandle;
use pezsc_transaction_pool_api::OffchainTransactionPoolFactory;
use pezsp_api::ProvideRuntimeApi;
use pezsp_core::traits::SpawnEssentialNamed;
use pezsp_inherents::CreateInherentDataProviders;
use pezsp_keystore::KeystorePtr;
use pezsp_runtime::{
	app_crypto::AppCrypto,
	traits::{Block as BlockT, Header as HeaderT, UniqueSaturatedInto},
};
use prometheus_endpoint::Registry;
use std::{marker::PhantomData, sync::Arc, time::Duration};

struct Verifier<Block, Client, AuraId> {
	client: Arc<Client>,
	aura_verifier: Box<dyn VerifierT<Block>>,
	relay_chain_verifier: Box<dyn VerifierT<Block>>,
	_phantom: PhantomData<AuraId>,
}

#[async_trait::async_trait]
impl<Block: BlockT, Client, AuraId> VerifierT<Block> for Verifier<Block, Client, AuraId>
where
	Client: ProvideRuntimeApi<Block> + Send + Sync,
	Client::Api: AuraRuntimeApi<Block, AuraId>,
	AuraId: AuraIdT + Sync,
{
	async fn verify(
		&self,
		block_import: BlockImportParams<Block>,
	) -> Result<BlockImportParams<Block>, String> {
		if self.client.runtime_api().has_aura_api(*block_import.header.parent_hash()) {
			self.aura_verifier.verify(block_import).await
		} else {
			self.relay_chain_verifier.verify(block_import).await
		}
	}
}

/// Build the import queue for teyrchain runtimes that started with relay chain consensus and
/// switched to aura.
pub(crate) struct BuildRelayToAuraImportQueue<Block, RuntimeApi, AuraId, BlockImport>(
	PhantomData<(Block, RuntimeApi, AuraId, BlockImport)>,
);

impl<Block: BlockT, RuntimeApi, AuraId, BlockImport>
	BuildImportQueue<Block, RuntimeApi, BlockImport>
	for BuildRelayToAuraImportQueue<Block, RuntimeApi, AuraId, BlockImport>
where
	RuntimeApi: ConstructNodeRuntimeApi<Block, TeyrchainClient<Block, RuntimeApi>>,
	RuntimeApi::RuntimeApi: AuraRuntimeApi<Block, AuraId>,
	AuraId: AuraIdT + Sync,
	BlockImport:
		pezsc_consensus::BlockImport<Block, Error = pezsp_consensus::Error> + Send + Sync + 'static,
{
	fn build_import_queue(
		client: Arc<TeyrchainClient<Block, RuntimeApi>>,
		block_import: TeyrchainBlockImport<Block, BlockImport>,
		config: &Configuration,
		telemetry_handle: Option<TelemetryHandle>,
		task_manager: &TaskManager,
	) -> pezsc_service::error::Result<DefaultImportQueue<Block>> {
		let inherent_data_providers = move |_, _| async move {
			Ok(pezsp_timestamp::InherentDataProvider::from_system_time())
		};
		let registry = config.prometheus_registry();
		let spawner = task_manager.spawn_essential_handle();

		let relay_chain_verifier =
			Box::new(RelayChainVerifier::new(client.clone(), inherent_data_providers));

		let equivocation_aura_verifier =
			EquivocationVerifier::<<AuraId as AppCrypto>::Pair, _, _, _>::new(
				client.clone(),
				inherent_data_providers,
				telemetry_handle,
			);

		let verifier = Verifier {
			client,
			aura_verifier: Box::new(equivocation_aura_verifier),
			relay_chain_verifier,
			_phantom: Default::default(),
		};

		Ok(BasicQueue::new(verifier, Box::new(block_import), None, &spawner, registry))
	}
}

/// Uses the lookahead collator to support async backing.
///
/// Start an aura powered teyrchain node. Some system chains use this.
pub(crate) struct AuraNode<Block, RuntimeApi, AuraId, StartConsensus, InitBlockImport>(
	pub PhantomData<(Block, RuntimeApi, AuraId, StartConsensus, InitBlockImport)>,
);

impl<Block, RuntimeApi, AuraId, StartConsensus, InitBlockImport> Default
	for AuraNode<Block, RuntimeApi, AuraId, StartConsensus, InitBlockImport>
{
	fn default() -> Self {
		Self(Default::default())
	}
}

impl<Block, RuntimeApi, AuraId, StartConsensus, InitBlockImport> BaseNodeSpec
	for AuraNode<Block, RuntimeApi, AuraId, StartConsensus, InitBlockImport>
where
	Block: NodeBlock,
	RuntimeApi: ConstructNodeRuntimeApi<Block, TeyrchainClient<Block, RuntimeApi>>,
	RuntimeApi::RuntimeApi: AuraRuntimeApi<Block, AuraId>
		+ pezpallet_transaction_payment_rpc::TransactionPaymentRuntimeApi<Block, Balance>
		+ bizinikiwi_frame_rpc_system::AccountNonceApi<Block, AccountId, Nonce>,
	AuraId: AuraIdT + Sync,
	InitBlockImport: self::InitBlockImport<Block, RuntimeApi> + Send,
	InitBlockImport::BlockImport:
		pezsc_consensus::BlockImport<Block, Error = pezsp_consensus::Error> + 'static,
{
	type Block = Block;
	type RuntimeApi = RuntimeApi;
	type BuildImportQueue =
		BuildRelayToAuraImportQueue<Block, RuntimeApi, AuraId, InitBlockImport::BlockImport>;
	type InitBlockImport = InitBlockImport;
}

impl<Block, RuntimeApi, AuraId, StartConsensus, InitBlockImport> NodeSpec
	for AuraNode<Block, RuntimeApi, AuraId, StartConsensus, InitBlockImport>
where
	Block: NodeBlock,
	RuntimeApi: ConstructNodeRuntimeApi<Block, TeyrchainClient<Block, RuntimeApi>>,
	RuntimeApi::RuntimeApi: AuraRuntimeApi<Block, AuraId>
		+ pezpallet_transaction_payment_rpc::TransactionPaymentRuntimeApi<Block, Balance>
		+ bizinikiwi_frame_rpc_system::AccountNonceApi<Block, AccountId, Nonce>,
	AuraId: AuraIdT + Sync,
	StartConsensus: self::StartConsensus<
			Block,
			RuntimeApi,
			InitBlockImport::BlockImport,
			InitBlockImport::BlockImportAuxiliaryData,
		> + 'static,
	InitBlockImport: self::InitBlockImport<Block, RuntimeApi> + Send + 'static,
	InitBlockImport::BlockImport:
		pezsc_consensus::BlockImport<Block, Error = pezsp_consensus::Error> + 'static,
{
	type BuildRpcExtensions = BuildTeyrchainRpcExtensions<Block, RuntimeApi>;
	type StartConsensus = StartConsensus;
	const SYBIL_RESISTANCE: CollatorSybilResistance = CollatorSybilResistance::Resistant;

	fn start_dev_node(
		mut config: Configuration,
		mode: DevSealMode,
	) -> pezsc_service::error::Result<TaskManager> {
		let PartialComponents {
			client,
			backend,
			mut task_manager,
			import_queue,
			keystore_container,
			select_chain: _,
			transaction_pool,
			other: (_, mut telemetry, _, _),
		} = Self::new_partial(&config)?;

		// Since this is a dev node, prevent it from connecting to peers.
		config.network.default_peers_set.in_peers = 0;
		config.network.default_peers_set.out_peers = 0;
		let net_config =
			FullNetworkConfiguration::<_, _, pezsc_network::Litep2pNetworkBackend>::new(
				&config.network,
				None,
			);

		let (network, system_rpc_tx, tx_handler_controller, sync_service) =
			pezsc_service::build_network(pezsc_service::BuildNetworkParams {
				config: &config,
				client: client.clone(),
				transaction_pool: transaction_pool.clone(),
				spawn_handle: task_manager.spawn_handle(),
				import_queue,
				net_config,
				block_announce_validator_builder: None,
				warp_sync_config: None,
				block_relay: None,
				metrics: NotificationMetrics::new(None),
			})?;

		if config.offchain_worker.enabled {
			let offchain_workers =
				pezsc_offchain::OffchainWorkers::new(pezsc_offchain::OffchainWorkerOptions {
					runtime_api_provider: client.clone(),
					keystore: Some(keystore_container.keystore()),
					offchain_db: backend.offchain_storage(),
					transaction_pool: Some(OffchainTransactionPoolFactory::new(
						transaction_pool.clone(),
					)),
					network_provider: Arc::new(network.clone()),
					is_validator: config.role.is_authority(),
					enable_http_requests: true,
					custom_extensions: move |_| vec![],
				})?;
			task_manager.spawn_handle().spawn(
				"offchain-workers-runner",
				"offchain-work",
				offchain_workers.run(client.clone(), task_manager.spawn_handle()).boxed(),
			);
		}

		let proposer = pezsc_basic_authorship::ProposerFactory::new(
			task_manager.spawn_handle(),
			client.clone(),
			transaction_pool.clone(),
			None,
			None,
		);

		// Note: Changing slot durations are currently not supported
		let slot_duration = pezsc_consensus_aura::slot_duration(&*client)
			.expect("slot_duration is always present; qed.");

		// The aura digest provider will provide digests that match the provided timestamp data.
		// Without this, the AURA teyrchain runtimes complain about slot mismatches.
		let aura_digest_provider = AuraConsensusDataProvider::new_with_slot_duration(slot_duration);

		let para_id =
			Self::teyrchain_id(&client, &config).ok_or("Failed to retrieve the teyrchain id")?;
		let create_inherent_data_providers =
			Self::create_dev_node_inherent_data_providers(client.clone(), para_id, slot_duration);

		match mode {
			DevSealMode::InstantSeal => {
				let params = pezsc_consensus_manual_seal::InstantSealParams {
					block_import: client.clone(),
					env: proposer,
					client: client.clone(),
					pool: transaction_pool.clone(),
					select_chain: LongestChain::new(backend.clone()),
					consensus_data_provider: Some(Box::new(aura_digest_provider)),
					create_inherent_data_providers,
				};

				let authorship_future = pezsc_consensus_manual_seal::run_instant_seal(params);
				task_manager.spawn_essential_handle().spawn_blocking(
					"instant-seal",
					None,
					authorship_future,
				);
			},
			DevSealMode::ManualSeal(block_time) => {
				let (manual_seal_sink, manual_seal_stream) = futures::channel::mpsc::channel(1024);
				let mut manual_seal_sink_clone = manual_seal_sink.clone();
				task_manager
					.spawn_essential_handle()
					.spawn("block_authoring", None, async move {
						loop {
							futures_timer::Delay::new(std::time::Duration::from_millis(block_time))
								.await;
							manual_seal_sink_clone
								.try_send(
									pezsc_consensus_manual_seal::EngineCommand::SealNewBlock {
										create_empty: true,
										finalize: true,
										parent_hash: None,
										sender: None,
									},
								)
								.unwrap();
						}
					});

				let params = pezsc_consensus_manual_seal::ManualSealParams {
					block_import: client.clone(),
					env: proposer,
					client: client.clone(),
					pool: transaction_pool.clone(),
					select_chain: LongestChain::new(backend.clone()),
					commands_stream: Box::pin(manual_seal_stream),
					consensus_data_provider: Some(Box::new(aura_digest_provider)),
					create_inherent_data_providers,
				};

				let authorship_future = pezsc_consensus_manual_seal::run_manual_seal(params);
				task_manager.spawn_essential_handle().spawn_blocking(
					"manual-seal",
					None,
					authorship_future,
				);
			},
		}

		let rpc_extensions_builder = {
			let client = client.clone();
			let transaction_pool = transaction_pool.clone();
			let backend_for_rpc = backend.clone();

			Box::new(move |_| {
				let module = Self::BuildRpcExtensions::build_rpc_extensions(
					client.clone(),
					backend_for_rpc.clone(),
					transaction_pool.clone(),
					None,
				)?;
				Ok(module)
			})
		};

		let _rpc_handlers = pezsc_service::spawn_tasks(pezsc_service::SpawnTasksParams {
			network,
			client,
			keystore: keystore_container.keystore(),
			task_manager: &mut task_manager,
			transaction_pool,
			rpc_builder: rpc_extensions_builder,
			backend,
			system_rpc_tx,
			tx_handler_controller,
			sync_service,
			config,
			telemetry: telemetry.as_mut(),
			tracing_execute_block: None,
		})?;

		Ok(task_manager)
	}
}

impl<Block, RuntimeApi, AuraId, StartConsensus, InitBlockImport>
	AuraNode<Block, RuntimeApi, AuraId, StartConsensus, InitBlockImport>
where
	Block: NodeBlock,
	RuntimeApi: ConstructNodeRuntimeApi<Block, TeyrchainClient<Block, RuntimeApi>>,
	RuntimeApi::RuntimeApi: AuraRuntimeApi<Block, AuraId>,
	AuraId: AuraIdT + Sync,
{
	/// Creates the inherent data providers for manual and instant seal consensus.
	///
	/// This function sets up the timestamp and teyrchain validation data providers
	/// required for dev seal block production in a teyrchain environment.
	fn create_dev_node_inherent_data_providers(
		client: Arc<TeyrchainClient<Block, RuntimeApi>>,
		para_id: ParaId,
		slot_duration: pezsp_consensus_aura::SlotDuration,
	) -> impl Fn(
		Hash,
		(),
	) -> future::Ready<
		Result<
			(pezsp_timestamp::InherentDataProvider, MockValidationDataInherentDataProvider<()>),
			Box<dyn std::error::Error + Send + Sync>,
		>,
	> + Send
	       + Sync {
		move |block: Hash, ()| {
			let current_para_head = client
				.header(block)
				.expect("Header lookup should succeed")
				.expect("Header passed in as parent should be present in backend.");

			let should_send_go_ahead = client
				.runtime_api()
				.collect_collation_info(block, &current_para_head)
				.map(|info| info.new_validation_code.is_some())
				.unwrap_or_default();

			let current_para_block_head =
				Some(pezkuwi_primitives::HeadData(current_para_head.encode()));
			let current_block_number =
				UniqueSaturatedInto::<u32>::unique_saturated_into(*current_para_head.number()) + 1;
			log::info!("Current block number: {current_block_number}");

			let mocked_teyrchain = MockValidationDataInherentDataProvider::<()> {
				current_para_block: current_block_number,
				para_id,
				current_para_block_head,
				relay_blocks_per_para_block: 1,
				para_blocks_per_relay_epoch: 10,
				upgrade_go_ahead: should_send_go_ahead.then(|| {
					log::info!("Detected pending validation code, sending go-ahead signal.");
					UpgradeGoAhead::GoAhead
				}),
				..Default::default()
			};

			let timestamp_provider = pezsp_timestamp::InherentDataProvider::new(
				(slot_duration.as_millis() * current_block_number as u64).into(),
			);

			futures::future::ready(Ok((timestamp_provider, mocked_teyrchain)))
		}
	}
}

pub fn new_aura_node_spec<Block, RuntimeApi, AuraId>(
	extra_args: &NodeExtraArgs,
) -> Box<dyn DynNodeSpec>
where
	Block: NodeBlock,
	RuntimeApi: ConstructNodeRuntimeApi<Block, TeyrchainClient<Block, RuntimeApi>>,
	RuntimeApi::RuntimeApi: AuraRuntimeApi<Block, AuraId>
		+ pezpallet_transaction_payment_rpc::TransactionPaymentRuntimeApi<Block, Balance>
		+ bizinikiwi_frame_rpc_system::AccountNonceApi<Block, AccountId, Nonce>
		+ GetTeyrchainInfo<Block>,
	AuraId: AuraIdT + Sync + Send,
	<AuraId as AppCrypto>::Pair: Send + Sync,
{
	if extra_args.authoring_policy == AuthoringPolicy::SlotBased {
		Box::new(AuraNode::<
			Block,
			RuntimeApi,
			AuraId,
			StartSlotBasedAuraConsensus<Block, RuntimeApi, AuraId>,
			StartSlotBasedAuraConsensus<Block, RuntimeApi, AuraId>,
		>::default())
	} else {
		Box::new(AuraNode::<
			Block,
			RuntimeApi,
			AuraId,
			StartLookaheadAuraConsensus<Block, RuntimeApi, AuraId>,
			ClientBlockImport,
		>::default())
	}
}

/// Start consensus using the lookahead aura collator.
pub(crate) struct StartSlotBasedAuraConsensus<Block, RuntimeApi, AuraId>(
	PhantomData<(Block, RuntimeApi, AuraId)>,
);

impl<Block: BlockT<Hash = DbHash>, RuntimeApi, AuraId>
	StartSlotBasedAuraConsensus<Block, RuntimeApi, AuraId>
where
	RuntimeApi: ConstructNodeRuntimeApi<Block, TeyrchainClient<Block, RuntimeApi>>,
	RuntimeApi::RuntimeApi: AuraRuntimeApi<Block, AuraId>,
	AuraId: AuraIdT + Sync + Send,
	<AuraId as AppCrypto>::Pair: Send + Sync,
{
	#[docify::export_content]
	fn launch_slot_based_collator<CIDP, CHP, Proposer, CS, Spawner>(
		params_with_export: SlotBasedParams<
			Block,
			TeyrchainBlockImport<
				Block,
				SlotBasedBlockImport<
					Block,
					Arc<TeyrchainClient<Block, RuntimeApi>>,
					TeyrchainClient<Block, RuntimeApi>,
				>,
			>,
			CIDP,
			TeyrchainClient<Block, RuntimeApi>,
			TeyrchainBackend<Block>,
			Arc<dyn RelayChainInterface>,
			CHP,
			Proposer,
			CS,
			Spawner,
		>,
	) where
		CIDP: CreateInherentDataProviders<Block, ()> + 'static,
		CIDP::InherentDataProviders: Send,
		CHP: pezcumulus_client_consensus_common::ValidationCodeHashProvider<Hash> + Send + 'static,
		Proposer: ProposerInterface<Block> + Send + Sync + 'static,
		CS: CollatorServiceInterface<Block> + Send + Sync + Clone + 'static,
		Spawner: SpawnEssentialNamed + Clone + 'static,
	{
		slot_based::run::<Block, <AuraId as AppCrypto>::Pair, _, _, _, _, _, _, _, _, _>(
			params_with_export,
		);
	}
}

impl<Block: BlockT<Hash = DbHash>, RuntimeApi, AuraId>
	StartConsensus<
		Block,
		RuntimeApi,
		SlotBasedBlockImport<
			Block,
			Arc<TeyrchainClient<Block, RuntimeApi>>,
			TeyrchainClient<Block, RuntimeApi>,
		>,
		SlotBasedBlockImportHandle<Block>,
	> for StartSlotBasedAuraConsensus<Block, RuntimeApi, AuraId>
where
	RuntimeApi: ConstructNodeRuntimeApi<Block, TeyrchainClient<Block, RuntimeApi>>,
	RuntimeApi::RuntimeApi: AuraRuntimeApi<Block, AuraId>,
	AuraId: AuraIdT + Sync + Send,
	<AuraId as AppCrypto>::Pair: Send + Sync,
{
	fn start_consensus(
		client: Arc<TeyrchainClient<Block, RuntimeApi>>,
		block_import: TeyrchainBlockImport<
			Block,
			SlotBasedBlockImport<
				Block,
				Arc<TeyrchainClient<Block, RuntimeApi>>,
				TeyrchainClient<Block, RuntimeApi>,
			>,
		>,
		prometheus_registry: Option<&Registry>,
		telemetry: Option<TelemetryHandle>,
		task_manager: &TaskManager,
		relay_chain_interface: Arc<dyn RelayChainInterface>,
		transaction_pool: Arc<TransactionPoolHandle<Block, TeyrchainClient<Block, RuntimeApi>>>,
		keystore: KeystorePtr,
		relay_chain_slot_duration: Duration,
		para_id: ParaId,
		collator_key: CollatorPair,
		collator_peer_id: PeerId,
		_overseer_handle: OverseerHandle,
		announce_block: Arc<dyn Fn(Hash, Option<Vec<u8>>) + Send + Sync>,
		backend: Arc<TeyrchainBackend<Block>>,
		node_extra_args: NodeExtraArgs,
		block_import_handle: SlotBasedBlockImportHandle<Block>,
	) -> Result<(), Error> {
		let proposer = pezsc_basic_authorship::ProposerFactory::with_proof_recording(
			task_manager.spawn_handle(),
			client.clone(),
			transaction_pool,
			prometheus_registry,
			telemetry.clone(),
		);

		let collator_service = CollatorService::new(
			client.clone(),
			Arc::new(task_manager.spawn_handle()),
			announce_block,
			client.clone(),
		);

		let client_for_aura = client.clone();
		let params = SlotBasedParams {
			create_inherent_data_providers: move |_, ()| async move { Ok(()) },
			block_import,
			para_client: client.clone(),
			para_backend: backend.clone(),
			relay_client: relay_chain_interface,
			relay_chain_slot_duration,
			code_hash_provider: move |block_hash| {
				client_for_aura.code_at(block_hash).ok().map(|c| ValidationCode::from(c).hash())
			},
			keystore,
			collator_key,
			collator_peer_id,
			para_id,
			proposer,
			collator_service,
			authoring_duration: Duration::from_millis(2000),
			reinitialize: false,
			slot_offset: Duration::from_secs(1),
			block_import_handle,
			spawner: task_manager.spawn_essential_handle(),
			export_pov: node_extra_args.export_pov,
			max_pov_percentage: node_extra_args.max_pov_percentage,
		};

		// We have a separate function only to be able to use `docify::export` on this piece of
		// code.

		Self::launch_slot_based_collator(params);

		Ok(())
	}
}

impl<Block: BlockT<Hash = DbHash>, RuntimeApi, AuraId> InitBlockImport<Block, RuntimeApi>
	for StartSlotBasedAuraConsensus<Block, RuntimeApi, AuraId>
where
	RuntimeApi: ConstructNodeRuntimeApi<Block, TeyrchainClient<Block, RuntimeApi>>,
	RuntimeApi::RuntimeApi: AuraRuntimeApi<Block, AuraId>,
	AuraId: AuraIdT + Sync,
{
	type BlockImport = SlotBasedBlockImport<
		Block,
		Arc<TeyrchainClient<Block, RuntimeApi>>,
		TeyrchainClient<Block, RuntimeApi>,
	>;
	type BlockImportAuxiliaryData = SlotBasedBlockImportHandle<Block>;

	fn init_block_import(
		client: Arc<TeyrchainClient<Block, RuntimeApi>>,
	) -> pezsc_service::error::Result<(Self::BlockImport, Self::BlockImportAuxiliaryData)> {
		Ok(SlotBasedBlockImport::new(client.clone(), client))
	}
}

/// Wait for the Aura runtime API to appear on chain.
/// This is useful for chains that started out without Aura. Components that
/// are depending on Aura functionality will wait until Aura appears in the runtime.
async fn wait_for_aura<Block: BlockT, RuntimeApi, AuraId>(
	client: Arc<TeyrchainClient<Block, RuntimeApi>>,
) where
	RuntimeApi: ConstructNodeRuntimeApi<Block, TeyrchainClient<Block, RuntimeApi>>,
	RuntimeApi::RuntimeApi: AuraRuntimeApi<Block, AuraId>,
	AuraId: AuraIdT + Sync,
{
	let finalized_hash = client.chain_info().finalized_hash;
	if client.runtime_api().has_aura_api(finalized_hash) {
		return;
	};

	let mut stream = client.finality_notification_stream();
	while let Some(notification) = stream.next().await {
		if client.runtime_api().has_aura_api(notification.hash) {
			return;
		}
	}
}

/// Start consensus using the lookahead aura collator.
pub(crate) struct StartLookaheadAuraConsensus<Block, RuntimeApi, AuraId>(
	PhantomData<(Block, RuntimeApi, AuraId)>,
);

impl<Block: BlockT<Hash = DbHash>, RuntimeApi, AuraId>
	StartConsensus<Block, RuntimeApi, Arc<TeyrchainClient<Block, RuntimeApi>>, ()>
	for StartLookaheadAuraConsensus<Block, RuntimeApi, AuraId>
where
	RuntimeApi: ConstructNodeRuntimeApi<Block, TeyrchainClient<Block, RuntimeApi>>,
	RuntimeApi::RuntimeApi: AuraRuntimeApi<Block, AuraId>,
	AuraId: AuraIdT + Sync + Send,
	<AuraId as AppCrypto>::Pair: Send + Sync,
{
	fn start_consensus(
		client: Arc<TeyrchainClient<Block, RuntimeApi>>,
		block_import: TeyrchainBlockImport<Block, Arc<TeyrchainClient<Block, RuntimeApi>>>,
		prometheus_registry: Option<&Registry>,
		telemetry: Option<TelemetryHandle>,
		task_manager: &TaskManager,
		relay_chain_interface: Arc<dyn RelayChainInterface>,
		transaction_pool: Arc<TransactionPoolHandle<Block, TeyrchainClient<Block, RuntimeApi>>>,
		keystore: KeystorePtr,
		relay_chain_slot_duration: Duration,
		para_id: ParaId,
		collator_key: CollatorPair,
		collator_peer_id: PeerId,
		overseer_handle: OverseerHandle,
		announce_block: Arc<dyn Fn(Hash, Option<Vec<u8>>) + Send + Sync>,
		backend: Arc<TeyrchainBackend<Block>>,
		node_extra_args: NodeExtraArgs,
		_: (),
	) -> Result<(), Error> {
		let proposer = pezsc_basic_authorship::ProposerFactory::with_proof_recording(
			task_manager.spawn_handle(),
			client.clone(),
			transaction_pool,
			prometheus_registry,
			telemetry.clone(),
		);
		let collator_service = CollatorService::new(
			client.clone(),
			Arc::new(task_manager.spawn_handle()),
			announce_block,
			client.clone(),
		);

		let params = aura::ParamsWithExport {
			export_pov: node_extra_args.export_pov,
			params: AuraParams {
				create_inherent_data_providers: move |_, ()| async move { Ok(()) },
				block_import,
				para_client: client.clone(),
				para_backend: backend,
				relay_client: relay_chain_interface,
				code_hash_provider: {
					let client = client.clone();
					move |block_hash| {
						client.code_at(block_hash).ok().map(|c| ValidationCode::from(c).hash())
					}
				},
				keystore,
				collator_key,
				collator_peer_id,
				para_id,
				overseer_handle,
				relay_chain_slot_duration,
				proposer,
				collator_service,
				authoring_duration: Duration::from_millis(2000),
				reinitialize: false,
				max_pov_percentage: node_extra_args.max_pov_percentage,
			},
		};

		let fut = async move {
			wait_for_aura(client).await;
			aura::run_with_export::<Block, <AuraId as AppCrypto>::Pair, _, _, _, _, _, _, _, _>(
				params,
			)
			.await;
		};
		task_manager.spawn_essential_handle().spawn("aura", None, fut);

		Ok(())
	}
}