consensus_core/

commit_observer.rs

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

use std::{sync::Arc, time::Duration};

use parking_lot::RwLock;
use tokio::time::Instant;
use tracing::info;

use crate::{
    CommitConsumerArgs, CommitConsumerMonitor, CommittedSubDag,
    block::{BlockAPI, VerifiedBlock},
    commit::{CommitAPI, load_committed_subdag_from_store},
    commit_consumer::ReplayMode,
    commit_finalizer::{CommitFinalizer, CommitFinalizerHandle},
    context::Context,
    dag_state::DagState,
    error::ConsensusResult,
    linearizer::Linearizer,
    storage::Store,
    task::spawn_blocking,
    transaction_vote_tracker::TransactionVoteTracker,
};

/// Role of CommitObserver
/// - Called by core when try_commit() returns newly committed leaders.
/// - The newly committed leaders are sent to commit observer and then commit observer
///   gets subdags for each leader via the commit interpreter (linearizer)
/// - The committed subdags are sent as consensus output via an unbounded tokio channel.
///
/// Live output has no flow control; consensus backpressure is applied earlier at consensus
/// input level and on commit sync. Full-epoch replay can optionally be paced by consumer
/// acknowledgements during recovery.
///
/// Commit is persisted in store before the CommittedSubDag is sent to the commit handler.
/// When Sui recovers, it blocks until the commits it knows about are recovered. So consensus
/// must be able to quickly recover the commits it has sent to Sui.
pub(crate) struct CommitObserver {
    context: Arc<Context>,
    dag_state: Arc<RwLock<DagState>>,
    /// Persistent storage for blocks, commits and other consensus data.
    store: Arc<dyn Store>,
    transaction_vote_tracker: TransactionVoteTracker,
    /// Component to deterministically collect subdags for committed leaders.
    commit_interpreter: Linearizer,
    /// Handle to an unbounded channel to send output commits.
    commit_finalizer_handle: CommitFinalizerHandle,
    consumer_monitor: Arc<CommitConsumerMonitor>,
}

impl CommitObserver {
    pub(crate) async fn new(
        context: Arc<Context>,
        commit_consumer: CommitConsumerArgs,
        dag_state: Arc<RwLock<DagState>>,
        transaction_vote_tracker: TransactionVoteTracker,
    ) -> Self {
        let store = dag_state.read().store();
        let commit_interpreter = Linearizer::new(context.clone(), dag_state.clone());
        let commit_finalizer_handle = CommitFinalizer::start(
            context.clone(),
            dag_state.clone(),
            transaction_vote_tracker.clone(),
            commit_consumer.commit_sender.clone(),
        );

        let mut observer = Self {
            context,
            dag_state,
            store,
            transaction_vote_tracker,
            commit_interpreter,
            commit_finalizer_handle,
            consumer_monitor: commit_consumer.monitor(),
        };
        observer.recover_and_send_commits(&commit_consumer).await;

        // Recover blocks needed for future commits (and block proposals).
        // Some blocks might have been recovered as committed blocks in recover_and_send_commits().
        // They will just be ignored.
        if let Err(e) = spawn_blocking({
            let transaction_vote_tracker = observer.transaction_vote_tracker.clone();
            let gc_round = observer.dag_state.read().gc_round();
            move || {
                transaction_vote_tracker.recover_blocks_after_round(gc_round);
            }
        })
        .await
        {
            info!("Skipping block recovery for transaction voting: {e}");
        }

        observer
    }

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

    /// Creates and returns a list of committed subdags containing committed blocks, from a sequence
    /// of selected leader blocks, and whether they come from local committer or commit sync remotely.
    ///
    /// Also, buffers the commits to DagState and forwards committed subdags to commit finalizer.
    pub(crate) fn handle_committed_leaders(
        &mut self,
        committed_leaders: Vec<VerifiedBlock>,
        local: bool,
    ) -> ConsensusResult<Vec<CommittedSubDag>> {
        let _s = self
            .context
            .metrics
            .node_metrics
            .scope_processing_time
            .with_label_values(&["CommitObserver::handle_committed_leaders"])
            .start_timer();

        let mut committed_sub_dags = self.commit_interpreter.handle_commit(committed_leaders);
        self.report_metrics(&committed_sub_dags);

        // Set if the commit is produced from local DAG, or received through commit sync.
        for subdag in committed_sub_dags.iter_mut() {
            subdag.decided_with_local_blocks = local;
        }

        for commit in committed_sub_dags.iter() {
            tracing::debug!(
                "Sending commit {} leader {} to finalization and execution.",
                commit.commit_ref,
                commit.leader
            );
            tracing::trace!("Committed subdag: {:#?}", commit);
            // Failures in sender.send() are assumed to be permanent
            self.send_to_finalizer(commit.clone())?;
        }

        self.dag_state
            .write()
            .add_scoring_subdags(committed_sub_dags.clone());

        Ok(committed_sub_dags)
    }

