forest/chain_sync/

chain_follower.rs

// Copyright 2019-2026 ChainSafe Systems
// SPDX-License-Identifier: Apache-2.0, MIT
//! This module contains the logic for driving Forest forward in the Filecoin
//! blockchain.
//!
//! Forest keeps track of the current heaviest tipset, and receives information
//! about new blocks and tipsets from peers as well as connected miners. The
//! state machine has the following rules:
//! - A tipset is invalid if its parent is invalid.
//! - If a tipset's parent isn't in our database, request it from the network.
//! - If a tipset's parent has been validated, validate the tipset.
//! - If a tipset is 1 day older than the heaviest tipset, the tipset is
//!   invalid. This prevents Forest from following forks that will never be
//!   accepted.
//!
//! The state machine does not do any network requests or validation. Those are
//! handled by an external actor.

use super::network_context::SyncNetworkContext;
use crate::{
    blocks::{Block, FullTipset, Tipset, TipsetKey},
    chain::{ChainStore, index::ResolveNullTipset},
    chain_sync::{
        ForkSyncInfo, ForkSyncStage, SyncStatus, SyncStatusReport, TipsetValidator,
        bad_block_cache::{BadBlockCache, SeenBlockCache},
        metrics,
        tipset_syncer::{TipsetSyncerError, validate_tipset},
        validation::GossipBlockValidator,
    },
    libp2p::{NetworkEvent, PubsubMessage, hello::HelloRequest},
    message_pool::MessagePool,
    networks::calculate_expected_epoch,
    prelude::*,
    shim::clock::ChainEpoch,
    state_manager::StateManager,
    utils::misc::env::env_or_default_logged,
};
use arc_swap::ArcSwap;
use chrono::Utc;
use hashbrown::{HashMap, HashSet};
use libp2p::PeerId;
use nonzero_ext::nonzero;
use parking_lot::Mutex;
use std::{
    borrow::Cow,
    sync::LazyLock,
    time::{Duration, Instant},
};
use tokio::{
    sync::{Notify, OwnedSemaphorePermit, Semaphore},
    task::JoinSet,
};
use tokio_util::sync::CancellationToken;

pub struct ChainFollower {
    /// Tasks
    tasks: Arc<Mutex<HashSet<SyncTask>>>,

    /// State machine
    state_machine: Arc<Mutex<SyncStateMachine>>,

    /// Syncing status of the chain
    pub sync_status: SyncStatus,

    /// manages retrieving and updates state objects
    pub state_manager: StateManager,

    /// Context to be able to send requests to P2P network
    pub network: SyncNetworkContext,

    /// Genesis tipset
    genesis: Tipset,

    /// Bad blocks cache, updates based on invalid state transitions.
    /// Will mark any invalid blocks and all children as bad in this bounded
    /// cache
    pub bad_blocks: Option<BadBlockCache>,

    /// Incoming network events to be handled by synchronizer
    net_handler: Arc<flume::Receiver<NetworkEvent>>,

    /// Tipset channel sender
    pub tipset_sender: flume::Sender<FullTipset>,

    /// Tipset channel receiver
    tipset_receiver: Arc<flume::Receiver<FullTipset>>,

    /// When `stateless_mode` is true, forest connects to the P2P network but
    /// does not execute any state transitions. This drastically reduces the
    /// memory and disk footprint of Forest but also means that Forest will not
    /// be able to validate the correctness of the chain.
    stateless_mode: bool,

    /// Message pool
    mem_pool: MessagePool<ChainStore>,
}

impl ShallowClone for ChainFollower {
    fn shallow_clone(&self) -> Self {
        Self {
            tasks: self.tasks.shallow_clone(),
            state_machine: self.state_machine.shallow_clone(),
            sync_status: self.sync_status.shallow_clone(),
            state_manager: self.state_manager.shallow_clone(),
            network: self.network.shallow_clone(),
            genesis: self.genesis.shallow_clone(),
            bad_blocks: self.bad_blocks.shallow_clone(),
            net_handler: self.net_handler.shallow_clone(),
            tipset_sender: self.tipset_sender.clone(),
            tipset_receiver: self.tipset_receiver.shallow_clone(),
            stateless_mode: self.stateless_mode,
            mem_pool: self.mem_pool.shallow_clone(),
        }
    }
}

impl ChainFollower {
    pub fn new(
        state_manager: StateManager,
        network: SyncNetworkContext,
        genesis: Tipset,
        net_handler: flume::Receiver<NetworkEvent>,
        stateless_mode: bool,
        mem_pool: MessagePool<ChainStore>,
    ) -> Self {
        crate::def_is_env_truthy!(cache_disabled, "FOREST_DISABLE_BAD_BLOCK_CACHE");
        let (tipset_sender, tipset_receiver) = flume::bounded(20);
        let tasks: Arc<Mutex<HashSet<SyncTask>>> = Arc::new(Mutex::new(HashSet::default()));
        let bad_blocks = if cache_disabled() {
            tracing::warn!("bad block cache is disabled by `FOREST_DISABLE_BAD_BLOCK_CACHE`");
            None
        } else {
            Some(Default::default())
        };
        let state_machine = Arc::new(Mutex::new(SyncStateMachine::new(
            state_manager.chain_store().shallow_clone(),
            bad_blocks.shallow_clone(),
            stateless_mode,
        )));

        crate::metrics::register_collector(Box::new(SyncTasks(tasks.shallow_clone())));
        crate::metrics::register_collector(Box::new(SyncStateMachineWrapper(
            state_machine.shallow_clone(),
        )));

        Self {
            tasks,
            state_machine,
            sync_status: Arc::new(ArcSwap::from_pointee(SyncStatusReport::init())),
            state_manager,
            network,
            genesis,
            bad_blocks,
            net_handler: net_handler.into(),
            tipset_sender,
            tipset_receiver: tipset_receiver.into(),
            stateless_mode,
            mem_pool,
        }
    }

