fn log_error<Blockhash: std::fmt::Debug>(

        &self,

        slot: Slot,

        para_id: ParaId,

        relay_parent: Blockhash,

    ) {

        match self {

                tracing::trace!(

                    "Para is not a parathread, skipping an attempt to buy core",

                tracing::trace!(

                    "Unable to claim slot for parathread, skipping attempt to buy the core.",

                tracing::error!(

                    "Unable to generate buy core proof as unable to find corresponding key",

                tracing::error!(

                    "Unable to calculate container chain slot drift from orchestrator chain's slot drift",

            BuyCoreError::ApiError(api_error) => {

                tracing::error!(

                    "Unable to call orchestrator runtime api",

            BuyCoreError::BuyingValidationError(buying_error) => {

                tracing::error!(

                    "Buying core is not allowed right now",

            BuyCoreError::UnableToCreateProof(proof_error) => {

                tracing::error!(

                    "Unable to generate buy core proof due to an error",

            BuyCoreError::TxSubmissionError(pool_error) => {

                tracing::error!(

                    "Unable to send buy core unsigned extrinsic through orchestrator tx pool",

    }

    fn from(buying_error: BuyingError<BlockNumber>) -> Self {

        BuyCoreError::BuyingValidationError(buying_error)

    }

    fn from(api_error: ApiError) -> Self {

        BuyCoreError::ApiError(api_error)

    }

    fn from(proof_creation_error: BuyCollatorProofCreationError) -> Self {

        BuyCoreError::UnableToCreateProof(proof_creation_error)

    }

pub async fn try_to_buy_core<Block, OBlock, OBlockNumber, P, CIDP, TxPool, OClient>(

    para_id: ParaId,

    aux_data: OrchestratorAuraWorkerAuxData<P>,

    inherent_providers: CIDP::InherentDataProviders,

    keystore: &KeystorePtr,

    orchestrator_client: Arc<OClient>,

    orchestrator_tx_pool: Arc<TxPool>,

    parent_header: <Block as BlockT>::Header,

    orchestrator_slot_duration: SlotDuration,

    container_slot_duration: SlotDuration,

) -> Result<

    <TxPool as TransactionPool>::Hash,

    BuyCoreError<

        <<OBlock as BlockT>::Header as HeaderT>::Number,

        <TxPool as TransactionPool>::Error,

    >,

>

where

    Block: BlockT,

    OBlock: BlockT,

    P: Pair<Public = NimbusId> + Sync + Send + 'static,

    P::Signature: TryFrom<Vec<u8>> + Member + Codec,

    CIDP: CreateInherentDataProviders<Block, (PHash, PersistedValidationData)>

        + Send

        + 'static

        + Clone,

    CIDP::InherentDataProviders: Send + InherentDataProviderExt,

    OClient: ProvideRuntimeApi<OBlock>

        + HeaderMetadata<OBlock, Error = sp_blockchain::Error>

        + HeaderBackend<OBlock>

        + BlockBackend<OBlock>

        + BlockIdTo<OBlock>

        + 'static,

    OClient::Api: TaggedTransactionQueue<OBlock>

        + XCMCoreBuyerApi<OBlock, <<OBlock as BlockT>::Header as HeaderT>::Number, ParaId, NimbusId>,

    TxPool: TransactionPool<Block = OBlock>,

{

    // We do nothing if this is not a parathread

    if aux_data.slot_freq.is_none() {

        return Err(BuyCoreError::NotAParathread);

    }

    let orchestrator_best_hash = orchestrator_client.info().best_hash;

    let orchestrator_runtime_api = orchestrator_client.runtime_api();

    let buy_core_slot_drift = orchestrator_client

        .runtime_api()

        .get_buy_core_slot_drift(orchestrator_best_hash)?;

    let buy_core_container_slot_drift = buy_core_slot_drift

        .checked_mul(orchestrator_slot_duration.as_millis())

        .and_then(|intermediate_result| {

            intermediate_result.checked_div(container_slot_duration.as_millis())

        })

        .ok_or(BuyCoreError::SlotDriftConversionOverflow)?;

    let current_container_slot = inherent_providers.slot();

    let slot_claim = tanssi_claim_slot::<P, Block>(

        aux_data,

        &parent_header,

        current_container_slot,

        ClaimMode::ParathreadCoreBuying {

            drift_permitted: buy_core_container_slot_drift.into(),

        },

        keystore,

    )

    .ok_or(BuyCoreError::UnableToClaimSlot)?;

    let pubkey = slot_claim.author_pub;

    orchestrator_runtime_api.is_core_buying_allowed(

        orchestrator_best_hash,

        para_id,

        pubkey.clone(),

    )??;

    let nonce =

        orchestrator_runtime_api.get_buy_core_signature_nonce(orchestrator_best_hash, para_id)?;

    let collator_buy_core_proof =

        BuyCoreCollatorProof::new_with_keystore(nonce, para_id, pubkey, keystore)?

            .ok_or(BuyCoreError::UnableToFindKeyForSigning)?;

    let extrinsic = orchestrator_runtime_api.create_buy_core_unsigned_extrinsic(

        orchestrator_best_hash,

        para_id,

        collator_buy_core_proof,

    )?;

    orchestrator_tx_pool

        .submit_one(orchestrator_best_hash, TransactionSource::Local, *extrinsic)

        .await

        .map_err(BuyCoreError::TxSubmissionError)

}

    fn clone(&self) -> Self {

        match self {

                orchestrator_tx_pool,

                orchestrator_client,

            } => Self::Orchestrator {

                orchestrator_tx_pool: orchestrator_tx_pool.clone(),

                orchestrator_client: orchestrator_client.clone(),

            },

            Self::Solochain {} => Self::Solochain {},

    }

pub fn run<

    GSD,

    Block,

    P,

    BI,

    CIDP,

    Client,

    Backend,

    RClient,

    CHP,

    SO,

    Proposer,

    CS,

    GOH,

    TxPool,

    OClient,

    OBlock,

>(

    mut params: Params<

        GSD,

        BI,

        CIDP,

        Client,

        Backend,

        RClient,

        CHP,

        SO,

        Proposer,

        CS,

        GOH,

        TxPool,

        OClient,

    >,

) -> (

    impl Future<Output = ()> + Send + 'static,

    oneshot::Receiver<()>,

)

where

    Block: BlockT,

    Client: ProvideRuntimeApi<Block>

        + BlockOf

        + AuxStore

        + HeaderBackend<Block>

        + BlockBackend<Block>

        + Send

        + Sync

        + 'static,

    Client::Api: UnincludedSegmentApi<Block>,

    Backend: sc_client_api::Backend<Block> + 'static,

    RClient: RelayChainInterface + Clone + 'static,

    CIDP: CreateInherentDataProviders<Block, (PHash, PersistedValidationData)>

        + Send

        + 'static

        + Clone,

    CIDP::InherentDataProviders: Send + InherentDataProviderExt,

    BI: BlockImport<Block> + Send + Sync + 'static,

    SO: SyncOracle + Send + Sync + Clone + 'static,

    Proposer: ProposerInterface<Block> + Send + Sync + 'static,

    CS: CollatorServiceInterface<Block> + Send + Sync + 'static,

    CHP: consensus_common::ValidationCodeHashProvider<Block::Hash> + Send + 'static,

    P: Pair<Public = NimbusId> + Sync + Send + 'static,

    P::Signature: TryFrom<Vec<u8>> + Member + Codec,

    GOH: RetrieveAuthoritiesFromOrchestrator<

            Block,

            (PHash, PersistedValidationData),

            OrchestratorAuraWorkerAuxData<P>,

        >

        + 'static

        + Sync

        + Send,

    OBlock: BlockT,

    OClient: ProvideRuntimeApi<OBlock>

        + HeaderMetadata<OBlock, Error = sp_blockchain::Error>

        + HeaderBackend<OBlock>

        + BlockBackend<OBlock>

        + BlockIdTo<OBlock>

        + 'static,

    OClient::Api: TaggedTransactionQueue<OBlock>

        + XCMCoreBuyerApi<OBlock, <<OBlock as BlockT>::Header as HeaderT>::Number, ParaId, NimbusId>,

    TxPool: TransactionPool<Block = OBlock> + 'static,

    GSD: Fn(<Block as BlockT>::Hash) -> SlotDuration + Send + 'static,

{

    let (exit_notification_sender, exit_notification_receiver) = oneshot::channel();

    let aura_fut = async move {

        cumulus_client_collator::initialize_collator_subsystems(

            &mut params.overseer_handle,

            params.collator_key,

            params.para_id,

            true,

        )

        .await;

        let mut import_notifications = match params.relay_client.import_notification_stream().await

            Ok(s) => s,

            Err(err) => {

                tracing::error!(

                    "Failed to initialize consensus: no relay chain import notification stream"

                return;

        let mut collator = {

            let params = collator_util::Params {

                create_inherent_data_providers: params.create_inherent_data_providers.clone(),

                block_import: params.block_import,

                relay_client: params.relay_client.clone(),

                keystore: params.keystore.clone(),

                para_id: params.para_id,

                proposer: params.proposer,

                collator_service: params.collator_service,

            };

            collator_util::Collator::<Block, P, _, _, _, _, _>::new(params)

            select! {

                maybe_relay_parent_header = import_notifications.next() => {

                    if maybe_relay_parent_header.is_none() {

                        break;

                    }

                    let relay_parent_header = maybe_relay_parent_header.expect("relay_parent_header must exists as we checked for None variant above; qed");

                    let relay_parent = relay_parent_header.hash();

                    let max_pov_size = match params

                        .relay_client

                        .persisted_validation_data(

                            relay_parent,

                            params.para_id,

                            OccupiedCoreAssumption::Included,

                        )

                        .await

                        Ok(None) => continue,

                        Ok(Some(pvd)) => pvd.max_pov_size,

                        Err(err) => {

                            tracing::error!(target: crate::LOG_TARGET, ?err, "Failed to gather information from relay-client");

                            continue;

                    let parent_search_params = ParentSearchParams {

                        relay_parent,

                        para_id: params.para_id,

                        ancestry_lookback: max_ancestry_lookback(relay_parent, &params.relay_client).await,

                    let potential_parents =

                        cumulus_client_consensus_common::find_potential_parents::<Block>(

                            parent_search_params,

                            &*params.para_backend,

                            &params.relay_client,

                        )

                        .await;

                    let mut potential_parents = match potential_parents {

                        Err(e) => {

                            tracing::error!(

                                "Could not fetch potential parents to build upon"

                            continue;

                        Ok(x) => x,

                    let included_block = match potential_parents.iter().find(|x| x.depth == 0) {

                        None => continue, // also serves as an `is_empty` check.

                        Some(b) => b.hash,

                    };

                    let para_client = &*params.para_client;

                    let keystore = &params.keystore;

                    let can_build_upon = |slot_now, block_hash, aux_data| {

                        can_build_upon::<_, _, P>(

                            slot_now,

                            aux_data,

                            block_hash,

                            included_block,

                            params.force_authoring,

                            para_client,

                            keystore,

                        )

                    };

                    potential_parents.sort_by_key(|a| a.depth);

                    let initial_parent = match potential_parents.pop() {

                        None => continue,

                        Some(p) => p,

                    };

                    // Build in a loop until not allowed. Note that the authorities can change

                    // at any block, so we need to re-claim our slot every time.

                    let mut parent_hash = initial_parent.hash;

                    let mut parent_header = initial_parent.header;

                    let core_indices = cores_scheduled_for_para(

                        relay_parent,

                        params.para_id,

                        &mut params.overseer_handle,

                        &params.relay_client,

                    )

                    .await;

                    let overseer_handle = &mut params.overseer_handle;

                    for n_built in 0..2 {

                        let validation_data = PersistedValidationData {

                            parent_head: parent_header.encode().into(),

                            relay_parent_number: *relay_parent_header.number(),

                            relay_parent_storage_root: *relay_parent_header.state_root(),

                            max_pov_size,

                        };

                        let aux_data = match params

                            .get_orchestrator_aux_data

                            .retrieve_authorities_from_orchestrator(

                                parent_hash,

                                (relay_parent_header.hash(), validation_data.clone()),

                            )

                            .await

                            Err(e) => {

                                tracing::error!(target: crate::LOG_TARGET, ?e);

                                break;

                            Ok(h) => h,

                        let inherent_providers = match params

                            .create_inherent_data_providers

                            .create_inherent_data_providers(

                                parent_hash,

                                (relay_parent_header.hash(), validation_data.clone()),

                            )

                            .await

                            Err(e) => {

                                tracing::error!(target: crate::LOG_TARGET, ?e);

                                break;

                            Ok(h) => h,

                        let (is_parachain, core_index) = match (&aux_data.slot_freq, core_indices.first()) {

                                tracing::warn!(target: crate::LOG_TARGET, para_id = ?params.para_id, "We are parachain and we do not have core allocated, nothing to do");

                                break;

                            (None, Some(core_index)) => {

                                tracing::trace!(target: crate::LOG_TARGET, para_id = ?params.para_id, ?core_index, "We are parachain and we core allocated, let's collate the block");

                                (true, core_index)

                            (Some(_slot_frequency), None) => { // We are parathread and core is not allocated. Let's try to buy core

                                tracing::trace!(target: crate::LOG_TARGET, para_id = ?params.para_id, "We are parathread and we do not have core allocated, let's try to buy the core");

                                let slot = inherent_providers.slot();

                                let container_chain_slot_duration = (params.get_current_slot_duration)(parent_header.hash());

                                let buy_core_result = match &params.buy_core_params {

                                        orchestrator_client,

                                        orchestrator_tx_pool,

                                    } => {

                                        try_to_buy_core::<_, _, <<OBlock as BlockT>::Header as HeaderT>::Number, _, CIDP, _, _>(params.para_id, aux_data, inherent_providers, &params.keystore, orchestrator_client.clone(), orchestrator_tx_pool.clone(), parent_header, params.orchestrator_slot_duration, container_chain_slot_duration).await

                                        log::warn!("Unimplemented: cannot buy core for parathread in solochain");

                                        break;

                                match buy_core_result {

                                    Ok(block_hash) => {

                                        tracing::trace!(target: crate::LOG_TARGET, ?block_hash, "Sent unsigned extrinsic to buy the core");

                                    Err(buy_core_error) => {

                                        buy_core_error.log_error(slot, params.para_id, relay_parent);

                                    }

                                break; // No point in continuing as we need to wait for few relay blocks in order for our core to be available.

                            (Some(_slot_frequency), Some(core_index)) => { // We are parathread and we do have core, let's continue

                                tracing::trace!(target: crate::LOG_TARGET, ?core_index, "We are parathread and we do have core allocated, let's collate the block");

                                (false, core_index)

                        let mut slot_claim = match can_build_upon(

                            inherent_providers.slot(),

                            parent_header.clone(),

                            aux_data,

                        )

                        .await

                            Ok(None) => break,

                            Err(e) => {

                                tracing::error!(target: crate::LOG_TARGET, ?e);

                                break;

                            Ok(Some(c)) => c,

                        };

                        tracing::debug!(

                            unincluded_segment_len = initial_parent.depth + n_built,

                            "Slot claimed. Building"

                        let (parachain_inherent_data, other_inherent_data) = match collator

                            .create_inherent_data(relay_parent, &validation_data, parent_hash, None)

                            .await

                            Err(err) => {

                                tracing::error!(target: crate::LOG_TARGET, ?err);

                                break;

                            Ok(x) => x,

                        let validation_code_hash = match params.code_hash_provider.code_hash_at(parent_hash)

                                tracing::error!(target: crate::LOG_TARGET, ?parent_hash, "Could not fetch validation code hash");

                                break;

                            Some(v) => v,

                        };

                        match collator

                            .collate(

                                &parent_header,

                                &mut slot_claim,

                                None,

                                (parachain_inherent_data, other_inherent_data),

                                params.authoring_duration,

                                // Set the block limit to 50% of the maximum PoV size.

                                //

                                // TODO: If we got benchmarking that includes the proof size,

                                // we should be able to use the maximum pov size.

                                (validation_data.max_pov_size / 2) as usize,

                            )

                            .await

                            Ok(Some((collation, block_data, new_block_hash))) => {

                                // Here we are assuming that the import logic protects against equivocations

                                // and provides sybil-resistance, as it should.

                                collator

                                    .collator_service()

                                    .announce_block(new_block_hash, None);

                                // Send a submit-collation message to the collation generation subsystem,

                                // which then distributes this to validators.

                                //

                                // Here we are assuming that the leaf is imported, as we've gotten an

                                // import notification.

                                overseer_handle

                                    .send_msg(

                                        CollationGenerationMessage::SubmitCollation(

                                            SubmitCollationParams {

                                                relay_parent,

                                                collation,

                                                parent_head: parent_header.encode().into(),

                                                validation_code_hash,

                                                result_sender: None,

                                                core_index: *core_index

                                            },

                                        ),

                                        "SubmitCollation",

                                    )

                                    .await;

                                parent_hash = new_block_hash;

                                parent_header = block_data.into_header();

                                tracing::debug!(target: crate::LOG_TARGET, "Lookahead collator: No block proposal");

                            Err(err) => {

                                tracing::error!(target: crate::LOG_TARGET, ?err);

                                break;

                        if !is_parachain {

                            tracing::trace!(target: crate::LOG_TARGET, "Not a parachain so terminated at {:?}", n_built);

                            break;

                        }

                _ = params.cancellation_token.cancelled() => {

                    log::info!("Stopping lookahead collator");

                    break;

        let _ = exit_notification_sender.send(());

    };

    (aura_fut, exit_notification_receiver)

}

async fn can_build_upon<Block: BlockT, Client, P>(

    slot: Slot,

    aux_data: OrchestratorAuraWorkerAuxData<P>,

    parent_header: Block::Header,

    included_block: <Block as BlockT>::Hash,

    force_authoring: bool,

    client: &Client,

    keystore: &KeystorePtr,

) -> Result<Option<SlotClaim<P::Public>>, Box<dyn Error>>

where

    Client: ProvideRuntimeApi<Block>,

    Client::Api: UnincludedSegmentApi<Block>,

    P: Pair + Send + Sync + 'static,

    P::Public: Codec + std::fmt::Debug,

    P::Signature: Codec,

{

    let runtime_api = client.runtime_api();

    let claim_mode = if force_authoring {

        ClaimMode::ForceAuthoring

        ClaimMode::NormalAuthoring

    let slot_claim =

        tanssi_claim_slot::<P, Block>(aux_data, &parent_header, slot, claim_mode, keystore);

    // Here we lean on the property that building on an empty unincluded segment must always

    // be legal. Skipping the runtime API query here allows us to seamlessly run this

    // collator against chains which have not yet upgraded their runtime.

    if parent_header.hash() != included_block

        && !runtime_api.can_build_upon(parent_header.hash(), included_block, slot)?

        return Ok(None);

    }

    Ok(slot_claim)

}

async fn max_ancestry_lookback(

    relay_parent: PHash,

    relay_client: &impl RelayChainInterface,

) -> usize {

    match load_abridged_host_configuration(relay_parent, relay_client).await {

        Ok(Some(config)) => config.async_backing_params.allowed_ancestry_len as usize,

            tracing::error!(

                "Active config is missing in relay chain storage",

            0

        Err(err) => {

            tracing::error!(

                "Failed to read active config from relay chain client",

            0

}

async fn cores_scheduled_for_para(

    relay_parent: PHash,

    para_id: ParaId,

    overseer_handle: &mut OverseerHandle,

    relay_client: &impl RelayChainInterface,

) -> Vec<CoreIndex> {

    let (tx, rx) = oneshot::channel();

    let request = RuntimeApiRequest::AvailabilityCores(tx);

    overseer_handle

        .send_msg(

            RuntimeApiMessage::Request(relay_parent, request),

            "LookaheadCollator",

        )

        .await;

    let max_candidate_depth = async_backing_params(relay_parent, relay_client)

        .await

        .map(|c| c.max_candidate_depth)

        .unwrap_or(0);

    let cores = match rx.await {

        Ok(Ok(cores)) => cores,

        Ok(Err(error)) => {

            tracing::error!(

                "Failed to query availability cores runtime API",

            return Vec::new();

            tracing::error!(

                "Sender for availability cores runtime request dropped",

            return Vec::new();

    cores

        .iter()

        .enumerate()

        .filter_map(|(index, core)| {

            let core_para_id = match core {

                CoreState::Scheduled(scheduled_core) => Some(scheduled_core.para_id),

                CoreState::Occupied(occupied_core) if max_candidate_depth >= 1 => occupied_core

                    .next_up_on_available

                    .as_ref()

                    .map(|scheduled_core| scheduled_core.para_id),

                CoreState::Free | CoreState::Occupied(_) => None,

            if core_para_id == Some(para_id) {

                Some(CoreIndex(index as u32))

                None

        })

        .collect()

}

async fn async_backing_params(

    relay_parent: PHash,

    relay_client: &impl RelayChainInterface,

) -> Option<AsyncBackingParams> {

    match load_abridged_host_configuration(relay_parent, relay_client).await {

        Ok(Some(config)) => Some(config.async_backing_params),

            tracing::error!(

                "Active config is missing in relay chain storage",

            None

        Err(err) => {

            tracing::error!(

                "Failed to read active config from relay chain client",

            None

}