consensus_core/

commit_syncer.rs

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

//! CommitSyncer implements efficient synchronization of committed data.
//!
//! During the operation of a committee of authorities for consensus, one or more authorities
//! can fall behind the quorum in their received and accepted blocks. This can happen due to
//! network disruptions, host crash, or other reasons. Authorities fell behind need to catch up to
//! the quorum to be able to vote on the latest leaders. So efficient synchronization is necessary
//! to minimize the impact of temporary disruptions and maintain smooth operations of the network.
//!  
//! CommitSyncer achieves efficient synchronization by relying on the following: when blocks
//! are included in commits with >= 2f+1 certifiers by stake, these blocks must have passed
//! verifications on some honest validators, so re-verifying them is unnecessary. In fact, the
//! quorum certified commits themselves can be trusted to be sent to Sui directly, but for
//! simplicity this is not done. Blocks from trusted commits still go through Core and committer.
//!
//! Another way CommitSyncer improves the efficiency of synchronization is parallel fetching:
//! commits have a simple dependency graph (linear), so it is easy to fetch ranges of commits
//! in parallel.
//!
//! Commit synchronization is an expensive operation, involving transferring large amount of data via
//! the network. And it is not on the critical path of block processing. So the heuristics for
//! synchronization, including triggers and retries, should be chosen to favor throughput and
//! efficient resource usage, over faster reactions.

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

use bytes::Bytes;
use consensus_types::block::{BlockRef, TransactionIndex};
use futures::{StreamExt as _, stream::FuturesOrdered};
use itertools::Itertools as _;
use mysten_common::ZipDebugEqIteratorExt;
use mysten_metrics::spawn_logged_monitored_task;
use parking_lot::RwLock;
use rand::{prelude::SliceRandom as _, rngs::ThreadRng};
use tokio::{
    sync::oneshot,
    task::{JoinHandle, JoinSet},
    time::{MissedTickBehavior, sleep},
};
use tracing::{debug, info, warn};

use crate::{
    CommitConsumerMonitor, CommitIndex,
    block::{BlockAPI, ExtendedBlock, SignedBlock, VerifiedBlock},
    block_verifier::BlockVerifier,
    commit::{
        CertifiedCommit, CertifiedCommits, Commit, CommitAPI as _, CommitDigest, CommitRange,
        CommitRef, TrustedCommit,
    },
    commit_vote_monitor::CommitVoteMonitor,
    context::Context,
    core_thread::CoreThreadDispatcher,
    dag_state::DagState,
    error::{ConsensusError, ConsensusResult},
    network::{CommitSyncerClient, ObserverNetworkClient, PeerId, ValidatorNetworkClient},
    peers_pool::PeersPool,
    round_tracker::RoundTracker,
    stake_aggregator::{QuorumThreshold, StakeAggregator},
    task::{join_and_propagate_panic, shutdown_join_set, spawn_blocking},
    transaction_vote_tracker::TransactionVoteTracker,
};

// Handle to stop the CommitSyncer loop.
pub(crate) struct CommitSyncerHandle {
    schedule_task: JoinHandle<()>,
    tx_shutdown: oneshot::Sender<()>,
}

impl CommitSyncerHandle {
    pub(crate) async fn stop(self) {
        let _ = self.tx_shutdown.send(());
        // Do not abort schedule task, which waits for fetches to shut down.
        join_and_propagate_panic(self.schedule_task).await;
    }
}

pub(crate) struct CommitSyncer<VC: ValidatorNetworkClient, OC: ObserverNetworkClient> {
    // States shared by scheduler and fetch tasks.

    // Shared components wrapper.
    inner: Arc<Inner<VC, OC>>,

    // States only used by the scheduler.

    // Inflight requests to fetch commits from different authorities.
    inflight_fetches: JoinSet<(u32, CertifiedCommits)>,
    // Additional ranges of commits to fetch.
    pending_fetches: BTreeSet<CommitRange>,
    // Fetched commits and blocks by commit range.
    fetched_ranges: BTreeMap<CommitRange, CertifiedCommits>,
    // Highest commit index among inflight and pending fetches.
    // Used to determine the start of new ranges to be fetched.
    highest_scheduled_index: Option<CommitIndex>,
    // Highest index among fetched commits, after commits and blocks are verified.
    // Used for metrics.
    highest_fetched_commit_index: CommitIndex,
    // The commit index that is the max of highest local commit index and commit index inflight to Core.
    // Used to determine if fetched blocks can be sent to Core without gaps.
    synced_commit_index: CommitIndex,
}

