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},
    transaction::{TransactionConsumer, TransactionConsumerPool},
};
use crate::{
    ancestor::AncestorStateManager,
    block::{BlockAPI, ExtendedBlock, GENESIS_ROUND, Slot, VerifiedBlock},
    block_manager::BlockManager,
    commit::{
        CertifiedCommit, CertifiedCommits, CommitAPI, CommittedSubDag, DecidedLeader, Decision,
        TrustedCommit,
    },
    commit_observer::CommitObserver,
    context::Context,
    dag_state::DagState,
    error::{ConsensusError, ConsensusResult},
    flex_committer::FlexCommitter,
    leader_schedule::LeaderSchedule,
    leader_schedule_v3::LeaderScheduleV3,
    leader_scoring::ReputationScores,
    proposer::{ProposalLeaderWaiter, Proposer, ValidatorProposer},
    round_tracker::RoundTracker,
    transaction::TransactionPool,
    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>,
    /// Scores validators using the DAG and schedules leader for next commit, in a sliding-window.
    leader_schedule_v3: Option<LeaderScheduleV3>,
    /// The commit observer is responsible for observing the commits and collecting
    /// + sending subdags over the consensus output channel.
    commit_observer: CommitObserver,
    /// The v3 commit pipeline helper. Set when `enable_v3` is on. Owns both
    /// the local decision rule via `try_commit` and the certified-commit
    /// sub-dag construction via `handle_certified_commit`.
    flex_committer: Option<FlexCommitter>,
    /// 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_pool: Arc<dyn TransactionPool>,
        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 leader_schedule_v3 = if context.protocol_config.enable_v3() {
            Some(LeaderScheduleV3::from_store(
                context.clone(),
                dag_state.clone(),
            ))
        } else {
            None
        };

        let flex_committer = if context.protocol_config.enable_v3() {
            Some(FlexCommitter::new(context.clone(), dag_state.clone()))
        } else {
            None
        };

        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 ancestor_state_manager = AncestorStateManager::new(context.clone(), dag_state.clone());

        // Create the ValidatorProposer.
        let leader_waiter = if let Some(schedule) = leader_schedule_v3.as_ref() {
            let next_commit_leader_schedule = schedule.next_commit_leader_schedule();
            info!(
                "Recovered next commit leaders: index={} min_round={} num={} allowed={:?}",
                next_commit_leader_schedule.next_commit_index,
                next_commit_leader_schedule.min_next_leader_round,
                next_commit_leader_schedule.num_leaders(),
                next_commit_leader_schedule.allowed_leaders,
            );
            ProposalLeaderWaiter::V3(next_commit_leader_schedule)
        } else {
            ProposalLeaderWaiter::V2(committer.clone())
        };
        let proposer = Some(Box::new(ValidatorProposer::new(
            dag_state.clone(),
            context.clone(),
            transaction_pool,
            transaction_vote_tracker.clone(),
            block_signer,
            last_known_proposed_round,
            ancestor_state_manager,
            round_tracker.clone(),
            leader_waiter,
        )) as Box<dyn Proposer>);

        let mut core = Self {
            context,
            last_signaled_round,
            last_decided_leader,
            leader_schedule,
            leader_schedule_v3,
            block_manager,
            committer,
            commit_observer,
            flex_committer,
            signals,
            dag_state,
            proposer,
        };

        // Initialize propagation scores for the proposer before recovery.
        let propagation_scores = core.current_reputation_scores();
        core.proposer
            .as_mut()
            .unwrap()
            .set_propagation_scores(propagation_scores);

        core.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 leader_schedule_v3 = if context.protocol_config.enable_v3() {
            Some(LeaderScheduleV3::from_store(
                context.clone(),
                dag_state.clone(),
            ))
        } else {
            None
        };

        let flex_committer = if context.protocol_config.enable_v3() {
            Some(FlexCommitter::new(context.clone(), dag_state.clone()))
        } else {
            None
        };

        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,
            leader_schedule_v3,
            block_manager,
            committer,
            commit_observer,
            flex_committer,
            signals,
            dag_state,
            proposer: None,
        }
        .recover_observer()
    }

    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_local().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
    }

    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_local().unwrap();

        self.try_signal_new_round();

        self
    }

    pub(crate) async fn stop(&mut self) {
        self.commit_observer.stop().await;
    }

    /// 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());
        }
        #[cfg(test)]
        sui_macros::handle_fail_point("consensus-after-accept-certified-blocks");
        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_local()?;

            // 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 certified commits synced from peers via the commit syncer. Local commit rule is
    /// skipped and the corresponding leaders and sub dags are committed directly. Blocks of
    /// the certified commits themselves are accepted inside `try_commit()` or
    /// `process_certified_commits()`.
    #[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);

        if self.context.protocol_config.enable_v3() {
            self.process_certified_commits(commits)?;
            self.try_commit_v3()?;
        } else {
            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 {
                    self.context
                        .metrics
                        .node_metrics
                        .core_certified_commits_processed
                        .with_label_values(&["skipped"])
                        .inc();
                    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_local()?;
            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.current_reputation_scores();
                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();

            // Selected certified leaders are guaranteed to be sequenced below,
            // so count them as committed here, keeping this metric consistent
            // with the v3 path (`process_certified_commits`).
            if !decided_certified_commits.is_empty() {
                self.context
                    .metrics
                    .node_metrics
                    .core_certified_commits_processed
                    .with_label_values(&["committed"])
                    .inc_by(decided_certified_commits.len() as u64);
            }

            // 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 = if local {
                self.commit_observer
                    .handle_committed_leaders(sequenced_leaders, true)?
            } else {
                let mut subdags = Vec::with_capacity(sequenced_leaders.len());
                for (leader, certified_commit) in sequenced_leaders
                    .into_iter()
                    .zip_eq(decided_certified_commits)
                {
                    // A certified subdag can omit ancestors at or below the preceding commit's
                    // GC round. Insert that commit before accepting the next subdag, so a
                    // concurrent finalizer flush cannot persist blocks without their GC boundary.
                    self.accept_committed_blocks(certified_commit.blocks().to_vec());
                    subdags.extend(
                        self.commit_observer
                            .handle_committed_leaders(vec![leader], false)?,
                    );
                }
                subdags
            };

            // 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)
    }

    // Processes certified commits that have been synced from peers. Each commit
    // is already quorum-certified, so the decision rule is skipped and the
    // corresponding leaders and sub dags are committed directly. Blocks for the
    // decided certified commits are accepted inside this function to avoid
    // flushing blocks that belong to certified commits which end up not being
    // linearized (see the comment on `accept_committed_blocks` below).
    fn process_certified_commits(
        &mut self,
        certified_commits: Vec<CertifiedCommit>,
    ) -> ConsensusResult<Vec<CommittedSubDag>> {
        if certified_commits.is_empty() {
            return Ok(vec![]);
        }

        let _s = self
            .context
            .metrics
            .node_metrics
            .scope_processing_time
            .with_label_values(&["Core::process_certified_commits"])
            .start_timer();
        info!(
            "Processing certified commits: {:?}",
            certified_commits
                .iter()
                .map(|c| (c.index(), c.leader()))
                .collect::<Vec<_>>()
        );

        let num_certified = certified_commits.len();
        let mut subdags = Vec::with_capacity(num_certified);
        for cert in certified_commits {
            // Accept blocks belonging to this commit before building its sub dag — the
            // sub dag is reconstructed by reading the commit's blocks from DagState.
            // Accepting per commit (rather than up front) keeps the read-after-write
            // coupling local and obvious.
            self.accept_committed_blocks(cert.blocks().to_vec());

            let commit: TrustedCommit = (*cert).clone();

            // A certified commit must extend the local commit chain. A mismatch means
            // local commits have diverged from the quorum-certified chain, which
            // should be surfaced immediately instead of corrupting the commit chain.
            assert_eq!(
                commit.previous_digest(),
                self.dag_state.read().last_commit_digest(),
                "Certified commit {} does not chain onto the last local commit",
                commit.reference(),
            );

            // Certified commits are not decided locally — mark blocks
            // committed and build the sub-dag in one shot.
            let subdag = self
                .flex_committer
                .as_ref()
                .expect("FlexCommitter must be initialized on the v3 path")
                .handle_certified_commit(&commit);

            self.post_commit(commit, subdag.clone())?;
            subdags.push(subdag);

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

        self.context
            .metrics
            .node_metrics
            .core_certified_commits_processed
            .with_label_values(&["committed"])
            .inc_by(num_certified as u64);

        Ok(subdags)
    }

    /// Runs the local commit decision rule on the path selected by `enable_v3`,
    /// without processing any certified commits.
    fn try_commit_local(&mut self) -> ConsensusResult<Vec<CommittedSubDag>> {
        if self.context.protocol_config.enable_v3() {
            self.try_commit_v3()
        } else {
            self.try_commit(vec![])
        }
    }

    // Runs the local commit decision rule on the current DAG and linearizes any
    // newly-decided sub dags. Does not process certified commits — see
    // `Self::process_certified_commits` for that path. Used when
    // `ConsensusProtocolConfig::enable_v3()` is enabled; otherwise the legacy
    // `try_commit` is used.
    fn try_commit_v3(&mut self) -> ConsensusResult<Vec<CommittedSubDag>> {
        // Deliberately reuses the legacy `Core::try_commit` label: only one of the
        // two paths is active for a given protocol config, and sharing the label
        // keeps dashboards working across the transition.
        let _s = self
            .context
            .metrics
            .node_metrics
            .scope_processing_time
            .with_label_values(&["Core::try_commit"])
            .start_timer();

        let mut committed_sub_dags = Vec::new();
        loop {
            let next_commit_leaders = self
                .leader_schedule_v3
                .as_ref()
                .expect("LeaderScheduleV3 must be initialized on the v3 path")
                .next_commit_leader_schedule();
            let Some((commit, subdag)) = self
                .flex_committer
                .as_mut()
                .expect("FlexCommitter must be initialized on the v3 path")
                .try_commit(next_commit_leaders)
            else {
                break;
            };

            self.post_commit(commit, subdag.clone())?;
            committed_sub_dags.push(subdag);

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

        Ok(committed_sub_dags)
    }

    // Post-processing for a single committed subdag on the v3 path: persists the
    // commit, forwards to the finalizer, updates bookkeeping, unsuspends blocks,
    // notifies the proposer about own blocks, and feeds v3 leader
    // scoring.
    fn post_commit(
        &mut self,
        commit: TrustedCommit,
        subdag: CommittedSubDag,
    ) -> ConsensusResult<()> {
        self.dag_state.write().add_commit(commit);

        self.commit_observer.report_commit_metrics(&subdag);
        self.commit_observer.send_to_finalizer(subdag.clone())?;

        self.last_decided_leader = subdag.leader.into();
        self.context
            .metrics
            .node_metrics
            .last_decided_leader_round
            .set(self.last_decided_leader.round as i64);

        self.block_manager
            .try_unsuspend_blocks_for_latest_gc_round();

        let committed_block_refs = subdag
            .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(),
            );
        }

        if let Some(schedule) = self.leader_schedule_v3.as_mut() {
            schedule.add_commit(subdag);
            let next = schedule.next_commit_leader_schedule();
            debug!(
                "Next v3 commit leaders: index={} min_round={} num={} allowed={:?}",
                next.next_commit_index,
                next.min_next_leader_round,
                next.num_leaders(),
                next.allowed_leaders,
            );
            // Push the refreshed schedule into the proposer so its
            // `allowed_leaders` waiting list tracks current scoring instead
            // of being frozen at the epoch-start (all-zero-scores) shuffle.
            if let Some(proposer) = self.proposer.as_mut() {
                proposer.set_next_commit_leader_schedule(next);
            }
        }

        // Refresh the proposer's propagation scores so score-based ancestor
        // exclusion tracks current v3 leader scoring, instead of staying frozen
        // at the scores from Core construction. The legacy path refreshes them
        // on leader schedule updates.
        let propagation_scores = self.current_reputation_scores();
        if let Some(proposer) = &mut self.proposer {
            proposer.set_propagation_scores(propagation_scores);
        }

        Ok(())
    }

    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())
    }

    /// Returns the current `ReputationScores` from the leader schedule.
    fn current_reputation_scores(&self) -> ReputationScores {
        if let Some(schedule) = self.leader_schedule_v3.as_ref() {
            schedule.current_reputation_scores()
        } else {
            self.leader_schedule
                .leader_swap_table
                .read()
                .reputation_scores
                .clone()
        }
    }

    // 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) commit_consumer_monitor: Arc<crate::CommitConsumerMonitor>,
    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 store = Arc::new(MemStore::new());
        if !prepopulated_blocks.is_empty() {
            store
                .write(WriteBatch::default().blocks(prepopulated_blocks))
                .expect("Storage error");
        }
        Self::new_with_store(
            context,
            authorities,
            own_index,
            sync_last_known_own_block,
            store,
        )
        .await
    }

    /// Variant of `new` that builds `DagState` and `Core` against the provided
    /// store. Used by tests exercising restart and recovery: constructing a
    /// second fixture from the same store simulates a node restart.
    async fn new_with_store(
        context: Context,
        authorities: Vec<Stake>,
        own_index: AuthorityIndex,
        sync_last_known_own_block: bool,
        store: Arc<MemStore>,
    ) -> 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 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, priority_tx_receiver) =
            TransactionClient::new(context.clone());
        let transaction_pool = Arc::new(TransactionConsumerPool::new(TransactionConsumer::new(
            tx_receiver,
            priority_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_consumer_monitor = commit_consumer.monitor();
        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_pool,
            transaction_vote_tracker.clone(),
            block_manager,
            commit_observer,
            signals,
            block_signer,
            dag_state.clone(),
            sync_last_known_own_block,
            round_tracker.clone(),
        );

        Self {
            core,
            commit_consumer_monitor,
            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::{BTreeMap, 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, max_transaction_vote_targets},
        block_verifier::NoopBlockVerifier,
        commit::CommitAPI,
        leader_schedule::LeaderSwapTable,
        leader_schedule_v3::NextCommitLeaderSchedule,
        leader_scoring::ReputationScores,
        storage::{Store, WriteBatch, mem_store::MemStore},
        test_dag_builder::DagBuilder,
        test_dag_parser::parse_dag,
        transaction::{BlockStatus, Priority, 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, priority_tx_receiver) =
            TransactionClient::new(context.clone());
        let transaction_pool = Arc::new(TransactionConsumerPool::new(TransactionConsumer::new(
            tx_receiver,
            priority_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_pool.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_pool,
            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], Priority::Normal)
                .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()
        );
        assert_eq!(
            fixture
                .core
                .context
                .metrics
                .node_metrics
                .proposed_block_transaction_bytes
                .get_sample_count(),
            1
        );
        assert_eq!(
            fixture
                .core
                .context
                .metrics
                .node_metrics
                .proposed_block_transaction_bytes
                .get_sample_sum(),
            total as f64
        );

        // 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_proposes_v3_block_with_gc_cutoff() {
        telemetry_subscribers::init_for_testing();
        const GC_DEPTH: u32 = 3;
        let (mut context, _) = Context::new_for_test(4);
        context.protocol_config.set_enable_v3_for_testing(true);
        context.protocol_config.set_gc_depth_for_testing(GC_DEPTH);

        let mut fixture = CoreTestFixture::new(
            context,
            vec![1, 1, 1, 1],
            AuthorityIndex::new_for_test(0),
            false,
        )
        .await;
        let first_proposal = fixture
            .block_receiver
            .recv()
            .await
            .expect("Recovery should propose the first v3 block");
        assert_eq!(
            first_proposal.block.transaction_votes_cutoff_round(),
            GENESIS_ROUND,
            "Nothing is committed yet, so the cutoff must be the genesis round"
        );

        // Commit enough rounds to move the GC round above the genesis round.
        // Core proposes the blocks of its own authority, so the builder must skip them.
        let mut builder = DagBuilder::new(fixture.core.context.clone());
        builder
            .layers(1..=10)
            .authorities(vec![fixture.core.context.own_index])
            .skip_block()
            .build();
        assert!(
            fixture
                .add_blocks(builder.blocks.values().cloned().collect())
                .unwrap()
                .is_empty()
        );
        let gc_round = fixture.dag_state.read().gc_round();
        assert!(gc_round > GENESIS_ROUND, "GC round should have advanced");

        fixture.core.try_propose(true).unwrap();
        let proposed = fixture.core.last_proposed_block();
        assert!(proposed.round() > first_proposal.block.round());
        assert_eq!(proposed.transaction_votes_cutoff_round(), gc_round);
    }

    // Adds 3 rounds of peer blocks. The local authority rejects transaction 0 in each block.
    // Then proposes a block.
    // Returns the fixture, the proposed block and the peer blocks.
    async fn propose_after_peer_blocks_with_reject_votes(
        context: Context,
    ) -> (CoreTestFixture, VerifiedBlock, Vec<VerifiedBlock>) {
        let mut fixture = CoreTestFixture::new(
            context,
            vec![1, 1, 1, 1],
            AuthorityIndex::new_for_test(0),
            false,
        )
        .await;
        let mut builder = DagBuilder::new(fixture.core.context.clone());
        builder
            .layers(1..=3)
            .authorities(vec![fixture.core.context.own_index])
            .skip_block()
            .build();
        let blocks = builder.blocks.values().cloned().collect::<Vec<_>>();
        fixture
            .transaction_vote_tracker
            .add_voted_blocks(blocks.iter().map(|b| (b.clone(), vec![0])).collect());
        assert!(fixture.core.add_blocks(blocks.clone()).unwrap().is_empty());

        fixture.core.try_propose(true).unwrap();
        let proposed = fixture.core.last_proposed_block();
        (fixture, proposed, blocks)
    }

    #[tokio::test]
    async fn test_core_v2_proposal_keeps_all_transaction_vote_targets() {
        telemetry_subscribers::init_for_testing();
        let (mut context, _) = Context::new_for_test(4);
        context.protocol_config.set_gc_depth_for_testing(1);
        let vote_target_limit = max_transaction_vote_targets(&context);

        let (_fixture, proposed, blocks) =
            propose_after_peer_blocks_with_reject_votes(context).await;

        // V2 blocks have no signed cutoff round, so the V3 limit must not remove their targets.
        assert!(blocks.len() > vote_target_limit);
        assert_eq!(proposed.transaction_votes().len(), blocks.len());
    }

    #[tokio::test]
    async fn test_core_v3_proposal_limits_transaction_vote_targets() {
        telemetry_subscribers::init_for_testing();
        let (mut context, _) = Context::new_for_test(4);
        context.protocol_config.set_enable_v3_for_testing(true);
        context.protocol_config.set_gc_depth_for_testing(1);
        let vote_target_limit = max_transaction_vote_targets(&context);

        let (fixture, proposed, blocks) =
            propose_after_peer_blocks_with_reject_votes(context).await;

        assert!(blocks.len() > vote_target_limit);
        assert!(!proposed.transaction_votes().is_empty());
        assert!(proposed.transaction_votes().len() <= vote_target_limit);
        // The signed cutoff moves above the GC round and covers every removed target. Only the
        // targets above the cutoff stay in the block.
        let cutoff_round = proposed.transaction_votes_cutoff_round();
        assert!(cutoff_round > fixture.dag_state.read().gc_round());
        let expected_targets = blocks
            .iter()
            .filter(|block| block.round() > cutoff_round)
            .count();
        assert_eq!(proposed.transaction_votes().len(), expected_targets);
    }

    #[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]);
        assert_eq!(
            context
                .metrics
                .node_metrics
                .proposed_block_transaction_vote_blocks
                .get_sample_count(),
            2
        );
        assert_eq!(
            context
                .metrics
                .node_metrics
                .proposed_block_transaction_vote_blocks
                .get_sample_sum(),
            1.0
        );
        assert_eq!(
            context
                .metrics
                .node_metrics
                .proposed_block_transaction_vote_entries
                .get_sample_count(),
            2
        );
        assert_eq!(
            context
                .metrics
                .node_metrics
                .proposed_block_transaction_vote_entries
                .get_sample_sum(),
            2.0
        );

        // 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, priority_tx_receiver) =
            TransactionClient::new(context.clone());
        let transaction_pool = Arc::new(TransactionConsumerPool::new(TransactionConsumer::new(
            tx_receiver,
            priority_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_pool.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_pool,
            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);
    }

    // Builds a core of 7 authorities and adds blocks up to round 14, with no block from
    // authority 1. The core proposed its last block at round 13. The first proposal of the
    // caller gives authority 1 the EXCLUDE state.
    async fn fixture_with_excluded_authority(
        enable_v3: bool,
    ) -> (CoreTestFixture, DagBuilder, AuthorityIndex) {
        let (mut context, _) = Context::new_for_test(7);
        context.protocol_config.set_enable_v3_for_testing(enable_v3);
        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 excluded = AuthorityIndex::new_for_test(1);

        // The DAG holds no block from this authority, so its propagation score stays zero and
        // the proposer moves it to the EXCLUDE state. The state lock keeps that state for the
        // next rounds.
        let mut builder = DagBuilder::new(fixture.core.context.clone());
        builder
            .layers(1..=12)
            .authorities(vec![excluded])
            .skip_block()
            .build();
        assert!(
            fixture
                .add_blocks(builder.blocks(1..=12))
                .unwrap()
                .is_empty()
        );
        fixture.core.set_last_known_proposed_round(12);
        let mut quorum_rounds = vec![vec![12; 7]; 7];
        quorum_rounds[excluded] = vec![0; 7];
        fixture
            .core
            .round_tracker_for_tests()
            .write()
            .update_from_probe(quorum_rounds.clone(), quorum_rounds);
        assert_eq!(
            fixture
                .core
                .try_propose(true)
                .expect("No error")
                .unwrap()
                .round(),
            13
        );

        builder
            .layers(13..=14)
            .authorities(vec![AuthorityIndex::new_for_test(0)])
            .skip_block()
            .build();
        assert!(
            fixture
                .add_blocks(builder.blocks(13..=14))
                .unwrap()
                .is_empty()
        );
        (fixture, builder, excluded)
    }

    // A v3 proposal must keep a leader ancestor which has the EXCLUDE state.
    #[tokio::test]
    async fn test_smart_ancestor_selection_keeps_v3_leader() {
        telemetry_subscribers::init_for_testing();
        let (mut fixture, mut builder, excluded) = fixture_with_excluded_authority(true).await;
        // Each commit sets the leader schedule of the proposer, so the schedule below must be set
        // after the last add_blocks() and before the proposal which reads it.
        let set_leader = |fixture: &mut CoreTestFixture, leader: AuthorityIndex, round: Round| {
            fixture
                .core
                .proposer
                .as_mut()
                .unwrap()
                .set_next_commit_leader_schedule(NextCommitLeaderSchedule {
                    next_commit_index: 1,
                    min_next_leader_round: round,
                    allowed_leaders: vec![leader],
                });
        };

        // Another authority is the leader of round 14, so the proposal excludes authority 1.
        set_leader(&mut fixture, AuthorityIndex::new_for_test(2), 14);
        let block = fixture.core.try_propose(true).expect("No error").unwrap();
        assert_eq!(block.round(), 15);
        assert!(
            block.ancestors().iter().all(|a| a.author != excluded),
            "The EXCLUDE authority is not a leader, so the proposal must not include it"
        );

        builder
            .layer(15)
            .authorities(vec![AuthorityIndex::new_for_test(0)])
            .skip_block()
            .build();
        assert!(
            fixture
                .add_blocks(builder.blocks(15..=15))
                .unwrap()
                .is_empty()
        );
        // The minimum round delay stops a proposal inside add_blocks(), so the leader schedule
        // below is the schedule of the round 16 proposal.
        assert_eq!(fixture.core.last_proposed_block().round(), 15);

        // Make the EXCLUDE authority the only allowed leader of round 15.
        set_leader(&mut fixture, excluded, 15);
        let proposed = fixture.core.try_propose(true).expect("No error").unwrap();
        assert_eq!(proposed.round(), 16);
        assert!(
            proposed.ancestors().iter().any(|a| a.author == excluded),
            "The proposal must include the leader block, even with its EXCLUDE state"
        );
    }

    // A v2 proposal must not keep a leader ancestor which has the EXCLUDE state.
    #[tokio::test]
    async fn test_smart_ancestor_selection_excludes_v2_leader() {
        telemetry_subscribers::init_for_testing();
        let (mut fixture, mut builder, excluded) = fixture_with_excluded_authority(false).await;

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

        builder
            .layer(15)
            .authorities(vec![AuthorityIndex::new_for_test(0)])
            .skip_block()
            .build();
        assert!(
            fixture
                .add_blocks(builder.blocks(15..=15))
                .unwrap()
                .is_empty()
        );

        // The minimum round delay stops a proposal inside add_blocks(), so the swap table below
        // is the table of the round 16 proposal.
        assert_eq!(fixture.core.last_proposed_block().round(), 15);

        // The leader schedule removes a bad node from the leader slots, so the default swap
        // table is necessary to make the EXCLUDE authority the leader of round 15. Test leader
        // election is `round % committee size`, which selects authority 1 for round 15. A commit
        // can update the swap table, so the table must be set after the last add_blocks().
        *fixture.core.leader_schedule.leader_swap_table.write() = LeaderSwapTable::default();
        assert_eq!(fixture.core.committer.get_leaders(15), vec![excluded]);

        let proposed = fixture.core.try_propose(true).expect("No error").unwrap();
        assert_eq!(proposed.round(), 16);
        assert!(
            proposed.ancestors().iter().all(|a| a.author != excluded),
            "V2 must not keep the EXCLUDE ancestor, even when it is the leader"
        );
    }

    #[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(start_paused = true)]
    async fn certified_commits_recover_after_flush_before_insertion() {
        use crate::linearizer::Linearizer;

        // G is catching up while E returns from an outage with its old self-link.
        // The other five authorities maintain quorum without committing E1.
        let (_, builder) = parse_dag("DAG {
            Round 0 : { 7 },
            Round 1 : { A -> [*], B -> [*], C -> [*], D -> [*], E -> [*], F -> [*] },
            Round 2 : { A -> [-E1], B -> [-E1], C -> [-E1], D -> [-E1], F -> [-E1] },
            Round 3 : { A -> [*], B -> [*], C -> [*], D -> [*], F -> [*] },
            Round 4 : { A -> [*], B -> [*], C -> [*], D -> [*], F -> [*] },
            Round 5 : { A -> [*], B -> [*], C -> [*], D -> [*], E -> [A4, B4, C4, D4, F4, E1], F -> [*] },
            Round 6 : { A -> [*], B -> [*], C -> [*], D -> [*], E -> [*], F -> [*] },
            Round 7 : { A -> [*], B -> [*], C -> [*], D -> [*], E -> [*], F -> [*] }
        }").unwrap();
        let mut context = (*builder.context).clone();
        context.protocol_config.set_gc_depth_for_testing(4);
        context.protocol_config.set_enable_v3_for_testing(false);
        let own_index = AuthorityIndex::new_for_test(6);
        let block = |round, author| {
            builder
                .all_blocks()
                .into_iter()
                .find(|b| b.round() == round && b.author() == AuthorityIndex::new_for_test(author))
                .unwrap()
        };
        let e1 = block(1, 4);
        let e5 = block(5, 4);
        let leaders = vec![
            block(1, 1),
            block(2, 2),
            block(3, 3),
            block(5, 5),
            block(7, 0),
        ];

        let source_store = Arc::new(MemStore::new());
        let source_dag = Arc::new(RwLock::new(DagState::new(
            Arc::new(context.clone()),
            source_store.clone(),
        )));
        source_dag.write().accept_blocks(builder.all_blocks());
        let subdags =
            Linearizer::new(Arc::new(context.clone()), source_dag.clone()).handle_commit(leaders);
        source_dag.write().flush();
        let certified_commits = source_store
            .scan_commits((1..=subdags.len() as u32).into())
            .unwrap()
            .into_iter()
            .zip_eq(&subdags)
            .map(|(commit, subdag)| CertifiedCommit::new_certified(commit, subdag.blocks.clone()))
            .collect::<Vec<_>>();
        assert!(subdags.iter().all(|subdag| {
            subdag
                .blocks
                .iter()
                .all(|b| b.reference() != e1.reference())
        }));
        assert!(
            subdags
                .last()
                .unwrap()
                .blocks
                .iter()
                .any(|b| b.reference() == e5.reference())
        );

        for crash_before_index in 2..=certified_commits.len() {
            let mut fixture =
                CoreTestFixture::new(context.clone(), vec![1; 7], own_index, true).await;
            // Commit sync verifies and votes on fetched blocks before forwarding them to Core.
            fixture.transaction_vote_tracker.add_voted_blocks(
                certified_commits
                    .iter()
                    .flat_map(|commit| commit.blocks().iter().cloned())
                    .map(|block| (block, vec![]))
                    .collect(),
            );
            let (_, missing) = fixture.core.accept_blocks(
                builder
                    .all_blocks()
                    .into_iter()
                    .filter(|block| block.round() <= 3 && block.reference() != e1.reference())
                    .collect(),
            );
            assert!(missing.is_empty());
            let prior = fixture.core.try_commit(vec![]).unwrap();
            assert_eq!(prior.len(), 1);
            assert!(prior[0].decided_with_local_blocks);
            assert_eq!(prior[0].commit_ref, certified_commits[0].reference());

            // A finalizer handling the prior commit can flush at any block-acceptance
            // boundary, before Core inserts the current certified commit.
            let dag = fixture.dag_state.clone();
            let accepted_leader = certified_commits[crash_before_index - 1].leader();
            let finalized_commit_ref = prior[0].commit_ref;
            let test_thread = std::thread::current().id();
            sui_macros::register_fail_point("consensus-after-accept-certified-blocks", move || {
                if std::thread::current().id() != test_thread {
                    return;
                }
                let mut dag = dag.write();
                if dag.last_commit_index() != crash_before_index as u32 - 1 {
                    return;
                }
                assert!(dag.contains_block(&accepted_leader));
                dag.add_finalized_commit(finalized_commit_ref, BTreeMap::new());
                dag.flush();
                panic!("crash after certified block acceptance");
            });
            let crash = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                fixture.core.try_commit(certified_commits[1..].to_vec())
            }));
            sui_macros::clear_fail_point("consensus-after-accept-certified-blocks");
            assert_eq!(
                crash.unwrap_err().downcast_ref::<&str>(),
                Some(&"crash after certified block acceptance")
            );
            let store = fixture.store.clone();
            assert_eq!(
                store.read_last_commit().unwrap().unwrap().index(),
                crash_before_index as u32 - 1
            );
            // Drain the fixture's tasks without adding the interrupted commit.
            fixture.core.stop().await;
            drop(fixture);

            // Exercise Core's actual startup decision rule, then resume commit sync.
            // Batch acceptance would persist E5 with GC=0 and panic on its absent E1 ancestor.
            let mut recovered = CoreTestFixture::new_with_store(
                context.clone(),
                vec![1; 7],
                own_index,
                true,
                store.clone(),
            )
            .await;
            assert!(!recovered.dag_state.read().contains_block(&e1.reference()));
            let remaining = recovered
                .core
                .filter_new_commits(certified_commits.clone())
                .unwrap();
            recovered.transaction_vote_tracker.add_voted_blocks(
                remaining
                    .iter()
                    .flat_map(|commit| commit.blocks().iter().cloned())
                    .map(|block| (block, vec![]))
                    .collect(),
            );
            let resumed = recovered.core.try_commit(remaining).unwrap();
            assert!(
                resumed
                    .iter()
                    .all(|subdag| !subdag.decided_with_local_blocks)
            );
            recovered.dag_state.write().flush();
            let commits = store
                .scan_commits((1..=certified_commits.len() as u32).into())
                .unwrap();
            assert_eq!(commits.len(), certified_commits.len());
            for (actual, expected) in commits.iter().zip_eq(&certified_commits) {
                assert_eq!(actual.reference(), expected.reference());
            }
            recovered.core.stop().await;
        }
    }

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

        let (mut context, _key_pairs) = Context::new_for_test(4);
        context.protocol_config.set_enable_v3_for_testing(true);
        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 monitor = core.commit_consumer_monitor.clone();
        let mut core = core.core;

        // Build a fully connected DAG and accept its blocks, without any commits yet.
        let mut dag_builder = DagBuilder::new(core.context.clone());
        dag_builder.layers(1..=12).build();
        core.dag_state
            .write()
            .accept_blocks(dag_builder.blocks(1..=12));

        let committed = core.try_commit_v3().unwrap();

        // With a fully connected DAG up to round 12, the direct rule can decide leaders
        // up to round 11 (quorum of votes at leader round + 1), one commit per leader round.
        assert_eq!(committed.len(), 11);
        for (i, subdag) in committed.iter().enumerate() {
            assert_eq!(subdag.commit_ref.index, (i + 1) as CommitIndex);
            assert_eq!(subdag.leader.round, (i + 1) as Round);
            assert!(subdag.decided_with_local_blocks);
        }
        assert_eq!(core.dag_state.read().last_commit_index(), 11);
        assert_eq!(monitor.progress().highest_committed_index, 11);
        assert_eq!(monitor.progress().highest_committed_round, 11);
        assert_eq!(monitor.highest_handled_commit(), 0);

        // Re-running the commit rule must be a no-op.
        assert!(core.try_commit_v3().unwrap().is_empty());
    }

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

        let (mut context, _key_pairs) = Context::new_for_test(4);
        context.protocol_config.set_enable_v3_for_testing(true);
        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;

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

        // Certified commits for leader rounds 1..=5. Their blocks have not been
        // accepted locally and are provided by the certified commits themselves.
        let certified_commits = dag_builder
            .get_sub_dag_and_certified_commits(1..=5)
            .into_iter()
            .map(|(_, c)| c)
            .collect::<Vec<_>>();

        core.add_certified_commits(CertifiedCommits::new(certified_commits.clone(), vec![]))
            .expect("Should not fail");

        assert_eq!(core.dag_state.read().last_commit_index(), 5);
        assert_eq!(
            core.context
                .metrics
                .node_metrics
                .core_certified_commits_processed
                .with_label_values(&["committed"])
                .get(),
            5
        );

        // The certified commits should be stored as-is.
        core.dag_state.write().flush();
        let commits = store.scan_commits((1..=5).into()).unwrap();
        assert_eq!(commits.len(), 5);
        for (i, commit) in commits.iter().enumerate() {
            assert_eq!(commit.reference(), certified_commits[i].reference());
        }

        // Accept the rest of the DAG. The local commit rule should continue from the
        // certified commits and commit the leaders of rounds 6..=11.
        core.dag_state
            .write()
            .accept_blocks(dag_builder.blocks(1..=12));
        let committed = core.try_commit_v3().unwrap();
        assert!(!committed.is_empty());
        assert!(committed.iter().all(|s| s.decided_with_local_blocks));
        assert_eq!(committed.last().unwrap().commit_ref.index, 11);
        assert_eq!(core.dag_state.read().last_commit_index(), 11);

        // Re-sending the same certified commits should skip all of them.
        core.add_certified_commits(CertifiedCommits::new(certified_commits, vec![]))
            .expect("Should not fail");
        assert_eq!(core.dag_state.read().last_commit_index(), 11);
        assert_eq!(
            core.context
                .metrics
                .node_metrics
                .core_certified_commits_processed
                .with_label_values(&["skipped"])
                .get(),
            5
        );
    }

    #[tokio::test]
    async fn add_multi_leader_certified_commits_v3() {
        const PREFIX_LAST_ROUND: Round = 5;
        const INITIAL_DAG_LAST_ROUND: Round = PREFIX_LAST_ROUND + 1;
        const CONTINUATION_DAG_LAST_ROUND: Round = INITIAL_DAG_LAST_ROUND + 2;

        telemetry_subscribers::init_for_testing();

        let (mut context, _key_pairs) = Context::new_for_test(4);
        context.protocol_config.set_enable_v3_for_testing(true);
        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);

        // Use the real v3 commit path as the producer. The legacy DagBuilder certified-commit
        // helper creates single-leader commits and cannot exercise this commit shape.
        let source_fixture =
            CoreTestFixture::new(context.clone(), vec![1, 1, 1, 1], authority_index, true).await;
        let source_store = source_fixture.store.clone();
        let mut source = source_fixture.core;

        let receiver_fixture =
            CoreTestFixture::new(context, vec![1, 1, 1, 1], authority_index, true).await;
        let receiver_store = receiver_fixture.store.clone();
        let mut receiver = receiver_fixture.core;

        let next_commit_leaders = source
            .leader_schedule_v3
            .as_ref()
            .expect("LeaderScheduleV3 must be initialized")
            .next_commit_leader_schedule();
        assert!(next_commit_leaders.num_leaders() > 1);
        assert!(
            source
                .context
                .protocol_config
                .leader_schedule_update_interval()
                > PREFIX_LAST_ROUND
        );
        let scheduled_leaders = next_commit_leaders
            .allowed_leaders
            .iter()
            .copied()
            .collect::<BTreeSet<_>>();

        let mut dag_builder = DagBuilder::new(source.context.clone());
        dag_builder.layers(1..=INITIAL_DAG_LAST_ROUND).build();
        source
            .dag_state
            .write()
            .accept_blocks(dag_builder.blocks(1..=INITIAL_DAG_LAST_ROUND));

        let source_subdags = source.try_commit_v3().unwrap();
        // Round 6 supplies direct votes for leader rounds through round 5.
        assert_eq!(source_subdags.len(), PREFIX_LAST_ROUND as usize);
        source.dag_state.write().flush();
        let source_commits = source_store
            .scan_commits((1..=PREFIX_LAST_ROUND).into())
            .unwrap();
        assert_eq!(source_commits.len(), source_subdags.len());

        let source_scores_at_prefix = source.current_reputation_scores();
        assert_ne!(source_scores_at_prefix, ReputationScores::default());

        let certified_leader_counts = |core: &Core| {
            core.context
                .committee
                .authorities()
                .map(|(authority_index, authority)| {
                    (
                        authority_index,
                        core.context
                            .metrics
                            .node_metrics
                            .committed_leaders_total
                            .with_label_values(&[&authority.hostname, "certified-commit"])
                            .get(),
                    )
                })
                .collect::<BTreeMap<_, _>>()
        };
        // The fixtures share a metrics registry. No source commit below uses the
        // `certified-commit` label, and the baseline is taken before receiver processing.
        let certified_leaders_before = certified_leader_counts(&receiver);

        let mut source_subdags_by_commit = source_subdags
            .into_iter()
            .map(|subdag| (subdag.commit_ref, subdag))
            .collect::<BTreeMap<_, _>>();
        let mut expected_certified_leaders = BTreeMap::<AuthorityIndex, u64>::new();
        let mut expected_receiver_subdags = Vec::new();
        let mut has_multi_leader_commit = false;
        let certified_commits = source_commits
            .iter()
            .map(|commit| {
                let subdag = source_subdags_by_commit
                    .remove(&commit.reference())
                    .expect("Every source commit must have a matching subdag");
                let block_refs = subdag
                    .blocks
                    .iter()
                    .map(|block| block.reference())
                    .collect::<Vec<_>>();
                assert_eq!(block_refs, commit.blocks());

                // V3 treats each scheduled block in the named leader round as a leader.
                let leader_blocks = subdag
                    .blocks
                    .iter()
                    .filter(|block| block.round() == commit.leader().round)
                    .map(|block| block.reference())
                    .collect::<Vec<_>>();
                assert!(leader_blocks.contains(&commit.leader()));
                assert!(
                    leader_blocks
                        .iter()
                        .all(|block| scheduled_leaders.contains(&block.author))
                );
                has_multi_leader_commit |= leader_blocks.len() > 1;
                for block in leader_blocks {
                    *expected_certified_leaders.entry(block.author).or_default() += 1;
                }

                let mut expected_receiver_subdag = subdag.clone();
                expected_receiver_subdag.decided_with_local_blocks = false;
                expected_receiver_subdags.push(expected_receiver_subdag);

                // Certification is tested in CommitSyncer. This test starts after verification
                // and checks multi-leader reconstruction and continued local commits.
                CertifiedCommit::new_certified(commit.clone(), subdag.blocks)
            })
            .collect::<Vec<_>>();
        assert!(source_subdags_by_commit.is_empty());
        assert!(has_multi_leader_commit);
        assert!(expected_certified_leaders.values().sum::<u64>() > source_commits.len() as u64);

        let receiver_subdags = receiver
            .process_certified_commits(certified_commits)
            .expect("Certified commits should be accepted");
        assert_eq!(receiver_subdags, expected_receiver_subdags);
        assert!(receiver.try_commit_v3().unwrap().is_empty());
        assert_eq!(
            receiver.dag_state.read().last_commit_index(),
            PREFIX_LAST_ROUND
        );
        assert_eq!(
            receiver.current_reputation_scores(),
            source_scores_at_prefix
        );

        receiver.dag_state.write().flush();
        let receiver_commits = receiver_store
            .scan_commits((1..=PREFIX_LAST_ROUND).into())
            .unwrap();
        assert_eq!(receiver_commits, source_commits);
        let certified_leaders_after = certified_leader_counts(&receiver);
        for (authority_index, expected) in expected_certified_leaders {
            assert_eq!(
                certified_leaders_after[&authority_index]
                    - certified_leaders_before[&authority_index],
                expected
            );
        }

        // Give both nodes the same remaining DAG. The receiver must produce the same next
        // commits as the source after the certified prefix.
        dag_builder
            .layers(INITIAL_DAG_LAST_ROUND + 1..=CONTINUATION_DAG_LAST_ROUND)
            .build();
        source.dag_state.write().accept_blocks(
            dag_builder.blocks(INITIAL_DAG_LAST_ROUND + 1..=CONTINUATION_DAG_LAST_ROUND),
        );
        // Re-accepting the prefix simulates normal block dissemination after commit sync.
        // DagState ignores blocks that the certified commits already marked committed.
        receiver
            .dag_state
            .write()
            .accept_blocks(dag_builder.blocks(1..=CONTINUATION_DAG_LAST_ROUND));

        let source_continuation = source.try_commit_v3().unwrap();
        let receiver_continuation = receiver.try_commit_v3().unwrap();
        assert_eq!(
            source_continuation
                .iter()
                .map(|subdag| subdag.leader.round)
                .collect::<Vec<_>>(),
            vec![PREFIX_LAST_ROUND + 1, PREFIX_LAST_ROUND + 2]
        );
        assert_eq!(receiver_continuation, source_continuation);

        source.dag_state.write().flush();
        receiver.dag_state.write().flush();
        let first_continuation_index = PREFIX_LAST_ROUND + 1;
        let last_continuation_index = source_continuation.last().unwrap().commit_ref.index;
        let source_continuation_commits = source_store
            .scan_commits((first_continuation_index..=last_continuation_index).into())
            .unwrap();
        let receiver_continuation_commits = receiver_store
            .scan_commits((first_continuation_index..=last_continuation_index).into())
            .unwrap();
        assert_eq!(receiver_continuation_commits, source_continuation_commits);
        assert_eq!(
            receiver_continuation_commits[0].previous_digest(),
            source_commits.last().unwrap().digest()
        );
    }

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

        let (mut context, _key_pairs) = Context::new_for_test(5);
        context.protocol_config.set_gc_depth_for_testing(GC_DEPTH);
        context.protocol_config.set_enable_v3_for_testing(true);
        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, 1], authority_index, true).await;
        let store = core.store.clone();
        let mut core = core.core;

        // Authority E proposes only at rounds 1 and 5: E1 stays outside the DAG
        // until E5 links back to it. E5 is first committed by the leader round 6
        // commit, whose gc bound (6 - 1 - GC_DEPTH = 2) excludes E1.
        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");

        // parse_dag creates the DagBuilder with a default context. Apply the same
        // gc_depth so certified commits are linearized with the same gc bound as
        // the receiving Core, like commits certified by peers sharing the
        // protocol config.
        {
            let mut builder_context = (*dag_builder.context).clone();
            builder_context
                .protocol_config
                .set_gc_depth_for_testing(GC_DEPTH);
            dag_builder.context = Arc::new(builder_context);
        }

        // Certified commits for the leader rounds in 1..=6 that have a leader
        // block. Rounds where the elected leader is authority E produce no
        // commit, so commit indices stay consecutive across leader round gaps.
        let certified_commits = dag_builder
            .get_sub_dag_and_certified_commits(1..=6)
            .into_iter()
            .map(|(_, c)| c)
            .collect::<Vec<_>>();
        let num_certified = certified_commits.len() as CommitIndex;
        assert!(num_certified > 0);

        core.add_certified_commits(CertifiedCommits::new(certified_commits, vec![]))
            .expect("Should not fail");

        assert_eq!(core.dag_state.read().last_commit_index(), num_certified);

        // E5 must be committed (every round 6 block links to it), while its
        // ancestor E1 must be excluded by the gc bound. This also exercises
        // accepting a committed block (E5) whose ancestor (E1) is gc'ed and
        // never accepted locally.
        core.dag_state.write().flush();
        let commits = store.scan_commits((1..=num_certified).into()).unwrap();
        assert_eq!(commits.len(), num_certified as usize);
        let authority_e = AuthorityIndex::new_for_test(4);
        let mut e5_committed = false;
        for commit in &commits {
            for block_ref in commit.blocks() {
                assert!(
                    !(block_ref.round == 1 && block_ref.author == authority_e),
                    "Did not expect to commit block E1"
                );
                if block_ref.round == 5 && block_ref.author == authority_e {
                    e5_committed = true;
                }
            }
        }
        assert!(e5_committed, "Expected block E5 to be committed");
    }

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

        let (mut context, _key_pairs) = Context::new_for_test(4);
        context.protocol_config.set_enable_v3_for_testing(true);
        let context = context.with_parameters(Parameters {
            sync_last_known_own_block_timeout: Duration::from_millis(2_000),
            ..Default::default()
        });
        let store = Arc::new(MemStore::new());
        let authority_index = AuthorityIndex::new_for_test(0);

        {
            let mut fixture = CoreTestFixture::new_with_store(
                context.clone(),
                vec![1, 1, 1, 1],
                authority_index,
                true,
                store.clone(),
            )
            .await;
            assert_eq!(
                fixture.core.current_reputation_scores(),
                ReputationScores::default()
            );

            let mut dag_builder = DagBuilder::new(fixture.core.context.clone());
            dag_builder.layers(1..=4).build();
            fixture
                .dag_state
                .write()
                .accept_blocks(dag_builder.blocks(1..=4));
            assert_eq!(fixture.core.try_commit_v3().unwrap().len(), 3);
            assert_eq!(
                fixture.core.current_reputation_scores(),
                ReputationScores::default()
            );
            fixture.dag_state.write().flush();
        }

        let fixture = CoreTestFixture::new_with_store(
            context,
            vec![1, 1, 1, 1],
            authority_index,
            true,
            store,
        )
        .await;
        assert_eq!(fixture.dag_state.read().last_commit_index(), 3);
        assert_eq!(
            fixture.core.current_reputation_scores(),
            ReputationScores::default()
        );
    }

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

        let (mut context, _key_pairs) = Context::new_for_test(4);
        context.protocol_config.set_enable_v3_for_testing(true);
        let context = context.with_parameters(Parameters {
            sync_last_known_own_block_timeout: Duration::from_millis(2_000),
            ..Default::default()
        });
        let store = Arc::new(MemStore::new());
        let authority_index = AuthorityIndex::new_for_test(0);

        // Commit with v3 and flush the resulting state to the store.
        let (last_commit_index, last_committed_rounds) = {
            let mut fixture = CoreTestFixture::new_with_store(
                context.clone(),
                vec![1, 1, 1, 1],
                authority_index,
                true,
                store.clone(),
            )
            .await;
            let mut dag_builder = DagBuilder::new(fixture.core.context.clone());
            dag_builder.layers(1..=12).build();
            fixture
                .dag_state
                .write()
                .accept_blocks(dag_builder.blocks(1..=12));
            let committed = fixture.core.try_commit_v3().unwrap();
            assert!(!committed.is_empty());
            fixture.dag_state.write().flush();
            (
                fixture.dag_state.read().last_commit_index(),
                fixture.dag_state.read().last_committed_rounds(),
            )
        };

        // "Restart" by recovering DagState and Core from the store. On the v3 path,
        // recovery does not read CommitInfo (never persisted with v3): committed
        // rounds are rebuilt from the bounded commit scan, and the scoring subdag
        // stays empty since only legacy leader scoring reads it.
        let fixture = CoreTestFixture::new_with_store(
            context,
            vec![1, 1, 1, 1],
            authority_index,
            true,
            store,
        )
        .await;
        assert_eq!(
            fixture.dag_state.read().last_commit_index(),
            last_commit_index
        );
        assert_eq!(
            fixture.dag_state.read().last_committed_rounds(),
            last_committed_rounds
        );
        assert_eq!(fixture.dag_state.read().scoring_subdags_count(), 0);
    }

    #[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, priority_tx_receiver) =
            TransactionClient::new(context.clone());
        let transaction_pool = Arc::new(TransactionConsumerPool::new(TransactionConsumer::new(
            tx_receiver,
            priority_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_pool,
            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()
    }
}