consensus_core/

core.rs

// Copyright (c) Mysten Labs, Inc.
// SPDX-License-Identifier: Apache-2.0

use std::{
    collections::{BTreeMap, BTreeSet},
    sync::Arc,
    vec,
};

use consensus_config::ProtocolKeyPair;
#[cfg(test)]
use consensus_config::{AuthorityIndex, Stake, local_committee_and_keys};
use consensus_types::block::{BlockRef, Round};
use itertools::Itertools as _;
#[cfg(test)]
use mysten_metrics::monitored_mpsc::UnboundedReceiver;
use mysten_metrics::monitored_scope;
use parking_lot::RwLock;
use sui_macros::fail_point;
use tokio::sync::{broadcast, watch};
use tracing::{debug, info, trace, warn};

#[cfg(test)]
use crate::{
    CommitConsumerArgs, TransactionClient,
    block_verifier::NoopBlockVerifier,
    storage::{Store, WriteBatch, mem_store::MemStore},
};
use crate::{
    ancestor::AncestorStateManager,
    block::{BlockAPI, ExtendedBlock, GENESIS_ROUND, Slot, VerifiedBlock},
    block_manager::BlockManager,
    commit::{
        CertifiedCommit, CertifiedCommits, CommitAPI, CommittedSubDag, DecidedLeader, Decision,
    },
    commit_observer::CommitObserver,
    context::Context,
    dag_state::DagState,
    error::{ConsensusError, ConsensusResult},
    leader_schedule::LeaderSchedule,
    proposer::{Proposer, ValidatorProposer},
    round_tracker::RoundTracker,
    transaction::TransactionConsumer,
    transaction_vote_tracker::TransactionVoteTracker,
    universal_committer::{
        UniversalCommitter, universal_committer_builder::UniversalCommitterBuilder,
    },
};

pub(crate) struct Core {
    context: Arc<Context>,
    /// The block manager which is responsible for keeping track of the DAG dependencies when processing new blocks
    /// and accept them or suspend if we are missing their causal history
    block_manager: BlockManager,
    /// Used to make commit decisions for leader blocks in the dag.
    committer: Arc<UniversalCommitter>,
    /// The last new round for which core has sent out a signal.
    last_signaled_round: Round,
    /// The last decided leader returned from the universal committer. Important to note
    /// that this does not signify that the leader has been persisted yet as it still has
    /// to go through CommitObserver and persist the commit in store. On recovery/restart
    /// the last_decided_leader will be set to the last_commit leader in dag state.
    last_decided_leader: Slot,
    /// The consensus leader schedule to be used to resolve the leader for a
    /// given round.
    leader_schedule: Arc<LeaderSchedule>,
    /// The commit observer is responsible for observing the commits and collecting
    /// + sending subdags over the consensus output channel.
    commit_observer: CommitObserver,
    /// Sender of outgoing signals from Core.
    signals: CoreSignals,
    /// Keeping track of state of the DAG, including blocks, commits and last committed rounds.
    dag_state: Arc<RwLock<DagState>>,
    /// Block proposal engine for Validator nodes only.
    /// Validators have a proposer to create blocks, Observers have None (they only receive blocks).
    proposer: Option<Box<dyn Proposer>>,
}

impl Core {
    /// Creates a new Core instance for a validator node that participates in consensus.
    pub(crate) fn new_validator(
        context: Arc<Context>,
        leader_schedule: Arc<LeaderSchedule>,
        transaction_consumer: TransactionConsumer,
        transaction_vote_tracker: TransactionVoteTracker,
        block_manager: BlockManager,
        commit_observer: CommitObserver,
        signals: CoreSignals,
        block_signer: ProtocolKeyPair,
        dag_state: Arc<RwLock<DagState>>,
        sync_last_known_own_block: bool,
        round_tracker: Arc<RwLock<RoundTracker>>,
    ) -> Self {
        let last_decided_leader = dag_state.read().last_commit_leader();
        let number_of_leaders = context.protocol_config.num_leaders_per_round().unwrap_or(1);
        let committer = Arc::new(
            UniversalCommitterBuilder::new(
                context.clone(),
                leader_schedule.clone(),
                dag_state.clone(),
            )
            .with_number_of_leaders(number_of_leaders)
            .with_pipeline(true)
            .build(),
        );

        let last_proposed_block = dag_state
            .read()
            .get_last_proposed_block()
            .expect("A block should have been returned");
        let last_signaled_round = last_proposed_block.round();

        // Recover the last included ancestor rounds based on the last proposed block. That will allow
        // to perform the next block proposal by using ancestor blocks of higher rounds and avoid
        // re-including blocks that have been already included in the last (or earlier) block proposal.
        // This is only strongly guaranteed for a quorum of ancestors. It is still possible to re-include
        // a block from an authority which hadn't been added as part of the last proposal hence its
        // latest included ancestor is not accurately captured here. This is considered a small deficiency,
        // and it mostly matters just for this next proposal without any actual penalties in performance
        // or block proposal.
        let mut last_included_ancestors = vec![None; context.committee.size()];
        for ancestor in last_proposed_block.ancestors() {
            last_included_ancestors[ancestor.author] = Some(*ancestor);
        }

        let last_known_proposed_round = if sync_last_known_own_block {
            None
        } else {
            // if the sync is disabled then we practically don't want to impose any restriction.
            Some(0)
        };

        let propagation_scores = leader_schedule
            .leader_swap_table
            .read()
            .reputation_scores
            .clone();
        let mut ancestor_state_manager =
            AncestorStateManager::new(context.clone(), dag_state.clone());
        ancestor_state_manager.set_propagation_scores(propagation_scores);

        // Create the ValidatorProposer
        let proposer = Some(Box::new(ValidatorProposer::new(
            dag_state.clone(),
            context.clone(),
            transaction_consumer,
            transaction_vote_tracker.clone(),
            block_signer,
            last_known_proposed_round,
            ancestor_state_manager,
            round_tracker.clone(),
            committer.clone(),
        )) as Box<dyn Proposer>);

        Self {
            context,
            last_signaled_round,
            last_decided_leader,
            leader_schedule,
            block_manager,
            committer,
            commit_observer,
            signals,
            dag_state,
            proposer,
        }
        .recover_validator()
    }

    /// Creates a new Core instance for an observer node that only processes blocks.
    pub(crate) fn new_observer(
        context: Arc<Context>,
        leader_schedule: Arc<LeaderSchedule>,
        block_manager: BlockManager,
        commit_observer: CommitObserver,
        signals: CoreSignals,
        dag_state: Arc<RwLock<DagState>>,
    ) -> Self {
        let last_decided_leader = dag_state.read().last_commit_leader();
        let number_of_leaders = context.protocol_config.num_leaders_per_round().unwrap_or(1);
        let committer = Arc::new(
            UniversalCommitterBuilder::new(
                context.clone(),
                leader_schedule.clone(),
                dag_state.clone(),
            )
            .with_number_of_leaders(number_of_leaders)
            .with_pipeline(true)
            .build(),
        );

        // For the Observer nodes let's consider the last signaled round as the latest threshold clock round.
        let last_signaled_round = dag_state.read().threshold_clock_round();

        Self {
            context,
            last_signaled_round,
            last_decided_leader,
            leader_schedule,
            block_manager,
            committer,
            commit_observer,
            signals,
            dag_state,
            proposer: None,
        }
        .recover_observer()
    }

    fn recover_observer(mut self) -> Self {
        let _s = self
            .context
            .metrics
            .node_metrics
            .scope_processing_time
            .with_label_values(&["Core::recover_observer"])
            .start_timer();

        // Try to commit, since they may not have run after the last storage write.
        self.try_commit(vec![]).unwrap();

        self.try_signal_new_round();

        self
    }

    fn recover_validator(mut self) -> Self {
        let _s = self
            .context
            .metrics
            .node_metrics
            .scope_processing_time
            .with_label_values(&["Core::recover_validator"])
            .start_timer();

        // Try to commit and propose, since they may not have run after the last storage write.
        self.try_commit(vec![]).unwrap();

        let last_proposed_block = if let Some(last_proposed_block) = self.try_propose(true).unwrap()
        {
            last_proposed_block
        } else {
            let proposer = self
                .proposer
                .as_ref()
                .expect("Validator must have proposer");
            let last_proposed_block = proposer.last_proposed_block();

            if proposer.should_propose() {
                assert!(
                    last_proposed_block.round() > GENESIS_ROUND,
                    "At minimum a block of round higher than genesis should have been produced during recovery"
                );
            }

            // if no new block proposed then just re-broadcast the last proposed one to ensure liveness.
            self.signals
                .new_block(ExtendedBlock {
                    block: last_proposed_block.clone(),
                    excluded_ancestors: vec![],
                })
                .unwrap();
            last_proposed_block
        };

        // Try to set up leader timeout if needed.
        // This needs to be called after try_commit() and try_propose(), which may
        // have advanced the threshold clock round.
        self.try_signal_new_round();

        info!(
            "Core recovery for validator completed with last proposed block {:?}",
            last_proposed_block
        );

        self
    }

    /// Calls `BlockManager::try_accept_blocks` and broadcasts each accepted block to any active
    /// observer subscribers. Returns the accepted blocks alongside any missing block refs.
    fn accept_blocks(
        &mut self,
        blocks: Vec<VerifiedBlock>,
    ) -> (Vec<VerifiedBlock>, BTreeSet<BlockRef>) {
        let (accepted_blocks, missing_block_refs) = self.block_manager.try_accept_blocks(blocks);
        for block in &accepted_blocks {
            tracing::trace!(
                "{} Core accepted round {}, author {} at timestamp: {}",
                self.context.own_index,
                block.round(),
                block.author(),
                block.timestamp_ms()
            );
            self.signals.new_accepted_block(block.clone());
        }
        (accepted_blocks, missing_block_refs)
    }

    /// Calls `BlockManager::try_accept_committed_blocks` and broadcasts each accepted block to any
    /// active observer subscribers. Returns all accepted blocks.
    fn accept_committed_blocks(&mut self, blocks: Vec<VerifiedBlock>) -> Vec<VerifiedBlock> {
        let accepted_blocks = self.block_manager.try_accept_committed_blocks(blocks);
        for block in &accepted_blocks {
            self.signals.new_accepted_block(block.clone());
        }
        accepted_blocks
    }

    /// Processes the provided blocks and accepts them if possible when their causal history exists.
    /// The method returns:
    /// - The references of ancestors missing their block
    #[tracing::instrument(skip_all)]
    pub(crate) fn add_blocks(
        &mut self,
        blocks: Vec<VerifiedBlock>,
    ) -> ConsensusResult<BTreeSet<BlockRef>> {
        let _scope = monitored_scope("Core::add_blocks");
        let _s = self
            .context
            .metrics
            .node_metrics
            .scope_processing_time
            .with_label_values(&["Core::add_blocks"])
            .start_timer();
        self.context
            .metrics
            .node_metrics
            .core_add_blocks_batch_size
            .observe(blocks.len() as f64);

        let (accepted_blocks, missing_block_refs) = self.accept_blocks(blocks);

        if !accepted_blocks.is_empty() {
            trace!(
                "Accepted blocks: {}",
                accepted_blocks
                    .iter()
                    .map(|b| b.reference().to_string())
                    .join(",")
            );

            // Try to commit the new blocks if possible.
            self.try_commit(vec![])?;

            // Try to propose now since there are new blocks accepted.
            self.try_propose(false)?;

            // Now set up leader timeout if needed.
            // This needs to be called after try_commit() and try_propose(), which may
            // have advanced the threshold clock round.
            self.try_signal_new_round();
        };

        if !missing_block_refs.is_empty() {
            trace!(
                "Missing block refs: {}",
                missing_block_refs.iter().map(|b| b.to_string()).join(", ")
            );
        }

        Ok(missing_block_refs)
    }

    // Adds the certified commits that have been synced via the commit syncer. We are using the commit info in order to skip running the decision
    // rule and immediately commit the corresponding leaders and sub dags. Pay attention that no block acceptance is happening here, but rather
    // internally in the `try_commit` method which ensures that everytime only the blocks corresponding to the certified commits that are about to
    // be committed are accepted.
    #[tracing::instrument(skip_all)]
    pub(crate) fn add_certified_commits(
        &mut self,
        certified_commits: CertifiedCommits,
    ) -> ConsensusResult<BTreeSet<BlockRef>> {
        let _scope = monitored_scope("Core::add_certified_commits");

        let votes = certified_commits.votes().to_vec();
        let commits = self
            .filter_new_commits(certified_commits.commits().to_vec())
            .expect("Certified commits validation failed");

        // Try to accept the certified commit votes.
        // Even if they may not be part of a future commit, these blocks are useful for certifying
        // commits when helping peers sync commits.
        let (_, missing_block_refs) = self.accept_blocks(votes);

        // Try to commit the new blocks. Take into account the trusted commit that has been provided.
        self.try_commit(commits)?;

        // Try to propose now since there are new blocks accepted.
        self.try_propose(false)?;

        // Now set up leader timeout if needed.
        // This needs to be called after try_commit() and try_propose(), which may
        // have advanced the threshold clock round.
        self.try_signal_new_round();

        Ok(missing_block_refs)
    }

    /// Checks if provided block refs have been accepted. If not, missing block refs are kept for synchronizations.
    /// Returns the references of missing blocks among the input blocks.
    pub(crate) fn check_block_refs(
        &mut self,
        block_refs: Vec<BlockRef>,
    ) -> ConsensusResult<BTreeSet<BlockRef>> {
        let _scope = monitored_scope("Core::check_block_refs");
        let _s = self
            .context
            .metrics
            .node_metrics
            .scope_processing_time
            .with_label_values(&["Core::check_block_refs"])
            .start_timer();
        self.context
            .metrics
            .node_metrics
            .core_check_block_refs_batch_size
            .observe(block_refs.len() as f64);

        // Try to find them via the block manager
        let missing_block_refs = self.block_manager.try_find_blocks(block_refs);

        if !missing_block_refs.is_empty() {
            trace!(
                "Missing block refs: {}",
                missing_block_refs.iter().map(|b| b.to_string()).join(", ")
            );
        }
        Ok(missing_block_refs)
    }