impl<VC, OC> CommitSyncer<VC, OC>
where
    VC: ValidatorNetworkClient,
    OC: ObserverNetworkClient,
{
    pub(crate) fn new(
        context: Arc<Context>,
        core_thread_dispatcher: Arc<dyn CoreThreadDispatcher>,
        commit_vote_monitor: Arc<CommitVoteMonitor>,
        commit_consumer_monitor: Arc<CommitConsumerMonitor>,
        block_verifier: Arc<dyn BlockVerifier>,
        transaction_vote_tracker: TransactionVoteTracker,
        round_tracker: Arc<RwLock<RoundTracker>>,
        network_client: Arc<CommitSyncerClient<VC, OC>>,
        dag_state: Arc<RwLock<DagState>>,
        peers_pool: Arc<PeersPool>,
    ) -> Self {
        let inner = Arc::new(Inner {
            context,
            core_thread_dispatcher,
            commit_vote_monitor,
            commit_consumer_monitor,
            block_verifier,
            transaction_vote_tracker,
            round_tracker,
            network_client,
            dag_state,
            peers_pool,
        });
        let synced_commit_index = inner.dag_state.read().last_commit_index();
        CommitSyncer {
            inner,
            inflight_fetches: JoinSet::new(),
            pending_fetches: BTreeSet::new(),
            fetched_ranges: BTreeMap::new(),
            highest_scheduled_index: None,
            highest_fetched_commit_index: 0,
            synced_commit_index,
        }
    }

    pub(crate) fn start(self) -> CommitSyncerHandle {
        let (tx_shutdown, rx_shutdown) = oneshot::channel();
        let schedule_task = spawn_logged_monitored_task!(self.schedule_loop(rx_shutdown,));
        CommitSyncerHandle {
            schedule_task,
            tx_shutdown,
        }
    }

    async fn schedule_loop(mut self, mut rx_shutdown: oneshot::Receiver<()>) {
        let mut interval = tokio::time::interval(Duration::from_secs(2));
        interval.set_missed_tick_behavior(MissedTickBehavior::Skip);

        loop {
            tokio::select! {
                // Periodically, schedule new fetches if the node is falling behind.
                _ = interval.tick() => {
                    self.try_schedule_once();
                }
                // Handles results from fetch tasks.
                Some(result) = self.inflight_fetches.join_next(), if !self.inflight_fetches.is_empty() => {
                    if let Err(e) = result {
                        if e.is_panic() {
                            std::panic::resume_unwind(e.into_panic());
                        }
                        warn!("Fetch cancelled. CommitSyncer shutting down: {}", e);
                        // If any fetch is cancelled or panicked, try to shutdown and exit the loop.
                        shutdown_join_set(&mut self.inflight_fetches).await;
                        return;
                    }
                    let (target_end, commits) = result.unwrap();
                    self.handle_fetch_result(target_end, commits).await;
                }
                _ = &mut rx_shutdown => {
                    // Shutdown requested.
                    info!("CommitSyncer shutting down ...");
                    shutdown_join_set(&mut self.inflight_fetches).await;
                    return;
                }
            }

            self.try_start_fetches();
        }
    }

    fn try_schedule_once(&mut self) {
        let quorum_commit_index = self.inner.commit_vote_monitor.quorum_commit_index();
        let local_commit_index = self.inner.dag_state.read().last_commit_index();
        let metrics = &self.inner.context.metrics.node_metrics;
        metrics
            .commit_sync_quorum_index
            .set(quorum_commit_index as i64);
        metrics
            .commit_sync_local_index
            .set(local_commit_index as i64);
        let highest_handled_index = self.inner.commit_consumer_monitor.highest_handled_commit();
        let highest_scheduled_index = self.highest_scheduled_index.unwrap_or(0);
        // Update synced_commit_index periodically to make sure it is no smaller than
        // local commit index.
        self.synced_commit_index = self.synced_commit_index.max(local_commit_index);
        let unhandled_commits_threshold = self.unhandled_commits_threshold();
        info!(
            "Checking to schedule fetches: synced_commit_index={}, highest_handled_index={}, highest_scheduled_index={}, quorum_commit_index={}, unhandled_commits_threshold={}",
            self.synced_commit_index,
            highest_handled_index,
            highest_scheduled_index,
            quorum_commit_index,
            unhandled_commits_threshold,
        );

        // TODO: cleanup inflight fetches that are no longer needed.
        let fetch_after_index = self
            .synced_commit_index
            .max(self.highest_scheduled_index.unwrap_or(0));
        // When the node is falling behind, schedule pending fetches which will be executed on later.
        for prev_end in (fetch_after_index..=quorum_commit_index)
            .step_by(self.inner.context.parameters.commit_sync_batch_size as usize)
        {
            // Create range with inclusive start and end.
            let range_start = prev_end + 1;
            let range_end = prev_end + self.inner.context.parameters.commit_sync_batch_size;
            // Commit range is not fetched when [range_start, range_end] contains less number of commits
            // than the target batch size. This is to avoid the cost of processing more and smaller batches.
            // Block broadcast, subscription and synchronization will help the node catchup.
            if quorum_commit_index < range_end {
                break;
            }
            // Pause scheduling new fetches when handling of commits is lagging.
            if highest_handled_index + unhandled_commits_threshold < range_end {
                warn!(
                    "Skip scheduling new commit fetches: consensus handler is lagging. highest_handled_index={}, highest_scheduled_index={}",
                    highest_handled_index, highest_scheduled_index
                );
                break;
            }
            self.pending_fetches
                .insert((range_start..=range_end).into());
            // quorum_commit_index should be non-decreasing, so highest_scheduled_index should not
            // decrease either.
            self.highest_scheduled_index = Some(range_end);
        }
    }

    async fn handle_fetch_result(
        &mut self,
        target_end: CommitIndex,
        certified_commits: CertifiedCommits,
    ) {
        assert!(!certified_commits.commits().is_empty());

        let (total_blocks_fetched, total_blocks_size_bytes) = certified_commits
            .commits()
            .iter()
            .fold((0, 0), |(blocks, bytes), c| {
                (
                    blocks + c.blocks().len(),
                    bytes
                        + c.blocks()
                            .iter()
                            .map(|b| b.serialized().len())
                            .sum::<usize>() as u64,
                )
            });

        let metrics = &self.inner.context.metrics.node_metrics;
        metrics
            .commit_sync_fetched_commits
            .inc_by(certified_commits.commits().len() as u64);
        metrics
            .commit_sync_fetched_blocks
            .inc_by(total_blocks_fetched as u64);
        metrics
            .commit_sync_total_fetched_blocks_size
            .inc_by(total_blocks_size_bytes);

        let (commit_start, commit_end) = (
            certified_commits.commits().first().unwrap().index(),
            certified_commits.commits().last().unwrap().index(),
        );
        self.highest_fetched_commit_index = self.highest_fetched_commit_index.max(commit_end);
        metrics
            .commit_sync_highest_fetched_index
            .set(self.highest_fetched_commit_index as i64);

        // Allow returning partial results, and try fetching the rest separately.
        if commit_end < target_end {
            self.pending_fetches
                .insert((commit_end + 1..=target_end).into());
        }
        // Make sure synced_commit_index is up to date.
        self.synced_commit_index = self
            .synced_commit_index
            .max(self.inner.dag_state.read().last_commit_index());
        // Only add new blocks if at least some of them are not already synced.
        if self.synced_commit_index < commit_end {
            self.fetched_ranges
                .insert((commit_start..=commit_end).into(), certified_commits);
        }
        // Try to process as many fetched blocks as possible.
        while let Some((fetched_commit_range, _commits)) = self.fetched_ranges.first_key_value() {
            // Only pop fetched_ranges if there is no gap with blocks already synced.
            // Note: start, end and synced_commit_index are all inclusive.
            let (fetched_commit_range, commits) =
                if fetched_commit_range.start() <= self.synced_commit_index + 1 {
                    self.fetched_ranges.pop_first().unwrap()
                } else {
                    // Found gap between earliest fetched block and latest synced block,
                    // so not sending additional blocks to Core.
                    metrics.commit_sync_gap_on_processing.inc();
                    break;
                };
            // Avoid sending to Core a whole batch of already synced blocks.
            if fetched_commit_range.end() <= self.synced_commit_index {
                continue;
            }

            debug!(
                "Fetched blocks for commit range {:?}: {}",
                fetched_commit_range,
                commits
                    .commits()
                    .iter()
                    .flat_map(|c| c.blocks())
                    .map(|b| b.reference().to_string())
                    .join(","),
            );

            // If core thread cannot handle the incoming blocks, it is ok to block here
            // to slow down the commit syncer.
            match self
                .inner
                .core_thread_dispatcher
                .add_certified_commits(commits)
                .await
            {
                // Missing ancestors are possible from certification blocks, but
                // it is unnecessary to try to sync their causal history. If they are required
                // for the progress of the DAG, they will be included in a future commit.
                Ok(missing) => {
                    if !missing.is_empty() {
                        info!(
                            "Certification blocks have missing ancestors: {} for commit range {:?}",
                            missing.iter().map(|b| b.to_string()).join(","),
                            fetched_commit_range,
                        );
                    }
                    for block_ref in missing {
                        let hostname = &self
                            .inner
                            .context
                            .committee
                            .authority(block_ref.author)
                            .hostname;
                        metrics
                            .commit_sync_fetch_missing_blocks
                            .with_label_values(&[hostname])
                            .inc();
                    }
                }
                Err(e) => {
                    info!("Failed to add blocks, shutting down: {}", e);
                    return;
                }
            };

            // Once commits and blocks are sent to Core, ratchet up synced_commit_index
            self.synced_commit_index = self.synced_commit_index.max(fetched_commit_range.end());
        }

        metrics
            .commit_sync_inflight_fetches
            .set(self.inflight_fetches.len() as i64);
        metrics
            .commit_sync_pending_fetches
            .set(self.pending_fetches.len() as i64);
        metrics
            .commit_sync_highest_synced_index
            .set(self.synced_commit_index as i64);
    }

    fn try_start_fetches(&mut self) {
        // Cap parallel fetches based on configured limit and known peers, to avoid overloading the network.
        // Also when there are too many fetched blocks that cannot be sent to Core before an earlier fetch
        // has not finished, reduce parallelism so the earlier fetch can retry on a better host and succeed.
        // For validators, use committee size for the calculation. For observers, don't apply the 2/3 limit.
        let known_peers_count = self.inner.peers_pool.get_known_peers().len();
        let target_parallel_fetches = if self.inner.context.is_validator() {
            self.inner
                .context
                .parameters
                .commit_sync_parallel_fetches
                .min(known_peers_count * 2 / 3)
                .min(
                    self.inner
                        .context
                        .parameters
                        .commit_sync_batches_ahead
                        .saturating_sub(self.fetched_ranges.len()),
                )
                .max(1)
        } else {
            // For observers, currently the node is probably connected only to another peer. In the future probably more observer peers might be available
            // to sync from. That's why we do not cap the number of parallel fetches by the number of known peers.
            self.inner
                .context
                .parameters
                .commit_sync_parallel_fetches
                .min(
                    self.inner
                        .context
                        .parameters
                        .commit_sync_batches_ahead
                        .saturating_sub(self.fetched_ranges.len()),
                )
                .max(1)
        };
        // Start new fetches if there are pending batches and available slots.
        loop {
            if self.inflight_fetches.len() >= target_parallel_fetches {
                break;
            }
            let Some(commit_range) = self.pending_fetches.pop_first() else {
                break;
            };
            self.inflight_fetches
                .spawn(Self::fetch_loop(self.inner.clone(), commit_range));
        }

        let metrics = &self.inner.context.metrics.node_metrics;
        metrics
            .commit_sync_inflight_fetches
            .set(self.inflight_fetches.len() as i64);
        metrics
            .commit_sync_pending_fetches
            .set(self.pending_fetches.len() as i64);
        metrics
            .commit_sync_highest_synced_index
            .set(self.synced_commit_index as i64);
    }

    // Retries fetching commits and blocks from available authorities, until a request succeeds
    // where at least a prefix of the commit range is fetched.
    // Returns the fetched commits and blocks referenced by the commits.
    async fn fetch_loop(
        inner: Arc<Inner<VC, OC>>,
        commit_range: CommitRange,
    ) -> (CommitIndex, CertifiedCommits) {
        let base_timeout = inner.context.parameters.commit_sync_request_timeout;
        // Max per-request timeout will be base timeout times a multiplier.
        // At the extreme, this means there will be 120s timeout to fetch max_blocks_per_fetch blocks.
        const MAX_TIMEOUT_MULTIPLIER: u32 = 12;
        // timeout * max number of targets should be reasonably small, so the
        // system can adjust to slow network or large data sizes quickly.
        const MAX_NUM_TARGETS: usize = 24;
        let mut timeout_multiplier = 0;
        let _timer = inner
            .context
            .metrics
            .node_metrics
            .commit_sync_fetch_loop_latency
            .start_timer();
        info!("Starting to fetch commits in {commit_range:?} ...",);
        loop {
            // Attempt to fetch commits and blocks through min(available peers count, MAX_NUM_TARGETS) peers.
            let mut target_peers = inner.peers_pool.get_known_peers();
            target_peers.shuffle(&mut ThreadRng::default());
            target_peers.truncate(MAX_NUM_TARGETS);
            // Increase timeout multiplier for each loop until MAX_TIMEOUT_MULTIPLIER.
            timeout_multiplier = (timeout_multiplier + 1).min(MAX_TIMEOUT_MULTIPLIER);
            let request_timeout = base_timeout * timeout_multiplier;
            // Give enough overall timeout for fetching commits and blocks.
            // - Timeout for fetching commits and commit certifying blocks.
            // - Timeout for fetching blocks referenced by the commits.
            // - Time spent on pipelining requests to fetch blocks.
            // - Another headroom to allow fetch_once() to timeout gracefully if possible.
            let fetch_timeout = request_timeout * 4;
            // Try fetching from selected target peers.
            for peer in target_peers {
                match tokio::time::timeout(
                    fetch_timeout,
                    Self::fetch_once(
                        inner.clone(),
                        peer.clone(),
                        commit_range.clone(),
                        request_timeout,
                    ),
                )
                .await
                {
                    Ok(Ok(commits)) => {
                        info!("Finished fetching commits in {commit_range:?}",);
                        return (commit_range.end(), commits);
                    }
                    Ok(Err(e)) => {
                        warn!(
                            "Failed to fetch {commit_range:?} from {}: {}",
                            peer.hostname(&inner.context),
                            e
                        );
                        inner
                            .context
                            .metrics
                            .node_metrics
                            .commit_sync_fetch_once_errors
                            .with_label_values(&[peer.labelname(&inner.context).as_str(), e.name()])
                            .inc();
                    }
                    Err(_) => {
                        warn!(
                            "Timed out fetching {commit_range:?} from {}",
                            peer.hostname(&inner.context)
                        );
                        inner
                            .context
                            .metrics
                            .node_metrics
                            .commit_sync_fetch_once_errors
                            .with_label_values(&[
                                peer.labelname(&inner.context).as_str(),
                                "FetchTimeout",
                            ])
                            .inc();
                    }
                }
            }
            // Avoid busy looping, by waiting for a while before retrying.
            sleep(base_timeout).await;
        }
    }

    // Fetches commits and blocks from a single peer. At a high level, first the commits are
    // fetched and verified. After that, blocks referenced in the certified commits are fetched
    // and sent to Core for processing.
    async fn fetch_once(
        inner: Arc<Inner<VC, OC>>,
        target_peer: PeerId,
        commit_range: CommitRange,
        timeout: Duration,
    ) -> ConsensusResult<CertifiedCommits> {
        let _timer = inner
            .context
            .metrics
            .node_metrics
            .commit_sync_fetch_once_latency
            .start_timer();

        // 0. Probe the target to check reachability before committing to the full fetch.
        // This skips unreachable and slow peers quickly.
        let probe_timeout = inner.context.parameters.commit_sync_probe_timeout;
        inner
            .network_client
            .probe_connectivity(target_peer.clone(), probe_timeout)
            .await?;

        // 1. Fetch commits in the commit range from the target authority.
        let (serialized_commits, serialized_blocks) = inner
            .network_client
            .fetch_commits(target_peer.clone(), commit_range.clone(), timeout)
            .await?;

        // 2. Verify the response contains blocks that can certify the last returned commit,
        // and the returned commits are chained by digests, so earlier commits are certified
        // as well.
        let (commits, commit_certifying_blocks_and_votes) = spawn_blocking({
            let context = inner.context.clone();
            let block_verifier = inner.block_verifier.clone();
            let peer = target_peer.clone();
            move || {
                Inner::<VC, OC>::verify_commits(
                    &context,
                    block_verifier.as_ref(),
                    peer,
                    commit_range,
                    serialized_commits,
                    serialized_blocks,
                )
            }
        })
        .await??;

        // Only the vote tracker needs the reject votes, so move them into it without cloning.
        // Cheap clones of the blocks are enough for the rest of the fetch handling.
        let commit_certifying_blocks: Vec<_> = commit_certifying_blocks_and_votes
            .iter()
            .map(|(block, _)| block.clone())
            .collect();
        if inner.context.protocol_config.transaction_voting_enabled() {
            inner
                .transaction_vote_tracker
                .add_voted_blocks(commit_certifying_blocks_and_votes);
        }

        // 3. Fetch blocks referenced by the commits, from the same peer where commits are fetched.
        let mut block_refs: Vec<_> = commits.iter().flat_map(|c| c.blocks()).cloned().collect();
        block_refs.sort();
        let num_chunks = block_refs
            .len()
            .div_ceil(inner.context.parameters.max_blocks_per_fetch)
            as u32;
        let mut requests: FuturesOrdered<_> = block_refs
            .chunks(inner.context.parameters.max_blocks_per_fetch)
            .enumerate()
            .map(|(i, request_block_refs)| {
                let inner = inner.clone();
                let peer = target_peer.clone();
                async move {
                    // 4. Send out pipelined fetch requests to avoid overloading the target authority.
                    sleep(timeout * i as u32 / num_chunks).await;
                    // TODO: add some retries.
                    let serialized_blocks = inner
                        .network_client
                        .fetch_blocks(
                            peer.clone(),
                            request_block_refs.to_vec(),
                            vec![],
                            false,
                            timeout,
                        )
                        .await?;
                    // 5. Verify the same number of blocks are returned as requested.
                    if request_block_refs.len() != serialized_blocks.len() {
                        return Err(ConsensusError::UnexpectedNumberOfBlocksFetched {
                            peer,
                            requested: request_block_refs.len(),
                            received: serialized_blocks.len(),
                        });
                    }
                    // 6. Verify returned blocks have valid formats.
                    let signed_blocks = serialized_blocks
                        .iter()
                        .map(|serialized| {
                            let block: SignedBlock = bcs::from_bytes(serialized)
                                .map_err(ConsensusError::MalformedBlock)?;
                            Ok(block)
                        })
                        .collect::<ConsensusResult<Vec<_>>>()?;
                    // 7. Verify the returned blocks match the requested block refs.
                    // If they do match, the returned blocks can be considered verified as well.
                    let mut blocks = Vec::new();
                    for ((requested_block_ref, signed_block), serialized) in request_block_refs
                        .iter()
                        .zip_debug_eq(signed_blocks)
                        .zip_debug_eq(serialized_blocks)
                    {
                        let signed_block_digest = VerifiedBlock::compute_digest(&serialized);
                        let received_block_ref = BlockRef::new(
                            signed_block.round(),
                            signed_block.author(),
                            signed_block_digest,
                        );
                        if *requested_block_ref != received_block_ref {
                            return Err(ConsensusError::UnexpectedBlockForCommit {
                                peer,
                                requested: *requested_block_ref,
                                received: received_block_ref,
                            });
                        }
                        blocks.push(VerifiedBlock::new_verified(signed_block, serialized));
                    }
                    Ok(blocks)
                }
            })
            .collect();

        let mut fetched_blocks = BTreeMap::new();
        while let Some(result) = requests.next().await {
            for block in result? {
                fetched_blocks.insert(block.reference(), block);
            }
        }

        // 8. Check if the block timestamps are lower than current time - this is for metrics only.
        for block in fetched_blocks
            .values()
            .chain(commit_certifying_blocks.iter())
        {
            let now_ms = inner.context.clock.timestamp_utc_ms();
            let forward_drift = block.timestamp_ms().saturating_sub(now_ms);
            if forward_drift == 0 {
                continue;
            };
            // Extract hostname based on peer type
            inner
                .context
                .metrics
                .node_metrics
                .block_timestamp_drift_ms
                .with_label_values(&[
                    target_peer.labelname(&inner.context).as_str(),
                    "commit_syncer",
                ])
                .inc_by(forward_drift);
        }

        // 9. Now create certified commits by assigning the blocks to each commit.
        let mut certified_commits = Vec::new();
        for commit in &commits {
            let blocks = commit
                .blocks()
                .iter()
                .map(|block_ref| {
                    fetched_blocks
                        .remove(block_ref)
                        .expect("Block should exist")
                })
                .collect::<Vec<_>>();
            certified_commits.push(CertifiedCommit::new_certified(commit.clone(), blocks));
        }

        // 10. Add blocks in certified commits to the transaction vote tracker.
        for commit in &certified_commits {
            for block in commit.blocks() {
                // Only account for reject votes in the block, since they may vote on uncommitted
                // blocks or transactions. It is unnecessary to vote on the committed blocks
                // themselves.
                if inner.context.protocol_config.transaction_voting_enabled() {
                    inner
                        .transaction_vote_tracker
                        .add_voted_blocks(vec![(block.clone(), vec![])]);
                }
            }
        }

        // 11. Record commit votes from the fetched blocks.
        for commit in &certified_commits {
            for block in commit.blocks() {
                inner.commit_vote_monitor.observe_block(block);
            }
        }
        for block in &commit_certifying_blocks {
            inner.commit_vote_monitor.observe_block(block);
        }

        // 12. Update round tracker from the fetched blocks. For fetched blocks,
        // excluded_ancestors are not available so we use an empty vector.
        {
            let mut tracker = inner.round_tracker.write();
            // Update from commit blocks
            for commit in &certified_commits {
                for block in commit.blocks() {
                    tracker.update_from_verified_block(&ExtendedBlock {
                        block: block.clone(),
                        excluded_ancestors: vec![],
                    });
                }
            }
            // Update from vote blocks
            for block in &commit_certifying_blocks {
                tracker.update_from_verified_block(&ExtendedBlock {
                    block: block.clone(),
                    excluded_ancestors: vec![],
                });
            }
        }

        Ok(CertifiedCommits::new(
            certified_commits,
            commit_certifying_blocks,
        ))
    }

    fn unhandled_commits_threshold(&self) -> CommitIndex {
        self.inner.context.parameters.commit_sync_batch_size
            * (self.inner.context.parameters.commit_sync_batches_ahead as u32)
    }

    #[cfg(test)]
    fn pending_fetches(&self) -> BTreeSet<CommitRange> {
        self.pending_fetches.clone()
    }

    #[cfg(test)]
    fn fetched_ranges(&self) -> BTreeMap<CommitRange, CertifiedCommits> {
        self.fetched_ranges.clone()
    }

    #[cfg(test)]
    fn highest_scheduled_index(&self) -> Option<CommitIndex> {
        self.highest_scheduled_index
    }

    #[cfg(test)]
    fn highest_fetched_commit_index(&self) -> CommitIndex {
        self.highest_fetched_commit_index
    }

    #[cfg(test)]
    fn synced_commit_index(&self) -> CommitIndex {
        self.synced_commit_index
    }
}