    /// Reset inner states
    pub fn reset(&self) {
        let start = Instant::now();
        self.tasks.lock().clear();
        self.state_manager.chain_store().validated_blocks.clear();
        self.state_machine.lock().tipsets.clear();
        if let Some(bad_blocks) = &self.bad_blocks {
            bad_blocks.clear();
        }
        tracing::info!(
            "chain follower reset, took {}",
            humantime::format_duration(start.elapsed())
        );
    }

    pub async fn run(&self) -> anyhow::Result<()> {
        chain_follower(
            &self.tasks,
            &self.state_machine,
            &self.state_manager,
            self.bad_blocks.shallow_clone(),
            self.net_handler.shallow_clone(),
            self.tipset_sender.clone(),
            self.tipset_receiver.shallow_clone(),
            &self.network,
            &self.mem_pool,
            &self.sync_status,
            &self.genesis,
            self.stateless_mode,
        )
        .await
    }

    /// Subscribe to validated tipsets.
    pub fn subscribe_validated_tipset(&self) -> tokio::sync::broadcast::Receiver<TipsetKey> {
        self.state_machine
            .lock()
            .validated_tipset_broadcast_tx
            .subscribe()
    }
}

#[allow(clippy::too_many_arguments)]
// We receive new full tipsets from the p2p swarm, and from miners that use Forest as their frontend.
async fn chain_follower(
    tasks: &Arc<Mutex<HashSet<SyncTask>>>,
    state_machine: &Arc<Mutex<SyncStateMachine>>,
    state_manager: &StateManager,
    bad_block_cache: Option<BadBlockCache>,
    network_rx: Arc<flume::Receiver<NetworkEvent>>,
    tipset_sender: flume::Sender<FullTipset>,
    tipset_receiver: Arc<flume::Receiver<FullTipset>>,
    network: &SyncNetworkContext,
    mem_pool: &MessagePool<ChainStore>,
    sync_status: &SyncStatus,
    genesis: &Tipset,
    stateless_mode: bool,
) -> anyhow::Result<()> {
    let state_changed = Arc::new(Notify::new());

    let seen_block_cache = SeenBlockCache::default();

    let hello_fetch_limiter = Arc::new(Semaphore::new(*MAX_CONCURRENT_HELLO_TRIGGERED_FETCHES));

    let mut set = JoinSet::new();
    let cancellation_token = CancellationToken::new();
    let _cancellation_token_drop_guard = cancellation_token.drop_guard_ref();

    // Increment metrics, update peer information, and forward tipsets to the state machine.
    set.spawn({
        let state_manager = state_manager.shallow_clone();
        let network = network.shallow_clone();
        let mem_pool = mem_pool.shallow_clone();
        let genesis = genesis.shallow_clone();
        let bad_block_cache = bad_block_cache.shallow_clone();
        let seen_block_cache = seen_block_cache.shallow_clone();
        let cancellation_token = cancellation_token.clone();
        let hello_fetch_limiter = hello_fetch_limiter.shallow_clone();
        let tipset_sender = tipset_sender.clone();
        async move {
            while let Ok(event) = network_rx.recv_async().await {
                inc_gossipsub_event_metrics(&event);

                update_peer_info(
                    &event,
                    &network,
                    state_manager.chain_store().shallow_clone(),
                    &genesis,
                    cancellation_token.clone(),
                );

                // Fetches are spawned, never awaited here: an unresponsive peer
                // would otherwise stall the loop and back up the unbounded event queue.
                match event {
                    NetworkEvent::HelloResponseOutbound { request, source } => {
                        let Ok(permit) = hello_fetch_limiter.shallow_clone().try_acquire_owned() else {
                            debug!(%source, "dropping hello-triggered tipset fetch: too many in flight");
                            continue;
                        };
                        spawn_tipset_fetch(
                            network.shallow_clone(),
                            state_manager.chain_store().shallow_clone(),
                            Some(source),
                            TipsetKey::from(request.heaviest_tip_set),
                            tipset_sender.clone(),
                            cancellation_token.clone(),
                            Some(permit),
                        );
                    }
                    NetworkEvent::PubsubMessage { message } => match message {
                        PubsubMessage::Block(b) => {
                            let cs = state_manager.chain_store();
                            let cfg = cs.chain_config();
                            if let Err(reason) = GossipBlockValidator::new(&b).validate_pre_fetch(
                                &genesis,
                                cfg.block_delay_secs,
                                cs.ec_calculator_finalized_epoch(), // Not using F3 finalized epoch as it could go above the chain head during catchup
                                bad_block_cache.as_ref(),
                                &seen_block_cache,
                            ) {
                                metrics::GOSSIP_BLOCK_REJECTED_TOTAL
                                    .get_or_create(&metrics::GossipRejectReasonLabel {
                                        reason: reason.label(),
                                    })
                                    .inc();
                                debug!("Rejected gossip block {}: {reason}", b.header.cid());
                                continue;
                            }
                            let key = TipsetKey::from(nunny::vec![*b.header.cid()]);
                            // Not rate-limited: gossip blocks are signed and peer-scored.
                            spawn_tipset_fetch(
                                network.shallow_clone(),
                                cs.shallow_clone(),
                                None,
                                key,
                                tipset_sender.clone(),
                                cancellation_token.clone(),
                                None,
                            );
                        }
                        PubsubMessage::Message(m) => {
                            if let Err(why) = mem_pool.add(m).await {
                                debug!("Received invalid GossipSub message: {}", why);
                            }
                        }
                    },
                    _ => {}
                }
            }
        }
    });

    // Forward tipsets from miners into the state machine.
    set.spawn({
        let state_changed = state_changed.clone();
        let state_machine = state_machine.clone();

        async move {
            while let Ok(tipset) = tipset_receiver.recv_async().await {
                state_machine
                    .lock()
                    .update(SyncEvent::NewFullTipsets(vec![tipset]));
                state_changed.notify_one();
            }
        }
    });

    // When the state machine is updated, we need to update the sync status and spawn tasks
    set.spawn({
        let state_manager = state_manager.shallow_clone();
        let state_machine = state_machine.shallow_clone();
        let network = network.shallow_clone();
        let sync_status = sync_status.shallow_clone();
        let state_changed = state_changed.shallow_clone();
        let tasks = tasks.shallow_clone();
        let bad_block_cache = bad_block_cache.shallow_clone();
        let cancellation_token = cancellation_token.clone();
        async move {
            const FORK_CLEANUP_INTERVAL: Duration = Duration::from_mins(1);
            let mut last_fork_cleanup = Instant::now();
            loop {
                state_changed.notified().await;

                let mut tasks_set = tasks.lock();
                if last_fork_cleanup + FORK_CLEANUP_INTERVAL < Instant::now() {
                    state_machine.lock().cleanup_dangling_forks();
                    last_fork_cleanup = Instant::now();
                }
                let (task_vec, current_active_forks) = state_machine.lock().tasks();

                // Update the sync states
                {
                    let old_status_report = sync_status.load().shallow_clone();
                    let new_status_report = old_status_report.update(
                        &state_manager,
                        current_active_forks,
                        stateless_mode,
                    );
                    sync_status.store(new_status_report.into());
                }

                for task in task_vec {
                    // insert task into tasks. If task is already in tasks, skip. If it is not, spawn it.
                    let new = tasks_set.insert(task.clone());
                    if new {
                        let action = task.clone().execute(
                            network.shallow_clone(),
                            state_manager.shallow_clone(),
                            stateless_mode,
                            bad_block_cache.shallow_clone(),
                        );
                        tokio::spawn({
                            let tasks = tasks.shallow_clone();
                            let state_machine = state_machine.shallow_clone();
                            let state_changed = state_changed.shallow_clone();
                            let cancellation_token = cancellation_token.clone();
                            async move {
                                cancellation_token
                                    .run_until_cancelled(async move {
                                        if let Some(event) = action.await {
                                            state_machine.lock().update(event);
                                            state_changed.notify_one();
                                        }
                                        let mut tasks = tasks.lock();
                                        tasks.remove(&task);
                                        tasks.shrink_to_fit();
                                    })
                                    .await
                            }
                        });
                    }
                }
            }
        }
    });

    // Periodically report progress while catching up to HEAD. Once we're in
    // steady-state (i.e. caught up to HEAD), and there are
    // no active forks, this will not report anything.
    set.spawn({
        let state_manager = state_manager.shallow_clone();
        let sync_status = sync_status.shallow_clone();
        async move {
            loop {
                tokio::time::sleep(tokio::time::Duration::from_secs(10)).await;
                let heaviest_tipset = state_manager.chain_store().heaviest_tipset();
                let heaviest_epoch = heaviest_tipset.epoch();
                let expected_head = calculate_expected_epoch(
                    Utc::now().timestamp() as u64,
                    state_manager.chain_store().genesis_block_header().timestamp,
                    state_manager.chain_config().block_delay_secs,
                );
                let diff = expected_head - heaviest_epoch;

                // Only print 'Catching up to HEAD' if we're more than 10 epochs
                // behind. Otherwise it can be too spammy.
                if diff <= 10 {
                    continue;
                }

                // The event loop refreshes this report on every state change,
                // so it is never staler than the state machine itself.
                let report = sync_status.load();
                let forks = &report.active_forks;
                // Once the nearest fork is validating, the remaining fetch
                // gaps belong to freshly gossiped forks near the clock head
                // and their min would mislead.
                let status: Cow<'static, str> = if forks
                    .iter()
                    .any(|fork| fork.stage == ForkSyncStage::ValidatingTipsets)
                {
                    ", validating tipsets".into()
                } else if forks
                    .iter()
                    .any(|fork| fork.stage == ForkSyncStage::FetchingHeaders)
                {
                    // The fork closest to the validated head connects first, so
                    // its gap approximates the header fetches left before
                    // validation can start.
                    let remaining = forks
                        .iter()
                        .filter(|fork| fork.stage == ForkSyncStage::FetchingHeaders)
                        .map(|fork| fork.target_sync_epoch_start - fork.validated_chain_head_epoch)
                        .filter(|gap| *gap > 0)
                        .min();
                    match remaining {
                        Some(remaining) => format!(", fetching headers (~{remaining})").into(),
                        // All fetches are at or below the validated head:
                        // fork-point searches, fetching a competing branch's
                        // ancestry down to where it forked off our chain.
                        None => ", fetching headers".into(),
                    }
                } else {
                    // waiting for peers to (re)seed tipsets (startup, no usable peers, or the buffer
                    // was just cleared after a failed validation).
                    ", waiting for tipsets".into()
                };
                info!(
                    "Catching up to HEAD: {heaviest_epoch}{} -> {expected_head} (diff: {diff}){status}",
                    heaviest_tipset.key()
                );
            }
        }
    });

    set.join_all().await;
    Ok(())
}