    /// Forwards a committed subdag to the commit finalizer. Used by `Core::post_commit`.
    pub(crate) fn send_to_finalizer(&self, subdag: CommittedSubDag) -> ConsensusResult<()> {
        // Both the legacy and v3 commit paths report progress before consumer work.
        self.consumer_monitor
            .report_committed(subdag.commit_ref.index, subdag.leader.round);
        self.commit_finalizer_handle.send(subdag)
    }

    async fn recover_and_send_commits(&mut self, commit_consumer: &CommitConsumerArgs) {
        let now = Instant::now();

        let replay_after_commit_index = match commit_consumer.replay_mode {
            ReplayMode::FromWatermark {
                replay_after_commit_index,
                ..
            } => replay_after_commit_index,
            ReplayMode::FullEpochPaced => 0,
        };

        let last_commit = self
            .store
            .read_last_commit()
            .expect("Reading the last commit should not fail");
        if let Some(commit) = &last_commit {
            self.consumer_monitor
                .report_committed(commit.index(), commit.leader().round);
        }
        // Storage ownership, target discovery, and commit decoding stay here.
        // Consumers rebuilding from empty never supply a second durable cursor.
        let replay_target = if matches!(commit_consumer.replay_mode, ReplayMode::FullEpochPaced) {
            let target = if self.context.protocol_config.transaction_voting_enabled() {
                self.store
                    .read_last_finalized_commit()
                    .expect("Reading the last finalized commit should not fail")
                    .map_or(0, |commit| commit.index)
            } else {
                last_commit.as_ref().map_or(0, |commit| commit.index())
            };
            assert!(
                target <= last_commit.as_ref().map_or(0, |commit| commit.index()),
                "finalized replay target cannot exceed the consensus store head"
            );
            self.consumer_monitor.set_replay_target(target);
            target
        } else {
            0
        };
        let Some(last_commit) = &last_commit else {
            assert_eq!(
                replay_after_commit_index, 0,
                "Commit replay should start at the beginning if there is no commit history"
            );
            info!("Nothing to recover for commit observer - starting new epoch");
            return;
        };

        let last_commit_index = last_commit.index();
        if last_commit_index == replay_after_commit_index {
            info!(
                "Nothing to recover for commit observer - replay is requested immediately after last commit index {last_commit_index}"
            );
            return;
        }
        assert!(last_commit_index > replay_after_commit_index);

        info!(
            "Recovering commit observer in the range [{}..={last_commit_index}]",
            replay_after_commit_index + 1,
        );

        // To avoid scanning too many commits at once and load in memory,
        // we limit the batch size to 250 and iterate over.
        const COMMIT_RECOVERY_BATCH_SIZE: u32 = if cfg!(test) { 3 } else { 250 };

        let mut last_sent_commit_index = replay_after_commit_index;

        // Make sure that there is no pending commits to be written to the store.
        self.dag_state.read().ensure_commits_to_write_is_empty();

        let mut seen_unfinalized_commit = false;
        for start_index in (replay_after_commit_index + 1..=last_commit_index)
            .step_by(COMMIT_RECOVERY_BATCH_SIZE as usize)
        {
            let end_index = start_index
                .saturating_add(COMMIT_RECOVERY_BATCH_SIZE - 1)
                .min(last_commit_index);

            let unsent_commits = self
                .store
                .scan_commits((start_index..=end_index).into())
                .expect("Scanning commits should not fail");
            assert_eq!(
                unsent_commits.len() as u32,
                end_index.checked_sub(start_index).unwrap() + 1,
                "Gap in scanned commits: start index: {start_index}, end index: {end_index}, commits: {:?}",
                unsent_commits,
            );

            // Buffered unsent commits in DAG state which is required to contain them when they are flushed
            // by CommitFinalizer.
            self.dag_state
                .write()
                .recover_commits_to_write(unsent_commits.clone());

            info!(
                "Recovering {} unsent commits in range [{start_index}..={end_index}]",
                unsent_commits.len()
            );

            // Resend all the committed subdags to the consensus output channel
            // for all the commits above the last processed index.
            for commit in unsent_commits.into_iter() {
                // Commit index must be continuous.
                last_sent_commit_index += 1;
                assert_eq!(commit.index(), last_sent_commit_index);

                let committed_sub_dag =
                    load_committed_subdag_from_store(self.store.as_ref(), commit);

                if matches!(commit_consumer.replay_mode, ReplayMode::FullEpochPaced)
                    && self.context.protocol_config.transaction_voting_enabled()
                    && committed_sub_dag.commit_ref.index <= replay_target
                {
                    // Finalized commits must have rejection metadata, even if empty.
                    // Without it, finalization may need later commits while replay
                    // waits for an acknowledgement, blocking startup.
                    assert!(
                        committed_sub_dag.recovered_rejected_transactions,
                        "missing rejected transactions below the finalized replay target"
                    );
                }

                if !committed_sub_dag.recovered_rejected_transactions && !seen_unfinalized_commit {
                    info!(
                        "Starting to recover unfinalized commit from {}",
                        committed_sub_dag.commit_ref
                    );
                    // When the commit has no associated storage entry for rejected transactions,
                    // not even an empty set, the commit is unfinalized.
                    seen_unfinalized_commit = true;
                }

                if seen_unfinalized_commit {
                    // After observing the first unfinalized commit, the rest of recovered commits should all be unfinalized.
                    assert!(!committed_sub_dag.recovered_rejected_transactions);
                    // All unfinalized commit cannot be assumed to be decided with local blocks, because they
                    // might have been received through commit sync.
                    assert!(!committed_sub_dag.decided_with_local_blocks);
                    // All unfinalized commits need to be processed by the CommitFinalizer, making it necessary to
                    // recover and vote on the blocks in this commit.
                    self.transaction_vote_tracker
                        .recover_and_vote_on_blocks(committed_sub_dag.blocks.clone());
                }

                self.commit_finalizer_handle
                    .send(committed_sub_dag)
                    .unwrap();

                self.context
                    .metrics
                    .node_metrics
                    .commit_observer_last_recovered_commit_index
                    .set(last_sent_commit_index as i64);

                tokio::task::yield_now().await;
            }
            if matches!(commit_consumer.replay_mode, ReplayMode::FullEpochPaced) {
                // Bound queued finalized history across BOTH the finalizer and
                // consumer channels. Yielding alone does not apply backpressure.
                // Do not await the unfinalized tail: it can require live consensus
                // to finalize, and startup must be allowed to reach that phase.
                tokio::select! {
                    biased;
                    () = self.consumer_monitor.wait_for_handled(end_index.min(replay_target)) => {},
                    () = commit_consumer.commit_sender.closed() => {
                        panic!("consensus commit consumer closed during startup replay");
                    }
                }
            }
        }

        assert_eq!(
            last_sent_commit_index, last_commit_index,
            "We should have sent all commits up to the last commit {}",
            last_commit_index
        );

        info!(
            "Commit observer recovery [{}..={}] completed, took {:?}",
            replay_after_commit_index + 1,
            last_commit_index,
            now.elapsed()
        );
    }