struct Inner<VC: ValidatorNetworkClient, OC: ObserverNetworkClient> {
    context: Arc<Context>,
    core_thread_dispatcher: Arc<dyn CoreThreadDispatcher>,
    commit_vote_monitor: Arc<CommitVoteMonitor>,
    commit_consumer_monitor: Arc<CommitConsumerMonitor>,
    block_verifier: Arc<dyn BlockVerifier>,
    transaction_vote_tracker: TransactionVoteTracker,
    round_tracker: Arc<RwLock<RoundTracker>>,
    network_client: Arc<CommitSyncerClient<VC, OC>>,
    dag_state: Arc<RwLock<DagState>>,
    peers_pool: Arc<PeersPool>,
}

impl<VC: ValidatorNetworkClient, OC: ObserverNetworkClient> Inner<VC, OC> {
    /// Verifies the commits and certifies them using the provided vote blocks for the last commit.
    /// Returns, in order:
    /// - the verified commit chain as trusted commits;
    /// - the verified blocks that certify the last commit, paired with locally rejected
    ///   transaction indices for transaction vote tracking.
    fn verify_commits(
        context: &Context,
        block_verifier: &dyn BlockVerifier,
        peer: PeerId,
        commit_range: CommitRange,
        serialized_commits: Vec<Bytes>,
        serialized_vote_blocks: Vec<Bytes>,
    ) -> ConsensusResult<(
        Vec<TrustedCommit>,
        Vec<(VerifiedBlock, Vec<TransactionIndex>)>,
    )> {
        // Parse and verify commits.
        let mut commits = Vec::new();
        for serialized in &serialized_commits {
            let commit: Commit =
                bcs::from_bytes(serialized).map_err(ConsensusError::MalformedCommit)?;
            let digest = TrustedCommit::compute_digest(serialized);
            if commits.is_empty() {
                // start is inclusive, so first commit must be at the start index.
                if commit.index() != commit_range.start() {
                    return Err(ConsensusError::UnexpectedStartCommit {
                        peer,
                        start: commit_range.start(),
                        commit: Box::new(commit),
                    });
                }
            } else {
                // Verify next commit increments index and references the previous digest.
                let (last_commit_digest, last_commit): &(CommitDigest, Commit) =
                    commits.last().unwrap();
                if commit.index() != last_commit.index() + 1
                    || &commit.previous_digest() != last_commit_digest
                {
                    return Err(ConsensusError::UnexpectedCommitSequence {
                        peer,
                        prev_commit: Box::new(last_commit.clone()),
                        curr_commit: Box::new(commit),
                    });
                }
            }
            // Do not process more commits past the end index.
            if commit.index() > commit_range.end() {
                break;
            }
            commits.push((digest, commit));
        }
        let Some((end_commit_digest, end_commit)) = commits.last() else {
            return Err(ConsensusError::NoCommitReceived { peer });
        };

        // Parse and verify blocks. Then accumulate votes on the end commit.
        let end_commit_ref = CommitRef::new(end_commit.index(), *end_commit_digest);
        let mut stake_aggregator = StakeAggregator::<QuorumThreshold>::new();
        let mut commit_certifying_blocks = Vec::new();
        for serialized in serialized_vote_blocks {
            let block: SignedBlock =
                bcs::from_bytes(&serialized).map_err(ConsensusError::MalformedBlock)?;
            // Only block signatures need to be verified, to verify commit votes.
            // But the blocks will be sent to Core, so they need to be fully verified.
            let (block, reject_transaction_votes) =
                block_verifier.verify_and_vote(block, serialized)?;
            for vote in block.commit_votes() {
                if *vote == end_commit_ref {
                    stake_aggregator.add(block.author(), &context.committee);
                }
            }
            commit_certifying_blocks.push((block, reject_transaction_votes));
        }

        // Check if the end commit has enough votes.
        if !stake_aggregator.reached_threshold(&context.committee) {
            return Err(ConsensusError::NotEnoughCommitVotes {
                stake: stake_aggregator.stake(),
                peer,
                commit: Box::new(end_commit.clone()),
            });
        }

        let trusted_commits = commits
            .into_iter()
            .zip_debug_eq(serialized_commits)
            .map(|((_d, c), s)| TrustedCommit::new_trusted(c, s))
            .collect();
        Ok((trusted_commits, commit_certifying_blocks))
    }
}