// Increment the gossipsub event metrics.
fn inc_gossipsub_event_metrics(event: &NetworkEvent) {
    let label = match event {
        NetworkEvent::HelloRequestInbound => metrics::values::HELLO_REQUEST_INBOUND,
        NetworkEvent::HelloResponseOutbound { .. } => metrics::values::HELLO_RESPONSE_OUTBOUND,
        NetworkEvent::HelloRequestOutbound => metrics::values::HELLO_REQUEST_OUTBOUND,
        NetworkEvent::HelloResponseInbound => metrics::values::HELLO_RESPONSE_INBOUND,
        NetworkEvent::PeerConnected(_) => metrics::values::PEER_CONNECTED,
        NetworkEvent::PeerDisconnected(_) => metrics::values::PEER_DISCONNECTED,
        NetworkEvent::PubsubMessage { message } => match message {
            PubsubMessage::Block(_) => metrics::values::PUBSUB_BLOCK,
            PubsubMessage::Message(_) => metrics::values::PUBSUB_MESSAGE,
        },
        NetworkEvent::ChainExchangeRequestOutbound => {
            metrics::values::CHAIN_EXCHANGE_REQUEST_OUTBOUND
        }
        NetworkEvent::ChainExchangeResponseInbound => {
            metrics::values::CHAIN_EXCHANGE_RESPONSE_INBOUND
        }
        NetworkEvent::ChainExchangeRequestInbound => {
            metrics::values::CHAIN_EXCHANGE_REQUEST_INBOUND
        }
        NetworkEvent::ChainExchangeResponseOutbound => {
            metrics::values::CHAIN_EXCHANGE_RESPONSE_OUTBOUND
        }
    };

    metrics::LIBP2P_MESSAGE_TOTAL.get_or_create(&label).inc();
}