    /// If needed, signals a new clock round and sets up leader timeout.
    fn try_signal_new_round(&mut self) {
        // Signal only when the threshold clock round is more advanced than the last signaled round.
        //
        // NOTE: a signal is still sent even when a block has been proposed at the new round.
        // We can consider changing this in the future.
        let new_clock_round = self.dag_state.read().threshold_clock_round();
        if new_clock_round <= self.last_signaled_round {
            return;
        }
        // Then send a signal to set up leader timeout.
        self.signals.new_round(new_clock_round);
        self.last_signaled_round = new_clock_round;

        // Report the threshold clock round
        self.context
            .metrics
            .node_metrics
            .threshold_clock_round
            .set(new_clock_round as i64);
    }

    /// Creating a new block for the dictated round. This is used when a leader timeout occurs, either
    /// when the min timeout expires or max. When `force = true` , then any checks like previous round
    /// leader existence will get skipped.
    pub(crate) fn new_block(
        &mut self,
        round: Round,
        force: bool,
    ) -> ConsensusResult<Option<VerifiedBlock>> {
        let _scope = monitored_scope("Core::new_block");
        if let Some(last_round) = self.last_proposed_round()
            && last_round < round
        {
            self.context
                .metrics
                .node_metrics
                .leader_timeout_total
                .with_label_values(&[&format!("{force}")])
                .inc();
            let result = self.try_propose(force);
            // The threshold clock round may have advanced, so a signal needs to be sent.
            self.try_signal_new_round();
            return result;
        }
        Ok(None)
    }

    /// Keeps only the certified commits that have a commit index > last commit index.
    /// It also ensures that the first commit in the list is the next one in line, otherwise it panics.
    fn filter_new_commits(
        &mut self,
        commits: Vec<CertifiedCommit>,
    ) -> ConsensusResult<Vec<CertifiedCommit>> {
        // Filter out the commits that have been already locally committed and keep only anything that is above the last committed index.
        let last_commit_index = self.dag_state.read().last_commit_index();
        let commits = commits
            .iter()
            .filter(|commit| {
                if commit.index() > last_commit_index {
                    true
                } else {
                    tracing::debug!(
                        "Skip commit for index {} as it is already committed with last commit index {}",
                        commit.index(),
                        last_commit_index
                    );
                    false
                }
            })
            .cloned()
            .collect::<Vec<_>>();

        // Make sure that the first commit we find is the next one in line and there is no gap.
        if let Some(commit) = commits.first()
            && commit.index() != last_commit_index + 1
        {
            return Err(ConsensusError::UnexpectedCertifiedCommitIndex {
                expected_commit_index: last_commit_index + 1,
                commit_index: commit.index(),
            });
        }

        Ok(commits)
    }

    // Attempts to create a new block, persist and propose it to all peers.
    // When force is true, ignore if leader from the last round exists among ancestors and if
    // the minimum round delay has passed.
    fn try_propose(&mut self, force: bool) -> ConsensusResult<Option<VerifiedBlock>> {
        if let Some(proposer) = &mut self.proposer
            && let Some(extended_block) = proposer.try_new_block(force)
        {
            self.signals.new_block(extended_block.clone())?;
            self.signals
                .new_accepted_block(extended_block.block.clone());

            fail_point!("consensus-after-propose");

            // The new block may help commit.
            self.try_commit(vec![])?;
            return Ok(Some(extended_block.block));
        }
        Ok(None)
    }

    /// Runs commit rule to attempt to commit additional blocks from the DAG. If any `certified_commits` are provided, then
    /// it will attempt to commit those first before trying to commit any further leaders.
    fn try_commit(
        &mut self,
        mut certified_commits: Vec<CertifiedCommit>,
    ) -> ConsensusResult<Vec<CommittedSubDag>> {
        let _s = self
            .context
            .metrics
            .node_metrics
            .scope_processing_time
            .with_label_values(&["Core::try_commit"])
            .start_timer();

        let mut certified_commits_map = BTreeMap::new();
        for c in &certified_commits {
            certified_commits_map.insert(c.index(), c.reference());
        }

        if !certified_commits.is_empty() {
            info!(
                "Processing synced commits: {:?}",
                certified_commits
                    .iter()
                    .map(|c| (c.index(), c.leader()))
                    .collect::<Vec<_>>()
            );
        }

        let mut committed_sub_dags = Vec::new();
        // TODO: Add optimization to abort early without quorum for a round.
        loop {
            // LeaderSchedule has a limit to how many sequenced leaders can be committed
            // before a change is triggered. Calling into leader schedule will get you
            // how many commits till next leader change. We will loop back and recalculate
            // any discarded leaders with the new schedule.
            let mut commits_until_update = self
                .leader_schedule
                .commits_until_leader_schedule_update(self.dag_state.clone());

            if commits_until_update == 0 {
                let last_commit_index = self.dag_state.read().last_commit_index();

                tracing::info!(
                    "Leader schedule change triggered at commit index {last_commit_index}"
                );

                self.leader_schedule
                    .update_leader_schedule_v2(&self.dag_state);

                let propagation_scores = self
                    .leader_schedule
                    .leader_swap_table
                    .read()
                    .reputation_scores
                    .clone();
                if let Some(proposer) = &mut self.proposer {
                    proposer.set_propagation_scores(propagation_scores);
                }

                commits_until_update = self
                    .leader_schedule
                    .commits_until_leader_schedule_update(self.dag_state.clone());

                fail_point!("consensus-after-leader-schedule-change");
            }
            assert!(commits_until_update > 0);

            // If there are certified commits to process, find out which leaders and commits from them
            // are decided and use them as the next commits.
            let (certified_leaders, decided_certified_commits): (
                Vec<DecidedLeader>,
                Vec<CertifiedCommit>,
            ) = self
                .try_select_certified_leaders(&mut certified_commits, commits_until_update)
                .into_iter()
                .unzip();

            // Only accept blocks for the certified commits that we are certain to sequence.
            // This ensures that only blocks corresponding to committed certified commits are flushed to disk.
            // Blocks from non-committed certified commits will not be flushed, preventing issues during crash-recovery.
            // This avoids scenarios where accepting and flushing blocks of non-committed certified commits could lead to
            // premature commit rule execution. Due to GC, this could cause a panic if the commit rule tries to access
            // missing causal history from blocks of certified commits.
            let blocks = decided_certified_commits
                .iter()
                .flat_map(|c| c.blocks())
                .cloned()
                .collect::<Vec<_>>();
            self.accept_committed_blocks(blocks);

            // If there is no certified commit to process, run the decision rule.
            let (decided_leaders, local) = if certified_leaders.is_empty() {
                // TODO: limit commits by commits_until_update for efficiency, which may be needed when leader schedule length is reduced.
                let mut decided_leaders = self.committer.try_decide(self.last_decided_leader);
                // Truncate the decided leaders to fit the commit schedule limit.
                if decided_leaders.len() >= commits_until_update {
                    let _ = decided_leaders.split_off(commits_until_update);
                }
                (decided_leaders, true)
            } else {
                (certified_leaders, false)
            };

            // If the decided leaders list is empty then just break the loop.
            let Some(last_decided) = decided_leaders.last().cloned() else {
                break;
            };

            self.last_decided_leader = last_decided.slot();
            self.context
                .metrics
                .node_metrics
                .last_decided_leader_round
                .set(self.last_decided_leader.round as i64);

            let sequenced_leaders = decided_leaders
                .into_iter()
                .filter_map(|leader| leader.into_committed_block())
                .collect::<Vec<_>>();
            // It's possible to reach this point as the decided leaders might all of them be "Skip" decisions. In this case there is no
            // leader to commit and we should break the loop.
            if sequenced_leaders.is_empty() {
                break;
            }
            tracing::info!(
                "Committing {} leaders: {}; {} commits before next leader schedule change",
                sequenced_leaders.len(),
                sequenced_leaders
                    .iter()
                    .map(|b| b.reference().to_string())
                    .join(","),
                commits_until_update,
            );

            // TODO: refcount subdags
            let subdags = self
                .commit_observer
                .handle_commit(sequenced_leaders, local)?;

            // Try to unsuspend blocks if gc_round has advanced.
            self.block_manager
                .try_unsuspend_blocks_for_latest_gc_round();

            committed_sub_dags.extend(subdags);

            fail_point!("consensus-after-handle-commit");
        }

        // Sanity check: for commits that have been linearized using the certified commits, ensure that the same sub dag has been committed.
        for sub_dag in &committed_sub_dags {
            if let Some(commit_ref) = certified_commits_map.remove(&sub_dag.commit_ref.index) {
                assert_eq!(
                    commit_ref, sub_dag.commit_ref,
                    "Certified commit has different reference than the committed sub dag"
                );
            }
        }

        // Notify about our own committed blocks
        let committed_block_refs = committed_sub_dags
            .iter()
            .flat_map(|sub_dag| sub_dag.blocks.iter())
            .filter_map(|block| {
                (block.author() == self.context.own_index).then_some(block.reference())
            })
            .collect::<Vec<_>>();
        if let Some(proposer) = &self.proposer {
            proposer.notify_own_blocks_committed(
                committed_block_refs,
                self.dag_state.read().gc_round(),
            );
        }

        Ok(committed_sub_dags)
    }

    pub(crate) fn get_missing_blocks(&self) -> BTreeSet<BlockRef> {
        let _scope = monitored_scope("Core::get_missing_blocks");
        self.block_manager.missing_blocks()
    }

    /// Sets the delay by round for propagating blocks to a quorum.
    pub(crate) fn set_propagation_delay(&mut self, delay: Round) {
        info!("Propagation round delay set to: {delay}");
        if let Some(proposer) = &mut self.proposer {
            proposer.set_propagation_delay(delay);
        }
    }

    /// Sets the min propose round for the proposer allowing to propose blocks only for round numbers
    /// `> last_known_proposed_round`. At the moment is allowed to call the method only once leading to a panic
    /// if attempt to do multiple times.
    pub(crate) fn set_last_known_proposed_round(&mut self, round: Round) {
        if let Some(proposer) = &mut self.proposer {
            if proposer.get_last_known_proposed_round().is_some() {
                panic!(
                    "Should not attempt to set the last known proposed round if that has been already set"
                );
            }
            proposer.set_last_known_proposed_round(round);
            info!("Last known proposed round set to {round}");
        }
    }

    /// Returns true if the node should propose blocks.
    /// Observers always return false since they don't have a proposer.
    pub(crate) fn should_propose(&self) -> bool {
        self.proposer
            .as_ref()
            .map(|p| p.should_propose())
            .unwrap_or(false)
    }

    /// Returns the last proposed round, or None if this is an observer node.
    pub(crate) fn last_proposed_round(&self) -> Option<Round> {
        self.proposer.as_ref().map(|p| p.last_proposed_round())
    }

    // Tries to select a prefix of certified commits to be committed next respecting the `limit`.
    // If provided `limit` is zero, it will panic.
    // The function returns a list of certified leaders and certified commits. If empty vector is returned, it means that
    // there are no certified commits to be committed, as input `certified_commits` is either empty or all of the certified
    // commits have been already committed.
    #[tracing::instrument(skip_all)]
    fn try_select_certified_leaders(
        &mut self,
        certified_commits: &mut Vec<CertifiedCommit>,
        limit: usize,
    ) -> Vec<(DecidedLeader, CertifiedCommit)> {
        assert!(limit > 0, "limit should be greater than 0");
        if certified_commits.is_empty() {
            return vec![];
        }

        let to_commit = if certified_commits.len() >= limit {
            // We keep only the number of leaders as dictated by the `limit`
            certified_commits.drain(..limit).collect::<Vec<_>>()
        } else {
            // Otherwise just take all of them and leave the `synced_commits` empty.
            std::mem::take(certified_commits)
        };

        tracing::debug!(
            "Selected {} certified leaders: {}",
            to_commit.len(),
            to_commit.iter().map(|c| c.leader().to_string()).join(",")
        );

        to_commit
            .into_iter()
            .map(|commit| {
                let leader = commit.blocks().last().expect("Certified commit should have at least one block");
                assert_eq!(leader.reference(), commit.leader(), "Last block of the committed sub dag should have the same digest as the leader of the commit");
                // There is no knowledge of direct commit with certified commits, so assuming indirect commit.
                let leader = DecidedLeader::Commit(leader.clone(), /* direct */ false);
                UniversalCommitter::update_metrics(&self.context, &leader, Decision::Certified);
                (leader, commit)
            })
            .collect::<Vec<_>>()
    }

    /// Helper method for tests to get the last proposed block from the proposer.
    #[cfg(test)]
    pub(crate) fn last_proposed_block(&self) -> VerifiedBlock {
        self.proposer
            .as_ref()
            .expect("Proposer should be present")
            .last_proposed_block()
    }

    /// Helper method for tests to get the round tracker from the proposer.
    /// Returns None if this is an observer node.
    #[cfg(test)]
    pub(crate) fn round_tracker_for_tests(&self) -> Arc<RwLock<RoundTracker>> {
        self.proposer
            .as_ref()
            .expect("Proposer should be present")
            .round_tracker_for_tests()
    }
}

/// Senders of signals from Core, for outputs and events (ex new block produced).
pub(crate) struct CoreSignals {
    tx_block_broadcast: broadcast::Sender<ExtendedBlock>,
    tx_accepted_block_broadcast: broadcast::Sender<VerifiedBlock>,
    new_round_sender: watch::Sender<Round>,
    context: Arc<Context>,
}