#[cfg(test)]
mod tests {
    use std::{sync::Arc, time::Duration};

    use bytes::Bytes;
    use consensus_config::{AuthorityIndex, NetworkKeyPair, Parameters};
    use consensus_types::block::{BlockRef, Round};
    use mysten_common::ZipDebugEqIteratorExt;
    use parking_lot::RwLock;

    use crate::{
        CommitConsumerMonitor, CommitDigest, CommitRef,
        block::{TestBlock, VerifiedBlock},
        block_verifier::NoopBlockVerifier,
        commit::CommitRange,
        commit_syncer::CommitSyncer,
        commit_vote_monitor::CommitVoteMonitor,
        context::Context,
        core_thread::MockCoreThreadDispatcher,
        dag_state::DagState,
        error::ConsensusResult,
        network::{BlockStream, CommitSyncerClient, ObserverNetworkClient, ValidatorNetworkClient},
        peers_pool::{PeerService, PeersPool},
        round_tracker::RoundTracker,
        storage::mem_store::MemStore,
        transaction_vote_tracker::TransactionVoteTracker,
    };

    #[derive(Default)]
    struct FakeNetworkClient {}

    #[async_trait::async_trait]
    impl ValidatorNetworkClient for FakeNetworkClient {
        async fn subscribe_blocks(
            &self,
            _peer: AuthorityIndex,
            _last_received: Round,
            _timeout: Duration,
        ) -> ConsensusResult<BlockStream> {
            unimplemented!("Unimplemented")
        }