// Keep our peer manager up to date.
fn update_peer_info(
    event: &NetworkEvent,
    network: &SyncNetworkContext,
    chain_store: ChainStore,
    genesis: &Tipset,
    cancellation_token: CancellationToken,
) {
    match event {
        NetworkEvent::PeerConnected(peer_id) => {
            let peer_id = *peer_id;
            let genesis_cid = *genesis.block_headers().first().cid();
            let network = network.shallow_clone();
            // Spawn and immediately move on to the next event
            tokio::task::spawn(async move {
                cancellation_token
                    .run_until_cancelled(handle_peer_connected_event(
                        network,
                        chain_store,
                        peer_id,
                        genesis_cid,
                    ))
                    .await
            });
        }
        NetworkEvent::PeerDisconnected(peer_id) => {
            handle_peer_disconnected_event(network, *peer_id);
        }
        _ => {}
    }
}

async fn handle_peer_connected_event(
    network: SyncNetworkContext,
    chain_store: ChainStore,
    peer_id: PeerId,
    genesis_block_cid: Cid,
) {
    // Query the heaviest TipSet from the store
    if network.peer_manager().is_peer_new(&peer_id) {
        // Since the peer is new, send them a hello request
        // Query the heaviest TipSet from the store
        let heaviest = chain_store.heaviest_tipset();
        let request = HelloRequest {
            heaviest_tip_set: heaviest.cids(),
            heaviest_tipset_height: heaviest.epoch(),
            heaviest_tipset_weight: heaviest.weight().clone().into(),
            genesis_cid: genesis_block_cid,
        };
        let (peer_id, moment_sent, response) = match network.hello_request(peer_id, request).await {
            Ok(response) => response,
            Err(e) => {
                debug!("Hello request failed: {}", e);
                return;
            }
        };
        let dur = Instant::now().duration_since(moment_sent);

        // Update the peer metadata based on the response
        match response {
            Some(_) => {
                network.peer_manager().log_success(&peer_id, dur);
            }
            None => {
                network.peer_manager().log_failure(&peer_id, dur);
            }
        }
    }
}

fn handle_peer_disconnected_event(network: &SyncNetworkContext, peer_id: PeerId) {
    network.peer_manager().remove_peer(&peer_id);
    network.peer_manager().unmark_peer_bad(&peer_id);
}

/// Concurrency cap for tipset fetches triggered by an inbound `hello`.
///
/// A `hello` is unauthenticated, and its sender controls both the tipset we
/// fetch and, by stalling the chain-exchange, how long the fetch blocks.
/// Uncapped, a flood spawns fetches without bound, each pinning a blocking-pool
/// thread for the chain-exchange timeout (up to 60 seconds). Excess is dropped, not queued.
static MAX_CONCURRENT_HELLO_TRIGGERED_FETCHES: LazyLock<usize> = LazyLock::new(|| {
    // `.min` keeps an oversized override from panicking `Semaphore::new`.
    env_or_default_logged(
        "FOREST_MAX_CONCURRENT_HELLO_TRIGGERED_FETCHES",
        nonzero!(16_usize),
    )
    .get()
    .min(Semaphore::MAX_PERMITS)
});

/// Fetches a tipset off the event loop, forwarding a success into `tipset_sender`
/// (the same channel miner tipsets use). Any `permit` is held for the fetch.
fn spawn_tipset_fetch(
    network: SyncNetworkContext,
    chain_store: ChainStore,
    peer_id: Option<PeerId>,
    tipset_keys: TipsetKey,
    tipset_sender: flume::Sender<FullTipset>,
    cancellation_token: CancellationToken,
    permit: Option<OwnedSemaphorePermit>,
) {
    tokio::spawn(async move {
        let _permit = permit;
        // Fetch and forward both run inside the cancellation scope so a shutdown
        // aborts a send blocked on a full channel, releasing the permit promptly.
        cancellation_token
            .run_until_cancelled(async {
                match get_full_tipset(&network, &chain_store, peer_id, &tipset_keys).await {
                    Ok(tipset) => {
                        let _ = tipset_sender.send_async(tipset).await;
                    }
                    Err(e) => debug!("Querying full tipset failed: {e}"),
                }
            })
            .await;
    });
}

pub async fn get_full_tipset(
    network: &SyncNetworkContext,
    chain_store: &ChainStore,
    peer_id: Option<PeerId>,
    tipset_keys: &TipsetKey,
) -> anyhow::Result<FullTipset> {
    // Attempt to load from the store
    if let Ok(full_tipset) = load_full_tipset(chain_store, tipset_keys) {
        return Ok(full_tipset);
    }
    // Load from the network
    let tipset = network
        .chain_exchange_full_tipset(peer_id, tipset_keys)
        .await
        .map_err(|e| anyhow::anyhow!(e))?;
    tipset.persist(chain_store.db())?;

    Ok(tipset)
}

async fn get_full_tipset_batch(
    network: &SyncNetworkContext,
    chain_store: &ChainStore,
    peer_id: Option<PeerId>,
    tipset_keys: &TipsetKey,
) -> anyhow::Result<Vec<FullTipset>> {
    // Attempt to load from the store
    if let Ok(full_tipset) = load_full_tipset(chain_store, tipset_keys) {
        return Ok(vec![full_tipset]);
    }
    // Load from the network
    let tipsets = network
        .chain_exchange_full_tipsets(peer_id, tipset_keys)
        .await
        .map_err(|e| anyhow::anyhow!(e))?;

    for tipset in tipsets.iter() {
        tipset.persist(chain_store.db())?;
    }

    Ok(tipsets)
}