    /// Reports per-commit metrics and logs the commit. Called for every commit on the
    /// legacy path via `report_metrics`, and directly by `Core::post_commit` on the v3 path.
    pub(crate) fn report_commit_metrics(&self, commit: &CommittedSubDag) {
        let metrics = &self.context.metrics.node_metrics;
        let utc_now = self.context.clock.timestamp_utc_ms();

        info!(
            "Consensus commit {} with leader {} has {} blocks",
            commit.commit_ref,
            commit.leader,
            commit.blocks.len()
        );

        metrics
            .last_committed_leader_round
            .set(commit.leader.round as i64);
        metrics
            .last_commit_index
            .set(commit.commit_ref.index as i64);
        metrics
            .blocks_per_commit_count
            .observe(commit.blocks.len() as f64);

        for block in &commit.blocks {
            let latency_ms = utc_now.saturating_sub(block.timestamp_ms());
            metrics
                .block_commit_latency
                .observe(Duration::from_millis(latency_ms).as_secs_f64());
            if block.author() == self.context.own_index {
                metrics
                    .proposed_block_commit_latency
                    .observe(Duration::from_millis(latency_ms).as_secs_f64());
            }
        }
    }

    fn report_metrics(&self, committed: &[CommittedSubDag]) {
        for commit in committed {
            self.report_commit_metrics(commit);
        }
        // Only the legacy path batches multiple subdags per call. The v3 path
        // handles one commit at a time, so this metric would always observe 1
        // and is intentionally not reported there.
        self.context
            .metrics
            .node_metrics
            .sub_dags_per_commit_count
            .observe(committed.len() as f64);
    }
}

#[cfg(test)]
mod tests {
    use consensus_types::block::BlockRef;
    use mysten_metrics::monitored_mpsc::UnboundedReceiver;
    use parking_lot::RwLock;
    use rstest::rstest;
    use tokio::time::timeout;

    use super::*;
    use crate::{
        CommitIndex, block_verifier::NoopBlockVerifier, context::Context, dag_state::DagState,
        linearizer::median_timestamp_by_stake, storage::mem_store::MemStore,
        test_dag_builder::DagBuilder,
    };