impl CoreSignals {
    pub fn new(context: Arc<Context>) -> (Self, CoreSignalsReceivers) {
        // Blocks buffered in broadcast channel should be roughly equal to thosed cached in dag state,
        // since the underlying blocks are ref counted so a lower buffer here will not reduce memory
        // usage significantly.
        let (tx_block_broadcast, rx_block_broadcast) = broadcast::channel::<ExtendedBlock>(
            context.parameters.dag_state_cached_rounds as usize,
        );
        let (tx_accepted_block_broadcast, rx_accepted_block_broadcast) =
            broadcast::channel::<VerifiedBlock>(
                2 * context.parameters.dag_state_cached_rounds as usize * context.committee.size(),
            );
        let (new_round_sender, new_round_receiver) = watch::channel(0);

        let me = Self {
            tx_block_broadcast,
            tx_accepted_block_broadcast,
            new_round_sender,
            context,
        };

        let receivers = CoreSignalsReceivers {
            rx_block_broadcast,
            rx_accepted_block_broadcast,
            new_round_receiver,
        };

        (me, receivers)
    }

    /// Sends a signal to all the waiters that a new block has been produced. The method will return
    /// true if block has reached even one subscriber, false otherwise.
    pub(crate) fn new_block(&self, extended_block: ExtendedBlock) -> ConsensusResult<()> {
        // When there is only one authority in committee, it is unnecessary to broadcast
        // the block which will fail anyway without subscribers to the signal.
        if self.context.committee.size() > 1 {
            if extended_block.block.round() == GENESIS_ROUND {
                debug!("Ignoring broadcasting genesis block to peers");
                return Ok(());
            }

            if let Err(err) = self.tx_block_broadcast.send(extended_block) {
                warn!("Couldn't broadcast the block to any receiver: {err}");
                return Err(ConsensusError::Shutdown);
            }
        } else {
            debug!(
                "Did not broadcast block {extended_block:?} to receivers as committee size is <= 1"
            );
        }
        Ok(())
    }

    /// Broadcasts a block that has been accepted into the local DAG to any active observer
    /// subscribers. Unlike `new_block()`, this covers blocks from all authorities, not just
    /// own proposals.
    /// Silently drops the send when there are no active observer subscribers.
    pub(crate) fn new_accepted_block(&self, block: VerifiedBlock) {
        // Ignoring send errors here: it is normal for there to be no observers subscribed.
        let _ = self.tx_accepted_block_broadcast.send(block);
    }

    /// Sends a signal that threshold clock has advanced to new round. The `round_number` is the round at which the
    /// threshold clock has advanced to.
    pub(crate) fn new_round(&mut self, round_number: Round) {
        let _ = self.new_round_sender.send_replace(round_number);
    }
}

/// Receivers of signals from Core.
/// Intentionally un-clonable. Comonents should only subscribe to channels they need.
pub(crate) struct CoreSignalsReceivers {
    rx_block_broadcast: broadcast::Receiver<ExtendedBlock>,
    rx_accepted_block_broadcast: broadcast::Receiver<VerifiedBlock>,
    new_round_receiver: watch::Receiver<Round>,
}

impl CoreSignalsReceivers {
    pub(crate) fn block_broadcast_receiver(&self) -> broadcast::Receiver<ExtendedBlock> {
        self.rx_block_broadcast.resubscribe()
    }

    pub(crate) fn accepted_block_broadcast_receiver(&self) -> broadcast::Receiver<VerifiedBlock> {
        self.rx_accepted_block_broadcast.resubscribe()
    }

    pub(crate) fn new_round_receiver(&self) -> watch::Receiver<Round> {
        self.new_round_receiver.clone()
    }
}

/// Creates cores for the specified number of authorities for their corresponding stakes. The method returns the
/// cores and their respective signal receivers are returned in `AuthorityIndex` order asc.
#[cfg(test)]
pub(crate) async fn create_cores(
    context: Context,
    authorities: Vec<Stake>,
) -> Vec<CoreTestFixture> {
    let mut cores = Vec::new();

    for index in 0..authorities.len() {
        let own_index = AuthorityIndex::new_for_test(index as u32);
        let core =
            CoreTestFixture::new(context.clone(), authorities.clone(), own_index, false).await;
        cores.push(core);
    }
    cores
}

#[cfg(test)]
pub(crate) struct CoreTestFixture {
    pub(crate) core: Core,
    pub(crate) transaction_vote_tracker: TransactionVoteTracker,
    pub(crate) signal_receivers: CoreSignalsReceivers,
    pub(crate) block_receiver: broadcast::Receiver<ExtendedBlock>,
    pub(crate) _commit_output_receiver: UnboundedReceiver<CommittedSubDag>,
    pub(crate) dag_state: Arc<RwLock<DagState>>,
    pub(crate) store: Arc<MemStore>,
    pub(crate) transaction_client: TransactionClient,
}

#[cfg(test)]
impl CoreTestFixture {
    async fn new(
        context: Context,
        authorities: Vec<Stake>,
        own_index: AuthorityIndex,
        sync_last_known_own_block: bool,
    ) -> Self {
        Self::new_with_prepopulated_blocks(
            context,
            authorities,
            own_index,
            sync_last_known_own_block,
            vec![],
        )
        .await
    }

    /// Variant of `new` that writes `prepopulated_blocks` into the store before
    /// constructing `DagState` and `Core`. Used by tests that recover from a
    /// pre-existing store state.
    async fn new_with_prepopulated_blocks(
        context: Context,
        authorities: Vec<Stake>,
        own_index: AuthorityIndex,
        sync_last_known_own_block: bool,
        prepopulated_blocks: Vec<VerifiedBlock>,
    ) -> Self {
        let (committee, mut signers) = local_committee_and_keys(0, authorities.clone());
        let mut context = context.clone();
        context = context
            .with_committee(committee)
            .with_authority_index(own_index);
        context
            .protocol_config
            .set_bad_nodes_stake_threshold_for_testing(33);

        let context = Arc::new(context);
        let store = Arc::new(MemStore::new());
        if !prepopulated_blocks.is_empty() {
            store
                .write(WriteBatch::default().blocks(prepopulated_blocks))
                .expect("Storage error");
        }
        let dag_state = Arc::new(RwLock::new(DagState::new(context.clone(), store.clone())));

        let block_manager = BlockManager::new(context.clone(), dag_state.clone());
        let leader_schedule = Arc::new(
            LeaderSchedule::from_store(context.clone(), dag_state.clone())
                .with_num_commits_per_schedule(10),
        );
        let (transaction_client, tx_receiver) = TransactionClient::new(context.clone());
        let transaction_consumer = TransactionConsumer::new(tx_receiver, context.clone());
        let transaction_vote_tracker = TransactionVoteTracker::new(
            context.clone(),
            Arc::new(NoopBlockVerifier {}),
            dag_state.clone(),
        );
        let (signals, signal_receivers) = CoreSignals::new(context.clone());
        // Need at least one subscriber to the block broadcast channel.
        let block_receiver = signal_receivers.block_broadcast_receiver();

        let (commit_consumer, commit_output_receiver) = CommitConsumerArgs::new(0, 0);
        let commit_observer = CommitObserver::new(
            context.clone(),
            commit_consumer,
            dag_state.clone(),
            transaction_vote_tracker.clone(),
        )
        .await;

        let block_signer = signers.remove(own_index.value()).1;

        let round_tracker = Arc::new(RwLock::new(RoundTracker::new(context.clone(), vec![])));
        let core = Core::new_validator(
            context.clone(),
            leader_schedule,
            transaction_consumer,
            transaction_vote_tracker.clone(),
            block_manager,
            commit_observer,
            signals,
            block_signer,
            dag_state.clone(),
            sync_last_known_own_block,
            round_tracker.clone(),
        );

        Self {
            core,
            transaction_vote_tracker,
            signal_receivers,
            block_receiver,
            _commit_output_receiver: commit_output_receiver,
            dag_state,
            store,
            transaction_client,
        }
    }

    pub(crate) fn add_blocks(
        &mut self,
        blocks: Vec<VerifiedBlock>,
    ) -> ConsensusResult<BTreeSet<BlockRef>> {
        self.transaction_vote_tracker
            .add_voted_blocks(blocks.iter().map(|b| (b.clone(), vec![])).collect());
        self.core.add_blocks(blocks)
    }
}

#[cfg(test)]
mod test {
    use std::{collections::BTreeSet, iter, time::Duration};

    use consensus_config::{AuthorityIndex, Parameters};
    use consensus_types::block::BlockTimestampMs;
    use futures::{StreamExt, stream::FuturesUnordered};
    use tokio::time::sleep;

    use super::*;
    use crate::{
        CommitConsumerArgs, CommitIndex,
        block::{TestBlock, genesis_blocks},
        block_verifier::NoopBlockVerifier,
        commit::CommitAPI,
        leader_scoring::ReputationScores,
        storage::{Store, WriteBatch, mem_store::MemStore},
        test_dag_builder::DagBuilder,
        test_dag_parser::parse_dag,
        transaction::{BlockStatus, TransactionClient},
    };

    /// Recover Core and continue proposing from the last round which forms a quorum.
    #[tokio::test]
    async fn test_core_recover_from_store_for_full_round() {
        telemetry_subscribers::init_for_testing();
        let (context, mut key_pairs) = Context::new_for_test(4);
        let context = Arc::new(context);
        let store = Arc::new(MemStore::new());
        let (_transaction_client, tx_receiver) = TransactionClient::new(context.clone());
        let transaction_consumer = TransactionConsumer::new(tx_receiver, context.clone());
        let mut block_status_subscriptions = FuturesUnordered::new();

        // Create test blocks for all the authorities for 4 rounds and populate them in store
        let mut last_round_blocks = genesis_blocks(&context);
        let mut all_blocks: Vec<VerifiedBlock> = last_round_blocks.clone();
        for round in 1..=4 {
            let mut this_round_blocks = Vec::new();
            for (index, _authority) in context.committee.authorities() {
                let block = VerifiedBlock::new_for_test(
                    TestBlock::new(round, index.value() as u32)
                        .set_ancestors(last_round_blocks.iter().map(|b| b.reference()).collect())
                        .build(),
                );

                // If it's round 1, that one will be committed later on, and it's our "own" block, then subscribe to listen for the block status.
                if round == 1 && index == context.own_index {
                    let subscription =
                        transaction_consumer.subscribe_for_block_status_testing(block.reference());
                    block_status_subscriptions.push(subscription);
                }

                this_round_blocks.push(block);
            }
            all_blocks.extend(this_round_blocks.clone());
            last_round_blocks = this_round_blocks;
        }
        // write them in store
        store
            .write(WriteBatch::default().blocks(all_blocks))
            .expect("Storage error");

        // create dag state after all blocks have been written to store
        let dag_state = Arc::new(RwLock::new(DagState::new(context.clone(), store.clone())));
        let block_manager = BlockManager::new(context.clone(), dag_state.clone());
        let leader_schedule = Arc::new(LeaderSchedule::from_store(
            context.clone(),
            dag_state.clone(),
        ));
        let transaction_vote_tracker = TransactionVoteTracker::new(
            context.clone(),
            Arc::new(NoopBlockVerifier {}),
            dag_state.clone(),
        );

        let (commit_consumer, _commit_receiver) = CommitConsumerArgs::new(0, 0);
        let commit_observer = CommitObserver::new(
            context.clone(),
            commit_consumer,
            dag_state.clone(),
            transaction_vote_tracker.clone(),
        )
        .await;

        // Check no commits have been persisted to dag_state or store.
        let last_commit = store.read_last_commit().unwrap();
        assert!(last_commit.is_none());
        assert_eq!(dag_state.read().last_commit_index(), 0);

        // Now spin up core
        let (signals, signal_receivers) = CoreSignals::new(context.clone());
        let transaction_vote_tracker = TransactionVoteTracker::new(
            context.clone(),
            Arc::new(NoopBlockVerifier {}),
            dag_state.clone(),
        );
        transaction_vote_tracker.recover_blocks_after_round(dag_state.read().gc_round());
        // Need at least one subscriber to the block broadcast channel.
        let mut block_receiver = signal_receivers.block_broadcast_receiver();
        let round_tracker = Arc::new(RwLock::new(RoundTracker::new(context.clone(), vec![])));
        let _core = Core::new_validator(
            context.clone(),
            leader_schedule,
            transaction_consumer,
            transaction_vote_tracker.clone(),
            block_manager,
            commit_observer,
            signals,
            key_pairs.remove(context.own_index.value()).1,
            dag_state.clone(),
            false,
            round_tracker,
        );

        // New round should be 5
        let mut new_round = signal_receivers.new_round_receiver();
        assert_eq!(*new_round.borrow_and_update(), 5);

        // Block for round 5 should have been proposed.
        let proposed_block = block_receiver
            .recv()
            .await
            .expect("A block should have been created");
        assert_eq!(proposed_block.block.round(), 5);
        let ancestors = proposed_block.block.ancestors();

        // Only ancestors of round 4 should be included.
        assert_eq!(ancestors.len(), 4);
        for ancestor in ancestors {
            assert_eq!(ancestor.round, 4);
        }

        // Flush the DAG state to storage.
        dag_state.write().flush();

        // There were no commits prior to the core starting up but there was completed
        // rounds up to and including round 4. So we should commit leaders in round 1 & 2
        // as soon as the new block for round 5 is proposed.
        let last_commit = store
            .read_last_commit()
            .unwrap()
            .expect("last commit should be set");
        assert_eq!(last_commit.index(), 2);
        assert_eq!(dag_state.read().last_commit_index(), 2);
        let all_stored_commits = store.scan_commits((0..=CommitIndex::MAX).into()).unwrap();
        assert_eq!(all_stored_commits.len(), 2);

        // And ensure that our "own" block 1 sent to TransactionConsumer as notification alongside with gc_round
        while let Some(result) = block_status_subscriptions.next().await {
            let status = result.unwrap();
            assert!(matches!(status, BlockStatus::Sequenced(_)));
        }
    }