pub fn load_full_tipset(
    chain_store: &ChainStore,
    tipset_keys: &TipsetKey,
) -> anyhow::Result<FullTipset> {
    // Retrieve tipset from store based on passed in TipsetKey
    let ts = chain_store
        .chain_index()
        .load_required_tipset(tipset_keys)?;
    let blocks: Vec<_> = ts
        .block_headers()
        .iter()
        .map(|header| -> anyhow::Result<Block> {
            let (bls_msgs, secp_msgs) = crate::chain::block_messages(chain_store.db(), header)?;
            Ok(Block {
                header: header.clone(),
                bls_messages: bls_msgs,
                secp_messages: secp_msgs,
            })
        })
        .try_collect()?;
    // Construct FullTipset
    let fts = FullTipset::new(blocks)?;
    Ok(fts)
}

enum SyncEvent {
    NewFullTipsets(Vec<FullTipset>),
    BadTipset(FullTipset),
    ValidatedTipset {
        tipset: FullTipset,
        is_proposed_head: bool,
    },
}

impl std::fmt::Display for SyncEvent {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        fn tss_to_string(tss: &[FullTipset]) -> String {
            format!(
                "epoch: {}-{}",
                tss.first().map(|ts| ts.epoch()).unwrap_or_default(),
                tss.last().map(|ts| ts.epoch()).unwrap_or_default()
            )
        }

        match self {
            Self::NewFullTipsets(tss) => write!(f, "NewFullTipsets({})", tss_to_string(tss)),
            Self::BadTipset(ts) => {
                write!(f, "BadTipset(epoch: {}, key: {})", ts.epoch(), ts.key())
            }
            Self::ValidatedTipset {
                tipset,
                is_proposed_head,
            } => write!(
                f,
                "ValidatedTipset(epoch: {}, key: {}, is_proposed_head: {is_proposed_head})",
                tipset.epoch(),
                tipset.key()
            ),
        }
    }
}

#[derive(derive_more::Debug)]
struct SyncStateMachine {
    #[debug(skip)]
    cs: ChainStore,
    bad_block_cache: Option<BadBlockCache>,
    // Map from TipsetKey to FullTipset
    tipsets: HashMap<TipsetKey, FullTipset>,
    stateless_mode: bool,
    /// Broadcast channel for validated tipsets, used to notify other components of new validated tipsets.
    validated_tipset_broadcast_tx: tokio::sync::broadcast::Sender<TipsetKey>,
}

impl SyncStateMachine {
    pub fn new(
        cs: ChainStore,
        bad_block_cache: Option<BadBlockCache>,
        stateless_mode: bool,
    ) -> Self {
        Self {
            cs,
            bad_block_cache,
            tipsets: HashMap::default(),
            stateless_mode,
            validated_tipset_broadcast_tx: tokio::sync::broadcast::Sender::new(1024),
        }
    }

    // Compute the list of chains from the tipsets map
    fn chains(&self) -> Vec<Vec<FullTipset>> {
        let mut chains = Vec::new();
        let mut remaining_tipsets = self.tipsets.clone();

        while let Some(heaviest) = remaining_tipsets
            .values()
            .max_by_key(|ts| ts.weight())
            .cloned()
        {
            // Build chain starting from heaviest
            let mut chain = Vec::new();
            let mut current = Some(heaviest);

            while let Some(tipset) = current.take() {
                remaining_tipsets.remove(tipset.key());

                // Find parent in tipsets map
                current = self.tipsets.get(tipset.parents()).cloned();

                chain.push(tipset);
            }
            chain.reverse();
            chains.push(chain);
        }

        chains
    }

    fn is_parent_validated(&self, tipset: &FullTipset) -> bool {
        let db = self.cs.db();
        self.stateless_mode || db.has(tipset.parent_state()).unwrap_or(false)
    }

    fn is_ready_for_validation(&self, tipset: &FullTipset) -> bool {
        if self.stateless_mode || tipset.key() == self.cs.genesis_tipset().key() {
            // Skip validation in stateless mode and for genesis tipset
            true
        } else if let Ok(parent_ts) = load_full_tipset(&self.cs, tipset.parents()) {
            let head_ts = self.cs.heaviest_tipset();
            // Treat post-head-epoch tipsets as not validated to fix <https://github.com/ChainSafe/forest/issues/5677>
            // basically, the follow task should always start from the current head which could be manually set
            // to an old one. When a post-head-epoch tipset is considered validated, it could mess up the state machine
            // in some edge cases and the node ends up being stuck with ever-empty sync task queue as reported
            // in <https://github.com/ChainSafe/forest/issues/5679>.
            if parent_ts.key() == head_ts.key() {
                true
            } else if parent_ts.epoch() >= head_ts.epoch() {
                false
            } else {
                self.is_parent_validated(tipset)
            }
        } else {
            false
        }
    }

    fn add_full_tipset(&mut self, tipset: FullTipset) {
        if let Err(why) = TipsetValidator(&tipset).validate(
            &self.cs,
            self.bad_block_cache.as_ref(),
            &self.cs.genesis_tipset(),
            self.cs.chain_config().block_delay_secs,
        ) {
            metrics::INVALID_TIPSET_TOTAL.inc();
            trace!("Skipping invalid tipset: {}", why);
            self.mark_bad_tipset(tipset);
            return;
        }

        // Check if tipset is outside the chain finality window.
        // Not using F3 finalized epoch as it could go above the chain head during catchup
        if tipset.epoch() < self.cs.ec_calculator_finalized_epoch() {
            self.mark_bad_tipset(tipset);
            return;
        }

        // Check if tipset already exists
        if self.tipsets.contains_key(tipset.key()) {
            return;
        }

        // Skip if tipset is part of the current chain.
        if let Ok(Some(ts)) = self.cs.chain_index().tipset_by_height_blocking(
            tipset.epoch(),
            self.cs.heaviest_tipset(),
            ResolveNullTipset::TakeOlder,
        ) && ts.key() == tipset.key()
        {
            return;
        }

        // Find any existing tipsets with same epoch and parents
        let mut to_remove = Vec::new();
        #[allow(clippy::mutable_key_type)]
        let mut merged_blocks: HashSet<_> = tipset.blocks().iter().cloned().collect();

        // Collect all parent references from existing tipsets
        let parent_refs: HashSet<_> = self
            .tipsets
            .values()
            .map(|ts| ts.parents().clone())
            .collect();

        for (key, existing_ts) in self.tipsets.iter() {
            if existing_ts.epoch() == tipset.epoch() && existing_ts.parents() == tipset.parents() {
                // Only mark for removal if not referenced as a parent
                if !parent_refs.contains(key) {
                    to_remove.push(key.clone());
                }
                // Add blocks from existing tipset - HashSet handles deduplication automatically
                merged_blocks.extend(existing_ts.blocks().iter().cloned());
            }
        }

        // Remove old tipsets that were merged and aren't referenced
        for key in to_remove {
            self.tipsets.remove(&key);
        }

        // Create and insert new merged tipset
        if let Ok(merged_tipset) = FullTipset::new(merged_blocks) {
            self.tipsets
                .insert(merged_tipset.key().clone(), merged_tipset);
        }

        self.tipsets.shrink_to_fit();
    }