    fn full_replay_fixture(
        commits: u32,
        finalized: u32,
        voting: bool,
    ) -> (Arc<Context>, Arc<RwLock<DagState>>, TransactionVoteTracker) {
        let mut config = consensus_config::ConsensusProtocolConfig::for_testing();
        config.set_transaction_voting_enabled_for_testing(voting);
        let context = Arc::new(Context::new_for_test(4).0.with_protocol_config(config));
        let store = Arc::new(MemStore::new());
        if commits > 0 {
            let mut builder = DagBuilder::new(context.clone());
            builder.layers(1..=(commits * 2 + 2)).build();
            let produced = builder.get_sub_dag_and_commits(1..=commits);
            assert!(produced.len() >= commits as usize);
            for (index, (subdag, commit)) in produced.into_iter().take(commits as usize).enumerate()
            {
                let finalized_rows = if (index as u32) < finalized {
                    vec![(commit.reference(), std::collections::BTreeMap::new())]
                } else {
                    Vec::new()
                };
                store
                    .write(crate::storage::WriteBatch::new(
                        subdag.blocks,
                        vec![commit],
                        Vec::new(),
                        finalized_rows,
                    ))
                    .unwrap();
            }
        }
        let dag_state = Arc::new(RwLock::new(DagState::new(context.clone(), store)));
        let tracker = TransactionVoteTracker::new(
            context.clone(),
            Arc::new(NoopBlockVerifier {}),
            dag_state.clone(),
        );
        (context, dag_state, tracker)
    }

    #[tokio::test]
    async fn supplied_watermark_replay_keeps_its_target_without_waiting_for_acknowledgements() {
        let (context, dag_state, tracker) = full_replay_fixture(7, 7, true);
        let (args, mut receiver) = CommitConsumerArgs::new(2, 4);
        let monitor = args.monitor();
        assert_eq!(monitor.replay_target(), Some(4));
        assert_eq!(monitor.highest_handled_commit(), 2);
        let mut observer = timeout(
            Duration::from_secs(5),
            CommitObserver::new(context, args, dag_state, tracker),
        )
        .await
        .expect("legacy replay must not require application acknowledgements during startup");
        assert_eq!(monitor.replay_target(), Some(4));
        assert_eq!(monitor.progress().highest_committed_index, 7);
        assert_eq!(monitor.highest_handled_commit(), 2);
        for index in 3..=7 {
            let commit = timeout(Duration::from_secs(5), receiver.recv())
                .await
                .unwrap()
                .unwrap();
            assert_eq!(commit.commit_ref.index, index);
        }
        monitor.set_highest_handled_commit(3);
        assert!(
            timeout(
                Duration::from_millis(20),
                monitor.replay_to_consumer_last_processed_commit_complete(),
            )
            .await
            .is_err()
        );
        monitor.set_highest_handled_commit(4);
        timeout(
            Duration::from_secs(1),
            monitor.replay_to_consumer_last_processed_commit_complete(),
        )
        .await
        .expect("legacy readiness uses the supplied watermark, not the stored head");
        observer.stop().await;
    }

    #[tokio::test]
    async fn full_replay_is_paced_by_applied_commits_even_without_finalized_rows() {
        let (context, dag_state, tracker) = full_replay_fixture(7, 0, false);
        let stored_head_round = dag_state.read().last_commit_round();
        let (args, mut receiver) = CommitConsumerArgs::new_with_full_replay();
        let monitor = args.monitor();
        let recovery = tokio::spawn(CommitObserver::new(context, args, dag_state, tracker));
        for index in 1..=3 {
            let commit = timeout(Duration::from_secs(5), receiver.recv())
                .await
                .unwrap()
                .unwrap();
            assert_eq!(commit.commit_ref.index, index);
            assert_eq!(monitor.replay_target(), Some(7));
            assert_eq!(monitor.progress().highest_committed_index, 7);
            assert_eq!(
                monitor.progress().highest_committed_round,
                stored_head_round
            );
            if index < 3 {
                monitor.set_highest_handled_commit(index);
            }
        }
        // Receiving is not applying. Withhold the last acknowledgement and
        // prove a second storage batch cannot get queued behind this one.
        assert!(
            timeout(Duration::from_millis(100), receiver.recv())
                .await
                .is_err(),
            "recovery must wait for the consumer to apply the current batch"
        );
        assert!(!recovery.is_finished());
        monitor.set_highest_handled_commit(3);
        for index in 4..=7 {
            let commit = timeout(Duration::from_secs(5), receiver.recv())
                .await
                .unwrap()
                .unwrap();
            assert_eq!(commit.commit_ref.index, index);
            monitor.set_highest_handled_commit(index);
        }
        let mut observer = timeout(Duration::from_secs(5), recovery)
            .await
            .unwrap()
            .unwrap();
        monitor
            .replay_to_consumer_last_processed_commit_complete()
            .await;
        observer.stop().await;
    }

