consensus_core/

synchronizer.rs

// Copyright (c) Mysten Labs, Inc.
// SPDX-License-Identifier: Apache-2.0
use std::{
    collections::{BTreeMap, BTreeSet},
    sync::Arc,
    time::Duration,
};

use bytes::Bytes;
use consensus_config::AuthorityIndex;
use consensus_types::block::{BlockRef, Round, TransactionIndex};
use futures::{StreamExt as _, stream::FuturesUnordered};
use itertools::Itertools as _;
use mysten_common::{ZipDebugEqIteratorExt, debug_fatal};
use mysten_metrics::{
    monitored_future,
    monitored_mpsc::{Receiver, Sender, channel},
    monitored_scope,
};
use parking_lot::{Mutex, RwLock};
use rand::{prelude::SliceRandom as _, rngs::ThreadRng};
use sui_macros::fail_point_async;
use tap::TapFallible;
use tokio::{
    sync::{mpsc::error::TrySendError, oneshot},
    task::JoinSet,
    time::{Instant, sleep, sleep_until, timeout},
};
use tracing::{debug, info, trace, warn};

use crate::{
    BlockAPI,
    block::{ExtendedBlock, SignedBlock, VerifiedBlock},
    block_verifier::BlockVerifier,
    commit::CommitIndex,
    commit_vote_monitor::{CommitVoteMonitor, is_commit_lagging},
    context::Context,
    dag_state::DagState,
    error::{ConsensusError, ConsensusResult},
    network::{ObserverNetworkClient, PeerId, SynchronizerClient, ValidatorNetworkClient},
    peers_pool::PeersPool,
    round_tracker::RoundTracker,
    task::{shutdown_join_set, spawn_blocking},
};
use crate::{core_thread::CoreThreadDispatcher, transaction_vote_tracker::TransactionVoteTracker};

/// The number of concurrent fetch blocks requests per authority
const FETCH_BLOCKS_CONCURRENCY: usize = 5;

/// Timeouts when fetching blocks.
const FETCH_REQUEST_TIMEOUT: Duration = Duration::from_millis(2_000);
const FETCH_FROM_PEERS_TIMEOUT: Duration = Duration::from_millis(4_000);

const MAX_AUTHORITIES_TO_FETCH_PER_BLOCK: usize = 2;

// Max number of peers to request missing blocks concurrently in periodic sync.
const MAX_PERIODIC_SYNC_PEERS: usize = 3;

/// How long commit must be stalled before periodic sync kicks in as fallback.
const COMMIT_PROGRESS_TIMEOUT: Duration = Duration::from_secs(10);

struct BlocksGuard {
    map: Arc<InflightBlocksMap>,
    block_refs: BTreeSet<BlockRef>,
    peer: PeerId,
}

impl Drop for BlocksGuard {
    fn drop(&mut self) {
        self.map.unlock_blocks(&self.block_refs, self.peer.clone());
    }
}

// Keeps a mapping between the missing blocks that have been instructed to be fetched and the peers
// that are currently fetching them. For a block ref there is a maximum number of peers that can
// concurrently fetch it. The peer ids that are currently fetching a block are set on the corresponding
// `BTreeSet` and basically they act as "locks".
struct InflightBlocksMap {
    inner: Mutex<BTreeMap<BlockRef, BTreeSet<PeerId>>>,
}

impl InflightBlocksMap {
    fn new() -> Arc<Self> {
        Arc::new(Self {
            inner: Mutex::new(BTreeMap::new()),
        })
    }

    /// Locks the blocks to be fetched for the assigned `peer`. We want to avoid re-fetching the
    /// missing blocks from too many peers at the same time, thus we limit the concurrency
    /// per block by attempting to lock per block. If a block is already fetched by the maximum allowed
    /// number of peers, then the block ref will not be included in the returned set. The method
    /// returns all the block refs that have been successfully locked and allowed to be fetched.
    fn lock_blocks(
        self: &Arc<Self>,
        missing_block_refs: BTreeSet<BlockRef>,
        peer: PeerId,
    ) -> Option<BlocksGuard> {
        let mut blocks = BTreeSet::new();
        let mut inner = self.inner.lock();

        for block_ref in missing_block_refs {
            // check that the number of peers that are already instructed to fetch the block is not
            // higher than the allowed and the `peer` has not already been instructed to do that.
            let peers = inner.entry(block_ref).or_default();
            if peers.len() < MAX_AUTHORITIES_TO_FETCH_PER_BLOCK && peers.get(&peer).is_none() {
                assert!(peers.insert(peer.clone()));
                blocks.insert(block_ref);
            }
        }

        if blocks.is_empty() {
            None
        } else {
            Some(BlocksGuard {
                map: self.clone(),
                block_refs: blocks,
                peer,
            })
        }
    }

    /// Unlocks the provided block references for the given `peer`. The unlocking is strict, meaning that
    /// if this method is called for a specific block ref and peer more times than the corresponding lock
    /// has been called, it will panic.
    fn unlock_blocks(self: &Arc<Self>, block_refs: &BTreeSet<BlockRef>, peer: PeerId) {
        // Now mark all the blocks as fetched from the map
        let mut blocks_to_fetch = self.inner.lock();
        for block_ref in block_refs {
            let peers = blocks_to_fetch
                .get_mut(block_ref)
                .expect("Should have found a non empty map");

            assert!(peers.remove(&peer), "Peer should be present!");

            // if the last one then just clean up
            if peers.is_empty() {
                blocks_to_fetch.remove(block_ref);
            }
        }
    }

    /// Drops the provided `blocks_guard` which will force to unlock the blocks, and lock now again the
    /// referenced block refs. The swap is best effort and there is no guarantee that the `peer` will
    /// be able to acquire the new locks.
    fn swap_locks(
        self: &Arc<Self>,
        blocks_guard: BlocksGuard,
        peer: PeerId,
    ) -> Option<BlocksGuard> {
        let block_refs = blocks_guard.block_refs.clone();

        // Explicitly drop the guard
        drop(blocks_guard);

        // Now create new guard
        self.lock_blocks(block_refs, peer)
    }

    #[cfg(test)]
    fn num_of_locked_blocks(self: &Arc<Self>) -> usize {
        let inner = self.inner.lock();
        inner.len()
    }
}

enum Command {
    FetchBlocks {
        missing_block_refs: BTreeSet<BlockRef>,
        peer: PeerId,
        result: oneshot::Sender<Result<(), ConsensusError>>,
    },
    FetchOwnLastBlock,
    KickOffScheduler,
    Shutdown {
        result: oneshot::Sender<()>,
    },
}

pub(crate) struct SynchronizerHandle {
    commands_sender: Sender<Command>,
    tasks: tokio::sync::Mutex<JoinSet<()>>,
}

impl SynchronizerHandle {
    /// Explicitly asks from the synchronizer to fetch the blocks - provided the block_refs set - from
    /// the peer.
    pub(crate) async fn fetch_blocks(
        &self,
        missing_block_refs: BTreeSet<BlockRef>,
        peer: PeerId,
    ) -> ConsensusResult<()> {
        let (sender, receiver) = oneshot::channel();
        self.commands_sender
            .send(Command::FetchBlocks {
                missing_block_refs,
                peer,
                result: sender,
            })
            .await
            .map_err(|_err| ConsensusError::Shutdown)?;
        receiver.await.map_err(|_err| ConsensusError::Shutdown)?
    }

    pub(crate) async fn stop(&self) {
        let (shutdown_sender, shutdown_receiver) = oneshot::channel();
        let _ = self
            .commands_sender
            .send(Command::Shutdown {
                result: shutdown_sender,
            })
            .await;
        let _ = shutdown_receiver.await;

        let mut tasks = self.tasks.lock().await;
        shutdown_join_set(&mut tasks).await;
    }

    #[cfg(test)]
    /// Creates a mock synchronizer handle for testing
    pub(crate) fn new_for_test() -> Arc<Self> {
        use tokio::task::JoinSet;
        let (tx, _rx) = channel("test_synchronizer", 1);
        Arc::new(Self {
            commands_sender: tx,
            tasks: tokio::sync::Mutex::new(JoinSet::new()),
        })
    }
}

/// `Synchronizer` oversees live block synchronization, crucial for node progress. Live synchronization
/// refers to the process of retrieving missing blocks, particularly those essential for advancing a node
/// when data from only a few rounds is absent. If a node significantly lags behind the network,
/// `commit_syncer` handles fetching missing blocks via a more efficient approach. `Synchronizer`
/// aims for swift catch-up employing two mechanisms:
///
/// 1. Explicitly requesting missing blocks from designated authorities via the "block send" path.
///    This includes attempting to fetch any missing ancestors necessary for processing a received block.
///    Such requests prioritize the block author, maximizing the chance of prompt retrieval.
///    A locking mechanism allows concurrent requests for missing blocks from up to two authorities
///    simultaneously, enhancing the chances of timely retrieval. Notably, if additional missing blocks
///    arise during block processing, requests to the same authority are deferred to the scheduler.
///
/// 2. Periodically requesting missing blocks via a scheduler. This primarily serves to retrieve
///    missing blocks that were not ancestors of a received block via the "block send" path.
///    The scheduler operates on either a fixed periodic basis or is triggered immediately
///    after explicit fetches described in (1), ensuring continued block retrieval if gaps persist.
///
/// Additionally to the above, the synchronizer can synchronize and fetch the last own proposed block
/// from the network peers as best effort approach to recover node from amnesia and avoid making the
/// node equivocate.
pub(crate) struct Synchronizer<
    V: BlockVerifier,
    D: CoreThreadDispatcher,
    VC: ValidatorNetworkClient,
    OC: ObserverNetworkClient,
> {
    context: Arc<Context>,
    commands_receiver: Receiver<Command>,
    fetch_block_senders: BTreeMap<PeerId, Sender<BlocksGuard>>,
    core_dispatcher: Arc<D>,
    commit_vote_monitor: Arc<CommitVoteMonitor>,
    dag_state: Arc<RwLock<DagState>>,
    fetch_blocks_scheduler_task: JoinSet<()>,
    fetch_own_last_block_task: JoinSet<()>,
    network_client: Arc<SynchronizerClient<VC, OC>>,
    block_verifier: Arc<V>,
    transaction_vote_tracker: TransactionVoteTracker,
    round_tracker: Arc<RwLock<RoundTracker>>,
    inflight_blocks_map: Arc<InflightBlocksMap>,
    commands_sender: Sender<Command>,
    last_changed_commit_index: CommitIndex,
    last_commit_change_time: Instant,
    // When commit is not progressing, commit sync fails over to periodic sync for catchup.
    commit_sync_failover: bool,
    peers_pool: Arc<PeersPool>,
}