    // Mark blocks in tipset as bad.
    // Mark all descendants of tipsets as bad.
    // Remove all bad tipsets from the tipset map.
    fn mark_bad_tipset(&mut self, tipset: FullTipset) {
        let mut stack = vec![tipset];
        while let Some(tipset) = stack.pop() {
            self.tipsets.remove(tipset.key());
            // Find all descendant tipsets (tipsets that have this tipset as a parent)
            let mut to_remove = Vec::new();
            let mut descendants = Vec::new();

            for (key, ts) in self.tipsets.iter() {
                if ts.parents() == tipset.key() {
                    to_remove.push(key.clone());
                    descendants.push(ts.clone());
                }
            }

            // Remove bad tipsets from the map
            for key in to_remove {
                self.tipsets.remove(&key);
            }

            // Mark descendants as bad
            stack.extend(descendants);
        }
    }

    fn try_mark_tipset_as_validated(&mut self, tipset: FullTipset, is_proposed_head: bool) -> bool {
        if !self.is_parent_validated(&tipset) {
            tracing::error!(epoch = %tipset.epoch(), tsk = %tipset.key(), parent_state = %tipset.parent_state(), "Parent tipset must be validated");
            return false;
        }

        self.tipsets.remove(tipset.key());
        let tipset = tipset.into_tipset();
        // cs.put_tipset requires state and doesn't work in stateless mode
        if self.stateless_mode {
            let epoch = tipset.epoch();
            let terse_key = tipset.key().terse();
            if self.cs.heaviest_tipset().weight() < tipset.weight() {
                if let Err(e) = self.cs.set_heaviest_tipset(tipset) {
                    error!("Error setting heaviest tipset: {}", e);
                    return false;
                } else {
                    info!("Heaviest tipset: {} ({})", epoch, terse_key);
                }
            }
        } else if is_proposed_head {
            if let Err(e) = self.cs.maybe_update_pending_head(&tipset) {
                error!("Error putting tipset: {e}");
                return false;
            }
        } else if let Err(e) = self.cs.set_heaviest_tipset(tipset) {
            error!("Error setting heaviest tipset: {e}");
            return false;
        }
        true
    }

    pub fn update(&mut self, event: SyncEvent) {
        tracing::trace!("update: {event}");
        match event {
            SyncEvent::NewFullTipsets(tipsets) => {
                for tipset in tipsets {
                    self.add_full_tipset(tipset);
                }
            }
            SyncEvent::BadTipset(tipset) => self.mark_bad_tipset(tipset),
            SyncEvent::ValidatedTipset {
                tipset,
                is_proposed_head,
            } => {
                if self.try_mark_tipset_as_validated(tipset, is_proposed_head)
                    && crate::utils::broadcast::has_subscribers(&self.validated_tipset_broadcast_tx)
                    // Sending the actual head key here as it could be expanded from the above tipset when `is_proposed_head` is `true`
                    && let Err(e) = self.validated_tipset_broadcast_tx.send(self.cs.heaviest_tipset().key().clone())
                {
                    warn!("Failed to broadcast validated tipset: {e}");
                }
            }
        }
    }

    pub fn tasks(&self) -> (Vec<SyncTask>, Vec<ForkSyncInfo>) {
        // Get the node's current validated head epoch once, as it's the same for all forks.
        let current_validated_epoch = self.cs.heaviest_tipset().epoch();
        let now = Utc::now();

        let mut active_sync_info = Vec::new();
        let mut tasks = Vec::new();
        for chain in self.chains() {
            if let Some(first_ts) = chain.first() {
                let last_ts = chain.last().expect("Infallible");
                let stage: ForkSyncStage;
                let start_time = Some(now);

                if !self.is_ready_for_validation(first_ts) {
                    stage = ForkSyncStage::FetchingHeaders;
                    tasks.push(SyncTask::FetchTipset(
                        first_ts.parents().clone(),
                        first_ts.epoch(),
                    ));
                } else {
                    stage = ForkSyncStage::ValidatingTipsets;
                    tasks.push(SyncTask::ValidateTipset {
                        tipset: first_ts.clone(),
                        is_proposed_head: chain.len() == 1,
                    });
                }

                let fork_info = ForkSyncInfo {
                    target_tipset_key: last_ts.key().clone(),
                    target_epoch: last_ts.epoch(),
                    target_sync_epoch_start: first_ts.epoch(),
                    stage,
                    validated_chain_head_epoch: current_validated_epoch,
                    start_time,
                    last_updated: Some(now),
                };

                active_sync_info.push(fork_info);
            }
        }
        (tasks, active_sync_info)
    }

    pub fn cleanup_dangling_forks(&mut self) {
        let finalized_epoch = self.cs.ec_calculator_finalized_epoch();
        for chain in self.chains() {
            // Cleanup dangling fork when its target epoch is finalized
            if let Some(target) = chain.last()
                && target.epoch() < finalized_epoch
            {
                chain.iter().for_each(|ts| {
                    self.tipsets.remove(ts.key());
                });
                tracing::info!(
                    "Cleaned up dangling fork from epoch {} to {}",
                    chain.first().map(|ts| ts.epoch()).unwrap_or_default(),
                    chain.last().map(|ts| ts.epoch()).unwrap_or_default(),
                );
            }
        }
    }
}