    #[rstest]
    #[case(0)]
    #[case(3)]
    #[tokio::test]
    async fn full_replay_does_not_wait_for_an_unfinalized_tail(#[case] finalized: u32) {
        let (context, dag_state, tracker) = full_replay_fixture(7, finalized, true);
        let (args, mut receiver) = CommitConsumerArgs::new_with_full_replay();
        let monitor = args.monitor();
        let recovery = tokio::spawn(CommitObserver::new(context, args, dag_state, tracker));
        for index in 1..=finalized {
            let commit = timeout(Duration::from_secs(5), receiver.recv())
                .await
                .unwrap()
                .unwrap();
            assert_eq!(commit.commit_ref.index, index);
            monitor.set_highest_handled_commit(index);
        }
        let mut observer = timeout(Duration::from_secs(5), recovery)
            .await
            .expect("unfinalized commits must not prevent live consensus startup")
            .unwrap();
        assert_eq!(monitor.replay_target(), Some(finalized));
        assert_eq!(monitor.highest_handled_commit(), finalized);
        monitor
            .replay_to_consumer_last_processed_commit_complete()
            .await;
        observer.stop().await;
    }

    #[tokio::test]
    async fn full_replay_of_an_empty_store_publishes_an_empty_target() {
        let (context, dag_state, tracker) = full_replay_fixture(0, 0, false);
        let (args, _receiver) = CommitConsumerArgs::new_with_full_replay();
        let monitor = args.monitor();
        assert_eq!(monitor.replay_target(), None);
        let mut observer = CommitObserver::new(context, args, dag_state, tracker).await;
        assert_eq!(monitor.replay_target(), Some(0));
        timeout(
            Duration::from_secs(1),
            monitor.replay_to_consumer_last_processed_commit_complete(),
        )
        .await
        .unwrap();
        observer.stop().await;
    }

    #[tokio::test]
    async fn a_closed_full_replay_consumer_does_not_leave_startup_waiting() {
        let (context, dag_state, tracker) = full_replay_fixture(7, 0, false);
        let (args, mut receiver) = CommitConsumerArgs::new_with_full_replay();
        let recovery = tokio::spawn(CommitObserver::new(context, args, dag_state, tracker));
        for _ in 0..3 {
            timeout(Duration::from_secs(5), receiver.recv())
                .await
                .unwrap()
                .unwrap();
        }
        drop(receiver);
        let error = timeout(Duration::from_secs(5), recovery)
            .await
            .unwrap()
            .err()
            .expect(
                "a vanished consumer must fail startup instead of waiting for its acknowledgement",
            );
        assert!(error.is_panic());
        assert!(
            error
                .to_string()
                .contains("consensus commit consumer closed during startup replay")
        );
    }

    #[tokio::test]
    #[should_panic(expected = "finalized replay target cannot exceed the consensus store head")]
    async fn a_finalized_target_beyond_stored_history_fails_replay() {
        let (context, dag_state, tracker) = full_replay_fixture(3, 3, true);
        let store = dag_state.read().store();
        let mut missing = store.read_last_commit().unwrap().unwrap().reference();
        missing.index += 1;
        store
            .write(crate::storage::WriteBatch::new(
                Vec::new(),
                Vec::new(),
                Vec::new(),
                vec![(missing, std::collections::BTreeMap::new())],
            ))
            .unwrap();
        let (args, _receiver) = CommitConsumerArgs::new_with_full_replay();
        CommitObserver::new(context, args, dag_state, tracker).await;
    }

    #[tokio::test]
    async fn a_finalization_hole_below_the_replay_target_fails_startup() {
        let (context, dag_state, tracker) = full_replay_fixture(5, 0, true);
        let store = dag_state.read().store();
        let finalized = store
            .scan_commits((1..=5).into())
            .unwrap()
            .into_iter()
            .filter(|commit| commit.index() != 3)
            .map(|commit| (commit.reference(), std::collections::BTreeMap::new()))
            .collect();
        store
            .write(crate::storage::WriteBatch::new(
                Vec::new(),
                Vec::new(),
                Vec::new(),
                finalized,
            ))
            .unwrap();
        let (args, mut receiver) = CommitConsumerArgs::new_with_full_replay();
        let monitor = args.monitor();
        let consume = tokio::spawn(async move {
            while let Some(commit) = receiver.recv().await {
                monitor.set_highest_handled_commit(commit.commit_ref.index);
            }
        });
        let recovery = tokio::spawn(CommitObserver::new(context, args, dag_state, tracker));
        let error = timeout(Duration::from_secs(5), recovery)
            .await
            .unwrap()
            .err()
            .expect("a hole in finalized history must fail startup");
        assert!(error.is_panic());
        assert!(
            error
                .to_string()
                .contains("missing rejected transactions below the finalized replay target")
        );
        consume.abort();
    }