        async fn fetch_blocks(
            &self,
            _peer: AuthorityIndex,
            _block_refs: Vec<BlockRef>,
            _fetch_after_rounds: Vec<Round>,
            _fetch_missing_ancestors: bool,
            _timeout: Duration,
        ) -> ConsensusResult<Vec<Bytes>> {
            unimplemented!("Unimplemented")
        }

        async fn fetch_commits(
            &self,
            _peer: AuthorityIndex,
            _commit_range: CommitRange,
            _timeout: Duration,
        ) -> ConsensusResult<(Vec<Bytes>, Vec<Bytes>)> {
            unimplemented!("Unimplemented")
        }

        async fn fetch_latest_blocks(
            &self,
            _peer: AuthorityIndex,
            _authorities: Vec<AuthorityIndex>,
            _timeout: Duration,
        ) -> ConsensusResult<Vec<Bytes>> {
            unimplemented!("Unimplemented")
        }

        async fn get_latest_rounds(
            &self,
            _peer: AuthorityIndex,
            _timeout: Duration,
        ) -> ConsensusResult<(Vec<Round>, Vec<Round>)> {
            unimplemented!("Unimplemented")
        }

        #[cfg(test)]
        async fn send_block(
            &self,
            _peer: AuthorityIndex,
            _block: &VerifiedBlock,
            _timeout: Duration,
        ) -> ConsensusResult<()> {
            unimplemented!("Unimplemented")
        }
    }