#[derive(PartialEq, Eq, Hash, Clone, Debug)]
enum SyncTask {
    ValidateTipset {
        tipset: FullTipset,
        is_proposed_head: bool,
    },
    FetchTipset(TipsetKey, ChainEpoch),
}

impl std::fmt::Display for SyncTask {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            SyncTask::ValidateTipset {
                tipset,
                is_proposed_head,
            } => write!(
                f,
                "ValidateTipset(epoch: {}, is_proposed_head: {is_proposed_head})",
                tipset.epoch()
            ),
            SyncTask::FetchTipset(key, epoch) => {
                let s = key.to_string();
                write!(
                    f,
                    "FetchTipset({}, epoch: {})",
                    &s[s.len().saturating_sub(8)..],
                    epoch
                )
            }
        }
    }
}

impl SyncTask {
    async fn execute(
        self,
        network: SyncNetworkContext,
        state_manager: StateManager,
        stateless_mode: bool,
        bad_block_cache: Option<BadBlockCache>,
    ) -> Option<SyncEvent> {
        tracing::trace!("SyncTask::execute {self}");
        match self {
            SyncTask::ValidateTipset {
                tipset,
                is_proposed_head,
            } if stateless_mode => Some(SyncEvent::ValidatedTipset {
                tipset,
                is_proposed_head,
            }),
            SyncTask::ValidateTipset {
                tipset,
                is_proposed_head,
            } => match validate_tipset(&state_manager, tipset.clone(), bad_block_cache).await {
                Ok(()) => Some(SyncEvent::ValidatedTipset {
                    tipset,
                    is_proposed_head,
                }),
                // If temporal drift error, don't mark as bad, just skip validation and try again
                // later. This mirrors internal logic where temporal drift doesn't mark a block as
                // bad permanently, since it could be valid later on. If not done, a single
                // time-traveling block could cause the node to be stuck without making progress.
                Err(e) if matches!(e, TipsetSyncerError::TimeTravellingBlock { .. }) => {
                    warn!("Time travelling block detected, skipping tipset for now: {e}");
                    None
                }
                // May stem from locally corrupted inputs rather than a bad block — repair
                // the lookup table and retry only if something was actually wrong.
                Err(e) if matches!(e, TipsetSyncerError::ParentChainStateMismatch(_)) => {
                    warn!("Error validating tipset: {e}");
                    let sm = state_manager.shallow_clone();
                    match tokio::task::spawn_blocking(move || sm.repair_tipset_lookup())
                        .await
                        .unwrap_or_else(|e| Err(e.into()))
                    {
                        // A verified-clean table means the mismatch is genuine.
                        Ok(0) => Some(SyncEvent::BadTipset(tipset)),
                        Ok(repaired) => {
                            warn!(
                                "Repaired {repaired} tipset lookup entries; retrying tipset validation at epoch {}",
                                tipset.epoch()
                            );
                            None
                        }
                        // The table could not be verified; retry later rather than risk
                        // marking a canonical tipset bad on incomplete knowledge.
                        Err(e) => {
                            warn!("Failed to repair tipset lookup table: {e:#}");
                            None
                        }
                    }
                }
                Err(e) => {
                    warn!("Error validating tipset: {e}");
                    Some(SyncEvent::BadTipset(tipset))
                }
            },
            SyncTask::FetchTipset(key, epoch) => {
                match get_full_tipset_batch(&network, state_manager.chain_store(), None, &key).await
                {
                    Ok(parents) => Some(SyncEvent::NewFullTipsets(parents)),
                    Err(e) => {
                        // It's not a massive error; could be a transient network issue or a fork.
                        tracing::debug!(%key, %epoch, "failed to fetch tipset: {e:#}");
                        None
                    }
                }
            }
        }
    }
}

#[derive(Debug)]
struct SyncTasks(Arc<Mutex<HashSet<SyncTask>>>);

#[derive(Debug)]
struct SyncStateMachineWrapper(Arc<Mutex<SyncStateMachine>>);

mod metrics_collection {
    use super::*;
    use prometheus_client::{
        collector::Collector,
        encoding::{DescriptorEncoder, EncodeMetric},
        metrics::gauge::Gauge,
        registry::Unit,
    };

    impl Collector for SyncTasks {
        fn encode(&self, mut encoder: DescriptorEncoder) -> Result<(), std::fmt::Error> {
            {
                let size_in_bytes = {
                    let g: Gauge = Default::default();
                    g.set(self.0.lock().allocation_size() as i64);
                    g
                };
                let size_metric_encoder = encoder.encode_descriptor(
                    "chain_follower_tasks_size",
                    "Size of the chain follower tasks in bytes",
                    Some(&Unit::Bytes),
                    size_in_bytes.metric_type(),
                )?;
                size_in_bytes.encode(size_metric_encoder)?;
            }
            {
                let len = {
                    let g: Gauge = Default::default();
                    g.set(self.0.lock().len() as i64);
                    g
                };
                let size_metric_encoder = encoder.encode_descriptor(
                    "chain_follower_tasks_len",
                    "Length of the chain follower tasks",
                    None,
                    len.metric_type(),
                )?;
                len.encode(size_metric_encoder)?;
            }
            {
                let cap = {
                    let g: Gauge = Default::default();
                    g.set(self.0.lock().capacity() as i64);
                    g
                };
                let size_metric_encoder = encoder.encode_descriptor(
                    "chain_follower_tasks_cap",
                    "Capacity of the chain follower tasks",
                    None,
                    cap.metric_type(),
                )?;
                cap.encode(size_metric_encoder)?;
            }

            Ok(())
        }
    }