    #[rstest]
    #[tokio::test]
    async fn test_handle_commit() {
        telemetry_subscribers::init_for_testing();
        let num_authorities = 4;
        let (context, _keys) = Context::new_for_test(num_authorities);
        let context = Arc::new(context);

        let mem_store = Arc::new(MemStore::new());
        let dag_state = Arc::new(RwLock::new(DagState::new(
            context.clone(),
            mem_store.clone(),
        )));
        let last_processed_commit_index = 0;
        let (commit_consumer, mut commit_receiver) =
            CommitConsumerArgs::new(0, last_processed_commit_index);
        let transaction_vote_tracker = TransactionVoteTracker::new(
            context.clone(),
            Arc::new(NoopBlockVerifier {}),
            dag_state.clone(),
        );

        let mut observer = CommitObserver::new(
            context.clone(),
            commit_consumer,
            dag_state.clone(),
            transaction_vote_tracker.clone(),
        )
        .await;

        // Populate fully connected test blocks for round 0 ~ 10, authorities 0 ~ 3.
        let num_rounds = 10;
        let mut builder = DagBuilder::new(context.clone());
        builder
            .layers(1..=num_rounds)
            .build()
            .persist_layers(dag_state.clone());
        transaction_vote_tracker.add_voted_blocks(
            builder
                .all_blocks()
                .iter()
                .map(|b| (b.clone(), vec![]))
                .collect(),
        );

        let leaders = builder
            .leader_blocks(1..=num_rounds)
            .into_iter()
            .map(Option::unwrap)
            .collect::<Vec<_>>();

        let commits = observer
            .handle_committed_leaders(leaders.clone(), true)
            .unwrap();
        let head = commits.last().unwrap();
        assert_eq!(
            observer.consumer_monitor.progress().highest_committed_index,
            head.commit_ref.index
        );
        assert_eq!(
            observer.consumer_monitor.progress().highest_committed_round,
            head.leader.round
        );
        assert_eq!(observer.consumer_monitor.highest_handled_commit(), 0);

        // Check commits are returned by CommitObserver::handle_committed_leaders is accurate
        let mut expected_stored_refs: Vec<BlockRef> = vec![];
        for (idx, subdag) in commits.iter().enumerate() {
            tracing::info!("{subdag:?}");
            assert_eq!(subdag.leader, leaders[idx].reference());

            let expected_ts = {
                let block_refs = leaders[idx]
                    .ancestors()
                    .iter()
                    .filter(|block_ref| block_ref.round == leaders[idx].round() - 1)
                    .cloned()
                    .collect::<Vec<_>>();
                let block_opts = dag_state.read().get_blocks(&block_refs);
                let blocks = block_opts.iter().map(|block_opt| {
                    block_opt
                        .as_ref()
                        .expect("We should have all blocks in dag state.")
                });
                median_timestamp_by_stake(&context, blocks).unwrap()
            };

            let expected_ts = if idx == 0 {
                expected_ts
            } else {
                expected_ts.max(commits[idx - 1].timestamp_ms)
            };

            assert_eq!(expected_ts, subdag.timestamp_ms);

            if idx == 0 {
                // First subdag includes the leader block plus all ancestor blocks
                // of the leader minus the genesis round blocks
                assert_eq!(subdag.blocks.len(), 1);
            } else {
                // Every subdag after will be missing the leader block from the previous
                // committed subdag
                assert_eq!(subdag.blocks.len(), num_authorities);
            }
            for block in subdag.blocks.iter() {
                expected_stored_refs.push(block.reference());
                assert!(block.round() <= leaders[idx].round());
            }
            assert_eq!(subdag.commit_ref.index, idx as CommitIndex + 1);
        }

        // Check commits sent over consensus output channel is accurate
        let mut processed_subdag_index = 0;
        while let Ok(Some(subdag)) = timeout(Duration::from_secs(1), commit_receiver.recv()).await {
            assert_eq!(subdag, commits[processed_subdag_index]);
            processed_subdag_index = subdag.commit_ref.index as usize;
            if processed_subdag_index == leaders.len() {
                break;
            }
        }
        assert_eq!(processed_subdag_index, leaders.len());

        // Own block latencies are observed once per committed & finalized block
        // authored by this authority.
        let own_committed_blocks = commits
            .iter()
            .flat_map(|commit| commit.blocks.iter())
            .filter(|block| block.author() == context.own_index)
            .count() as u64;
        assert!(own_committed_blocks > 0);
        assert_eq!(
            context
                .metrics
                .node_metrics
                .proposed_block_commit_latency
                .get_sample_count(),
            own_committed_blocks
        );
        assert_eq!(
            context
                .metrics
                .node_metrics
                .proposed_block_finalization_latency
                .get_sample_count(),
            own_committed_blocks
        );

        verify_channel_empty(&mut commit_receiver).await;

        // Check commits have been persisted to storage
        let last_commit = mem_store.read_last_commit().unwrap().unwrap();
        assert_eq!(
            last_commit.index(),
            commits.last().unwrap().commit_ref.index
        );
        let all_stored_commits = mem_store
            .scan_commits((0..=CommitIndex::MAX).into())
            .unwrap();
        assert_eq!(all_stored_commits.len(), leaders.len());
        let blocks_existence = mem_store.contains_blocks(&expected_stored_refs).unwrap();
        assert!(blocks_existence.iter().all(|exists| *exists));
    }