    #[async_trait::async_trait]
    impl ObserverNetworkClient for FakeNetworkClient {
        async fn stream_blocks(
            &self,
            _peer: crate::network::PeerId,
            _highest_round_per_authority: Vec<Round>,
            _timeout: Duration,
        ) -> ConsensusResult<crate::network::ObserverBlockStream> {
            unimplemented!("Unimplemented")
        }

        async fn fetch_blocks(
            &self,
            _peer: crate::network::PeerId,
            _block_refs: Vec<BlockRef>,
            _highest_accepted_rounds: Vec<Round>,
            _breadth_first: bool,
            _timeout: Duration,
        ) -> ConsensusResult<Vec<Bytes>> {
            unimplemented!("Unimplemented")
        }

        async fn fetch_commits(
            &self,
            _peer: crate::network::PeerId,
            _commit_range: CommitRange,
            _timeout: Duration,
        ) -> ConsensusResult<(Vec<Bytes>, Vec<Bytes>)> {
            unimplemented!("Unimplemented")
        }
    }

    #[tokio::test(flavor = "current_thread", start_paused = true)]
    async fn commit_syncer_observer_node_basic() {
        // Test basic Observer node behavior for commit syncing
        // An Observer node should be able to sync from both validator and observer peers

        // Create an Observer node context (own_index = MAX)
        let (mut context, _) = Context::new_for_test(4);
        context.own_index = AuthorityIndex::MAX; // Mark this as an observer node
        context.parameters = Parameters {
            commit_sync_batch_size: 5,
            commit_sync_batches_ahead: 10,
            commit_sync_parallel_fetches: 5,
            max_blocks_per_fetch: 5,
            ..context.parameters
        };
        let context = Arc::new(context);

        // Setup observer node commit syncer
        let block_verifier = Arc::new(NoopBlockVerifier {});
        let core_thread_dispatcher = Arc::new(MockCoreThreadDispatcher::default());
        let mock_client = Arc::new(FakeNetworkClient::default());
        let network_client = Arc::new(CommitSyncerClient::new(
            context.clone(),
            Some(mock_client.clone()),
            Some(mock_client.clone()),
        ));
        let store = Arc::new(MemStore::new());
        let dag_state = Arc::new(RwLock::new(DagState::new(context.clone(), store)));
        let transaction_vote_tracker =
            TransactionVoteTracker::new(context.clone(), block_verifier.clone(), dag_state.clone());
        let commit_vote_monitor = Arc::new(CommitVoteMonitor::new(context.clone()));
        let commit_consumer_monitor = Arc::new(CommitConsumerMonitor::new(0, 0));
        let round_tracker = Arc::new(RwLock::new(RoundTracker::new(context.clone(), vec![])));

        // Create PeersPool - Observer typically connects to one validator
        let peers_pool = Arc::new(PeersPool::new(context.clone()));
        // Register the validator peer that the observer connects to
        peers_pool
            .register_validator(
                AuthorityIndex::new_for_test(0),
                vec![PeerService::Validator, PeerService::Observer],
            )
            .unwrap();

        let mut commit_syncer = CommitSyncer::new(
            context.clone(),
            core_thread_dispatcher,
            commit_vote_monitor.clone(),
            commit_consumer_monitor.clone(),
            block_verifier,
            transaction_vote_tracker,
            round_tracker,
            network_client,
            dag_state,
            peers_pool.clone(),
        );

        // Verify this is recognized as an observer
        assert!(!context.is_validator(), "Should be an observer node");

        // Simulate the observer seeing commits from its connected validator
        for i in 0..3 {
            let test_block = TestBlock::new(10, i)
                .set_commit_votes(vec![CommitRef::new(5, CommitDigest::MIN)])
                .build();
            let block = VerifiedBlock::new_for_test(test_block);
            commit_vote_monitor.observe_block(&block);
        }

        // Observer should be able to schedule commit fetches
        commit_syncer.try_schedule_once();
        assert_eq!(commit_syncer.pending_fetches().len(), 1);
        assert_eq!(commit_syncer.highest_scheduled_index(), Some(5));

        // Start fetches - observer should be able to fetch from its single peer
        commit_syncer.try_start_fetches();
        assert_eq!(
            commit_syncer.inflight_fetches.len(),
            1,
            "Should start fetch from single peer"
        );
    }