    impl Collector for SyncStateMachineWrapper {
        fn encode(&self, mut encoder: DescriptorEncoder) -> Result<(), std::fmt::Error> {
            {
                let size_in_bytes = {
                    let g: Gauge = Default::default();
                    g.set(self.0.lock().tipsets.allocation_size() as i64);
                    g
                };
                let size_metric_encoder = encoder.encode_descriptor(
                    "chain_follower_tipsets_size",
                    "Size of the chain follower tipsets in bytes",
                    Some(&Unit::Bytes),
                    size_in_bytes.metric_type(),
                )?;
                size_in_bytes.encode(size_metric_encoder)?;
            }
            {
                let len = {
                    let g: Gauge = Default::default();
                    g.set(self.0.lock().tipsets.len() as i64);
                    g
                };
                let size_metric_encoder = encoder.encode_descriptor(
                    "chain_follower_tipsets_len",
                    "Length of the chain follower tipsets",
                    None,
                    len.metric_type(),
                )?;
                len.encode(size_metric_encoder)?;
            }
            {
                let cap = {
                    let g: Gauge = Default::default();
                    g.set(self.0.lock().tipsets.capacity() as i64);
                    g
                };
                let size_metric_encoder = encoder.encode_descriptor(
                    "chain_follower_tipsets_cap",
                    "Capacity of the chain follower tipsets",
                    None,
                    cap.metric_type(),
                )?;
                cap.encode(size_metric_encoder)?;
            }

            Ok(())
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::blocks::{Chain4U, HeaderBuilder, chain4u};
    use crate::db::MemoryDB;
    use crate::utils::db::CborStoreExt as _;
    use num_bigint::BigInt;
    use num_traits::ToPrimitive;
    use std::sync::Arc;
    use tracing::level_filters::LevelFilter;
    use tracing_subscriber::EnvFilter;

    fn setup() -> (ChainStore, Chain4U<Arc<MemoryDB>>) {
        // Initialize test logger
        let _ = tracing_subscriber::fmt()
            .without_time()
            .with_env_filter(
                EnvFilter::builder()
                    .with_default_directive(LevelFilter::DEBUG.into())
                    .from_env()
                    .unwrap(),
            )
            .try_init();

        let db = Arc::new(MemoryDB::default());

        // Create a chain of 5 tipsets using Chain4U
        let c4u = Chain4U::with_blockstore(db.clone());
        chain4u! {
            in c4u;
            [genesis_header = dummy_node(&db, 0)]
        };

        let cs = ChainStore::new(db, Default::default(), genesis_header).unwrap();

        cs.set_heaviest_tipset(cs.genesis_tipset()).unwrap();

        (cs, c4u)
    }

    fn dummy_state(db: impl Blockstore, i: ChainEpoch) -> Cid {
        db.put_cbor_default(&i).unwrap()
    }

    fn dummy_node(db: impl Blockstore, i: ChainEpoch) -> HeaderBuilder {
        HeaderBuilder {
            state_root: dummy_state(db, i).into(),
            weight: BigInt::from(i).into(),
            epoch: i.into(),
            ..Default::default()
        }
    }

    #[test]
    fn test_state_machine_validation_order() {
        let (cs, c4u) = setup();
        let db = cs.db_owned();

        chain4u! {
            from [genesis_header] in c4u;
            [a = dummy_node(&db, 1)] -> [b = dummy_node(&db, 2)] -> [c = dummy_node(&db, 3)] -> [d = dummy_node(&db, 4)] -> [e = dummy_node(&db, 5)]
        };

        // Create the state machine
        let mut state_machine = SyncStateMachine::new(cs.shallow_clone(), Default::default(), true);

        // Insert tipsets in random order
        let tipsets = vec![e, b, d, c, a];

        // Convert each block into a FullTipset and add it to the state machine
        for block in tipsets {
            let full_tipset = FullTipset::new(vec![Block {
                header: block.clone().into(),
                bls_messages: vec![],
                secp_messages: vec![],
            }])
            .unwrap();
            state_machine.update(SyncEvent::NewFullTipsets(vec![full_tipset]));
        }

        // Record validation order by processing all validation tasks in each iteration
        let mut validation_tasks = Vec::new();
        loop {
            let (tasks, _) = state_machine.tasks();

            // Find all validation tasks
            let validation_tipsets: Vec<_> = tasks
                .into_iter()
                .filter_map(|task| {
                    if let SyncTask::ValidateTipset {
                        tipset,
                        is_proposed_head,
                    } = task
                    {
                        Some((tipset, is_proposed_head))
                    } else {
                        None
                    }
                })
                .collect();

            if validation_tipsets.is_empty() {
                break;
            }

            // Record and mark all tipsets as validated
            for (ts, is_proposed_head) in validation_tipsets {
                validation_tasks.push(ts.epoch());
                db.put_cbor_default(&ts.epoch()).unwrap();
                state_machine.try_mark_tipset_as_validated(ts, is_proposed_head);
            }
        }

        // We expect validation tasks for epochs 1 through 5 in order
        assert_eq!(validation_tasks, vec![1, 2, 3, 4, 5]);
    }

    #[test]
    fn test_sync_state_machine_chain_fragments() {
        let (cs, c4u) = setup();
        let db = cs.db();

        // Create a forked chain
        // genesis -> a -> b
        //            \--> c
        chain4u! {
            in c4u;
            [a = dummy_node(db, 1)] -> [b = dummy_node(db, 2)]
        };
        chain4u! {
            from [a] in c4u;
            [c = dummy_node(db, 3)]
        };

        // Create the state machine
        let mut state_machine = SyncStateMachine::new(cs, Default::default(), false);

        // Convert each block into a FullTipset and add it to the state machine
        for block in [a, b, c] {
            let full_tipset = FullTipset::new(vec![Block {
                header: block.clone().into(),
                bls_messages: vec![],
                secp_messages: vec![],
            }])
            .unwrap();
            state_machine.update(SyncEvent::NewFullTipsets(vec![full_tipset]));
        }

        let chains = state_machine
            .chains()
            .into_iter()
            .map(|v| {
                v.into_iter()
                    .map(|ts| ts.weight().to_i64().unwrap_or(0))
                    .collect_vec()
            })
            .collect_vec();

        // Both chains should start at the same tipset
        assert_eq!(chains, vec![vec![1, 3], vec![1, 2]]);
    }
}