    /// Recover Core and continue proposing when having a partial last round which doesn't form a quorum and we haven't
    /// proposed for that round yet.
    #[tokio::test]
    async fn test_core_recover_from_store_for_partial_round() {
        telemetry_subscribers::init_for_testing();

        let (context, _) = Context::new_for_test(4);

        // Create test blocks for all authorities except our's (index = 0).
        let mut last_round_blocks = genesis_blocks(&context);
        let mut all_blocks = last_round_blocks.clone();
        for round in 1..=4 {
            let mut this_round_blocks = Vec::new();

            // For round 4 only produce f+1 blocks. Skip our validator 0 and that of position 1 from creating blocks.
            let authorities_to_skip = if round == 4 {
                context.committee.validity_threshold() as usize
            } else {
                // otherwise always skip creating a block for our authority
                1
            };

            for (index, _authority) in context.committee.authorities().skip(authorities_to_skip) {
                let block = TestBlock::new(round, index.value() as u32)
                    .set_ancestors(last_round_blocks.iter().map(|b| b.reference()).collect())
                    .build();
                this_round_blocks.push(VerifiedBlock::new_for_test(block));
            }
            all_blocks.extend(this_round_blocks.clone());
            last_round_blocks = this_round_blocks;
        }

        let mut fixture = CoreTestFixture::new_with_prepopulated_blocks(
            context,
            vec![1, 1, 1, 1],
            AuthorityIndex::new_for_test(0),
            false,
            all_blocks,
        )
        .await;

        // Clock round should have advanced to 5 during recovery because
        // a quorum has formed in round 4.
        let mut new_round = fixture.signal_receivers.new_round_receiver();
        assert_eq!(*new_round.borrow_and_update(), 5);

        // During recovery, round 4 block should have been proposed.
        let proposed_block = fixture
            .block_receiver
            .recv()
            .await
            .expect("A block should have been created");
        assert_eq!(proposed_block.block.round(), 4);
        let ancestors = proposed_block.block.ancestors();

        assert_eq!(ancestors.len(), 4);
        let own_index = fixture.core.context.own_index;
        for ancestor in ancestors {
            if ancestor.author == own_index {
                assert_eq!(ancestor.round, 0);
            } else {
                assert_eq!(ancestor.round, 3);
            }
        }

        // Run commit rule.
        fixture.core.try_commit(vec![]).ok();

        // Flush the DAG state to storage.
        fixture.core.dag_state.write().flush();

        // There were no commits prior to the core starting up but there was completed
        // rounds up to round 4. So we should commit leaders in round 1 & 2 as soon
        // as the new block for round 4 is proposed.
        let last_commit = fixture
            .store
            .read_last_commit()
            .unwrap()
            .expect("last commit should be set");
        assert_eq!(last_commit.index(), 2);
        assert_eq!(fixture.dag_state.read().last_commit_index(), 2);
        let all_stored_commits = fixture
            .store
            .scan_commits((0..=CommitIndex::MAX).into())
            .unwrap();
        assert_eq!(all_stored_commits.len(), 2);
    }

    #[tokio::test]
    async fn test_core_propose_after_genesis() {
        telemetry_subscribers::init_for_testing();
        let (mut context, _) = Context::new_for_test(4);
        context
            .protocol_config
            .set_max_transaction_size_bytes_for_testing(2_000);
        context
            .protocol_config
            .set_max_transactions_in_block_bytes_for_testing(2_000);

        let mut fixture = CoreTestFixture::new(
            context,
            vec![1, 1, 1, 1],
            AuthorityIndex::new_for_test(0),
            false,
        )
        .await;

        // Send some transactions
        let mut total = 0;
        let mut index = 0;
        loop {
            let transaction =
                bcs::to_bytes(&format!("Transaction {index}")).expect("Shouldn't fail");
            total += transaction.len();
            index += 1;
            let _w = fixture
                .transaction_client
                .submit_no_wait(vec![transaction])
                .await
                .unwrap();

            // Create total size of transactions up to 1KB
            if total >= 1_000 {
                break;
            }
        }

        // a new block should have been created during recovery.
        let extended_block = fixture
            .block_receiver
            .recv()
            .await
            .expect("A new block should have been created");

        // A new block created - assert the details
        assert_eq!(extended_block.block.round(), 1);
        assert_eq!(extended_block.block.author().value(), 0);
        assert_eq!(extended_block.block.ancestors().len(), 4);

        let mut total = 0;
        for (i, transaction) in extended_block.block.transactions().iter().enumerate() {
            total += transaction.data().len() as u64;
            let transaction: String = bcs::from_bytes(transaction.data()).unwrap();
            assert_eq!(format!("Transaction {i}"), transaction);
        }
        assert!(
            total
                <= fixture
                    .core
                    .context
                    .protocol_config
                    .max_transactions_in_block_bytes()
        );

        // genesis blocks should be referenced
        let all_genesis = genesis_blocks(&fixture.core.context);

        for ancestor in extended_block.block.ancestors() {
            all_genesis
                .iter()
                .find(|block| block.reference() == *ancestor)
                .expect("Block should be found amongst genesis blocks");
        }

        // Try to propose again - with or without ignore leaders check, it will not return any block
        assert!(fixture.core.try_propose(false).unwrap().is_none());
        assert!(fixture.core.try_propose(true).unwrap().is_none());

        // Flush the DAG state to storage.
        fixture.dag_state.write().flush();

        // Check no commits have been persisted to dag_state & store
        let last_commit = fixture.store.read_last_commit().unwrap();
        assert!(last_commit.is_none());
        assert_eq!(fixture.dag_state.read().last_commit_index(), 0);
    }

    #[tokio::test]
    async fn test_core_propose_once_receiving_a_quorum() {
        telemetry_subscribers::init_for_testing();
        let (context, _key_pairs) = Context::new_for_test(4);
        let mut core_fixture = CoreTestFixture::new(
            context.clone(),
            vec![1, 1, 1, 1],
            AuthorityIndex::new_for_test(0),
            false,
        )
        .await;
        let transaction_vote_tracker = &core_fixture.transaction_vote_tracker;
        let store = &core_fixture.store;
        let dag_state = &core_fixture.dag_state;
        let core = &mut core_fixture.core;

        let mut expected_ancestors = BTreeSet::new();

        // Adding one block now will trigger the creation of new block for round 1
        let block_1 = VerifiedBlock::new_for_test(TestBlock::new(1, 1).build());
        expected_ancestors.insert(block_1.reference());
        // Wait for min round delay to allow blocks to be proposed.
        sleep(context.parameters.min_round_delay).await;
        // add blocks to trigger proposal.
        transaction_vote_tracker.add_voted_blocks(vec![(block_1.clone(), vec![])]);
        _ = core.add_blocks(vec![block_1]);

        assert_eq!(core.last_proposed_round(), Some(1));
        expected_ancestors.insert(core.last_proposed_block().reference());
        // attempt to create a block - none will be produced.
        assert!(core.try_propose(false).unwrap().is_none());

        // Adding another block now forms a quorum for round 1, so block at round 2 will proposed
        let block_2 = VerifiedBlock::new_for_test(TestBlock::new(1, 2).build());
        expected_ancestors.insert(block_2.reference());
        // Wait for min round delay to allow blocks to be proposed.
        sleep(context.parameters.min_round_delay).await;
        // add blocks to trigger proposal.
        transaction_vote_tracker.add_voted_blocks(vec![(block_2.clone(), vec![1, 4])]);
        _ = core.add_blocks(vec![block_2.clone()]);

        assert_eq!(core.last_proposed_round(), Some(2));

        let proposed_block = core.last_proposed_block();
        assert_eq!(proposed_block.round(), 2);
        assert_eq!(proposed_block.author(), context.own_index);
        assert_eq!(proposed_block.ancestors().len(), 3);
        let ancestors = proposed_block.ancestors();
        let ancestors = ancestors.iter().cloned().collect::<BTreeSet<_>>();
        assert_eq!(ancestors, expected_ancestors);

        let transaction_votes = proposed_block.transaction_votes();
        assert_eq!(transaction_votes.len(), 1);
        let transaction_vote = transaction_votes.first().unwrap();
        assert_eq!(transaction_vote.block_ref, block_2.reference());
        assert_eq!(transaction_vote.rejects, vec![1, 4]);

        // Flush the DAG state to storage.
        dag_state.write().flush();

        // Check no commits have been persisted to dag_state & store
        let last_commit = store.read_last_commit().unwrap();
        assert!(last_commit.is_none());
        assert_eq!(dag_state.read().last_commit_index(), 0);
    }

    #[tokio::test]
    async fn test_commit_and_notify_for_block_status() {
        telemetry_subscribers::init_for_testing();
        let (mut context, mut key_pairs) = Context::new_for_test(4);
        const GC_DEPTH: u32 = 2;

        context.protocol_config.set_gc_depth_for_testing(GC_DEPTH);

        let context = Arc::new(context);

        let store = Arc::new(MemStore::new());
        let (_transaction_client, tx_receiver) = TransactionClient::new(context.clone());
        let transaction_consumer = TransactionConsumer::new(tx_receiver, context.clone());
        let mut block_status_subscriptions = FuturesUnordered::new();

        let dag_str = "DAG {
            Round 0 : { 4 },
            Round 1 : { * },
            Round 2 : { * },
            Round 3 : {
                A -> [*],
                B -> [-A2],
                C -> [-A2],
                D -> [-A2],
            },
            Round 4 : { 
                B -> [-A3],
                C -> [-A3],
                D -> [-A3],
            },
            Round 5 : { 
                A -> [A3, B4, C4, D4]
                B -> [*],
                C -> [*],
                D -> [*],
            },
            Round 6 : { * },
            Round 7 : { * },
            Round 8 : { * },
        }";

        let (_, dag_builder) = parse_dag(dag_str).expect("Invalid dag");
        dag_builder.print();

        // Subscribe to all created "own" blocks. We know that for our node (A) we'll be able to commit up to round 5.
        for block in dag_builder.blocks(1..=5) {
            if block.author() == context.own_index {
                let subscription =
                    transaction_consumer.subscribe_for_block_status_testing(block.reference());
                block_status_subscriptions.push(subscription);
            }
        }

        // write them in store
        store
            .write(WriteBatch::default().blocks(dag_builder.blocks(1..=8)))
            .expect("Storage error");

        // create dag state after all blocks have been written to store
        let dag_state = Arc::new(RwLock::new(DagState::new(context.clone(), store.clone())));
        let block_manager = BlockManager::new(context.clone(), dag_state.clone());
        let leader_schedule = Arc::new(LeaderSchedule::from_store(
            context.clone(),
            dag_state.clone(),
        ));
        let transaction_vote_tracker = TransactionVoteTracker::new(
            context.clone(),
            Arc::new(NoopBlockVerifier {}),
            dag_state.clone(),
        );

        let (commit_consumer, _commit_receiver) = CommitConsumerArgs::new(0, 0);
        let commit_observer = CommitObserver::new(
            context.clone(),
            commit_consumer,
            dag_state.clone(),
            transaction_vote_tracker.clone(),
        )
        .await;

        // Flush the DAG state to storage.
        dag_state.write().flush();

        // Check no commits have been persisted to dag_state or store.
        let last_commit = store.read_last_commit().unwrap();
        assert!(last_commit.is_none());
        assert_eq!(dag_state.read().last_commit_index(), 0);

        // Now recover Core and other components.
        let (signals, signal_receivers) = CoreSignals::new(context.clone());
        let transaction_vote_tracker = TransactionVoteTracker::new(
            context.clone(),
            Arc::new(NoopBlockVerifier {}),
            dag_state.clone(),
        );
        transaction_vote_tracker.recover_blocks_after_round(dag_state.read().gc_round());
        // Need at least one subscriber to the block broadcast channel.
        let _block_receiver = signal_receivers.block_broadcast_receiver();
        let round_tracker = Arc::new(RwLock::new(RoundTracker::new(context.clone(), vec![])));
        let _core = Core::new_validator(
            context.clone(),
            leader_schedule,
            transaction_consumer,
            transaction_vote_tracker,
            block_manager,
            commit_observer,
            signals,
            key_pairs.remove(context.own_index.value()).1,
            dag_state.clone(),
            false,
            round_tracker,
        );

        // Flush the DAG state to storage.
        dag_state.write().flush();

        let last_commit = store
            .read_last_commit()
            .unwrap()
            .expect("last commit should be set");

        assert_eq!(last_commit.index(), 5);