    #[tokio::test(flavor = "current_thread", start_paused = true)]
    async fn commit_syncer_observer_with_multiple_peers() {
        // Test Observer node behavior when connected to multiple peers
        // This simulates an observer that can sync from multiple sources

        let (mut context, _) = Context::new_for_test(4);
        context.own_index = AuthorityIndex::MAX; // Observer node
        context.parameters = Parameters {
            commit_sync_batch_size: 5,
            commit_sync_batches_ahead: 10,
            commit_sync_parallel_fetches: 8, // Allow more parallelism
            max_blocks_per_fetch: 5,
            ..context.parameters
        };
        let context = Arc::new(context);

        // Setup
        let block_verifier = Arc::new(NoopBlockVerifier {});
        let core_thread_dispatcher = Arc::new(MockCoreThreadDispatcher::default());
        let network_client = Arc::new(CommitSyncerClient::new(
            context.clone(),
            Some(Arc::new(FakeNetworkClient::default())),
            Some(Arc::new(FakeNetworkClient::default())),
        ));
        let store = Arc::new(MemStore::new());
        let dag_state = Arc::new(RwLock::new(DagState::new(context.clone(), store)));
        let transaction_vote_tracker =
            TransactionVoteTracker::new(context.clone(), block_verifier.clone(), dag_state.clone());
        let commit_vote_monitor = Arc::new(CommitVoteMonitor::new(context.clone()));
        let commit_consumer_monitor = Arc::new(CommitConsumerMonitor::new(0, 0));
        let round_tracker = Arc::new(RwLock::new(RoundTracker::new(context.clone(), vec![])));

        // Create PeersPool with multiple peers (validators and another observer)
        let peers_pool = Arc::new(PeersPool::new(context.clone()));
        // Register multiple validator peers
        peers_pool
            .register_validator(
                AuthorityIndex::new_for_test(0),
                vec![PeerService::Validator, PeerService::Observer],
            )
            .unwrap();
        peers_pool
            .register_validator(
                AuthorityIndex::new_for_test(1),
                vec![PeerService::Validator, PeerService::Observer],
            )
            .unwrap();
        peers_pool
            .register_validator(
                AuthorityIndex::new_for_test(2),
                vec![PeerService::Validator, PeerService::Observer],
            )
            .unwrap();

        // Now register another observer peer (simulating observer-to-observer sync)
        let observer_peer = NetworkKeyPair::generate(&mut rand::thread_rng()).public();
        peers_pool.register_observer(observer_peer);

        let mut commit_syncer = CommitSyncer::new(
            context.clone(),
            core_thread_dispatcher,
            commit_vote_monitor.clone(),
            commit_consumer_monitor.clone(),
            block_verifier,
            transaction_vote_tracker,
            round_tracker,
            network_client,
            dag_state,
            peers_pool.clone(),
        );

        // Simulate heavy commit load that requires parallel fetching
        for i in 0..3 {
            let test_block = TestBlock::new(100, i)
                .set_commit_votes(vec![CommitRef::new(50, CommitDigest::MIN)])
                .build();
            let block = VerifiedBlock::new_for_test(test_block);
            commit_vote_monitor.observe_block(&block);
        }

        commit_syncer.try_schedule_once();

        // Should schedule multiple batches
        let pending_fetches = commit_syncer.pending_fetches().len();
        assert!(pending_fetches > 0, "Should schedule fetches");

        // Start fetches - observer should utilize multiple peers in parallel
        commit_syncer.try_start_fetches();

        // Observer with 4 peers (3 validators + 1 observer) should be able to
        // fetch from multiple peers in parallel, not limited by 2/3 rule
        let inflight = commit_syncer.inflight_fetches.len();
        let known_peers = peers_pool.get_known_peers().len();

        assert_eq!(known_peers, 4, "Should have 3 validators + 1 observer peer");

        // Observer should be able to use full parallelism up to configured limit
        // Not restricted by the validator's 2/3 limitation
        let max_parallel = context
            .parameters
            .commit_sync_parallel_fetches
            .min(pending_fetches)
            .min(context.parameters.commit_sync_batches_ahead);

        assert!(
            inflight <= max_parallel,
            "Observer should respect configured parallelism limit: {} <= {}",
            inflight,
            max_parallel
        );

        // Verify observer can potentially use more parallelism than a validator would
        // A validator with 4 peers would be limited to 4 * 2/3 = 2 parallel fetches
        // But an observer can use more
        if pending_fetches >= 3 {
            assert!(
                max_parallel > 2,
                "Observer should be able to use more parallelism than validator's 2/3 limit"
            );
        }
    }