    #[tokio::test]
    async fn test_recover_and_send_commits() {
        telemetry_subscribers::init_for_testing();
        let num_authorities = 4;
        let context = Arc::new(Context::new_for_test(num_authorities).0);
        let mem_store = Arc::new(MemStore::new());
        let dag_state = Arc::new(RwLock::new(DagState::new(
            context.clone(),
            mem_store.clone(),
        )));
        let transaction_vote_tracker = TransactionVoteTracker::new(
            context.clone(),
            Arc::new(NoopBlockVerifier {}),
            dag_state.clone(),
        );
        let last_processed_commit_index = 0;
        let (commit_consumer, mut commit_receiver) =
            CommitConsumerArgs::new(0, last_processed_commit_index);

        let mut observer = CommitObserver::new(
            context.clone(),
            commit_consumer,
            dag_state.clone(),
            transaction_vote_tracker.clone(),
        )
        .await;

        // Populate fully connected test blocks for round 0 ~ 10, authorities 0 ~ 3.
        let num_rounds = 10;
        let mut builder = DagBuilder::new(context.clone());
        builder
            .layers(1..=num_rounds)
            .build()
            .persist_layers(dag_state.clone());
        transaction_vote_tracker.add_voted_blocks(
            builder
                .all_blocks()
                .iter()
                .map(|b| (b.clone(), vec![]))
                .collect(),
        );

        let leaders = builder
            .leader_blocks(1..=num_rounds)
            .into_iter()
            .map(Option::unwrap)
            .collect::<Vec<_>>();

        // Commit first batch of leaders (2) and "receive" the subdags as the
        // consumer of the consensus output channel.
        let expected_last_processed_index: usize = 2;
        let mut commits = observer
            .handle_committed_leaders(leaders[..expected_last_processed_index].to_vec(), true)
            .unwrap();

        // Check commits sent over consensus output channel is accurate
        let mut processed_subdag_index = 0;
        while let Ok(Some(subdag)) = timeout(Duration::from_secs(1), commit_receiver.recv()).await {
            tracing::info!("Processed {subdag}");
            assert_eq!(subdag, commits[processed_subdag_index]);
            processed_subdag_index = subdag.commit_ref.index as usize;
            if processed_subdag_index == expected_last_processed_index {
                break;
            }
        }
        assert_eq!(processed_subdag_index, expected_last_processed_index);

        verify_channel_empty(&mut commit_receiver).await;

        // Check last stored commit is correct
        let last_commit = mem_store.read_last_commit().unwrap().unwrap();
        assert_eq!(
            last_commit.index(),
            expected_last_processed_index as CommitIndex
        );

        // Handle next batch of leaders (10 - 2 = 8), these will be sent by consensus but not
        // "processed" by consensus output channel. Simulating something happened on
        // the consumer side where the commits were not persisted.
        commits.append(
            &mut observer
                .handle_committed_leaders(leaders[expected_last_processed_index..].to_vec(), true)
                .unwrap(),
        );

        let expected_last_sent_index = num_rounds as usize;
        while let Ok(Some(subdag)) = timeout(Duration::from_secs(1), commit_receiver.recv()).await {
            tracing::info!("{subdag} was sent but not processed by consumer");
            assert_eq!(subdag, commits[processed_subdag_index]);
            assert!(subdag.decided_with_local_blocks);
            processed_subdag_index = subdag.commit_ref.index as usize;
            if processed_subdag_index == expected_last_sent_index {
                break;
            }
        }
        assert_eq!(processed_subdag_index, expected_last_sent_index);

        verify_channel_empty(&mut commit_receiver).await;

        // Check last stored commit is correct. We should persist the last commit
        // that was sent over the channel regardless of how the consumer handled
        // the commit on their end.
        let last_commit = mem_store.read_last_commit().unwrap().unwrap();
        assert_eq!(last_commit.index(), expected_last_sent_index as CommitIndex);

        // Replay commits after index 2. And use last processed index by consumer at 10, which is the last persisted commit.
        {
            let replay_after_commit_index = 2;
            let consumer_last_processed_commit_index = 10;
            let dag_state = Arc::new(RwLock::new(DagState::new(
                context.clone(),
                mem_store.clone(),
            )));
            let (commit_consumer, mut commit_receiver) = CommitConsumerArgs::new(
                replay_after_commit_index,
                consumer_last_processed_commit_index,
            );
            let _observer = CommitObserver::new(
                context.clone(),
                commit_consumer,
                dag_state.clone(),
                transaction_vote_tracker.clone(),
            )
            .await;

            let mut processed_subdag_index = replay_after_commit_index;
            while let Ok(Some(mut subdag)) =
                timeout(Duration::from_secs(1), commit_receiver.recv()).await
            {
                tracing::info!("Received {subdag} on recovery");
                assert_eq!(subdag.commit_ref.index, processed_subdag_index + 1);
                assert!(subdag.recovered_rejected_transactions);

                // Allow comparison with committed subdag before recovery.
                subdag.recovered_rejected_transactions = false;
                assert_eq!(subdag, commits[processed_subdag_index as usize]);

                assert!(subdag.decided_with_local_blocks);
                processed_subdag_index = subdag.commit_ref.index;
                if processed_subdag_index == consumer_last_processed_commit_index {
                    break;
                }
            }
            assert_eq!(processed_subdag_index, consumer_last_processed_commit_index);

            verify_channel_empty(&mut commit_receiver).await;
        }

        // Replay commits from index 10, which is the last persisted commit.
        {
            let replay_after_commit_index = 10;
            let consumer_last_processed_commit_index = 10;
            let dag_state = Arc::new(RwLock::new(DagState::new(
                context.clone(),
                mem_store.clone(),
            )));
            // Re-create commit observer starting after index 10 which represents the
            // last processed index from the consumer over consensus output channel
            let (commit_consumer, mut commit_receiver) = CommitConsumerArgs::new(
                replay_after_commit_index,
                consumer_last_processed_commit_index,
            );
            let _observer = CommitObserver::new(
                context.clone(),
                commit_consumer,
                dag_state.clone(),
                transaction_vote_tracker.clone(),
            )
            .await;

            // No commits should be resubmitted as consensus store's last commit index
            // is equal to replay after index by consumer
            verify_channel_empty(&mut commit_receiver).await;
        }

        // Replay commits after index 2. And use last processed index by consumer at 4, less than the last persisted commit.
        {
            let replay_after_commit_index = 2;
            let consumer_last_processed_commit_index = 4;
            let dag_state = Arc::new(RwLock::new(DagState::new(
                context.clone(),
                mem_store.clone(),
            )));
            let (commit_consumer, mut commit_receiver) = CommitConsumerArgs::new(
                replay_after_commit_index,
                consumer_last_processed_commit_index,
            );
            let _observer = CommitObserver::new(
                context.clone(),
                commit_consumer,
                dag_state.clone(),
                transaction_vote_tracker.clone(),
            )
            .await;

            // Checks that commits up to expected_last_sent_index are recovered as finalized.
            // The fact that they are finalized and have been recorded in the store.
            let mut processed_subdag_index = replay_after_commit_index;
            while let Ok(Some(subdag)) =
                timeout(Duration::from_secs(1), commit_receiver.recv()).await
            {
                tracing::info!("Received {subdag} on recovery");
                assert_eq!(subdag.commit_ref.index, processed_subdag_index + 1);
                assert!(subdag.decided_with_local_blocks);
                processed_subdag_index = subdag.commit_ref.index;
                if processed_subdag_index == expected_last_sent_index as CommitIndex {
                    break;
                }
            }
            assert_eq!(
                processed_subdag_index,
                expected_last_sent_index as CommitIndex
            );

            verify_channel_empty(&mut commit_receiver).await;
        }

        // Replay commits after index 2. And use last processed index by consumer at 20,
        // which is greater than the last persisted commit.
        // This allows removing from the store a suffix of commits which have not been part of certified checkpoints.
        {
            let replay_after_commit_index = 2;
            let consumer_last_processed_commit_index = 20;
            let dag_state = Arc::new(RwLock::new(DagState::new(
                context.clone(),
                mem_store.clone(),
            )));
            let (commit_consumer, mut commit_receiver) = CommitConsumerArgs::new(
                replay_after_commit_index,
                consumer_last_processed_commit_index,
            );
            let _observer = CommitObserver::new(
                context.clone(),
                commit_consumer,
                dag_state.clone(),
                transaction_vote_tracker.clone(),
            )
            .await;

            // Check commits sent over consensus output channel is accurate starting
            // from last processed index of 2 and finishing at last sent index of 10.
            let mut processed_subdag_index = replay_after_commit_index;
            while let Ok(Some(mut subdag)) =
                timeout(Duration::from_secs(1), commit_receiver.recv()).await
            {
                tracing::info!("Received {subdag} on recovery");
                assert_eq!(subdag.commit_ref.index, processed_subdag_index + 1);
                assert!(subdag.recovered_rejected_transactions);

                // Allow comparison with committed subdag before recovery.
                subdag.recovered_rejected_transactions = false;
                assert_eq!(subdag, commits[processed_subdag_index as usize]);

                assert!(subdag.decided_with_local_blocks);
                processed_subdag_index = subdag.commit_ref.index;
                if processed_subdag_index == expected_last_sent_index as CommitIndex {
                    break;
                }
            }
            assert_eq!(
                processed_subdag_index,
                expected_last_sent_index as CommitIndex
            );
            assert_eq!(10, expected_last_sent_index);

            verify_channel_empty(&mut commit_receiver).await;
        }
    }

    /// After receiving all expected subdags, ensure channel is empty
    async fn verify_channel_empty(receiver: &mut UnboundedReceiver<CommittedSubDag>) {
        if let Ok(Some(_)) = timeout(Duration::from_secs(1), receiver.recv()).await {
            panic!("Expected the consensus output channel to be empty, but found more subdags.")
        }
    }
}