        while let Some(result) = block_status_subscriptions.next().await {
            let status = result.unwrap();

            match status {
                BlockStatus::Sequenced(block_ref) => {
                    assert!(block_ref.round == 1 || block_ref.round == 5);
                }
                BlockStatus::GarbageCollected(block_ref) => {
                    assert!(block_ref.round == 2 || block_ref.round == 3);
                }
            }
        }
    }

    // Tests that the threshold clock advances when blocks get unsuspended due to GC'ed blocks and newly created blocks are always higher
    // than the last advanced gc round.
    #[tokio::test]
    async fn test_multiple_commits_advance_threshold_clock() {
        telemetry_subscribers::init_for_testing();
        let (mut context, _) = Context::new_for_test(4);
        const GC_DEPTH: u32 = 2;

        context.protocol_config.set_gc_depth_for_testing(GC_DEPTH);

        // On round 1 we do produce the block for authority D but we do not link it until round 6. This is making round 6 unable to get processed
        // until leader of round 3 is committed where round 1 gets garbage collected.
        // Then we add more rounds so we can trigger a commit for leader of round 9 which will move the gc round to 7.
        let dag_str = "DAG {
            Round 0 : { 4 },
            Round 1 : { * },
            Round 2 : {
                B -> [-D1],
                C -> [-D1],
                D -> [-D1],
            },
            Round 3 : {
                B -> [*],
                C -> [*]
                D -> [*],
            },
            Round 4 : {
                A -> [*],
                B -> [*],
                C -> [*]
                D -> [*],
            },
            Round 5 : {
                A -> [*],
                B -> [*],
                C -> [*],
                D -> [*],
            },
            Round 6 : {
                B -> [A5, B5, C5, D1],
                C -> [A5, B5, C5, D1],
                D -> [A5, B5, C5, D1],
            },
            Round 7 : {
                B -> [*],
                C -> [*],
                D -> [*],
            },
            Round 8 : {
                B -> [*],
                C -> [*],
                D -> [*],
            },
            Round 9 : {
                B -> [*],
                C -> [*],
                D -> [*],
            },
            Round 10 : {
                B -> [*],
                C -> [*],
                D -> [*],
            },
            Round 11 : {
                B -> [*],
                C -> [*],
                D -> [*],
            },
        }";

        let (_, dag_builder) = parse_dag(dag_str).expect("Invalid dag");
        dag_builder.print();

        let mut fixture = CoreTestFixture::new(
            context,
            vec![1, 1, 1, 1],
            AuthorityIndex::new_for_test(0),
            true,
        )
        .await;

        // Check no commits have been persisted to dag_state or store.
        let last_commit = fixture.store.read_last_commit().unwrap();
        assert!(last_commit.is_none());
        assert_eq!(fixture.dag_state.read().last_commit_index(), 0);

        // We set the last known round to 4 so we avoid creating new blocks until then - otherwise it will crash as the already created DAG contains blocks for this
        // authority.
        fixture.core.set_last_known_proposed_round(4);

        // We add all the blocks except D1. The only ones we can immediately accept are the ones up to round 5 as they don't have a dependency on D1. Rest of blocks do have causal dependency
        // to D1 so they can't be processed until the leader of round 3 can get committed and gc round moves to 1. That will make all the blocks that depend to D1 get accepted.
        // However, our threshold clock is now at round 6 as the last quorum that we managed to process was the round 5.
        // As commits happen blocks of later rounds get accepted and more leaders get committed. Eventually the leader of round 9 gets committed and gc is moved to 9 - 2 = 7.
        // If our node attempts to produce a block for the threshold clock 6, that will make the acceptance checks fail as now gc has moved far past this round.
        let mut all_blocks = dag_builder.blocks(1..=11);
        all_blocks.sort_by_key(|b| b.round());
        // Register votes for every block - including the one withheld from Core -
        // so transaction_vote_tracker's state matches the full DAG.
        fixture
            .transaction_vote_tracker
            .add_voted_blocks(all_blocks.iter().map(|b| (b.clone(), vec![])).collect());
        let blocks: Vec<VerifiedBlock> = all_blocks
            .into_iter()
            .filter(|b| !(b.round() == 1 && b.author() == AuthorityIndex::new_for_test(3)))
            .collect();
        fixture.core.add_blocks(blocks).expect("Should not fail");

        assert_eq!(fixture.core.last_proposed_round(), Some(12));
    }

    #[tokio::test]
    async fn test_core_set_min_propose_round() {
        telemetry_subscribers::init_for_testing();
        let (context, _) = Context::new_for_test(4);
        let context = context.with_parameters(Parameters {
            sync_last_known_own_block_timeout: Duration::from_millis(2_000),
            ..Default::default()
        });
        let mut fixture = CoreTestFixture::new(
            context,
            vec![1, 1, 1, 1],
            AuthorityIndex::new_for_test(0),
            true,
        )
        .await;

        // No new block should have been produced
        assert_eq!(
            fixture.core.last_proposed_round(),
            Some(GENESIS_ROUND),
            "No block should have been created other than genesis"
        );

        // Trying to explicitly propose a block will not produce anything
        assert!(fixture.core.try_propose(true).unwrap().is_none());

        // Create blocks for the whole network - even "our" node in order to replicate an "amnesia" recovery.
        let mut builder = DagBuilder::new(fixture.core.context.clone());
        builder.layers(1..=10).build();

        let blocks = builder.blocks.values().cloned().collect::<Vec<_>>();

        // Process all the blocks
        assert!(fixture.add_blocks(blocks).unwrap().is_empty());

        fixture
            .core
            .round_tracker_for_tests()
            .write()
            .update_from_probe(
                vec![
                    vec![10, 10, 10, 10],
                    vec![10, 10, 10, 10],
                    vec![10, 10, 10, 10],
                    vec![10, 10, 10, 10],
                ],
                vec![
                    vec![10, 10, 10, 10],
                    vec![10, 10, 10, 10],
                    vec![10, 10, 10, 10],
                    vec![10, 10, 10, 10],
                ],
            );

        // Try to propose - no block should be produced.
        assert!(fixture.core.try_propose(true).unwrap().is_none());

        // Now set the last known proposed round which is the highest round for which the network informed
        // us that we do have proposed a block about.
        fixture.core.set_last_known_proposed_round(10);

        let block = fixture.core.try_propose(true).expect("No error").unwrap();
        assert_eq!(block.round(), 11);
        assert_eq!(block.ancestors().len(), 4);

        let our_ancestor_included = block.ancestors()[0];
        assert_eq!(our_ancestor_included.author, fixture.core.context.own_index);
        assert_eq!(our_ancestor_included.round, 10);
    }

    #[tokio::test(flavor = "current_thread", start_paused = true)]
    async fn test_core_try_new_block_leader_timeout() {
        telemetry_subscribers::init_for_testing();

        // Since we run the test with started_paused = true, any time-dependent operations using Tokio's time
        // facilities, such as tokio::time::sleep or tokio::time::Instant, will not advance. So practically each
        // Core's clock will have initialised potentially with different values but it never advances.
        // To ensure that blocks won't get rejected by cores we'll need to manually wait for the time
        // diff before processing them. By calling the `tokio::time::sleep` we implicitly also advance the
        // tokio clock.
        async fn wait_blocks(blocks: &[VerifiedBlock], context: &Context) {
            // Simulate the time wait before processing a block to ensure that block.timestamp <= now
            let now = context.clock.timestamp_utc_ms();
            let max_timestamp = blocks
                .iter()
                .max_by_key(|block| block.timestamp_ms() as BlockTimestampMs)
                .map(|block| block.timestamp_ms())
                .unwrap_or(0);

            let wait_time = Duration::from_millis(max_timestamp.saturating_sub(now));
            sleep(wait_time).await;
        }

        let (context, _) = Context::new_for_test(4);
        // Create the cores for all authorities
        let mut all_cores = create_cores(context, vec![1, 1, 1, 1]).await;

        // Create blocks for rounds 1..=3 from all Cores except last Core of authority 3, so we miss the block from it. As
        // it will be the leader of round 3 then no-one will be able to progress to round 4 unless we explicitly trigger
        // the block creation.
        // create the cores and their signals for all the authorities
        let (_last_core, cores) = all_cores.split_last_mut().unwrap();

        // Now iterate over a few rounds and ensure the corresponding signals are created while network advances
        let mut last_round_blocks = Vec::<VerifiedBlock>::new();
        for round in 1..=3 {
            let mut this_round_blocks = Vec::new();

            for core_fixture in cores.iter_mut() {
                wait_blocks(&last_round_blocks, &core_fixture.core.context).await;

                core_fixture.add_blocks(last_round_blocks.clone()).unwrap();

                // Only when round > 1 and using non-genesis parents.
                if let Some(r) = last_round_blocks.first().map(|b| b.round()) {
                    assert_eq!(round - 1, r);
                    if core_fixture.core.last_proposed_round() == Some(r) {
                        // Force propose new block regardless of min round delay.
                        core_fixture
                            .core
                            .try_propose(true)
                            .unwrap()
                            .unwrap_or_else(|| {
                                panic!("Block should have been proposed for round {}", round)
                            });
                    }
                }

                assert_eq!(core_fixture.core.last_proposed_round(), Some(round));

                this_round_blocks.push(core_fixture.core.last_proposed_block().clone());
            }

            last_round_blocks = this_round_blocks;
        }

        // Try to create the blocks for round 4 by calling the try_propose() method. No block should be created as the
        // leader - authority 3 - hasn't proposed any block.
        for core_fixture in cores.iter_mut() {
            wait_blocks(&last_round_blocks, &core_fixture.core.context).await;

            core_fixture.add_blocks(last_round_blocks.clone()).unwrap();
            assert!(core_fixture.core.try_propose(false).unwrap().is_none());
        }

        // Now try to create the blocks for round 4 via the leader timeout method which should
        // ignore any leader checks or min round delay.
        for core_fixture in cores.iter_mut() {
            assert!(core_fixture.core.new_block(4, true).unwrap().is_some());
            assert_eq!(core_fixture.core.last_proposed_round(), Some(4));

            // Flush the DAG state to storage.
            core_fixture.dag_state.write().flush();

            // Check commits have been persisted to store
            let last_commit = core_fixture
                .store
                .read_last_commit()
                .unwrap()
                .expect("last commit should be set");
            // There are 1 leader rounds with rounds completed up to and including
            // round 4
            assert_eq!(last_commit.index(), 1);
            let all_stored_commits = core_fixture
                .store
                .scan_commits((0..=CommitIndex::MAX).into())
                .unwrap();
            assert_eq!(all_stored_commits.len(), 1);
        }
    }

    #[tokio::test(flavor = "current_thread", start_paused = true)]
    async fn test_core_try_new_block_with_leader_timeout_and_low_scoring_authority() {
        telemetry_subscribers::init_for_testing();

        // Since we run the test with started_paused = true, any time-dependent operations using Tokio's time
        // facilities, such as tokio::time::sleep or tokio::time::Instant, will not advance. So practically each
        // Core's clock will have initialised potentially with different values but it never advances.
        // To ensure that blocks won't get rejected by cores we'll need to manually wait for the time
        // diff before processing them. By calling the `tokio::time::sleep` we implicitly also advance the
        // tokio clock.
        async fn wait_blocks(blocks: &[VerifiedBlock], context: &Context) {
            // Simulate the time wait before processing a block to ensure that block.timestamp <= now
            let now = context.clock.timestamp_utc_ms();
            let max_timestamp = blocks
                .iter()
                .max_by_key(|block| block.timestamp_ms() as BlockTimestampMs)
                .map(|block| block.timestamp_ms())
                .unwrap_or(0);

            let wait_time = Duration::from_millis(max_timestamp.saturating_sub(now));
            sleep(wait_time).await;
        }

        let (mut context, _) = Context::new_for_test(5);
        context
            .protocol_config
            .set_bad_nodes_stake_threshold_for_testing(33);

        // Create the cores for all authorities
        let mut all_cores = create_cores(context, vec![1, 1, 1, 1, 1]).await;
        let (_last_core, cores) = all_cores.split_last_mut().unwrap();

        // Create blocks for rounds 1..=30 from all Cores except last Core of authority 4.
        let mut last_round_blocks = Vec::<VerifiedBlock>::new();
        for round in 1..=30 {
            let mut this_round_blocks = Vec::new();

            for core_fixture in cores.iter_mut() {
                wait_blocks(&last_round_blocks, &core_fixture.core.context).await;

                core_fixture.add_blocks(last_round_blocks.clone()).unwrap();

                core_fixture
                    .core
                    .round_tracker_for_tests()
                    .write()
                    .update_from_probe(
                        vec![
                            vec![round, round, round, round, 0],
                            vec![round, round, round, round, 0],
                            vec![round, round, round, round, 0],
                            vec![round, round, round, round, 0],
                            vec![0, 0, 0, 0, 0],
                        ],
                        vec![
                            vec![round, round, round, round, 0],
                            vec![round, round, round, round, 0],
                            vec![round, round, round, round, 0],
                            vec![round, round, round, round, 0],
                            vec![0, 0, 0, 0, 0],
                        ],
                    );

                // Only when round > 1 and using non-genesis parents.
                if let Some(r) = last_round_blocks.first().map(|b| b.round()) {
                    assert_eq!(round - 1, r);
                    if core_fixture.core.last_proposed_round() == Some(r) {
                        // Force propose new block regardless of min round delay.
                        core_fixture
                            .core
                            .try_propose(true)
                            .unwrap()
                            .unwrap_or_else(|| {
                                panic!("Block should have been proposed for round {}", round)
                            });
                    }
                }

                assert_eq!(core_fixture.core.last_proposed_round(), Some(round));

                this_round_blocks.push(core_fixture.core.last_proposed_block().clone());
            }

            last_round_blocks = this_round_blocks;
        }

        // Now produce blocks for all Cores
        for round in 31..=40 {
            let mut this_round_blocks = Vec::new();

            for core_fixture in all_cores.iter_mut() {
                wait_blocks(&last_round_blocks, &core_fixture.core.context).await;

                core_fixture.add_blocks(last_round_blocks.clone()).unwrap();

                // Don't update probed rounds for authority 3 so it will remain
                // excluded
                core_fixture
                    .core
                    .round_tracker_for_tests()
                    .write()
                    .update_from_probe(
                        vec![
                            vec![round, round, round, round, 0],
                            vec![round, round, round, round, 0],
                            vec![round, round, round, round, 0],
                            vec![round, round, round, round, 0],
                            vec![0, 0, 0, 0, 0],
                        ],
                        vec![
                            vec![round, round, round, round, 0],
                            vec![round, round, round, round, 0],
                            vec![round, round, round, round, 0],
                            vec![round, round, round, round, 0],
                            vec![0, 0, 0, 0, 0],
                        ],
                    );

                // Only when round > 1 and using non-genesis parents.
                if let Some(r) = last_round_blocks.first().map(|b| b.round()) {
                    assert_eq!(round - 1, r);
                    if core_fixture.core.last_proposed_round() == Some(r) {
                        // Force propose new block regardless of min round delay.
                        core_fixture
                            .core
                            .try_propose(true)
                            .unwrap()
                            .unwrap_or_else(|| {
                                panic!("Block should have been proposed for round {}", round)
                            });
                    }
                }

                this_round_blocks.push(core_fixture.core.last_proposed_block().clone());

                for block in this_round_blocks.iter() {
                    if block.author() != AuthorityIndex::new_for_test(4) {
                        // Assert blocks created include only 4 ancestors per block as one
                        // should be excluded
                        assert_eq!(block.ancestors().len(), 4);
                    } else {
                        // Authority 3 is the low scoring authority so it will still include
                        // its own blocks.
                        assert_eq!(block.ancestors().len(), 5);
                    }
                }
            }

            last_round_blocks = this_round_blocks;
        }
    }

    #[tokio::test]
    async fn test_smart_ancestor_selection() {
        telemetry_subscribers::init_for_testing();
        let (context, _) = Context::new_for_test(7);
        let context = context.with_parameters(Parameters {
            sync_last_known_own_block_timeout: Duration::from_millis(2_000),
            ..Default::default()
        });
        let mut fixture =
            CoreTestFixture::new(context, vec![1; 7], AuthorityIndex::new_for_test(0), true).await;
        let min_round_delay = fixture.core.context.parameters.min_round_delay;

        // No new block should have been produced
        assert_eq!(
            fixture.core.last_proposed_round(),
            Some(GENESIS_ROUND),
            "No block should have been created other than genesis"
        );

        // Trying to explicitly propose a block will not produce anything
        assert!(fixture.core.try_propose(true).unwrap().is_none());

        // Create blocks for the whole network but not for authority 1
        let mut builder = DagBuilder::new(fixture.core.context.clone());
        builder
            .layers(1..=12)
            .authorities(vec![AuthorityIndex::new_for_test(1)])
            .skip_block()
            .build();
        let blocks = builder.blocks(1..=12);
        // Process all the blocks
        assert!(fixture.add_blocks(blocks).unwrap().is_empty());
        fixture.core.set_last_known_proposed_round(12);

        fixture
            .core
            .round_tracker_for_tests()
            .write()
            .update_from_probe(
                vec![
                    vec![12, 12, 12, 12, 12, 12, 12],
                    vec![0, 0, 0, 0, 0, 0, 0],
                    vec![12, 12, 12, 12, 12, 12, 12],
                    vec![12, 12, 12, 12, 12, 12, 12],
                    vec![12, 12, 12, 12, 12, 12, 12],
                    vec![12, 12, 12, 12, 12, 12, 12],
                    vec![12, 12, 12, 12, 12, 12, 12],
                ],
                vec![
                    vec![12, 12, 12, 12, 12, 12, 12],
                    vec![0, 0, 0, 0, 0, 0, 0],
                    vec![12, 12, 12, 12, 12, 12, 12],
                    vec![12, 12, 12, 12, 12, 12, 12],
                    vec![12, 12, 12, 12, 12, 12, 12],
                    vec![12, 12, 12, 12, 12, 12, 12],
                    vec![12, 12, 12, 12, 12, 12, 12],
                ],
            );

        let block = fixture.core.try_propose(true).expect("No error").unwrap();
        assert_eq!(block.round(), 13);
        assert_eq!(block.ancestors().len(), 7);

        // Build blocks for rest of the network other than own index
        builder
            .layers(13..=14)
            .authorities(vec![AuthorityIndex::new_for_test(0)])
            .skip_block()
            .build();
        let blocks = builder.blocks(13..=14);
        assert!(fixture.add_blocks(blocks).unwrap().is_empty());

        // We now have triggered a leader schedule change so we should have
        // one EXCLUDE authority (1) when we go to select ancestors for the next proposal
        let block = fixture.core.try_propose(true).expect("No error").unwrap();
        assert_eq!(block.round(), 15);
        assert_eq!(block.ancestors().len(), 6);

        // Build blocks for a quorum of the network including the EXCLUDE authority (1)
        // which will trigger smart select and we will not propose a block
        let round_14_ancestors = builder.last_ancestors.clone();
        builder
            .layer(15)
            .authorities(vec![
                AuthorityIndex::new_for_test(0),
                AuthorityIndex::new_for_test(5),
                AuthorityIndex::new_for_test(6),
            ])
            .skip_block()
            .build();
        let blocks = builder.blocks(15..=15);
        let authority_1_excluded_block_reference = blocks
            .iter()
            .find(|block| block.author() == AuthorityIndex::new_for_test(1))
            .unwrap()
            .reference();
        // Wait for min round delay to allow blocks to be proposed.
        sleep(min_round_delay).await;
        // Smart select should be triggered and no block should be proposed.
        assert!(fixture.add_blocks(blocks).unwrap().is_empty());
        assert_eq!(fixture.core.last_proposed_block().round(), 15);

        builder
            .layer(15)
            .authorities(vec![
                AuthorityIndex::new_for_test(0),
                AuthorityIndex::new_for_test(1),
                AuthorityIndex::new_for_test(2),
                AuthorityIndex::new_for_test(3),
                AuthorityIndex::new_for_test(4),
            ])
            .skip_block()
            .override_last_ancestors(round_14_ancestors)
            .build();
        let blocks = builder.blocks(15..=15);
        let round_15_ancestors: Vec<BlockRef> = blocks
            .iter()
            .filter(|block| block.round() == 15)
            .map(|block| block.reference())
            .collect();
        let included_block_references = iter::once(&fixture.core.last_proposed_block())
            .chain(blocks.iter())
            .filter(|block| block.author() != AuthorityIndex::new_for_test(1))
            .map(|block| block.reference())
            .collect::<Vec<_>>();

        // Have enough ancestor blocks to propose now.
        assert!(fixture.add_blocks(blocks).unwrap().is_empty());
        assert_eq!(fixture.core.last_proposed_block().round(), 16);

        // Check that a new block has been proposed & signaled.
        let extended_block = loop {
            let extended_block =
                tokio::time::timeout(Duration::from_secs(1), fixture.block_receiver.recv())
                    .await
                    .unwrap()
                    .unwrap();
            if extended_block.block.round() == 16 {
                break extended_block;
            }
        };
        assert_eq!(extended_block.block.round(), 16);
        assert_eq!(
            extended_block.block.author(),
            fixture.core.context.own_index
        );
        assert_eq!(extended_block.block.ancestors().len(), 6);
        assert_eq!(extended_block.block.ancestors(), included_block_references);
        assert_eq!(extended_block.excluded_ancestors.len(), 1);
        assert_eq!(
            extended_block.excluded_ancestors[0],
            authority_1_excluded_block_reference
        );

        // Build blocks for a quorum of the network including the EXCLUDE ancestor
        // which will trigger smart select and we will not propose a block.
        // This time we will force propose by hitting the leader timeout after which
        // should cause us to include this EXCLUDE ancestor.
        builder
            .layer(16)
            .authorities(vec![
                AuthorityIndex::new_for_test(0),
                AuthorityIndex::new_for_test(5),
                AuthorityIndex::new_for_test(6),
            ])
            .skip_block()
            .override_last_ancestors(round_15_ancestors)
            .build();
        let blocks = builder.blocks(16..=16);
        // Wait for leader timeout to force blocks to be proposed.
        sleep(min_round_delay).await;
        // Smart select should be triggered and no block should be proposed.
        assert!(fixture.add_blocks(blocks).unwrap().is_empty());
        assert_eq!(fixture.core.last_proposed_block().round(), 16);

        // Simulate a leader timeout and a force proposal where we will include
        // one EXCLUDE ancestor when we go to select ancestors for the next proposal
        let block = fixture.core.try_propose(true).expect("No error").unwrap();
        assert_eq!(block.round(), 17);
        assert_eq!(block.ancestors().len(), 5);

        // Check that a new block has been proposed & signaled.
        let extended_block =
            tokio::time::timeout(Duration::from_secs(1), fixture.block_receiver.recv())
                .await
                .unwrap()
                .unwrap();
        assert_eq!(extended_block.block.round(), 17);
        assert_eq!(
            extended_block.block.author(),
            fixture.core.context.own_index
        );
        assert_eq!(extended_block.block.ancestors().len(), 5);
        assert_eq!(extended_block.excluded_ancestors.len(), 0);

        // Excluded authority is locked until round 20, simulate enough rounds to
        // unlock
        builder
            .layers(17..=22)
            .authorities(vec![AuthorityIndex::new_for_test(0)])
            .skip_block()
            .build();
        let blocks = builder.blocks(17..=22);

        // Simulate updating received and accepted rounds from prober.
        // New quorum rounds for authority can then be computed which will unlock
        // the Excluded authority (1) and then we should be able to create a new
        // layer of blocks which will then all be included as ancestors for the
        // next proposal
        fixture
            .core
            .round_tracker_for_tests()
            .write()
            .update_from_probe(
                vec![
                    vec![22, 22, 22, 22, 22, 22, 22],
                    vec![22, 22, 22, 22, 22, 22, 22],
                    vec![22, 22, 22, 22, 22, 22, 22],
                    vec![22, 22, 22, 22, 22, 22, 22],
                    vec![22, 22, 22, 22, 22, 22, 22],
                    vec![22, 22, 22, 22, 22, 22, 22],
                    vec![22, 22, 22, 22, 22, 22, 22],
                ],
                vec![
                    vec![22, 22, 22, 22, 22, 22, 22],
                    vec![22, 22, 22, 22, 22, 22, 22],
                    vec![22, 22, 22, 22, 22, 22, 22],
                    vec![22, 22, 22, 22, 22, 22, 22],
                    vec![22, 22, 22, 22, 22, 22, 22],
                    vec![22, 22, 22, 22, 22, 22, 22],
                    vec![22, 22, 22, 22, 22, 22, 22],
                ],
            );

        let own_index = fixture.core.context.own_index;
        let included_block_references = iter::once(&fixture.core.last_proposed_block())
            .chain(blocks.iter())
            .filter(|block| block.round() == 22 || block.author() == own_index)
            .map(|block| block.reference())
            .collect::<Vec<_>>();

        // Have enough ancestor blocks to propose now.
        sleep(min_round_delay).await;
        assert!(fixture.add_blocks(blocks).unwrap().is_empty());
        assert_eq!(fixture.core.last_proposed_block().round(), 23);

        // Check that a new block has been proposed & signaled.
        let extended_block =
            tokio::time::timeout(Duration::from_secs(1), fixture.block_receiver.recv())
                .await
                .unwrap()
                .unwrap();
        assert_eq!(extended_block.block.round(), 23);
        assert_eq!(
            extended_block.block.author(),
            fixture.core.context.own_index
        );
        assert_eq!(extended_block.block.ancestors().len(), 7);
        assert_eq!(extended_block.block.ancestors(), included_block_references);
        assert_eq!(extended_block.excluded_ancestors.len(), 0);
    }

    #[tokio::test]
    async fn test_excluded_ancestor_limit() {
        telemetry_subscribers::init_for_testing();
        let (context, _) = Context::new_for_test(4);
        let context = context.with_parameters(Parameters {
            sync_last_known_own_block_timeout: Duration::from_millis(2_000),
            ..Default::default()
        });
        let mut fixture = CoreTestFixture::new(
            context,
            vec![1, 1, 1, 1],
            AuthorityIndex::new_for_test(0),
            true,
        )
        .await;

        // No new block should have been produced
        assert_eq!(
            fixture.core.last_proposed_round(),
            Some(GENESIS_ROUND),
            "No block should have been created other than genesis"
        );

        // Create blocks for the whole network
        let mut builder = DagBuilder::new(fixture.core.context.clone());
        builder.layers(1..=3).build();

        // This will equivocate 9 blocks for authority 1 which will be excluded on
        // the proposal but because of the limits set will be dropped and not included
        // as part of the ExtendedBlock structure sent to the rest of the network
        builder
            .layer(4)
            .authorities(vec![AuthorityIndex::new_for_test(1)])
            .equivocate(9)
            .build();
        let blocks = builder.blocks(1..=4);

        // Process all the blocks
        assert!(fixture.add_blocks(blocks).unwrap().is_empty());
        fixture.core.set_last_known_proposed_round(3);

        let block = fixture.core.try_propose(true).expect("No error").unwrap();
        assert_eq!(block.round(), 5);
        assert_eq!(block.ancestors().len(), 4);

        // Check that a new block has been proposed & signaled.
        let extended_block =
            tokio::time::timeout(Duration::from_secs(1), fixture.block_receiver.recv())
                .await
                .unwrap()
                .unwrap();
        assert_eq!(extended_block.block.round(), 5);
        assert_eq!(
            extended_block.block.author(),
            fixture.core.context.own_index
        );
        assert_eq!(extended_block.block.ancestors().len(), 4);
        assert_eq!(extended_block.excluded_ancestors.len(), 8);
    }

    #[tokio::test]
    async fn test_core_set_propagation_delay_per_authority() {
        telemetry_subscribers::init_for_testing();
        let (context, _) = Context::new_for_test(4);
        let mut fixture = CoreTestFixture::new(
            context,
            vec![1, 1, 1, 1],
            AuthorityIndex::new_for_test(0),
            false,
        )
        .await;

        // Use a large propagation delay to disable proposing.
        // This is done by accepting an own block at round 1000 to dag state and
        // then simulating updating round tracker received rounds from probe where
        // low quorum round for own index should get calculated to round 0.
        let test_block = VerifiedBlock::new_for_test(TestBlock::new(1000, 0).build());
        fixture
            .transaction_vote_tracker
            .add_voted_blocks(vec![(test_block.clone(), vec![])]);
        // Force accepting the block to dag state because its causal history is incomplete.
        fixture.dag_state.write().accept_block(test_block);

        fixture
            .core
            .round_tracker_for_tests()
            .write()
            .update_from_probe(
                vec![
                    vec![0, 0, 0, 0],
                    vec![0, 0, 0, 0],
                    vec![0, 0, 0, 0],
                    vec![0, 0, 0, 0],
                ],
                vec![
                    vec![0, 0, 0, 0],
                    vec![0, 0, 0, 0],
                    vec![0, 0, 0, 0],
                    vec![0, 0, 0, 0],
                ],
            );

        // There is no proposal even with forced proposing.
        assert!(fixture.core.try_propose(true).unwrap().is_none());

        // Let Core know there is no propagation delay.
        // This is done by simulating updating round tracker recieved rounds from probe
        // where low quorum round for own index should get calculated to round 1000.
        fixture
            .core
            .round_tracker_for_tests()
            .write()
            .update_from_probe(
                vec![
                    vec![1000, 1000, 1000, 1000],
                    vec![1000, 1000, 1000, 1000],
                    vec![1000, 1000, 1000, 1000],
                    vec![1000, 1000, 1000, 1000],
                ],
                vec![
                    vec![1000, 1000, 1000, 1000],
                    vec![1000, 1000, 1000, 1000],
                    vec![1000, 1000, 1000, 1000],
                    vec![1000, 1000, 1000, 1000],
                ],
            );

        // Also add the necessary blocks from round 1000 so core will propose for
        // round 1001
        for author in 1..4 {
            let block = VerifiedBlock::new_for_test(TestBlock::new(1000, author).build());
            fixture
                .transaction_vote_tracker
                .add_voted_blocks(vec![(block.clone(), vec![])]);
            // Force accepting the block to dag state because its causal history is incomplete.
            fixture.dag_state.write().accept_block(block);
        }

        // Proposing now would succeed.
        assert!(fixture.core.try_propose(true).unwrap().is_some());
    }

    #[tokio::test(flavor = "current_thread", start_paused = true)]
    async fn test_leader_schedule_change() {
        telemetry_subscribers::init_for_testing();
        let default_params = Parameters::default();

        let (context, _) = Context::new_for_test(4);
        // create the cores and their signals for all the authorities
        let mut cores = create_cores(context, vec![1, 1, 1, 1]).await;

        // Now iterate over a few rounds and ensure the corresponding signals are created while network advances
        let mut last_round_blocks = Vec::new();
        for round in 1..=30 {
            let mut this_round_blocks = Vec::new();

            // Wait for min round delay to allow blocks to be proposed.
            sleep(default_params.min_round_delay).await;

            for core_fixture in &mut cores {
                // add the blocks from last round
                // this will trigger a block creation for the round and a signal should be emitted
                core_fixture.add_blocks(last_round_blocks.clone()).unwrap();

                core_fixture
                    .core
                    .round_tracker_for_tests()
                    .write()
                    .update_from_probe(
                        vec![
                            vec![round, round, round, round],
                            vec![round, round, round, round],
                            vec![round, round, round, round],
                            vec![round, round, round, round],
                        ],
                        vec![
                            vec![round, round, round, round],
                            vec![round, round, round, round],
                            vec![round, round, round, round],
                            vec![round, round, round, round],
                        ],
                    );

                // A "new round" signal should be received given that all the blocks of previous round have been processed
                let new_round = receive(
                    Duration::from_secs(1),
                    core_fixture.signal_receivers.new_round_receiver(),
                )
                .await;
                assert_eq!(new_round, round);

                // Check that a new block has been proposed.
                let extended_block = tokio::time::timeout(
                    Duration::from_secs(1),
                    core_fixture.block_receiver.recv(),
                )
                .await
                .unwrap()
                .unwrap();
                assert_eq!(extended_block.block.round(), round);
                assert_eq!(
                    extended_block.block.author(),
                    core_fixture.core.context.own_index
                );

                // append the new block to this round blocks
                this_round_blocks.push(core_fixture.core.last_proposed_block().clone());

                let block = core_fixture.core.last_proposed_block();

                // ensure that produced block is referring to the blocks of last_round
                assert_eq!(
                    block.ancestors().len(),
                    core_fixture.core.context.committee.size()
                );
                for ancestor in block.ancestors() {
                    if block.round() > 1 {
                        // don't bother with round 1 block which just contains the genesis blocks.
                        assert!(
                            last_round_blocks
                                .iter()
                                .any(|block| block.reference() == *ancestor),
                            "Reference from previous round should be added"
                        );
                    }
                }
            }

            last_round_blocks = this_round_blocks;
        }

        for core_fixture in cores {
            // Flush the DAG state to storage.
            core_fixture.dag_state.write().flush();

            // Check commits have been persisted to store
            let last_commit = core_fixture
                .store
                .read_last_commit()
                .unwrap()
                .expect("last commit should be set");
            // There are 28 leader rounds with rounds completed up to and including
            // round 29. Round 30 blocks will only include their own blocks, so the
            // 28th leader will not be committed.
            assert_eq!(last_commit.index(), 27);
            let all_stored_commits = core_fixture
                .store
                .scan_commits((0..=CommitIndex::MAX).into())
                .unwrap();
            assert_eq!(all_stored_commits.len(), 27);
            assert_eq!(
                core_fixture
                    .core
                    .leader_schedule
                    .leader_swap_table
                    .read()
                    .bad_nodes
                    .len(),
                1
            );
            assert_eq!(
                core_fixture
                    .core
                    .leader_schedule
                    .leader_swap_table
                    .read()
                    .good_nodes
                    .len(),
                1
            );
            let expected_reputation_scores =
                ReputationScores::new((11..=20).into(), vec![29, 29, 29, 29]);
            assert_eq!(
                core_fixture
                    .core
                    .leader_schedule
                    .leader_swap_table
                    .read()
                    .reputation_scores,
                expected_reputation_scores
            );
        }
    }

    #[tokio::test]
    async fn test_filter_new_commits() {
        telemetry_subscribers::init_for_testing();

        let (context, _key_pairs) = Context::new_for_test(4);
        let context = context.with_parameters(Parameters {
            sync_last_known_own_block_timeout: Duration::from_millis(2_000),
            ..Default::default()
        });

        let authority_index = AuthorityIndex::new_for_test(0);
        let core = CoreTestFixture::new(context, vec![1, 1, 1, 1], authority_index, true).await;
        let mut core = core.core;

        // No new block should have been produced
        assert_eq!(
            core.last_proposed_round(),
            Some(GENESIS_ROUND),
            "No block should have been created other than genesis"
        );

        // create a DAG of 12 rounds
        let mut dag_builder = DagBuilder::new(core.context.clone());
        dag_builder.layers(1..=12).build();

        // Store all blocks up to round 6 which should be enough to decide up to leader 4
        dag_builder.print();
        let blocks = dag_builder.blocks(1..=6);
        core.dag_state.write().accept_blocks(blocks);

        // Get all the committed sub dags up to round 10
        let sub_dags_and_commits = dag_builder.get_sub_dag_and_certified_commits(1..=10);

        // Now try to commit up to the latest leader (round = 4). Do not provide any certified commits.
        let committed_sub_dags = core.try_commit(vec![]).unwrap();

        // We should have committed up to round 4
        assert_eq!(committed_sub_dags.len(), 4);

        // Now validate the certified commits. We'll try 3 different scenarios:
        println!("Case 1. Provide certified commits that are all before the last committed round.");

        // Highest certified commit should be for leader of round 4.
        let certified_commits = sub_dags_and_commits
            .iter()
            .take(4)
            .map(|(_, c)| c)
            .cloned()
            .collect::<Vec<_>>();
        assert!(
            certified_commits.last().unwrap().index()
                <= committed_sub_dags.last().unwrap().commit_ref.index,
            "Highest certified commit should older than the highest committed index."
        );

        let certified_commits = core.filter_new_commits(certified_commits).unwrap();

        // No commits should be processed
        assert!(certified_commits.is_empty());

        println!("Case 2. Provide certified commits that are all after the last committed round.");

        // Highest certified commit should be for leader of round 4.
        let certified_commits = sub_dags_and_commits
            .iter()
            .take(5)
            .map(|(_, c)| c.clone())
            .collect::<Vec<_>>();

        let certified_commits = core.filter_new_commits(certified_commits.clone()).unwrap();

        // The certified commit of index 5 should be processed.
        assert_eq!(certified_commits.len(), 1);
        assert_eq!(certified_commits.first().unwrap().reference().index, 5);

        println!(
            "Case 3. Provide certified commits where the first certified commit index is not the last_commited_index + 1."
        );

        // Highest certified commit should be for leader of round 4.
        let certified_commits = sub_dags_and_commits
            .iter()
            .skip(5)
            .take(1)
            .map(|(_, c)| c.clone())
            .collect::<Vec<_>>();

        let err = core
            .filter_new_commits(certified_commits.clone())
            .unwrap_err();
        match err {
            ConsensusError::UnexpectedCertifiedCommitIndex {
                expected_commit_index: 5,
                commit_index: 6,
            } => (),
            _ => panic!("Unexpected error: {:?}", err),
        }
    }

    #[tokio::test]
    async fn test_add_certified_commits() {
        telemetry_subscribers::init_for_testing();

        let (context, _key_pairs) = Context::new_for_test(4);
        let context = context.with_parameters(Parameters {
            sync_last_known_own_block_timeout: Duration::from_millis(2_000),
            ..Default::default()
        });

        let authority_index = AuthorityIndex::new_for_test(0);
        let core = CoreTestFixture::new(context, vec![1, 1, 1, 1], authority_index, true).await;
        let store = core.store.clone();
        let mut core = core.core;

        // No new block should have been produced
        assert_eq!(
            core.last_proposed_round(),
            Some(GENESIS_ROUND),
            "No block should have been created other than genesis"
        );

        // create a DAG of 12 rounds
        let mut dag_builder = DagBuilder::new(core.context.clone());
        dag_builder.layers(1..=12).build();

        // Store all blocks up to round 6 which should be enough to decide up to leader 4
        dag_builder.print();
        let blocks = dag_builder.blocks(1..=6);
        core.dag_state.write().accept_blocks(blocks);

        // Get all the committed sub dags up to round 10
        let sub_dags_and_commits = dag_builder.get_sub_dag_and_certified_commits(1..=10);

        // Now try to commit up to the latest leader (round = 4). Do not provide any certified commits.
        let committed_sub_dags = core.try_commit(vec![]).unwrap();

        // We should have committed up to round 4
        assert_eq!(committed_sub_dags.len(), 4);

        // Flush the DAG state to storage.
        core.dag_state.write().flush();

        {
            println!("Case 1. Provide no certified commits. No commit should happen.");
            // Now try to commit up to the latest leader (round = 4). Do not provide any certified commits.
            let committed_sub_dags = core.try_commit(vec![]).unwrap();
            assert!(committed_sub_dags.is_empty());
            let last_commit = store
                .read_last_commit()
                .unwrap()
                .expect("Last commit should be set");
            assert_eq!(last_commit.reference().index, 4);
        }

        println!(
            "Case 2. Provide certified commits that before and after the last committed round and also there are additional blocks so can run the direct decide rule as well."
        );

        // The commits of leader rounds 5-8 should be committed via the certified commits.
        let certified_commits = sub_dags_and_commits
            .iter()
            .skip(3)
            .take(5)
            .map(|(_, c)| c.clone())
            .collect::<Vec<_>>();

        // Now only add the blocks of rounds 7..=12.
        let blocks = dag_builder.blocks(7..=12);
        core.dag_state.write().accept_blocks(blocks);

        // The corresponding blocks of the certified commits should be accepted and stored before linearizing and committing the DAG.
        core.add_certified_commits(CertifiedCommits::new(certified_commits.clone(), vec![]))
            .expect("Should not fail");

        // Flush the DAG state to storage.
        core.dag_state.write().flush();

        let commits = store.scan_commits((6..=10).into()).unwrap();

        // We expect all the sub dags up to leader round 10 to be committed.
        assert_eq!(commits.len(), 5);

        for i in 6..=10 {
            let commit = &commits[i - 6];
            assert_eq!(commit.reference().index, i as u32);
        }
    }

    #[tokio::test]
    async fn try_commit_with_certified_commits_gced_blocks() {
        const GC_DEPTH: u32 = 3;
        telemetry_subscribers::init_for_testing();

        let (mut context, mut key_pairs) = Context::new_for_test(5);
        context.protocol_config.set_gc_depth_for_testing(GC_DEPTH);
        let context = Arc::new(context.with_parameters(Parameters {
            sync_last_known_own_block_timeout: Duration::from_millis(2_000),
            ..Default::default()
        }));

        let store = Arc::new(MemStore::new());
        let dag_state = Arc::new(RwLock::new(DagState::new(context.clone(), store.clone())));

        let block_manager = BlockManager::new(context.clone(), dag_state.clone());
        let leader_schedule = Arc::new(
            LeaderSchedule::from_store(context.clone(), dag_state.clone())
                .with_num_commits_per_schedule(10),
        );

        let (_transaction_client, tx_receiver) = TransactionClient::new(context.clone());
        let transaction_consumer = TransactionConsumer::new(tx_receiver, context.clone());
        let (signals, signal_receivers) = CoreSignals::new(context.clone());
        let transaction_vote_tracker = TransactionVoteTracker::new(
            context.clone(),
            Arc::new(NoopBlockVerifier {}),
            dag_state.clone(),
        );
        // Need at least one subscriber to the block broadcast channel.
        let _block_receiver = signal_receivers.block_broadcast_receiver();

        let (commit_consumer, _commit_receiver) = CommitConsumerArgs::new(0, 0);
        let commit_observer = CommitObserver::new(
            context.clone(),
            commit_consumer,
            dag_state.clone(),
            transaction_vote_tracker.clone(),
        )
        .await;

        let round_tracker = Arc::new(RwLock::new(RoundTracker::new(context.clone(), vec![])));
        let mut core = Core::new_validator(
            context.clone(),
            leader_schedule,
            transaction_consumer,
            transaction_vote_tracker.clone(),
            block_manager,
            commit_observer,
            signals,
            key_pairs.remove(context.own_index.value()).1,
            dag_state.clone(),
            true,
            round_tracker,
        );

        // No new block should have been produced
        assert_eq!(
            core.last_proposed_round(),
            Some(GENESIS_ROUND),
            "No block should have been created other than genesis"
        );

        let dag_str = "DAG {
            Round 0 : { 5 },
            Round 1 : { * },
            Round 2 : { 
                A -> [-E1],
                B -> [-E1],
                C -> [-E1],
                D -> [-E1],
            },
            Round 3 : {
                A -> [*],
                B -> [*],
                C -> [*],
                D -> [*],
            },
            Round 4 : { 
                A -> [*],
                B -> [*],
                C -> [*],
                D -> [*],
            },
            Round 5 : { 
                A -> [*],
                B -> [*],
                C -> [*],
                D -> [*],
                E -> [A4, B4, C4, D4, E1]
            },
            Round 6 : { * },
            Round 7 : { * },
        }";

        let (_, mut dag_builder) = parse_dag(dag_str).expect("Invalid dag");
        dag_builder.print();

        // Now get all the committed sub dags from the DagBuilder
        let (_sub_dags, certified_commits): (Vec<_>, Vec<_>) = dag_builder
            .get_sub_dag_and_certified_commits(1..=5)
            .into_iter()
            .unzip();

        // Now try to commit up to the latest leader (round = 5) with the provided certified commits. Not that we have not accepted any
        // blocks. That should happen during the commit process.
        let committed_sub_dags = core.try_commit(certified_commits).unwrap();

        // We should have committed up to round 4
        assert_eq!(committed_sub_dags.len(), 4);
        for (index, committed_sub_dag) in committed_sub_dags.iter().enumerate() {
            assert_eq!(committed_sub_dag.commit_ref.index as usize, index + 1);

            // ensure that block from E1 node has not been committed
            for block in committed_sub_dag.blocks.iter() {
                if block.round() == 1 && block.author() == AuthorityIndex::new_for_test(5) {
                    panic!("Did not expect to commit block E1");
                }
            }
        }
    }

    #[tokio::test(flavor = "current_thread", start_paused = true)]
    async fn test_commit_on_leader_schedule_change_boundary_without_multileader() {
        parameterized_test_commit_on_leader_schedule_change_boundary(Some(1)).await;
    }

    #[tokio::test(flavor = "current_thread", start_paused = true)]
    async fn test_commit_on_leader_schedule_change_boundary_with_multileader() {
        parameterized_test_commit_on_leader_schedule_change_boundary(None).await;
    }

    async fn parameterized_test_commit_on_leader_schedule_change_boundary(
        num_leaders_per_round: Option<usize>,
    ) {
        telemetry_subscribers::init_for_testing();
        let default_params = Parameters::default();

        let (mut context, _) = Context::new_for_test(6);
        context
            .protocol_config
            .set_num_leaders_per_round_for_testing(num_leaders_per_round);
        // create the cores and their signals for all the authorities
        let mut cores = create_cores(context, vec![1, 1, 1, 1, 1, 1]).await;

        // Now iterate over a few rounds and ensure the corresponding signals are created while network advances
        let mut last_round_blocks: Vec<VerifiedBlock> = Vec::new();
        for round in 1..=33 {
            let mut this_round_blocks = Vec::new();

            // Wait for min round delay to allow blocks to be proposed.
            sleep(default_params.min_round_delay).await;

            for core_fixture in &mut cores {
                // add the blocks from last round
                // this will trigger a block creation for the round and a signal should be emitted
                core_fixture.add_blocks(last_round_blocks.clone()).unwrap();

                core_fixture
                    .core
                    .round_tracker_for_tests()
                    .write()
                    .update_from_probe(
                        vec![
                            vec![round, round, round, round, round, round],
                            vec![round, round, round, round, round, round],
                            vec![round, round, round, round, round, round],
                            vec![round, round, round, round, round, round],
                            vec![round, round, round, round, round, round],
                            vec![round, round, round, round, round, round],
                        ],
                        vec![
                            vec![round, round, round, round, round, round],
                            vec![round, round, round, round, round, round],
                            vec![round, round, round, round, round, round],
                            vec![round, round, round, round, round, round],
                            vec![round, round, round, round, round, round],
                            vec![round, round, round, round, round, round],
                        ],
                    );

                // A "new round" signal should be received given that all the blocks of previous round have been processed
                let new_round = receive(
                    Duration::from_secs(1),
                    core_fixture.signal_receivers.new_round_receiver(),
                )
                .await;
                assert_eq!(new_round, round);

                // Check that a new block has been proposed.
                let extended_block = tokio::time::timeout(
                    Duration::from_secs(1),
                    core_fixture.block_receiver.recv(),
                )
                .await
                .unwrap()
                .unwrap();
                assert_eq!(extended_block.block.round(), round);
                assert_eq!(
                    extended_block.block.author(),
                    core_fixture.core.context.own_index
                );

                // append the new block to this round blocks
                this_round_blocks.push(core_fixture.core.last_proposed_block().clone());

                let block = core_fixture.core.last_proposed_block();

                // ensure that produced block is referring to the blocks of last_round
                assert_eq!(
                    block.ancestors().len(),
                    core_fixture.core.context.committee.size()
                );
                for ancestor in block.ancestors() {
                    if block.round() > 1 {
                        // don't bother with round 1 block which just contains the genesis blocks.
                        assert!(
                            last_round_blocks
                                .iter()
                                .any(|block| block.reference() == *ancestor),
                            "Reference from previous round should be added"
                        );
                    }
                }
            }

            last_round_blocks = this_round_blocks;
        }

        for core_fixture in cores {
            // There are 31 leader rounds with rounds completed up to and including
            // round 33. Round 33 blocks will only include their own blocks, so there
            // should only be 30 commits.
            // However on a leader schedule change boundary its is possible for a
            // new leader to get selected for the same round if the leader elected
            // gets swapped allowing for multiple leaders to be committed at a round.
            // Meaning with multi leader per round explicitly set to 1 we will have 30,
            // otherwise 31.
            // NOTE: We used 31 leader rounds to specifically trigger the scenario
            // where the leader schedule boundary occurred AND we had a swap to a new
            // leader for the same round
            let expected_commit_count = match num_leaders_per_round {
                Some(1) => 30,
                _ => 31,
            };

            // Flush the DAG state to storage.
            core_fixture.dag_state.write().flush();

            // Check commits have been persisted to store
            let last_commit = core_fixture
                .store
                .read_last_commit()
                .unwrap()
                .expect("last commit should be set");
            assert_eq!(last_commit.index(), expected_commit_count);
            let all_stored_commits = core_fixture
                .store
                .scan_commits((0..=CommitIndex::MAX).into())
                .unwrap();
            assert_eq!(all_stored_commits.len(), expected_commit_count as usize);
            assert_eq!(
                core_fixture
                    .core
                    .leader_schedule
                    .leader_swap_table
                    .read()
                    .bad_nodes
                    .len(),
                1
            );
            assert_eq!(
                core_fixture
                    .core
                    .leader_schedule
                    .leader_swap_table
                    .read()
                    .good_nodes
                    .len(),
                1
            );
            let expected_reputation_scores =
                ReputationScores::new((21..=30).into(), vec![43, 43, 43, 43, 43, 43]);
            assert_eq!(
                core_fixture
                    .core
                    .leader_schedule
                    .leader_swap_table
                    .read()
                    .reputation_scores,
                expected_reputation_scores
            );
        }
    }

    #[tokio::test]
    async fn test_core_signals() {
        telemetry_subscribers::init_for_testing();
        let default_params = Parameters::default();

        let (context, _) = Context::new_for_test(4);
        // create the cores and their signals for all the authorities
        let mut cores = create_cores(context, vec![1, 1, 1, 1]).await;

        // Now iterate over a few rounds and ensure the corresponding signals are created while network advances
        let mut last_round_blocks = Vec::new();
        for round in 1..=10 {
            let mut this_round_blocks = Vec::new();

            // Wait for min round delay to allow blocks to be proposed.
            sleep(default_params.min_round_delay).await;

            for core_fixture in &mut cores {
                // add the blocks from last round
                // this will trigger a block creation for the round and a signal should be emitted
                core_fixture.add_blocks(last_round_blocks.clone()).unwrap();

                core_fixture
                    .core
                    .round_tracker_for_tests()
                    .write()
                    .update_from_probe(
                        vec![
                            vec![round, round, round, round],
                            vec![round, round, round, round],
                            vec![round, round, round, round],
                            vec![round, round, round, round],
                        ],
                        vec![
                            vec![round, round, round, round],
                            vec![round, round, round, round],
                            vec![round, round, round, round],
                            vec![round, round, round, round],
                        ],
                    );

                // A "new round" signal should be received given that all the blocks of previous round have been processed
                let new_round = receive(
                    Duration::from_secs(1),
                    core_fixture.signal_receivers.new_round_receiver(),
                )
                .await;
                assert_eq!(new_round, round);

                // Check that a new block has been proposed.
                let extended_block = tokio::time::timeout(
                    Duration::from_secs(1),
                    core_fixture.block_receiver.recv(),
                )
                .await
                .unwrap()
                .unwrap();
                assert_eq!(extended_block.block.round(), round);
                assert_eq!(
                    extended_block.block.author(),
                    core_fixture.core.context.own_index
                );

                // append the new block to this round blocks
                this_round_blocks.push(core_fixture.core.last_proposed_block().clone());

                let block = core_fixture.core.last_proposed_block();

                // ensure that produced block is referring to the blocks of last_round
                assert_eq!(
                    block.ancestors().len(),
                    core_fixture.core.context.committee.size()
                );
                for ancestor in block.ancestors() {
                    if block.round() > 1 {
                        // don't bother with round 1 block which just contains the genesis blocks.
                        assert!(
                            last_round_blocks
                                .iter()
                                .any(|block| block.reference() == *ancestor),
                            "Reference from previous round should be added"
                        );
                    }
                }
            }

            last_round_blocks = this_round_blocks;
        }

        for core_fixture in cores {
            // Flush the DAG state to storage.
            core_fixture.dag_state.write().flush();
            // Check commits have been persisted to store
            let last_commit = core_fixture
                .store
                .read_last_commit()
                .unwrap()
                .expect("last commit should be set");
            // There are 8 leader rounds with rounds completed up to and including
            // round 9. Round 10 blocks will only include their own blocks, so the
            // 8th leader will not be committed.
            assert_eq!(last_commit.index(), 7);
            let all_stored_commits = core_fixture
                .store
                .scan_commits((0..=CommitIndex::MAX).into())
                .unwrap();
            assert_eq!(all_stored_commits.len(), 7);
        }
    }

    #[tokio::test]
    async fn test_core_compress_proposal_references() {
        telemetry_subscribers::init_for_testing();
        let default_params = Parameters::default();

        let (context, _) = Context::new_for_test(4);
        // create the cores and their signals for all the authorities
        let mut cores = create_cores(context, vec![1, 1, 1, 1]).await;

        let mut last_round_blocks = Vec::new();
        let mut all_blocks = Vec::new();

        let excluded_authority = AuthorityIndex::new_for_test(3);

        for round in 1..=10 {
            let mut this_round_blocks = Vec::new();

            for core_fixture in &mut cores {
                // do not produce any block for authority 3
                if core_fixture.core.context.own_index == excluded_authority {
                    continue;
                }

                // try to propose to ensure that we are covering the case where we miss the leader authority 3
                core_fixture.add_blocks(last_round_blocks.clone()).unwrap();
                core_fixture
                    .core
                    .round_tracker_for_tests()
                    .write()
                    .update_from_probe(
                        vec![
                            vec![round, round, round, round],
                            vec![round, round, round, round],
                            vec![round, round, round, round],
                            vec![round, round, round, round],
                        ],
                        vec![
                            vec![round, round, round, round],
                            vec![round, round, round, round],
                            vec![round, round, round, round],
                            vec![round, round, round, round],
                        ],
                    );
                core_fixture.core.new_block(round, true).unwrap();

                let block = core_fixture.core.last_proposed_block();
                assert_eq!(block.round(), round);

                // append the new block to this round blocks
                this_round_blocks.push(block.clone());
            }

            last_round_blocks = this_round_blocks.clone();
            all_blocks.extend(this_round_blocks);
        }

        // Now send all the produced blocks to core of authority 3. It should produce a new block. If no compression would
        // be applied the we should expect all the previous blocks to be referenced from round 0..=10. However, since compression
        // is applied only the last round's (10) blocks should be referenced + the authority's block of round 0.
        let core_fixture = &mut cores[excluded_authority];
        // Wait for min round delay to allow blocks to be proposed.
        sleep(default_params.min_round_delay).await;
        // add blocks to trigger proposal.
        core_fixture.add_blocks(all_blocks).unwrap();

        // Assert that a block has been created for round 11 and it references to blocks of round 10 for the other peers, and
        // to round 1 for its own block (created after recovery).
        let block = core_fixture.core.last_proposed_block();
        assert_eq!(block.round(), 11);
        assert_eq!(block.ancestors().len(), 4);
        for block_ref in block.ancestors() {
            if block_ref.author == excluded_authority {
                assert_eq!(block_ref.round, 1);
            } else {
                assert_eq!(block_ref.round, 10);
            }
        }

        // Flush the DAG state to storage.
        core_fixture.dag_state.write().flush();

        // Check commits have been persisted to store
        let last_commit = core_fixture
            .store
            .read_last_commit()
            .unwrap()
            .expect("last commit should be set");
        // There are 8 leader rounds with rounds completed up to and including
        // round 10. However because there were no blocks produced for authority 3
        // 2 leader rounds will be skipped.
        assert_eq!(last_commit.index(), 6);
        let all_stored_commits = core_fixture
            .store
            .scan_commits((0..=CommitIndex::MAX).into())
            .unwrap();
        assert_eq!(all_stored_commits.len(), 6);
    }

    #[tokio::test]
    async fn try_select_certified_leaders() {
        // GIVEN
        telemetry_subscribers::init_for_testing();

        let (context, _) = Context::new_for_test(4);

        let authority_index = AuthorityIndex::new_for_test(0);
        let core =
            CoreTestFixture::new(context.clone(), vec![1, 1, 1, 1], authority_index, true).await;
        let mut core = core.core;

        let mut dag_builder = DagBuilder::new(Arc::new(context.clone()));
        dag_builder.layers(1..=12).build();

        let limit = 2;

        let blocks = dag_builder.blocks(1..=12);
        core.dag_state.write().accept_blocks(blocks);

        // WHEN
        let sub_dags_and_commits = dag_builder.get_sub_dag_and_certified_commits(1..=4);
        let mut certified_commits = sub_dags_and_commits
            .into_iter()
            .map(|(_, commit)| commit)
            .collect::<Vec<_>>();

        let leaders = core.try_select_certified_leaders(&mut certified_commits, limit);

        // THEN
        assert_eq!(leaders.len(), 2);
        assert_eq!(certified_commits.len(), 2);
    }

    pub(crate) async fn receive<T: Copy>(timeout: Duration, mut receiver: watch::Receiver<T>) -> T {
        tokio::time::timeout(timeout, receiver.changed())
            .await
            .expect("Timeout while waiting to read from receiver")
            .expect("Signal receive channel shouldn't be closed");
        *receiver.borrow_and_update()
    }
}