    #[tokio::test(flavor = "current_thread", start_paused = true)]
    async fn commit_syncer_start_and_pause_scheduling() {
        // SETUP
        let (context, _) = Context::new_for_test(4);
        // Use smaller batches and fetch limits for testing.
        let context = Context {
            own_index: AuthorityIndex::new_for_test(3),
            parameters: Parameters {
                commit_sync_batch_size: 5,
                commit_sync_batches_ahead: 5,
                commit_sync_parallel_fetches: 5,
                max_blocks_per_fetch: 5,
                ..context.parameters
            },
            ..context
        };
        let context = Arc::new(context);
        let block_verifier = Arc::new(NoopBlockVerifier {});
        let core_thread_dispatcher = Arc::new(MockCoreThreadDispatcher::default());
        let mock_client = Arc::new(FakeNetworkClient::default());
        let network_client = Arc::new(CommitSyncerClient::new(
            context.clone(),
            Some(mock_client.clone()),
            Some(mock_client.clone()),
        ));
        let store = Arc::new(MemStore::new());
        let dag_state = Arc::new(RwLock::new(DagState::new(context.clone(), store)));
        let transaction_vote_tracker =
            TransactionVoteTracker::new(context.clone(), block_verifier.clone(), dag_state.clone());
        let commit_vote_monitor = Arc::new(CommitVoteMonitor::new(context.clone()));
        let commit_consumer_monitor = Arc::new(CommitConsumerMonitor::new(0, 0));
        let round_tracker = Arc::new(RwLock::new(RoundTracker::new(context.clone(), vec![])));
        let peers_pool = Arc::new(PeersPool::new(context.clone()));
        let mut commit_syncer = CommitSyncer::new(
            context,
            core_thread_dispatcher,
            commit_vote_monitor.clone(),
            commit_consumer_monitor.clone(),
            block_verifier,
            transaction_vote_tracker,
            round_tracker,
            network_client,
            dag_state,
            peers_pool,
        );

        // Check initial state.
        assert!(commit_syncer.pending_fetches().is_empty());
        assert!(commit_syncer.fetched_ranges().is_empty());
        assert!(commit_syncer.highest_scheduled_index().is_none());
        assert_eq!(commit_syncer.highest_fetched_commit_index(), 0);
        assert_eq!(commit_syncer.synced_commit_index(), 0);

        // Observe round 15 blocks voting for commit 10 from authorities 0 to 2 in CommitVoteMonitor
        for i in 0..3 {
            let test_block = TestBlock::new(15, i)
                .set_commit_votes(vec![CommitRef::new(10, CommitDigest::MIN)])
                .build();
            let block = VerifiedBlock::new_for_test(test_block);
            commit_vote_monitor.observe_block(&block);
        }

        // Fetches should be scheduled after seeing progress of other validators.
        commit_syncer.try_schedule_once();

        // Verify state.
        assert_eq!(commit_syncer.pending_fetches().len(), 2);
        assert!(commit_syncer.fetched_ranges().is_empty());
        assert_eq!(commit_syncer.highest_scheduled_index(), Some(10));
        assert_eq!(commit_syncer.highest_fetched_commit_index(), 0);
        assert_eq!(commit_syncer.synced_commit_index(), 0);

        // Observe round 40 blocks voting for commit 35 from authorities 0 to 2 in CommitVoteMonitor
        for i in 0..3 {
            let test_block = TestBlock::new(40, i)
                .set_commit_votes(vec![CommitRef::new(35, CommitDigest::MIN)])
                .build();
            let block = VerifiedBlock::new_for_test(test_block);
            commit_vote_monitor.observe_block(&block);
        }

        // Fetches should be scheduled until the unhandled commits threshold.
        commit_syncer.try_schedule_once();

        // Verify commit syncer is paused after scheduling 15 commits to index 25.
        assert_eq!(commit_syncer.unhandled_commits_threshold(), 25);
        assert_eq!(commit_syncer.highest_scheduled_index(), Some(25));
        let pending_fetches = commit_syncer.pending_fetches();
        assert_eq!(pending_fetches.len(), 5);

        // Indicate commit index 25 is consumed, and try to schedule again.
        commit_consumer_monitor.set_highest_handled_commit(25);
        commit_syncer.try_schedule_once();

        // Verify commit syncer schedules fetches up to index 35.
        assert_eq!(commit_syncer.highest_scheduled_index(), Some(35));
        let pending_fetches = commit_syncer.pending_fetches();
        assert_eq!(pending_fetches.len(), 7);

        // Verify contiguous ranges are scheduled.
        for (range, start) in pending_fetches.iter().zip_debug_eq((1..35).step_by(5)) {
            assert_eq!(range.start(), start);
            assert_eq!(range.end(), start + 4);
        }
    }
}