impl<V, D, VC, OC> Synchronizer<V, D, VC, OC>
where
    V: BlockVerifier,
    D: CoreThreadDispatcher,
    VC: ValidatorNetworkClient,
    OC: ObserverNetworkClient,
{
    pub(crate) fn start(
        network_client: Arc<SynchronizerClient<VC, OC>>,
        context: Arc<Context>,
        core_dispatcher: Arc<D>,
        commit_vote_monitor: Arc<CommitVoteMonitor>,
        block_verifier: Arc<V>,
        transaction_vote_tracker: TransactionVoteTracker,
        round_tracker: Arc<RwLock<RoundTracker>>,
        dag_state: Arc<RwLock<DagState>>,
        peers_pool: Arc<PeersPool>,
        sync_last_known_own_block: bool,
    ) -> Arc<SynchronizerHandle> {
        let (commands_sender, commands_receiver) =
            channel("consensus_synchronizer_commands", 1_000);
        let inflight_blocks_map = InflightBlocksMap::new();

        // Spawn the tasks to fetch the blocks from the others
        let mut fetch_block_senders = BTreeMap::new();
        let mut tasks = JoinSet::new();

        // Create fetch tasks for all known peers (validators and observers)
        // TODO: refactor to update the sender tasks based on the registered/removed pool peers.
        let known_peers = peers_pool.get_known_peers();
        for peer in known_peers {
            let (sender, receiver) =
                channel("consensus_synchronizer_fetches", FETCH_BLOCKS_CONCURRENCY);
            let fetch_blocks_from_peer_async = Self::fetch_blocks_from_peer(
                peer.clone(),
                network_client.clone(),
                block_verifier.clone(),
                transaction_vote_tracker.clone(),
                commit_vote_monitor.clone(),
                context.clone(),
                core_dispatcher.clone(),
                dag_state.clone(),
                receiver,
                commands_sender.clone(),
                round_tracker.clone(),
                peers_pool.clone(),
            );
            tasks.spawn(monitored_future!(fetch_blocks_from_peer_async));
            fetch_block_senders.insert(peer, sender);
        }

        let commands_sender_clone = commands_sender.clone();

        if sync_last_known_own_block {
            commands_sender
                .try_send(Command::FetchOwnLastBlock)
                .expect("Failed to sync our last block");
        }

        // Spawn the task to listen to the requests & periodic runs
        tasks.spawn(monitored_future!(async move {
            let mut s = Self {
                context,
                commands_receiver,
                fetch_block_senders,
                core_dispatcher,
                commit_vote_monitor,
                fetch_blocks_scheduler_task: JoinSet::new(),
                fetch_own_last_block_task: JoinSet::new(),
                network_client,
                block_verifier,
                transaction_vote_tracker,
                inflight_blocks_map,
                commands_sender: commands_sender_clone,
                dag_state,
                round_tracker,
                last_changed_commit_index: 0,
                last_commit_change_time: Instant::now(),
                commit_sync_failover: false,
                peers_pool,
            };
            s.run().await;
        }));

        Arc::new(SynchronizerHandle {
            commands_sender,
            tasks: tokio::sync::Mutex::new(tasks),
        })
    }

    // The main loop to listen for the submitted commands.
    async fn run(&mut self) {
        // We want the synchronizer to run periodically every 200ms to fetch any missing blocks.
        const PERIODIC_FETCH_INTERVAL: Duration = Duration::from_millis(200);
        let scheduler_timeout = sleep_until(Instant::now() + PERIODIC_FETCH_INTERVAL);

        tokio::pin!(scheduler_timeout);

        loop {
            tokio::select! {
                Some(command) = self.commands_receiver.recv() => {
                    match command {
                        Command::FetchBlocks{ missing_block_refs, peer, result } => {
                            // Check if peer is available. This check also makes sure that we are not trying to fetch from ourselves.
                            if !self.peers_pool.is_peer_known(&peer) {
                                result.send(Err(ConsensusError::PeerUnavailable(format!("{:?}", peer)))).ok();
                                continue;
                            }

                            // Keep only the max allowed blocks to request. Additional missing blocks
                            // will be fetched via periodic sync.
                            // Fetch from the lowest to highest round, to ensure progress.
                            let missing_block_refs = missing_block_refs
                                .into_iter()
                                .take(self.context.parameters.max_blocks_per_sync)
                                .collect();

                            let blocks_guard = self.inflight_blocks_map.lock_blocks(missing_block_refs, peer.clone());
                            let Some(blocks_guard) = blocks_guard else {
                                result.send(Ok(())).ok();
                                continue;
                            };

                            // We don't block if the corresponding peer task is saturated - but we rather drop the request. That's ok as the periodic
                            // synchronization task will handle any still missing blocks in next run.
                            let r = self
                                .fetch_block_senders
                                .get(&peer)
                                .ok_or(ConsensusError::PeerNotFound(format!("Peer {} not found in fetch_block_senders", peer)))
                                .and_then(|sender| {
                                    sender
                                        .try_send(blocks_guard)
                                        .map_err(|err| {
                                            match err {
                                                TrySendError::Full(_) => {
                                                    let peer_name = peer.labelname(&self.context);
                                                    self.context
                                                        .metrics
                                                        .node_metrics
                                                        .synchronizer_skipped_fetch_requests
                                                        .with_label_values(&[peer_name])
                                                        .inc();
                                                    ConsensusError::SynchronizerSaturated(format!("{:?}", peer))
                                                },
                                                TrySendError::Closed(_) => ConsensusError::Shutdown
                                            }
                                        })
                                });

                            result.send(r).ok();
                        }
                        Command::FetchOwnLastBlock => {
                            if self.fetch_own_last_block_task.is_empty() {
                                self.start_fetch_own_last_block_task();
                            }
                        }
                        Command::KickOffScheduler => {
                            // just reset the scheduler timeout timer to run immediately if not already running.
                            // If the scheduler is already running then just reduce the remaining time to run.
                            let timeout = if self.fetch_blocks_scheduler_task.is_empty() {
                                Instant::now()
                            } else {
                                Instant::now() + PERIODIC_FETCH_INTERVAL.checked_div(2).unwrap()
                            };

                            // only reset if it is earlier than the next deadline
                            if timeout < scheduler_timeout.deadline() {
                                scheduler_timeout.as_mut().reset(timeout);
                            }
                        }
                        Command::Shutdown { result } => {
                            self.shutdown_tasks().await;
                            self.fetch_block_senders.clear();
                            let _ = result.send(());
                            return;
                        }
                    }
                },
                Some(result) = self.fetch_own_last_block_task.join_next(), if !self.fetch_own_last_block_task.is_empty() => {
                    match result {
                        Ok(()) => {},
                        Err(e) => {
                            if e.is_cancelled() {
                            } else if e.is_panic() {
                                std::panic::resume_unwind(e.into_panic());
                            } else {
                                panic!("fetch our last block task failed: {e}");
                            }
                        },
                    };
                },
                Some(result) = self.fetch_blocks_scheduler_task.join_next(), if !self.fetch_blocks_scheduler_task.is_empty() => {
                    match result {
                        Ok(()) => {},
                        Err(e) => {
                            if e.is_cancelled() {
                            } else if e.is_panic() {
                                std::panic::resume_unwind(e.into_panic());
                            } else {
                                panic!("fetch blocks scheduler task failed: {e}");
                            }
                        },
                    };
                },
                () = &mut scheduler_timeout => {
                    // we want to start a new task only if the previous one has already finished.
                    // TODO: consider starting backup fetches in parallel, when a fetch takes too long?
                    if self.fetch_blocks_scheduler_task.is_empty()
                        && let Err(err) = self.start_fetch_missing_blocks_task().await {
                            debug!("Core is shutting down, synchronizer is shutting down: {err:?}");
                            self.shutdown_tasks().await;
                            return;
                        };

                    scheduler_timeout
                        .as_mut()
                        .reset(Instant::now() + PERIODIC_FETCH_INTERVAL);
                }
            }
        }
    }

    // Must be called before exiting run(), so no task is dropped without being awaited
    // and task panics propagate.
    async fn shutdown_tasks(&mut self) {
        shutdown_join_set(&mut self.fetch_own_last_block_task).await;
        shutdown_join_set(&mut self.fetch_blocks_scheduler_task).await;
    }

    async fn fetch_blocks_from_peer(
        peer: PeerId,
        network_client: Arc<SynchronizerClient<VC, OC>>,
        block_verifier: Arc<V>,
        transaction_vote_tracker: TransactionVoteTracker,
        commit_vote_monitor: Arc<CommitVoteMonitor>,
        context: Arc<Context>,
        core_dispatcher: Arc<D>,
        dag_state: Arc<RwLock<DagState>>,
        mut receiver: Receiver<BlocksGuard>,
        commands_sender: Sender<Command>,
        round_tracker: Arc<RwLock<RoundTracker>>,
        _peers_pool: Arc<PeersPool>,
    ) {
        const MAX_RETRIES: u32 = 3;
        let mut requests = FuturesUnordered::new();

        loop {
            tokio::select! {
                Some(blocks_guard) = receiver.recv(), if requests.len() < FETCH_BLOCKS_CONCURRENCY => {
                    let fetch_after_rounds = Self::get_fetch_after_rounds(&context, dag_state.clone());

                    requests.push(Self::fetch_blocks_request(network_client.clone(), peer.clone(), blocks_guard, fetch_after_rounds, true, FETCH_REQUEST_TIMEOUT, 1))
                },
                Some((response, blocks_guard, retries, _peer, fetch_after_rounds)) = requests.next() => {
                    match response {
                        Ok(blocks) => {
                            if let Err(err) = Self::process_fetched_blocks(blocks,
                                peer.clone(),
                                blocks_guard,
                                core_dispatcher.clone(),
                                block_verifier.clone(),
                                transaction_vote_tracker.clone(),
                                commit_vote_monitor.clone(),
                                context.clone(),
                                commands_sender.clone(),
                                round_tracker.clone(),
                                "live"
                            ).await {
                                warn!("Error while processing fetched blocks from peer {}: {err}", peer.hostname(&context));
                                context.metrics.node_metrics.synchronizer_process_fetched_failures.with_label_values(&[peer.labelname(&context).as_str(), "live"]).inc();
                            }
                        },
                        Err(_) => {
                            context.metrics.node_metrics.synchronizer_fetch_failures.with_label_values(&[peer.labelname(&context).as_str(), "live"]).inc();
                            if retries <= MAX_RETRIES {
                                requests.push(Self::fetch_blocks_request(network_client.clone(), peer.clone(), blocks_guard, fetch_after_rounds, true, FETCH_REQUEST_TIMEOUT, retries))
                            } else {
                                warn!("Max retries {retries} reached while trying to fetch blocks from peer {}.", peer.hostname(&context));
                                // we don't necessarily need to do, but dropping the guard here to unlock the blocks
                                drop(blocks_guard);
                            }
                        }
                    }
                },
                else => {
                    info!("Fetching blocks from peer {} task will now abort.", peer.hostname(&context));
                    break;
                }
            }
        }
    }

    /// Processes the requested raw fetched blocks from peer. If no error is returned then
    /// the verified blocks are immediately sent to Core for processing.
    async fn process_fetched_blocks(
        mut serialized_blocks: Vec<Bytes>,
        peer: PeerId,
        requested_blocks_guard: BlocksGuard,
        core_dispatcher: Arc<D>,
        block_verifier: Arc<V>,
        transaction_vote_tracker: TransactionVoteTracker,
        commit_vote_monitor: Arc<CommitVoteMonitor>,
        context: Arc<Context>,
        commands_sender: Sender<Command>,
        round_tracker: Arc<RwLock<RoundTracker>>,
        sync_method: &str,
    ) -> ConsensusResult<()> {
        if serialized_blocks.is_empty() {
            return Ok(());
        }

        // Limit the number of the returned blocks processed.
        serialized_blocks.truncate(context.parameters.max_blocks_per_sync);

        // Verify all the fetched blocks
        let (blocks, voted_blocks) = spawn_blocking({
            let block_verifier = block_verifier.clone();
            let context = context.clone();
            let peer = peer.clone();
            move || Self::verify_blocks(serialized_blocks, block_verifier, &context, peer)
        })
        .await??;

        if context.protocol_config.transaction_voting_enabled() {
            transaction_vote_tracker.add_voted_blocks(voted_blocks);
        }

        // Record commit votes from the verified blocks.
        for block in &blocks {
            commit_vote_monitor.observe_block(block);
        }

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

        let metrics = &context.metrics.node_metrics;
        metrics
            .synchronizer_fetched_blocks_by_peer
            .with_label_values(&[peer.labelname(&context).as_str(), sync_method])
            .inc_by(blocks.len() as u64);
        for block in &blocks {
            let block_hostname = &context.committee.authority(block.author()).hostname;
            metrics
                .synchronizer_fetched_blocks_by_authority
                .with_label_values(&[block_hostname.as_str(), sync_method])
                .inc();
        }

        debug!(
            "Synced {} missing blocks from peer {:?}: {}",
            blocks.len(),
            peer,
            blocks.iter().map(|b| b.reference().to_string()).join(", "),
        );

        // Now send them to core for processing. Ignore the returned missing blocks as we don't want
        // this mechanism to keep feedback looping on fetching more blocks. The periodic synchronization
        // will take care of that.
        let missing_blocks = core_dispatcher
            .add_blocks(blocks)
            .await
            .map_err(|_| ConsensusError::Shutdown)?;

        // now release all the locked blocks as they have been fetched, verified & processed
        drop(requested_blocks_guard);

        // kick off immediately the scheduled synchronizer
        if !missing_blocks.is_empty() {
            // do not block here, so we avoid any possible cycles.
            if let Err(TrySendError::Full(_)) = commands_sender.try_send(Command::KickOffScheduler)
            {
                warn!("Commands channel is full")
            }
        }

        context
            .metrics
            .node_metrics
            .missing_blocks_after_fetch_total
            .inc_by(missing_blocks.len() as u64);

        Ok(())
    }

    fn get_fetch_after_rounds(
        context: &Arc<Context>,
        dag_state: Arc<RwLock<DagState>>,
    ) -> Vec<Round> {
        let (blocks, gc_round) = {
            let dag_state = dag_state.read();
            (
                dag_state.get_last_cached_block_per_authority(Round::MAX),
                dag_state.gc_round(),
            )
        };
        assert_eq!(blocks.len(), context.committee.size());

        blocks
            .into_iter()
            .map(|(block, _)| block.round().max(gc_round))
            .collect::<Vec<_>>()
    }

    fn verify_blocks(
        serialized_blocks: Vec<Bytes>,
        block_verifier: Arc<V>,
        context: &Context,
        peer: PeerId,
    ) -> ConsensusResult<(
        Vec<VerifiedBlock>,
        Vec<(VerifiedBlock, Vec<TransactionIndex>)>,
    )> {
        let mut verified_blocks = Vec::new();
        let mut voted_blocks = Vec::new();
        for serialized_block in serialized_blocks {
            let signed_block: SignedBlock =
                bcs::from_bytes(&serialized_block).map_err(ConsensusError::MalformedBlock)?;

            // TODO: cache received and verified block refs to avoid duplicated work.
            let (verified_block, reject_txn_votes) = block_verifier
                .verify_and_vote(signed_block, serialized_block)
                .tap_err(|e| {
                    let peer_label = peer.labelname(context);
                    context
                        .metrics
                        .node_metrics
                        .invalid_blocks
                        .with_label_values(&[peer_label.as_str(), "synchronizer", e.clone().name()])
                        .inc();
                    info!("Invalid block received from {}: {}", peer, e);
                })?;

            // TODO: improve efficiency, maybe suspend and continue processing the block asynchronously.
            let now = context.clock.timestamp_utc_ms();
            let drift = verified_block.timestamp_ms().saturating_sub(now);
            if drift > 0 {
                let peer_hostname = &context
                    .committee
                    .authority(verified_block.author())
                    .hostname;
                context
                    .metrics
                    .node_metrics
                    .block_timestamp_drift_ms
                    .with_label_values(&[peer_hostname.as_str(), "synchronizer"])
                    .inc_by(drift);

                trace!(
                    "Synced block {} timestamp {} is in the future (now={}).",
                    verified_block.reference(),
                    verified_block.timestamp_ms(),
                    now
                );
            }

            verified_blocks.push(verified_block.clone());
            voted_blocks.push((verified_block, reject_txn_votes));
        }

        Ok((verified_blocks, voted_blocks))
    }

    async fn fetch_blocks_request(
        network_client: Arc<SynchronizerClient<VC, OC>>,
        peer: PeerId,
        blocks_guard: BlocksGuard,
        fetch_after_rounds: Vec<Round>,
        fetch_missing_ancestors: bool,
        request_timeout: Duration,
        mut retries: u32,
    ) -> (
        ConsensusResult<Vec<Bytes>>,
        BlocksGuard,
        u32,
        PeerId,
        Vec<Round>,
    ) {
        let start = Instant::now();
        let resp = timeout(
            request_timeout,
            network_client.fetch_blocks(
                peer.clone(),
                blocks_guard
                    .block_refs
                    .clone()
                    .into_iter()
                    .collect::<Vec<_>>(),
                fetch_after_rounds.clone().into_iter().collect::<Vec<_>>(),
                fetch_missing_ancestors,
                request_timeout,
            ),
        )
        .await;

        fail_point_async!("consensus-delay");

        let resp = match resp {
            Ok(Err(err)) => {
                // Add a delay before retrying - if that is needed. If request has timed out then eventually
                // this will be a no-op.
                sleep_until(start + request_timeout).await;
                retries += 1;
                Err(err)
            } // network error
            Err(err) => {
                // timeout
                sleep_until(start + request_timeout).await;
                retries += 1;
                Err(ConsensusError::NetworkRequestTimeout(err.to_string()))
            }
            Ok(result) => result,
        };
        (resp, blocks_guard, retries, peer, fetch_after_rounds)
    }

    fn start_fetch_own_last_block_task(&mut self) {
        const FETCH_OWN_BLOCK_RETRY_DELAY: Duration = Duration::from_millis(1_000);
        const MAX_RETRY_DELAY_STEP: Duration = Duration::from_millis(4_000);

        let context = self.context.clone();
        let dag_state = self.dag_state.clone();
        let network_client = self.network_client.clone();
        let block_verifier = self.block_verifier.clone();
        let core_dispatcher = self.core_dispatcher.clone();

        self.fetch_own_last_block_task
            .spawn(monitored_future!(async move {
                let _scope = monitored_scope("FetchOwnLastBlockTask");

                let fetch_own_block = |authority_index: AuthorityIndex, fetch_own_block_delay: Duration| {
                    let network_client_cloned = network_client.clone();
                    let own_index = context.own_index;
                    async move {
                        sleep(fetch_own_block_delay).await;
                        let r = network_client_cloned.fetch_latest_blocks(authority_index, vec![own_index], FETCH_REQUEST_TIMEOUT).await;
                        (r, authority_index)
                    }
                };


                let process_blocks = |blocks: Vec<Bytes>, authority_index: AuthorityIndex| -> ConsensusResult<Vec<VerifiedBlock>> {
                    let mut result = Vec::new();
                    for serialized_block in blocks {
                        let signed_block = bcs::from_bytes(&serialized_block).map_err(ConsensusError::MalformedBlock)?;
                        let (verified_block, _) = block_verifier.verify_and_vote(signed_block, serialized_block).tap_err(|err|{
                            let hostname = context.committee.authority(authority_index).hostname.clone();
                            context
                                .metrics
                                .node_metrics
                                .invalid_blocks
                                .with_label_values(&[hostname.as_str(), "synchronizer_own_block", err.clone().name()])
                                .inc();
                            warn!("Invalid block received from {}: {}", authority_index, err);
                        })?;

                        if verified_block.author() != context.own_index {
                            return Err(ConsensusError::UnexpectedLastOwnBlock { index: authority_index, block_ref: verified_block.reference()});
                        }
                        result.push(verified_block);
                    }
                    Ok(result)
                };

                // Get the highest of all the results. Retry until at least `f+1` results have been gathered.
                let mut highest_round;
                let mut retries = 0;
                let mut retry_delay_step = Duration::from_millis(500);
                'main:loop {
                    if context.committee.size() == 1 {
                        highest_round = dag_state.read().get_last_proposed_block().expect("Last proposed block should be returned on validators").round();
                        info!("Only one node in the network, will not try fetching own last block from peers.");
                        break 'main;
                    }

                    let mut total_stake = 0;
                    highest_round = 0;

                    // Ask all the other peers about our last block
                    let mut results = FuturesUnordered::new();

                    for (authority_index, _authority) in context.committee.authorities() {
                        if authority_index != context.own_index {
                            results.push(fetch_own_block(authority_index, Duration::from_millis(0)));
                        }
                    }

                    // Gather the results but wait to timeout as well
                    let timer = sleep_until(Instant::now() + context.parameters.sync_last_known_own_block_timeout);
                    tokio::pin!(timer);

                    'inner: loop {
                        tokio::select! {
                            result = results.next() => {
                                let Some((result, authority_index)) = result else {
                                    break 'inner;
                                };
                                match result {
                                    Ok(result) => {
                                        match process_blocks(result, authority_index) {
                                            Ok(blocks) => {
                                                let max_round = blocks.into_iter().map(|b|b.round()).max().unwrap_or(0);
                                                highest_round = highest_round.max(max_round);

                                                total_stake += context.committee.stake(authority_index);
                                            },
                                            Err(err) => {
                                                warn!("Invalid result returned from {authority_index} while fetching last own block: {err}");
                                            }
                                        }
                                    },
                                    Err(err) => {
                                        warn!("Error {err} while fetching our own block from peer {authority_index}. Will retry.");
                                        results.push(fetch_own_block(authority_index, FETCH_OWN_BLOCK_RETRY_DELAY));
                                    }
                                }
                            },
                            () = &mut timer => {
                                info!("Timeout while trying to sync our own last block from peers");
                                break 'inner;
                            }
                        }
                    }

                    // Request at least f+1 stake to have replied back.
                    if context.committee.reached_validity(total_stake) {
                        info!("{} out of {} total stake returned acceptable results for our own last block with highest round {}, with {retries} retries.", total_stake, context.committee.total_stake(), highest_round);
                        break 'main;
                    }

                    retries += 1;
                    context.metrics.node_metrics.sync_last_known_own_block_retries.inc();
                    warn!("Not enough stake: {} out of {} total stake returned acceptable results for our own last block with highest round {}. Will now retry {retries}.", total_stake, context.committee.total_stake(), highest_round);

                    sleep(retry_delay_step).await;

                    retry_delay_step = Duration::from_secs_f64(retry_delay_step.as_secs_f64() * 1.5);
                    retry_delay_step = retry_delay_step.min(MAX_RETRY_DELAY_STEP);
                }

                // Update the Core with the highest detected round
                context.metrics.node_metrics.last_known_own_block_round.set(highest_round as i64);

                if let Err(err) = core_dispatcher.set_last_known_proposed_round(highest_round) {
                    warn!("Error received while calling dispatcher, probably dispatcher is shutting down, will now exit: {err:?}");
                }
            }));
    }

    async fn start_fetch_missing_blocks_task(&mut self) -> ConsensusResult<()> {
        if self.context.committee.size() == 1 {
            trace!(
                "Only one node in the network, will not try fetching missing blocks from peers."
            );
            return Ok(());
        }

        // If commit is lagging and commit sync is making progress, skip periodic sync.
        // Commit syncer fetches certified commits with all necessary causal history.
        // If commit sync is not making progress, periodic sync resumes as a fallback.
        if !self.should_run_periodic_sync() {
            return Ok(());
        }

        let context = self.context.clone();
        let network_client = self.network_client.clone();
        let block_verifier = self.block_verifier.clone();
        let transaction_vote_tracker = self.transaction_vote_tracker.clone();
        let commit_vote_monitor = self.commit_vote_monitor.clone();
        let core_dispatcher = self.core_dispatcher.clone();
        let blocks_to_fetch = self.inflight_blocks_map.clone();
        let commands_sender = self.commands_sender.clone();
        let dag_state = self.dag_state.clone();
        let round_tracker = self.round_tracker.clone();
        let peers_pool = self.peers_pool.clone();

        let mut missing_blocks = self
            .core_dispatcher
            .get_missing_blocks()
            .await
            .map_err(|_err| ConsensusError::Shutdown)?;
        if self.commit_sync_failover {
            // Keep missing blocks to those that must be included in fetch request.
            // Filtered out missing blocks that will eventually be fetched with fetch_after_rounds.
            let fetch_after_rounds = Self::get_fetch_after_rounds(&context, dag_state.clone());
            missing_blocks.retain(|block| block.round <= fetch_after_rounds[block.author.value()]);
        } else if missing_blocks.is_empty() {
            return Ok(());
        }

        self.fetch_blocks_scheduler_task
            .spawn(monitored_future!(async move {
                let _scope = monitored_scope("FetchMissingBlocksScheduler");
                context
                    .metrics
                    .node_metrics
                    .fetch_blocks_scheduler_inflight
                    .inc();
                let total_requested = missing_blocks.len();

                let results = if missing_blocks.is_empty() {
                    let _scope = monitored_scope("BlockSync::Periodic::HighestAcceptedRounds");
                    // Fetch blocks from a random peer using highest accepted rounds (commit sync failover)
                    Self::fetch_blocks_with_fetch_after_rounds(
                        context.clone(),
                        blocks_to_fetch.clone(),
                        network_client,
                        dag_state,
                        peers_pool,
                    )
                    .await
                } else {
                    let _scope = monitored_scope("BlockSync::Periodic::MissingBlocks");
                    // Fetch blocks from 1 to MAX_PERIODIC_SYNC_PEERS peers
                    Self::fetch_blocks_from_peers(
                        context.clone(),
                        blocks_to_fetch.clone(),
                        network_client,
                        missing_blocks,
                        dag_state,
                        peers_pool,
                    )
                    .await
                };
                context
                    .metrics
                    .node_metrics
                    .fetch_blocks_scheduler_inflight
                    .dec();
                if results.is_empty() {
                    return;
                }

                fail_point_async!("consensus-delay");

                // Now process the returned results
                let mut total_fetched = 0;
                for (blocks_guard, fetched_blocks, peer) in results {
                    total_fetched += fetched_blocks.len();

                    if let Err(err) = Self::process_fetched_blocks(
                        fetched_blocks,
                        peer.clone(),
                        blocks_guard,
                        core_dispatcher.clone(),
                        block_verifier.clone(),
                        transaction_vote_tracker.clone(),
                        commit_vote_monitor.clone(),
                        context.clone(),
                        commands_sender.clone(),
                        round_tracker.clone(),
                        "periodic",
                    )
                    .await
                    {
                        warn!(
                            "Error occurred while processing fetched blocks from peer {:?}: {err}",
                            peer
                        );
                        let peer_name = peer.labelname(&context);
                        context
                            .metrics
                            .node_metrics
                            .synchronizer_process_fetched_failures
                            .with_label_values(&[peer_name.as_str(), "periodic"])
                            .inc();
                    }
                }

                debug!(
                    "Total blocks requested to fetch: {}, total fetched: {}",
                    total_requested, total_fetched
                );
            }));

        Ok(())
    }

    fn should_run_periodic_sync(&mut self) -> bool {
        let current_commit_index = self.dag_state.read().last_commit_index();
        let quorum_commit_index = self.commit_vote_monitor.quorum_commit_index();
        let now = Instant::now();
        let metrics = &self.context.metrics.node_metrics;

        // Commit is not lagging.
        if !is_commit_lagging(
            self.context.as_ref(),
            current_commit_index,
            quorum_commit_index,
        ) {
            metrics
                .synchronizer_periodic_sync_decision
                .with_label_values(&["true", "default"])
                .inc();
            // Reset last commit state.
            self.last_changed_commit_index = current_commit_index;
            self.last_commit_change_time = now;
            self.commit_sync_failover = false;
            // Run periodic sync.
            return true;
        }
        // Commit is lagging.

        // When in commit sync failover, check if enough progress has been made.
        if self.commit_sync_failover {
            if current_commit_index
                < self.last_changed_commit_index + self.context.parameters.commit_sync_batch_size
            {
                metrics
                    .synchronizer_periodic_sync_decision
                    .with_label_values(&["true", "commit_catchup::run"])
                    .inc();
                // Not enough progress has been made yet. Keep running periodic sync.
                // Do not update last commit state yet.
                // Run periodic sync.
                return true;
            } else {
                metrics
                    .synchronizer_periodic_sync_decision
                    .with_label_values(&["false", "commit_catchup::end"])
                    .inc();
                // Enough progress has been made. Disable commit sync failover and reset last commit state.
                self.last_changed_commit_index = current_commit_index;
                self.last_commit_change_time = now;
                self.commit_sync_failover = false;
                // Skip periodic sync because of commit lag.
                return false;
            }
        }

        // The node is commit lagging and not in commit sync failover yet.
        if current_commit_index == self.last_changed_commit_index {
            // Enter commit sync failover if not enough progress has been made.
            if now.duration_since(self.last_commit_change_time) >= COMMIT_PROGRESS_TIMEOUT {
                metrics
                    .synchronizer_periodic_sync_decision
                    .with_label_values(&["true", "commit_catchup::start"])
                    .inc();
                self.commit_sync_failover = true;
                // Run periodic sync.
                return true;
            }
        } else {
            // IMPORTANT: Only update last commit state when commit index is changing.
            self.last_changed_commit_index = current_commit_index;
            self.last_commit_change_time = now;
        }

        metrics
            .synchronizer_periodic_sync_decision
            .with_label_values(&["false", "commit_lag"])
            .inc();
        false
    }

    /// Fetches blocks from a random peer using only fetch_after_rounds (no specific missing blocks).
    /// Used during commit sync failover to make progress when commit sync is stalled.
    async fn fetch_blocks_with_fetch_after_rounds(
        context: Arc<Context>,
        inflight_blocks: Arc<InflightBlocksMap>,
        network_client: Arc<SynchronizerClient<VC, OC>>,
        dag_state: Arc<RwLock<DagState>>,
        peers_pool: Arc<PeersPool>,
    ) -> Vec<(BlocksGuard, Vec<Bytes>, PeerId)> {
        let fetch_after_rounds = Self::get_fetch_after_rounds(&context, dag_state.clone());

        // Pick a random peer (excluding self).
        // Get available peers from the PeersPool
        let mut peers = peers_pool.get_known_peers();

        // TODO: in the future it would be possible, temporarily, for an Observer node to not have peers to fetch from.
        // We should change this assertion to allow for this case.
        assert!(!peers.is_empty(), "No known peers to fetch blocks from");

        if cfg!(not(test)) {
            peers.shuffle(&mut ThreadRng::default());
        }

        let peer = peers.first().unwrap().clone();

        let response = timeout(
            FETCH_REQUEST_TIMEOUT,
            network_client.fetch_blocks(
                peer.clone(),
                vec![],
                fetch_after_rounds,
                false,
                FETCH_REQUEST_TIMEOUT,
            ),
        )
        .await;

        let serialized_blocks = match response {
            Ok(Ok(blocks)) => blocks,
            Ok(Err(err)) => {
                debug!("Failed to fetch blocks with fetch_after_rounds from peer {peer}: {err}");
                return vec![];
            }
            Err(_) => {
                debug!("Timed out fetching blocks with fetch_after_rounds from peer {peer}");
                return vec![];
            }
        };

        let blocks_guard = BlocksGuard {
            map: inflight_blocks,
            block_refs: BTreeSet::new(),
            peer: peer.clone(),
        };

        vec![(blocks_guard, serialized_blocks, peer)]
    }

    /// Fetches the `missing_blocks` from peers. Requests the same number of authorities with missing blocks from each peer.
    /// Each response from peer can contain the requested blocks, and additional blocks from the last accepted round for
    /// authorities with missing blocks.
    /// Each element of the vector is a tuple which contains the requested missing block refs, the returned blocks and
    /// the peer.
    async fn fetch_blocks_from_peers(
        context: Arc<Context>,
        inflight_blocks: Arc<InflightBlocksMap>,
        network_client: Arc<SynchronizerClient<VC, OC>>,
        missing_blocks: BTreeSet<BlockRef>,
        dag_state: Arc<RwLock<DagState>>,
        peers_pool: Arc<PeersPool>,
    ) -> Vec<(BlocksGuard, Vec<Bytes>, PeerId)> {
        // Preliminary truncation of missing blocks to fetch. Since each peer can have different
        // number of missing blocks and the fetching is batched by peer, so keep more than max_blocks_per_fetch
        // per peer on average.
        let missing_blocks = missing_blocks
            .into_iter()
            .take(2 * MAX_PERIODIC_SYNC_PEERS * context.parameters.max_blocks_per_fetch)
            .collect::<Vec<_>>();

        // Maps authorities to the missing blocks they have.
        let mut authorities = BTreeMap::<AuthorityIndex, Vec<BlockRef>>::new();
        for block_ref in &missing_blocks {
            authorities
                .entry(block_ref.author)
                .or_default()
                .push(*block_ref);
        }

        // Get known peers from the PeersPool
        let mut peers = peers_pool.get_known_peers();

        // Distribute the same number of authorities into each peer to sync.
        // Use the number of known peers from the pool, capped at MAX_PERIODIC_SYNC_PEERS
        // TODO: in the future it would be possible, temporarily, for an Observer node to not have peers to fetch from.
        // We should change this assertion to allow for this case.
        assert!(!peers.is_empty(), "No known peers to fetch blocks from");

        let num_authorities_per_peer = authorities
            .len()
            .div_ceil(peers.len().min(MAX_PERIODIC_SYNC_PEERS));

        // Update metrics related to missing blocks.
        let mut missing_blocks_per_authority = vec![0; context.committee.size()];
        for (authority, blocks) in &authorities {
            missing_blocks_per_authority[*authority] += blocks.len();
        }
        for (missing, (_, authority)) in missing_blocks_per_authority
            .into_iter()
            .zip_debug_eq(context.committee.authorities())
        {
            context
                .metrics
                .node_metrics
                .synchronizer_missing_blocks_by_authority
                .with_label_values(&[&authority.hostname])
                .inc_by(missing as u64);
            context
                .metrics
                .node_metrics
                .synchronizer_current_missing_blocks_by_authority
                .with_label_values(&[&authority.hostname])
                .set(missing as i64);
        }

        // TODO: probably inject the RNG to allow unit testing - this is a work around for now.
        if cfg!(not(test)) {
            // Shuffle the peers
            peers.shuffle(&mut ThreadRng::default());
        }

        let mut peers = peers.into_iter();
        let mut request_futures = FuturesUnordered::new();

        // Shuffle the authorities for each request.
        let mut authorities = authorities.into_values().collect::<Vec<_>>();
        if cfg!(not(test)) {
            // Shuffle the authorities
            authorities.shuffle(&mut ThreadRng::default());
        }

        let fetch_after_rounds = Self::get_fetch_after_rounds(&context, dag_state.clone());

        // Send the fetch requests
        for batch in authorities.chunks(num_authorities_per_peer) {
            let Some(peer) = peers.next() else {
                debug_fatal!("No more peers left to fetch blocks!");
                break;
            };
            let peer_name = peer.hostname(&context);
            // Fetch from the lowest round missing blocks to ensure progress.
            // This may reduce efficiency and increase the chance of duplicated data transfer in edge cases.
            let block_refs = batch
                .iter()
                .flatten()
                .cloned()
                .collect::<BTreeSet<_>>()
                .into_iter()
                .take(context.parameters.max_blocks_per_fetch)
                .collect::<BTreeSet<_>>();

            // lock the blocks to be fetched. If no lock can be acquired for any of the blocks then don't bother
            if let Some(blocks_guard) =
                inflight_blocks.lock_blocks(block_refs.clone(), peer.clone())
            {
                info!(
                    "Periodic sync of {} missing blocks from peer {} {:?}: {}",
                    peer_name.as_str(),
                    block_refs.len(),
                    peer,
                    block_refs
                        .iter()
                        .map(|b| b.to_string())
                        .collect::<Vec<_>>()
                        .join(", ")
                );
                request_futures.push(Self::fetch_blocks_request(
                    network_client.clone(),
                    peer,
                    blocks_guard,
                    fetch_after_rounds.clone(),
                    false,
                    FETCH_REQUEST_TIMEOUT,
                    1,
                ));
            }
        }

        let mut results = Vec::new();
        let fetcher_timeout = sleep(FETCH_FROM_PEERS_TIMEOUT);

        tokio::pin!(fetcher_timeout);

        loop {
            tokio::select! {
                Some((response, blocks_guard, _retries, peer, fetch_after_rounds)) = request_futures.next() => {
                    match response {
                        Ok(fetched_blocks) => {
                            results.push((blocks_guard, fetched_blocks, peer));

                            // no more pending requests are left, just break the loop
                            if request_futures.is_empty() {
                                break;
                            }
                        },
                        Err(_) => {
                            let peer_name = peer.labelname(&context);
                            context.metrics.node_metrics.synchronizer_fetch_failures.with_label_values(&[peer_name.as_str(), "periodic"]).inc();
                            // try again if there is any peer left
                            if let Some(next_peer) = peers.next() {
                                // do best effort to lock guards. If we can't lock then don't bother at this run.
                                if let Some(blocks_guard) = inflight_blocks.swap_locks(blocks_guard, next_peer.clone()) {
                                    info!(
                                        "Retrying syncing {} missing blocks from peer {:?}: {}",
                                        blocks_guard.block_refs.len(),
                                        next_peer,
                                        blocks_guard.block_refs
                                            .iter()
                                            .map(|b| b.to_string())
                                            .collect::<Vec<_>>()
                                            .join(", ")
                                    );
                                    request_futures.push(Self::fetch_blocks_request(
                                        network_client.clone(),
                                        next_peer,
                                        blocks_guard,
                                        fetch_after_rounds,
                                        false,
                                        FETCH_REQUEST_TIMEOUT,
                                        1,
                                    ));
                                } else {
                                    debug!("Couldn't acquire locks to fetch blocks from peer {:?}.", next_peer)
                                }
                            } else {
                                debug!("No more peers left to fetch blocks");
                            }
                        }
                    }
                },
                _ = &mut fetcher_timeout => {
                    debug!("Timed out while fetching missing blocks");
                    break;
                }
            }
        }

        results
    }
}

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

    use async_trait::async_trait;
    use bytes::Bytes;
    use consensus_config::{AuthorityIndex, Parameters};
    use consensus_types::block::{BlockDigest, BlockRef, Round};
    use mysten_metrics::monitored_mpsc;
    use parking_lot::RwLock;
    use tokio::{sync::Mutex, time::sleep};

    use crate::commit::{CommitVote, TrustedCommit};
    use crate::{
        CommitDigest, CommitIndex,
        block::{TestBlock, VerifiedBlock},
        block_verifier::NoopBlockVerifier,
        commit_vote_monitor::CommitVoteMonitor,
        context::Context,
        core_thread::CoreThreadDispatcher,
        dag_state::DagState,
        error::{ConsensusError, ConsensusResult},
        network::{
            BlockStream, ObserverNetworkClient, PeerId, SynchronizerClient, ValidatorNetworkClient,
        },
        storage::mem_store::MemStore,
        synchronizer::{
            COMMIT_PROGRESS_TIMEOUT, FETCH_BLOCKS_CONCURRENCY, FETCH_REQUEST_TIMEOUT,
            InflightBlocksMap, Synchronizer,
        },
    };
    use crate::{
        commit_vote_monitor::COMMIT_LAG_MULTIPLIER, core_thread::MockCoreThreadDispatcher,
        peers_pool::PeersPool, round_tracker::RoundTracker,
        transaction_vote_tracker::TransactionVoteTracker,
    };

    type FetchRequestKey = (Vec<BlockRef>, AuthorityIndex);
    type FetchRequestResponse = (Vec<VerifiedBlock>, Option<Duration>);
    type FetchLatestBlockKey = (AuthorityIndex, Vec<AuthorityIndex>);
    type FetchLatestBlockResponse = (Vec<VerifiedBlock>, Option<Duration>);

    #[derive(Default)]
    struct MockNetworkClient {
        fetch_blocks_requests: Mutex<BTreeMap<FetchRequestKey, FetchRequestResponse>>,
        fetch_latest_blocks_requests:
            Mutex<BTreeMap<FetchLatestBlockKey, Vec<FetchLatestBlockResponse>>>,
    }

    impl MockNetworkClient {
        async fn stub_fetch_blocks(
            &self,
            blocks: Vec<VerifiedBlock>,
            peer: AuthorityIndex,
            latency: Option<Duration>,
        ) {
            let mut lock = self.fetch_blocks_requests.lock().await;
            let block_refs = blocks
                .iter()
                .map(|block| block.reference())
                .collect::<Vec<_>>();
            lock.insert((block_refs, peer), (blocks, latency));
        }

        async fn stub_fetch_blocks_for_key(
            &self,
            key_refs: Vec<BlockRef>,
            response_blocks: Vec<VerifiedBlock>,
            peer: AuthorityIndex,
            latency: Option<Duration>,
        ) {
            let mut lock = self.fetch_blocks_requests.lock().await;
            lock.insert((key_refs, peer), (response_blocks, latency));
        }

        async fn stub_fetch_latest_blocks(
            &self,
            blocks: Vec<VerifiedBlock>,
            peer: AuthorityIndex,
            authorities: Vec<AuthorityIndex>,
            latency: Option<Duration>,
        ) {
            let mut lock = self.fetch_latest_blocks_requests.lock().await;
            lock.entry((peer, authorities))
                .or_default()
                .push((blocks, latency));
        }

        async fn fetch_latest_blocks_pending_calls(&self) -> usize {
            let lock = self.fetch_latest_blocks_requests.lock().await;
            lock.len()
        }
    }

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

        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>> {
            let mut lock = self.fetch_blocks_requests.lock().await;
            let response = lock.remove(&(block_refs.clone(), peer)).unwrap_or_else(|| {
                panic!(
                    "Unexpected fetch blocks request made: {:?} {}. Current lock: {:?}",
                    block_refs, peer, lock
                );
            });

            let serialised = response
                .0
                .into_iter()
                .map(|block| block.serialized().clone())
                .collect::<Vec<_>>();

            drop(lock);

            if let Some(latency) = response.1 {
                sleep(latency).await;
            }

            Ok(serialised)
        }

        async fn fetch_commits(
            &self,
            _peer: AuthorityIndex,
            _commit_range: crate::commit::CommitRange,
            _timeout: Duration,
        ) -> ConsensusResult<(Vec<Bytes>, Vec<Bytes>)> {
            unimplemented!("fetch_commits not implemented in mock")
        }

        async fn fetch_latest_blocks(
            &self,
            peer: AuthorityIndex,
            authorities: Vec<AuthorityIndex>,
            _timeout: Duration,
        ) -> ConsensusResult<Vec<Bytes>> {
            let mut lock = self.fetch_latest_blocks_requests.lock().await;
            let mut responses = lock
                .remove(&(peer, authorities.clone()))
                .expect("Unexpected fetch blocks request made");

            let response = responses.remove(0);
            let serialised = response
                .0
                .into_iter()
                .map(|block| block.serialized().clone())
                .collect::<Vec<_>>();

            if !responses.is_empty() {
                lock.insert((peer, authorities), responses);
            }

            drop(lock);

            if let Some(latency) = response.1 {
                sleep(latency).await;
            }

            Ok(serialised)
        }

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

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

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

        async fn fetch_blocks(
            &self,
            _peer: crate::network::PeerId,
            _block_refs: Vec<BlockRef>,
            _fetch_after_rounds: Vec<Round>,
            _fetch_missing_ancestors: bool,
            _timeout: Duration,
        ) -> ConsensusResult<Vec<Bytes>> {
            unimplemented!("Observer fetch_blocks not implemented in mock")
        }

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

    #[test]
    fn test_inflight_blocks_map() {
        // GIVEN
        let map = InflightBlocksMap::new();
        let some_block_refs = [
            BlockRef::new(1, AuthorityIndex::new_for_test(0), BlockDigest::MIN),
            BlockRef::new(10, AuthorityIndex::new_for_test(0), BlockDigest::MIN),
            BlockRef::new(12, AuthorityIndex::new_for_test(3), BlockDigest::MIN),
            BlockRef::new(15, AuthorityIndex::new_for_test(2), BlockDigest::MIN),
        ];
        let missing_block_refs = some_block_refs.iter().cloned().collect::<BTreeSet<_>>();

        // Lock & unlock blocks
        {
            let mut all_guards = Vec::new();

            // Try to acquire the block locks for authorities 1 & 2
            for i in 1..=2 {
                let authority = AuthorityIndex::new_for_test(i);
                let peer = PeerId::Validator(authority);

                let guard = map.lock_blocks(missing_block_refs.clone(), peer.clone());
                let guard = guard.expect("Guard should be created");
                assert_eq!(guard.block_refs.len(), 4);

                all_guards.push(guard);

                // trying to acquire any of them again will not succeed
                let guard = map.lock_blocks(missing_block_refs.clone(), peer);
                assert!(guard.is_none());
            }

            // Trying to acquire for authority 3 it will fail - as we have maxed out the number of allowed peers
            let authority_3 = AuthorityIndex::new_for_test(3);
            let peer_3 = PeerId::Validator(authority_3);

            let guard = map.lock_blocks(missing_block_refs.clone(), peer_3.clone());
            assert!(guard.is_none());

            // Explicitly drop the guard of authority 1 and try for authority 3 again - it will now succeed
            drop(all_guards.remove(0));

            let guard = map.lock_blocks(missing_block_refs.clone(), peer_3);
            let guard = guard.expect("Guard should be successfully acquired");

            assert_eq!(guard.block_refs, missing_block_refs);

            // Dropping all guards should unlock on the block refs
            drop(guard);
            drop(all_guards);

            assert_eq!(map.num_of_locked_blocks(), 0);
        }

        // Swap locks
        {
            // acquire a lock for authority 1
            let authority_1 = AuthorityIndex::new_for_test(1);
            let peer_1 = PeerId::Validator(authority_1);
            let guard = map.lock_blocks(missing_block_refs.clone(), peer_1).unwrap();

            // Now swap the locks for authority 2
            let authority_2 = AuthorityIndex::new_for_test(2);
            let peer_2 = PeerId::Validator(authority_2);
            let guard = map.swap_locks(guard, peer_2);

            assert_eq!(guard.unwrap().block_refs, missing_block_refs);
        }
    }

    #[tokio::test]
    async fn successful_fetch_blocks_from_peer() {
        // GIVEN
        let (context, _) = Context::new_for_test(4);
        let context = Arc::new(context);
        let block_verifier = Arc::new(NoopBlockVerifier {});
        let core_dispatcher = Arc::new(MockCoreThreadDispatcher::default());
        let commit_vote_monitor = Arc::new(CommitVoteMonitor::new(context.clone()));
        let mock_client = Arc::new(MockNetworkClient::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 round_tracker = Arc::new(RwLock::new(RoundTracker::new(context.clone(), vec![])));

        let network_client = Arc::new(SynchronizerClient::new(
            context.clone(),
            Some(mock_client.clone()),
            Some(mock_client.clone()),
        ));
        let peers_pool = Arc::new(PeersPool::new(context.clone()));
        let handle = Synchronizer::start(
            network_client,
            context.clone(),
            core_dispatcher.clone(),
            commit_vote_monitor,
            block_verifier,
            transaction_vote_tracker,
            round_tracker,
            dag_state,
            peers_pool.clone(),
            false,
        );

        // Create some test blocks
        let expected_blocks = (0..10)
            .map(|round| VerifiedBlock::new_for_test(TestBlock::new(round, 0).build()))
            .collect::<Vec<_>>();
        let missing_blocks = expected_blocks
            .iter()
            .map(|block| block.reference())
            .collect::<BTreeSet<_>>();

        // AND stub the fetch_blocks request from peer 1
        let peer = AuthorityIndex::new_for_test(1);
        mock_client
            .stub_fetch_blocks(expected_blocks.clone(), peer, None)
            .await;

        // WHEN request missing blocks from peer 1
        assert!(
            handle
                .fetch_blocks(missing_blocks, PeerId::Validator(peer))
                .await
                .is_ok()
        );

        // Wait a little bit until those have been added in core
        sleep(Duration::from_millis(1_000)).await;

        // THEN ensure those ended up in Core
        let added_blocks = core_dispatcher.get_add_blocks().await;
        assert_eq!(added_blocks, expected_blocks);
    }

    #[tokio::test]
    async fn saturate_fetch_blocks_from_peer() {
        // GIVEN
        let (context, _) = Context::new_for_test(4);
        let context = Arc::new(context);
        let block_verifier = Arc::new(NoopBlockVerifier {});
        let commit_vote_monitor = Arc::new(CommitVoteMonitor::new(context.clone()));
        let core_dispatcher = Arc::new(MockCoreThreadDispatcher::default());
        let mock_client = Arc::new(MockNetworkClient::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 round_tracker = Arc::new(RwLock::new(RoundTracker::new(context.clone(), vec![])));

        let network_client = Arc::new(SynchronizerClient::new(
            context.clone(),
            Some(mock_client.clone()),
            Some(mock_client.clone()),
        ));
        let peers_pool = Arc::new(PeersPool::new(context.clone()));
        let handle = Synchronizer::start(
            network_client,
            context.clone(),
            core_dispatcher.clone(),
            commit_vote_monitor,
            block_verifier,
            transaction_vote_tracker,
            round_tracker,
            dag_state,
            peers_pool.clone(),
            false,
        );

        // Create some test blocks
        let expected_blocks = (0..=2 * FETCH_BLOCKS_CONCURRENCY)
            .map(|round| VerifiedBlock::new_for_test(TestBlock::new(round as Round, 0).build()))
            .collect::<Vec<_>>();

        // Now start sending requests to fetch blocks by trying to saturate peer 1 task
        let peer = AuthorityIndex::new_for_test(1);
        let mut iter = expected_blocks.iter().peekable();
        while let Some(block) = iter.next() {
            // stub the fetch_blocks request from peer 1 and give some high response latency so requests
            // can start blocking the peer task.
            mock_client
                .stub_fetch_blocks(
                    vec![block.clone()],
                    peer,
                    Some(Duration::from_millis(5_000)),
                )
                .await;

            let mut missing_blocks = BTreeSet::new();
            missing_blocks.insert(block.reference());

            // WHEN requesting to fetch the blocks, it should not succeed for the last request and get
            // an error with "saturated" synchronizer
            if iter.peek().is_none() {
                match handle
                    .fetch_blocks(missing_blocks, PeerId::Validator(peer))
                    .await
                {
                    Err(ConsensusError::SynchronizerSaturated(peer_str)) => {
                        assert_eq!(peer_str, format!("{:?}", PeerId::Validator(peer)));
                    }
                    _ => panic!("A saturated synchronizer error was expected"),
                }
            } else {
                assert!(
                    handle
                        .fetch_blocks(missing_blocks, PeerId::Validator(peer))
                        .await
                        .is_ok()
                );
            }
        }
    }

    #[tokio::test(flavor = "current_thread", start_paused = true)]
    async fn synchronizer_periodic_task_fetch_blocks() {
        // GIVEN
        let (context, _) = Context::new_for_test(4);
        let context = Arc::new(context);
        let block_verifier = Arc::new(NoopBlockVerifier {});
        let commit_vote_monitor = Arc::new(CommitVoteMonitor::new(context.clone()));
        let core_dispatcher = Arc::new(MockCoreThreadDispatcher::default());
        let mock_client = Arc::new(MockNetworkClient::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 round_tracker = Arc::new(RwLock::new(RoundTracker::new(context.clone(), vec![])));

        // Create some test blocks
        let expected_blocks = (0..10)
            .map(|round| VerifiedBlock::new_for_test(TestBlock::new(round, 0).build()))
            .collect::<Vec<_>>();
        let missing_blocks = expected_blocks
            .iter()
            .map(|block| block.reference())
            .collect::<BTreeSet<_>>();

        // AND stub the missing blocks
        core_dispatcher
            .stub_missing_blocks(missing_blocks.clone())
            .await;

        // AND stub the requests for authority 1 & 2
        // Make the first authority timeout, so the second will be called. "We" are authority = 0, so
        // we are skipped anyways.
        mock_client
            .stub_fetch_blocks(
                expected_blocks.clone(),
                AuthorityIndex::new_for_test(1),
                Some(FETCH_REQUEST_TIMEOUT),
            )
            .await;
        mock_client
            .stub_fetch_blocks(
                expected_blocks.clone(),
                AuthorityIndex::new_for_test(2),
                None,
            )
            .await;

        // WHEN start the synchronizer and wait for a couple of seconds
        let network_client = Arc::new(SynchronizerClient::new(
            context.clone(),
            Some(mock_client.clone()),
            Some(mock_client.clone()),
        ));
        let peers_pool = Arc::new(PeersPool::new(context.clone()));
        let _handle = Synchronizer::start(
            network_client,
            context.clone(),
            core_dispatcher.clone(),
            commit_vote_monitor,
            block_verifier,
            transaction_vote_tracker,
            round_tracker,
            dag_state,
            peers_pool.clone(),
            false,
        );

        sleep(2 * FETCH_REQUEST_TIMEOUT).await;

        // THEN the missing blocks should now be fetched and added to core
        let added_blocks = core_dispatcher.get_add_blocks().await;
        assert_eq!(added_blocks, expected_blocks);

        // AND missing blocks should have been consumed by the stub
        assert!(
            core_dispatcher
                .get_missing_blocks()
                .await
                .unwrap()
                .is_empty()
        );
    }

    #[tokio::test(flavor = "current_thread", start_paused = true)]
    async fn synchronizer_periodic_task_when_commit_lagging_gets_disabled() {
        // GIVEN
        let (context, _) = Context::new_for_test(4);
        let context = Arc::new(context);
        let block_verifier = Arc::new(NoopBlockVerifier {});
        let core_dispatcher = Arc::new(MockCoreThreadDispatcher::default());
        let mock_client = Arc::new(MockNetworkClient::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 round_tracker = Arc::new(RwLock::new(RoundTracker::new(context.clone(), vec![])));

        // AND stub some missing blocks. The highest accepted round is 0. Create blocks that are above the sync threshold.
        let sync_missing_block_round_threshold = context.parameters.commit_sync_batch_size;
        let stub_blocks = (sync_missing_block_round_threshold * 2
            ..sync_missing_block_round_threshold * 3)
            .map(|round| VerifiedBlock::new_for_test(TestBlock::new(round, 0).build()))
            .collect::<Vec<_>>();
        let missing_blocks = stub_blocks
            .iter()
            .map(|block| block.reference())
            .collect::<BTreeSet<_>>();
        core_dispatcher
            .stub_missing_blocks(missing_blocks.clone())
            .await;

        // AND stub the requests for authority 1 & 2
        // Make the first authority timeout, so the second will be called. "We" are authority = 0, so
        // we are skipped anyways. Stub all blocks since the full set is sent in one request.
        // Only the first max_blocks_per_sync blocks will be processed by process_fetched_blocks.
        mock_client
            .stub_fetch_blocks(
                stub_blocks.clone(),
                AuthorityIndex::new_for_test(1),
                Some(FETCH_REQUEST_TIMEOUT),
            )
            .await;
        mock_client
            .stub_fetch_blocks(stub_blocks.clone(), AuthorityIndex::new_for_test(2), None)
            .await;
        let mut expected_blocks = stub_blocks
            .iter()
            .take(context.parameters.max_blocks_per_sync)
            .cloned()
            .collect::<Vec<_>>();

        // Now create some blocks to simulate a commit lag
        let round = context.parameters.commit_sync_batch_size * COMMIT_LAG_MULTIPLIER * 2;
        let commit_index: CommitIndex = round - 1;
        let blocks = (0..4)
            .map(|authority| {
                let commit_votes = vec![CommitVote::new(commit_index, CommitDigest::MIN)];
                let block = TestBlock::new(round, authority)
                    .set_commit_votes(commit_votes)
                    .build();

                VerifiedBlock::new_for_test(block)
            })
            .collect::<Vec<_>>();

        // Pass them through the commit vote monitor - so now there will be a big commit lag to prevent
        // the scheduled synchronizer from running
        for block in blocks {
            commit_vote_monitor.observe_block(&block);
        }

        // WHEN start the synchronizer and wait for a couple of seconds where normally the synchronizer should have kicked in.
        let network_client = Arc::new(SynchronizerClient::new(
            context.clone(),
            Some(mock_client.clone()),
            Some(mock_client.clone()),
        ));
        let peers_pool = Arc::new(PeersPool::new(context.clone()));
        let _handle = Synchronizer::start(
            network_client,
            context.clone(),
            core_dispatcher.clone(),
            commit_vote_monitor.clone(),
            block_verifier,
            transaction_vote_tracker,
            round_tracker,
            dag_state.clone(),
            peers_pool.clone(),
            false,
        );

        // Wait long enough for periodic sync to have run, but stay under COMMIT_PROGRESS_TIMEOUT
        // to avoid triggering commit sync failover.
        sleep(COMMIT_PROGRESS_TIMEOUT / 2).await;

        // Since we should be in commit lag mode none of the missed blocks should have been fetched - hence nothing should be
        // sent to core for processing.
        let added_blocks = core_dispatcher.get_add_blocks().await;
        assert_eq!(added_blocks, vec![]);

        // AND advance now the local commit index by adding a new commit that matches the commit index
        // of quorum
        {
            let mut d = dag_state.write();
            for index in 1..=commit_index {
                let commit =
                    TrustedCommit::new_for_test(index, CommitDigest::MIN, 0, BlockRef::MIN, vec![]);

                d.add_commit(commit);
            }

            assert_eq!(
                d.last_commit_index(),
                commit_vote_monitor.quorum_commit_index()
            );
        }

        // Now stub again the missing blocks to fetch the exact same ones.
        core_dispatcher
            .stub_missing_blocks(missing_blocks.clone())
            .await;

        sleep(2 * FETCH_REQUEST_TIMEOUT).await;

        // THEN the missing blocks should now be fetched and added to core
        let mut added_blocks = core_dispatcher.get_add_blocks().await;

        added_blocks.sort_by_key(|block| block.reference());
        expected_blocks.sort_by_key(|block| block.reference());

        assert_eq!(added_blocks, expected_blocks);
    }

    #[tokio::test(flavor = "current_thread", start_paused = true)]
    async fn synchronizer_periodic_sync_resumes_when_commit_sync_stalled() {
        // GIVEN
        let (context, _) = Context::new_for_test(4);
        let context = Arc::new(context);
        let block_verifier = Arc::new(NoopBlockVerifier {});
        let core_dispatcher = Arc::new(MockCoreThreadDispatcher::default());
        let mock_client = Arc::new(MockNetworkClient::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 round_tracker = Arc::new(RwLock::new(RoundTracker::new(context.clone(), vec![])));

        // AND create missing blocks to fetch
        let sync_missing_block_round_threshold = context.parameters.commit_sync_batch_size;
        let stub_blocks = (sync_missing_block_round_threshold * 2
            ..sync_missing_block_round_threshold * 3)
            .map(|round| VerifiedBlock::new_for_test(TestBlock::new(round, 0).build()))
            .collect::<Vec<_>>();
        let missing_blocks = stub_blocks
            .iter()
            .map(|block| block.reference())
            .collect::<BTreeSet<_>>();
        core_dispatcher
            .stub_missing_blocks(missing_blocks.clone())
            .await;

        // AND stub the fetch responses for authority 1 & 2.
        // Stub all blocks since the full set is sent in one request.
        mock_client
            .stub_fetch_blocks(
                stub_blocks.clone(),
                AuthorityIndex::new_for_test(1),
                Some(FETCH_REQUEST_TIMEOUT),
            )
            .await;
        mock_client
            .stub_fetch_blocks(stub_blocks.clone(), AuthorityIndex::new_for_test(2), None)
            .await;
        let mut expected_blocks = stub_blocks
            .iter()
            .take(context.parameters.max_blocks_per_sync)
            .cloned()
            .collect::<Vec<_>>();

        // AND create commit lag by observing high commit votes
        let round = context.parameters.commit_sync_batch_size * COMMIT_LAG_MULTIPLIER * 2;
        let commit_index: CommitIndex = round - 1;
        let blocks = (0..4)
            .map(|authority| {
                let commit_votes = vec![CommitVote::new(commit_index, CommitDigest::MIN)];
                let block = TestBlock::new(round, authority)
                    .set_commit_votes(commit_votes)
                    .build();
                VerifiedBlock::new_for_test(block)
            })
            .collect::<Vec<_>>();
        for block in blocks {
            commit_vote_monitor.observe_block(&block);
        }

        // WHEN start the synchronizer
        let network_client = Arc::new(SynchronizerClient::new(
            context.clone(),
            Some(mock_client.clone()),
            Some(mock_client.clone()),
        ));
        let peers_pool = Arc::new(PeersPool::new(context.clone()));
        let _handle = Synchronizer::start(
            network_client,
            context.clone(),
            core_dispatcher.clone(),
            commit_vote_monitor.clone(),
            block_verifier,
            transaction_vote_tracker,
            round_tracker,
            dag_state.clone(),
            peers_pool.clone(),
            false,
        );

        // Wait just under COMMIT_PROGRESS_TIMEOUT — sync should be skipped (commit lagging).
        sleep(COMMIT_PROGRESS_TIMEOUT - Duration::from_millis(100)).await;
        let added_blocks = core_dispatcher.get_add_blocks().await;
        assert_eq!(added_blocks, vec![]);

        // Re-stub missing blocks (consumed by earlier get_missing_blocks calls).
        core_dispatcher
            .stub_missing_blocks(missing_blocks.clone())
            .await;
        // Stub the failover fetch (empty block_refs key, peer 1) before the stall triggers.
        // Use latency to prevent immediate completion and re-triggering.
        mock_client
            .stub_fetch_blocks_for_key(
                vec![],
                expected_blocks.clone(),
                AuthorityIndex::new_for_test(1),
                Some(Duration::from_millis(500)),
            )
            .await;

        // Now sleep past the stall trigger + time for the fetch to complete.
        sleep(Duration::from_millis(200) + FETCH_REQUEST_TIMEOUT).await;

        let mut added_blocks = core_dispatcher.get_add_blocks().await;
        assert!(
            !added_blocks.is_empty(),
            "Expected periodic sync to resume after commit sync stall"
        );
        added_blocks.sort_by_key(|block| block.reference());
        expected_blocks.sort_by_key(|block| block.reference());
        assert_eq!(added_blocks, expected_blocks);

        // AND advance commit index enough to reach commit sync batch size from stall start.
        // This should resolve the failover and skip periodic sync again.
        core_dispatcher.get_add_blocks().await;
        {
            let current = dag_state.read().last_commit_index();
            let mut d = dag_state.write();
            // Advance well past the recovery threshold.
            for index in (current + 1)..=(current + context.parameters.commit_sync_batch_size) {
                let commit =
                    TrustedCommit::new_for_test(index, CommitDigest::MIN, 0, BlockRef::MIN, vec![]);
                d.add_commit(commit);
            }
        }

        // Re-stub missing blocks and fetch response for another round of checking
        core_dispatcher
            .stub_missing_blocks(missing_blocks.clone())
            .await;
        mock_client
            .stub_fetch_blocks(
                stub_blocks
                    .iter()
                    .take(context.parameters.max_blocks_per_sync)
                    .cloned()
                    .collect::<Vec<_>>(),
                AuthorityIndex::new_for_test(2),
                None,
            )
            .await;

        // Wait for a sync cycle — periodic sync should be skipped again since stall resolved
        // but commit is still lagging
        sleep(2 * FETCH_REQUEST_TIMEOUT).await;
        let added_blocks = core_dispatcher.get_add_blocks().await;
        assert_eq!(
            added_blocks,
            vec![],
            "Expected periodic sync to be skipped after stall resolved"
        );
    }

    #[tokio::test(flavor = "current_thread", start_paused = true)]
    async fn synchronizer_fetch_own_last_block() {
        // GIVEN
        let (context, _) = Context::new_for_test(4);
        let context = Arc::new(context.with_parameters(Parameters {
            sync_last_known_own_block_timeout: Duration::from_millis(2_000),
            ..Default::default()
        }));
        let block_verifier = Arc::new(NoopBlockVerifier {});
        let core_dispatcher = Arc::new(MockCoreThreadDispatcher::default());
        let mock_client = Arc::new(MockNetworkClient::default());
        let commit_vote_monitor = Arc::new(CommitVoteMonitor::new(context.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 round_tracker = Arc::new(RwLock::new(RoundTracker::new(context.clone(), vec![])));
        let our_index = AuthorityIndex::new_for_test(0);

        // Create some test blocks
        let mut expected_blocks = (9..=10)
            .map(|round| VerifiedBlock::new_for_test(TestBlock::new(round, 0).build()))
            .collect::<Vec<_>>();

        // Now set different latest blocks for the peers
        // For peer 1 we give the block of round 10 (highest)
        let block_1 = expected_blocks.pop().unwrap();
        mock_client
            .stub_fetch_latest_blocks(
                vec![block_1.clone()],
                AuthorityIndex::new_for_test(1),
                vec![our_index],
                None,
            )
            .await;
        mock_client
            .stub_fetch_latest_blocks(
                vec![block_1],
                AuthorityIndex::new_for_test(1),
                vec![our_index],
                None,
            )
            .await;

        // For peer 2 we give the block of round 9
        let block_2 = expected_blocks.pop().unwrap();
        mock_client
            .stub_fetch_latest_blocks(
                vec![block_2.clone()],
                AuthorityIndex::new_for_test(2),
                vec![our_index],
                Some(Duration::from_secs(10)),
            )
            .await;
        mock_client
            .stub_fetch_latest_blocks(
                vec![block_2],
                AuthorityIndex::new_for_test(2),
                vec![our_index],
                None,
            )
            .await;

        // For peer 3 we don't give any block - and it should return an empty vector
        mock_client
            .stub_fetch_latest_blocks(
                vec![],
                AuthorityIndex::new_for_test(3),
                vec![our_index],
                Some(Duration::from_secs(10)),
            )
            .await;
        mock_client
            .stub_fetch_latest_blocks(
                vec![],
                AuthorityIndex::new_for_test(3),
                vec![our_index],
                None,
            )
            .await;

        // WHEN start the synchronizer and wait for a couple of seconds
        let network_client = Arc::new(SynchronizerClient::new(
            context.clone(),
            Some(mock_client.clone()),
            Some(mock_client.clone()),
        ));
        let peers_pool = Arc::new(PeersPool::new(context.clone()));
        let handle = Synchronizer::start(
            network_client,
            context.clone(),
            core_dispatcher.clone(),
            commit_vote_monitor,
            block_verifier,
            transaction_vote_tracker,
            round_tracker,
            dag_state,
            peers_pool.clone(),
            true,
        );

        // Wait at least for the timeout time
        sleep(context.parameters.sync_last_known_own_block_timeout * 2).await;

        // Assert that core has been called to set the min propose round
        assert_eq!(
            core_dispatcher.get_last_own_proposed_round().await,
            vec![10]
        );

        // Ensure that all the requests have been called
        assert_eq!(mock_client.fetch_latest_blocks_pending_calls().await, 0);

        // And we got one retry
        assert_eq!(
            context
                .metrics
                .node_metrics
                .sync_last_known_own_block_retries
                .get(),
            1
        );

        // Ensure that no panic occurred: stop() propagates panics from synchronizer tasks.
        handle.stop().await;
    }

    #[tokio::test]
    async fn test_process_fetched_blocks() {
        // GIVEN
        let (context, _) = Context::new_for_test(4);
        let context = Arc::new(context);
        let block_verifier = Arc::new(NoopBlockVerifier {});
        let core_dispatcher = Arc::new(MockCoreThreadDispatcher::default());
        let commit_vote_monitor = Arc::new(CommitVoteMonitor::new(context.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 round_tracker = Arc::new(RwLock::new(RoundTracker::new(context.clone(), vec![])));
        let (commands_sender, _commands_receiver) =
            monitored_mpsc::channel("consensus_synchronizer_commands", 1000);

        // Create input test blocks:
        // - Authority 0 block at round 60.
        // - Authority 1 blocks from round 30 to 60.
        let mut expected_blocks = vec![VerifiedBlock::new_for_test(TestBlock::new(60, 0).build())];
        expected_blocks.extend(
            (30..=60).map(|round| VerifiedBlock::new_for_test(TestBlock::new(round, 1).build())),
        );
        assert_eq!(
            expected_blocks.len(),
            context.parameters.max_blocks_per_sync
        );

        let expected_serialized_blocks = expected_blocks
            .iter()
            .map(|b| b.serialized().clone())
            .collect::<Vec<_>>();

        let expected_block_refs = expected_blocks
            .iter()
            .map(|b| b.reference())
            .collect::<BTreeSet<_>>();

        // GIVEN peer to fetch blocks from
        let peer_index = AuthorityIndex::new_for_test(2);
        let peer = PeerId::Validator(peer_index);

        // Create blocks_guard
        let inflight_blocks_map = InflightBlocksMap::new();
        let blocks_guard = inflight_blocks_map
            .lock_blocks(expected_block_refs.clone(), peer.clone())
            .expect("Failed to lock blocks");

        assert_eq!(
            inflight_blocks_map.num_of_locked_blocks(),
            expected_block_refs.len()
        );

        // Create a Synchronizer
        let result = Synchronizer::<
            NoopBlockVerifier,
            MockCoreThreadDispatcher,
            MockNetworkClient,
            MockNetworkClient,
        >::process_fetched_blocks(
            expected_serialized_blocks,
            peer,
            blocks_guard, // The guard is consumed here
            core_dispatcher.clone(),
            block_verifier,
            transaction_vote_tracker,
            commit_vote_monitor,
            context.clone(),
            commands_sender,
            round_tracker,
            "test",
        )
        .await;

        // THEN
        assert!(result.is_ok());

        // Check blocks were sent to core
        let added_blocks = core_dispatcher.get_add_blocks().await;
        assert_eq!(
            added_blocks
                .iter()
                .map(|b| b.reference())
                .collect::<BTreeSet<_>>(),
            expected_block_refs,
        );

        // Check blocks were unlocked
        assert_eq!(inflight_blocks_map.num_of_locked_blocks(), 0);
    }
}