sui_node/

lib.rs

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

use anemo::Network;
use anemo::PeerId;
use anemo_tower::callback::CallbackLayer;
use anemo_tower::trace::DefaultMakeSpan;
use anemo_tower::trace::DefaultOnFailure;
use anemo_tower::trace::TraceLayer;
use anyhow::Context;
use anyhow::Result;
use anyhow::anyhow;
use arc_swap::ArcSwap;
use fastcrypto_zkp::bn254::zk_login::JwkId;
use fastcrypto_zkp::bn254::zk_login::OIDCProvider;
use futures::future::BoxFuture;
use mysten_common::debug_fatal;
use prometheus::Registry;
use std::collections::{BTreeSet, HashMap, HashSet};
use std::fmt;
use std::future::Future;
use std::path::PathBuf;
use std::str::FromStr;
#[cfg(msim)]
use std::sync::atomic::Ordering;
use std::sync::{Arc, Weak};
use std::time::Duration;
use sui_core::authority::ExecutionEnv;
use sui_core::authority::RandomnessRoundReceiver;
use sui_core::authority::authority_store_tables::{
    AuthorityPerpetualTablesOptions, AuthorityPrunerTables,
};
use sui_core::authority::backpressure::BackpressureManager;
use sui_core::authority::epoch_start_configuration::EpochFlag;
use sui_core::authority::execution_time_estimator::ExecutionTimeObserver;
use sui_core::authority::shared_object_version_manager::Schedulable;
use sui_core::consensus_adapter::ConsensusClient;
use sui_core::consensus_manager::UpdatableConsensusClient;
use sui_core::epoch::randomness::RandomnessManager;
use sui_core::execution_cache::build_execution_cache;
use sui_network::validator::server::SUI_TLS_SERVER_NAME;
use sui_types::full_checkpoint_content::Checkpoint;

use sui_core::execution_scheduler::SchedulingSource;
use sui_core::global_state_hasher::GlobalStateHashMetrics;
use sui_core::storage::RestReadStore;
use sui_json_rpc::bridge_api::BridgeReadApi;
use sui_json_rpc_api::JsonRpcMetrics;
use sui_network::randomness;
use sui_rpc_api::RpcMetrics;
use sui_rpc_api::ServerVersion;
use sui_rpc_api::subscription::SubscriptionService;
use sui_types::base_types::ConciseableName;
use sui_types::crypto::RandomnessRound;
use sui_types::digests::{
    ChainIdentifier, CheckpointDigest, TransactionDigest, TransactionEffectsDigest,
};
use sui_types::executable_transaction::VerifiedExecutableTransaction;
use sui_types::messages_consensus::AuthorityCapabilitiesV2;
use sui_types::messages_consensus::ConsensusTransactionKind;
use sui_types::sui_system_state::SuiSystemState;
use sui_types::transaction::VerifiedCertificate;
use tap::tap::TapFallible;
use tokio::sync::oneshot;
use tokio::sync::{Mutex, broadcast, mpsc, watch};
use tokio::task::JoinHandle;
use tower::ServiceBuilder;
use tracing::{Instrument, error_span, info};
use tracing::{debug, error, warn};

use fastcrypto_zkp::bn254::zk_login::JWK;
pub use handle::SuiNodeHandle;
use mysten_metrics::{RegistryService, spawn_monitored_task};
use mysten_service::server_timing::server_timing_middleware;
use sui_config::node::{DBCheckpointConfig, RunWithRange};
use sui_config::node::{ForkCrashBehavior, ForkRecoveryConfig};
use sui_config::node_config_metrics::NodeConfigMetrics;
use sui_config::{ConsensusConfig, NodeConfig};
use sui_core::authority::authority_per_epoch_store::AuthorityPerEpochStore;
use sui_core::authority::authority_store_tables::AuthorityPerpetualTables;
use sui_core::authority::epoch_start_configuration::EpochStartConfigTrait;
use sui_core::authority::epoch_start_configuration::EpochStartConfiguration;
use sui_core::authority::submitted_transaction_cache::SubmittedTransactionCacheMetrics;
use sui_core::authority_aggregator::{AuthAggMetrics, AuthorityAggregator};
use sui_core::authority_server::{ValidatorService, ValidatorServiceMetrics};
use sui_core::checkpoints::checkpoint_executor::metrics::CheckpointExecutorMetrics;
use sui_core::checkpoints::checkpoint_executor::{CheckpointExecutor, StopReason};
use sui_core::checkpoints::{
    CheckpointMetrics, CheckpointService, CheckpointStore, SendCheckpointToStateSync,
    SubmitCheckpointToConsensus,
};
use sui_core::consensus_adapter::{
    CheckConnection, ConnectionMonitorStatus, ConsensusAdapter, ConsensusAdapterMetrics,
};
use sui_core::consensus_manager::ConsensusManager;
use sui_core::consensus_throughput_calculator::{
    ConsensusThroughputCalculator, ConsensusThroughputProfiler, ThroughputProfileRanges,
};
use sui_core::consensus_validator::{SuiTxValidator, SuiTxValidatorMetrics};
use sui_core::db_checkpoint_handler::DBCheckpointHandler;
use sui_core::epoch::committee_store::CommitteeStore;
use sui_core::epoch::consensus_store_pruner::ConsensusStorePruner;
use sui_core::epoch::epoch_metrics::EpochMetrics;
use sui_core::epoch::reconfiguration::ReconfigurationInitiator;
use sui_core::global_state_hasher::GlobalStateHasher;
use sui_core::jsonrpc_index::IndexStore;
use sui_core::module_cache_metrics::ResolverMetrics;
use sui_core::overload_monitor::overload_monitor;
use sui_core::rpc_index::RpcIndexStore;
use sui_core::signature_verifier::SignatureVerifierMetrics;
use sui_core::storage::RocksDbStore;
use sui_core::transaction_orchestrator::TransactionOrchestrator;
use sui_core::{
    authority::{AuthorityState, AuthorityStore},
    authority_client::NetworkAuthorityClient,
};
use sui_json_rpc::JsonRpcServerBuilder;
use sui_json_rpc::coin_api::CoinReadApi;
use sui_json_rpc::governance_api::GovernanceReadApi;
use sui_json_rpc::indexer_api::IndexerApi;
use sui_json_rpc::move_utils::MoveUtils;
use sui_json_rpc::read_api::ReadApi;
use sui_json_rpc::transaction_builder_api::TransactionBuilderApi;
use sui_json_rpc::transaction_execution_api::TransactionExecutionApi;
use sui_macros::fail_point;
use sui_macros::{fail_point_async, replay_log};
use sui_network::api::ValidatorServer;
use sui_network::discovery;
use sui_network::discovery::TrustedPeerChangeEvent;
use sui_network::state_sync;
use sui_network::validator::server::ServerBuilder;
use sui_protocol_config::{Chain, ProtocolConfig, ProtocolVersion};
use sui_snapshot::uploader::StateSnapshotUploader;
use sui_storage::{
    http_key_value_store::HttpKVStore,
    key_value_store::{FallbackTransactionKVStore, TransactionKeyValueStore},
    key_value_store_metrics::KeyValueStoreMetrics,
};
use sui_types::base_types::{AuthorityName, EpochId};
use sui_types::committee::Committee;
use sui_types::crypto::KeypairTraits;
use sui_types::error::{SuiError, SuiResult};
use sui_types::messages_consensus::{
    AuthorityCapabilitiesV1, ConsensusTransaction, check_total_jwk_size,
};
use sui_types::sui_system_state::SuiSystemStateTrait;
use sui_types::sui_system_state::epoch_start_sui_system_state::EpochStartSystemState;
use sui_types::sui_system_state::epoch_start_sui_system_state::EpochStartSystemStateTrait;
use sui_types::supported_protocol_versions::SupportedProtocolVersions;
use typed_store::DBMetrics;
use typed_store::rocks::default_db_options;

use crate::metrics::{GrpcMetrics, SuiNodeMetrics};

pub mod admin;
mod handle;
pub mod metrics;

pub struct ValidatorComponents {
    validator_server_handle: SpawnOnce,
    validator_overload_monitor_handle: Option<JoinHandle<()>>,
    consensus_manager: Arc<ConsensusManager>,
    consensus_store_pruner: ConsensusStorePruner,
    consensus_adapter: Arc<ConsensusAdapter>,
    checkpoint_metrics: Arc<CheckpointMetrics>,
    sui_tx_validator_metrics: Arc<SuiTxValidatorMetrics>,
}
pub struct P2pComponents {
    p2p_network: Network,
    known_peers: HashMap<PeerId, String>,
    discovery_handle: discovery::Handle,
    state_sync_handle: state_sync::Handle,
    randomness_handle: randomness::Handle,
}

#[cfg(msim)]
mod simulator {
    use std::sync::atomic::AtomicBool;
    use sui_types::error::SuiErrorKind;

    use super::*;
    pub(super) struct SimState {
        pub sim_node: sui_simulator::runtime::NodeHandle,
        pub sim_safe_mode_expected: AtomicBool,
        _leak_detector: sui_simulator::NodeLeakDetector,
    }

    impl Default for SimState {
        fn default() -> Self {
            Self {
                sim_node: sui_simulator::runtime::NodeHandle::current(),
                sim_safe_mode_expected: AtomicBool::new(false),
                _leak_detector: sui_simulator::NodeLeakDetector::new(),
            }
        }
    }

    type JwkInjector = dyn Fn(AuthorityName, &OIDCProvider) -> SuiResult<Vec<(JwkId, JWK)>>
        + Send
        + Sync
        + 'static;

    fn default_fetch_jwks(
        _authority: AuthorityName,
        _provider: &OIDCProvider,
    ) -> SuiResult<Vec<(JwkId, JWK)>> {
        use fastcrypto_zkp::bn254::zk_login::parse_jwks;
        // Just load a default Twitch jwk for testing.
        parse_jwks(
            sui_types::zk_login_util::DEFAULT_JWK_BYTES,
            &OIDCProvider::Twitch,
            true,
        )
        .map_err(|_| SuiErrorKind::JWKRetrievalError.into())
    }

    thread_local! {
        static JWK_INJECTOR: std::cell::RefCell<Arc<JwkInjector>> = std::cell::RefCell::new(Arc::new(default_fetch_jwks));
    }

    pub(super) fn get_jwk_injector() -> Arc<JwkInjector> {
        JWK_INJECTOR.with(|injector| injector.borrow().clone())
    }

    pub fn set_jwk_injector(injector: Arc<JwkInjector>) {
        JWK_INJECTOR.with(|cell| *cell.borrow_mut() = injector);
    }
}

#[cfg(msim)]
pub use simulator::set_jwk_injector;
#[cfg(msim)]
use simulator::*;
use sui_core::authority::authority_store_pruner::{ObjectsCompactionFilter, PrunerWatermarks};
use sui_core::{
    consensus_handler::ConsensusHandlerInitializer, safe_client::SafeClientMetricsBase,
    validator_tx_finalizer::ValidatorTxFinalizer,
};

const DEFAULT_GRPC_CONNECT_TIMEOUT: Duration = Duration::from_secs(60);

pub struct SuiNode {
    config: NodeConfig,
    validator_components: Mutex<Option<ValidatorComponents>>,

    /// The http servers responsible for serving RPC traffic (gRPC and JSON-RPC)
    #[allow(unused)]
    http_servers: HttpServers,

    state: Arc<AuthorityState>,
    transaction_orchestrator: Option<Arc<TransactionOrchestrator<NetworkAuthorityClient>>>,
    registry_service: RegistryService,
    metrics: Arc<SuiNodeMetrics>,
    checkpoint_metrics: Arc<CheckpointMetrics>,

    _discovery: discovery::Handle,
    _connection_monitor_handle: mysten_network::anemo_connection_monitor::ConnectionMonitorHandle,
    state_sync_handle: state_sync::Handle,
    randomness_handle: randomness::Handle,
    checkpoint_store: Arc<CheckpointStore>,
    global_state_hasher: Mutex<Option<Arc<GlobalStateHasher>>>,
    connection_monitor_status: Arc<ConnectionMonitorStatus>,

    /// Broadcast channel to send the starting system state for the next epoch.
    end_of_epoch_channel: broadcast::Sender<SuiSystemState>,

    /// Broadcast channel to notify state-sync for new validator peers.
    trusted_peer_change_tx: watch::Sender<TrustedPeerChangeEvent>,

    backpressure_manager: Arc<BackpressureManager>,

    _db_checkpoint_handle: Option<tokio::sync::broadcast::Sender<()>>,

    #[cfg(msim)]
    sim_state: SimState,

    _state_snapshot_uploader_handle: Option<broadcast::Sender<()>>,
    // Channel to allow signaling upstream to shutdown sui-node
    shutdown_channel_tx: broadcast::Sender<Option<RunWithRange>>,

    /// AuthorityAggregator of the network, created at start and beginning of each epoch.
    /// Use ArcSwap so that we could mutate it without taking mut reference.
    // TODO: Eventually we can make this auth aggregator a shared reference so that this
    // update will automatically propagate to other uses.
    auth_agg: Arc<ArcSwap<AuthorityAggregator<NetworkAuthorityClient>>>,

    subscription_service_checkpoint_sender: Option<tokio::sync::mpsc::Sender<Checkpoint>>,
}

impl fmt::Debug for SuiNode {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.debug_struct("SuiNode")
            .field("name", &self.state.name.concise())
            .finish()
    }
}

static MAX_JWK_KEYS_PER_FETCH: usize = 100;

impl SuiNode {
    pub async fn start(
        config: NodeConfig,
        registry_service: RegistryService,
    ) -> Result<Arc<SuiNode>> {
        Self::start_async(
            config,
            registry_service,
            ServerVersion::new("sui-node", "unknown"),
        )
        .await
    }

    fn start_jwk_updater(
        config: &NodeConfig,
        metrics: Arc<SuiNodeMetrics>,
        authority: AuthorityName,
        epoch_store: Arc<AuthorityPerEpochStore>,
        consensus_adapter: Arc<ConsensusAdapter>,
    ) {
        let epoch = epoch_store.epoch();

        let supported_providers = config
            .zklogin_oauth_providers
            .get(&epoch_store.get_chain_identifier().chain())
            .unwrap_or(&BTreeSet::new())
            .iter()
            .map(|s| OIDCProvider::from_str(s).expect("Invalid provider string"))
            .collect::<Vec<_>>();

        let fetch_interval = Duration::from_secs(config.jwk_fetch_interval_seconds);

        info!(
            ?fetch_interval,
            "Starting JWK updater tasks with supported providers: {:?}", supported_providers
        );

        fn validate_jwk(
            metrics: &Arc<SuiNodeMetrics>,
            provider: &OIDCProvider,
            id: &JwkId,
            jwk: &JWK,
        ) -> bool {
            let Ok(iss_provider) = OIDCProvider::from_iss(&id.iss) else {
                warn!(
                    "JWK iss {:?} (retrieved from {:?}) is not a valid provider",
                    id.iss, provider
                );
                metrics
                    .invalid_jwks
                    .with_label_values(&[&provider.to_string()])
                    .inc();
                return false;
            };

            if iss_provider != *provider {
                warn!(
                    "JWK iss {:?} (retrieved from {:?}) does not match provider {:?}",
                    id.iss, provider, iss_provider
                );
                metrics
                    .invalid_jwks
                    .with_label_values(&[&provider.to_string()])
                    .inc();
                return false;
            }

            if !check_total_jwk_size(id, jwk) {
                warn!("JWK {:?} (retrieved from {:?}) is too large", id, provider);
                metrics
                    .invalid_jwks
                    .with_label_values(&[&provider.to_string()])
                    .inc();
                return false;
            }

            true
        }

        // metrics is:
        //  pub struct SuiNodeMetrics {
        //      pub jwk_requests: IntCounterVec,
        //      pub jwk_request_errors: IntCounterVec,
        //      pub total_jwks: IntCounterVec,
        //      pub unique_jwks: IntCounterVec,
        //  }

        for p in supported_providers.into_iter() {
            let provider_str = p.to_string();
            let epoch_store = epoch_store.clone();
            let consensus_adapter = consensus_adapter.clone();
            let metrics = metrics.clone();
            spawn_monitored_task!(epoch_store.clone().within_alive_epoch(
                async move {
                    // note: restart-safe de-duplication happens after consensus, this is
                    // just best-effort to reduce unneeded submissions.
                    let mut seen = HashSet::new();
                    loop {
                        info!("fetching JWK for provider {:?}", p);
                        metrics.jwk_requests.with_label_values(&[&provider_str]).inc();
                        match Self::fetch_jwks(authority, &p).await {
                            Err(e) => {
                                metrics.jwk_request_errors.with_label_values(&[&provider_str]).inc();
                                warn!("Error when fetching JWK for provider {:?} {:?}", p, e);
                                // Retry in 30 seconds
                                tokio::time::sleep(Duration::from_secs(30)).await;
                                continue;
                            }
                            Ok(mut keys) => {
                                metrics.total_jwks
                                    .with_label_values(&[&provider_str])
                                    .inc_by(keys.len() as u64);

                                keys.retain(|(id, jwk)| {
                                    validate_jwk(&metrics, &p, id, jwk) &&
                                    !epoch_store.jwk_active_in_current_epoch(id, jwk) &&
                                    seen.insert((id.clone(), jwk.clone()))
                                });

                                metrics.unique_jwks
                                    .with_label_values(&[&provider_str])
                                    .inc_by(keys.len() as u64);

                                // prevent oauth providers from sending too many keys,
                                // inadvertently or otherwise
                                if keys.len() > MAX_JWK_KEYS_PER_FETCH {
                                    warn!("Provider {:?} sent too many JWKs, only the first {} will be used", p, MAX_JWK_KEYS_PER_FETCH);
                                    keys.truncate(MAX_JWK_KEYS_PER_FETCH);
                                }

                                for (id, jwk) in keys.into_iter() {
                                    info!("Submitting JWK to consensus: {:?}", id);

                                    let txn = ConsensusTransaction::new_jwk_fetched(authority, id, jwk);
                                    consensus_adapter.submit(txn, None, &epoch_store, None, None)
                                        .tap_err(|e| warn!("Error when submitting JWKs to consensus {:?}", e))
                                        .ok();
                                }
                            }
                        }
                        tokio::time::sleep(fetch_interval).await;
                    }
                }
                .instrument(error_span!("jwk_updater_task", epoch)),
            ));
        }
    }

    pub async fn start_async(
        config: NodeConfig,
        registry_service: RegistryService,
        server_version: ServerVersion,
    ) -> Result<Arc<SuiNode>> {
        NodeConfigMetrics::new(&registry_service.default_registry()).record_metrics(&config);
        let mut config = config.clone();
        if config.supported_protocol_versions.is_none() {
            info!(
                "populating config.supported_protocol_versions with default {:?}",
                SupportedProtocolVersions::SYSTEM_DEFAULT
            );
            config.supported_protocol_versions = Some(SupportedProtocolVersions::SYSTEM_DEFAULT);
        }

        let run_with_range = config.run_with_range;
        let is_validator = config.consensus_config().is_some();
        let is_full_node = !is_validator;
        let prometheus_registry = registry_service.default_registry();

        info!(node =? config.protocol_public_key(),
            "Initializing sui-node listening on {}", config.network_address
        );

        // Initialize metrics to track db usage before creating any stores
        DBMetrics::init(registry_service.clone());

        // Initialize Mysten metrics.
        mysten_metrics::init_metrics(&prometheus_registry);
        // Unsupported (because of the use of static variable) and unnecessary in simtests.
        #[cfg(not(msim))]
        mysten_metrics::thread_stall_monitor::start_thread_stall_monitor();

        let genesis = config.genesis()?.clone();

        let secret = Arc::pin(config.protocol_key_pair().copy());
        let genesis_committee = genesis.committee()?;
        let committee_store = Arc::new(CommitteeStore::new(
            config.db_path().join("epochs"),
            &genesis_committee,
            None,
        ));

        let pruner_watermarks = Arc::new(PrunerWatermarks::default());
        let checkpoint_store = CheckpointStore::new(
            &config.db_path().join("checkpoints"),
            pruner_watermarks.clone(),
        );
        let checkpoint_metrics = CheckpointMetrics::new(&registry_service.default_registry());

        Self::check_and_recover_forks(
            &checkpoint_store,
            &checkpoint_metrics,
            is_validator,
            config.fork_recovery.as_ref(),
        )
        .await?;

        let mut pruner_db = None;
        if config
            .authority_store_pruning_config
            .enable_compaction_filter
        {
            pruner_db = Some(Arc::new(AuthorityPrunerTables::open(
                &config.db_path().join("store"),
            )));
        }
        let compaction_filter = pruner_db
            .clone()
            .map(|db| ObjectsCompactionFilter::new(db, &prometheus_registry));

        // By default, only enable write stall on validators for perpetual db.
        let enable_write_stall = config.enable_db_write_stall.unwrap_or(is_validator);
        let perpetual_tables_options = AuthorityPerpetualTablesOptions {
            enable_write_stall,
            compaction_filter,
        };
        let perpetual_tables = Arc::new(AuthorityPerpetualTables::open(
            &config.db_path().join("store"),
            Some(perpetual_tables_options),
            Some(pruner_watermarks.epoch_id.clone()),
        ));
        let is_genesis = perpetual_tables
            .database_is_empty()
            .expect("Database read should not fail at init.");

        let backpressure_manager =
            BackpressureManager::new_from_checkpoint_store(&checkpoint_store);

        let store =
            AuthorityStore::open(perpetual_tables, &genesis, &config, &prometheus_registry).await?;

        let cur_epoch = store.get_recovery_epoch_at_restart()?;
        let committee = committee_store
            .get_committee(&cur_epoch)?
            .expect("Committee of the current epoch must exist");
        let epoch_start_configuration = store
            .get_epoch_start_configuration()?
            .expect("EpochStartConfiguration of the current epoch must exist");
        let cache_metrics = Arc::new(ResolverMetrics::new(&prometheus_registry));
        let signature_verifier_metrics = SignatureVerifierMetrics::new(&prometheus_registry);

        let cache_traits = build_execution_cache(
            &config.execution_cache,
            &prometheus_registry,
            &store,
            backpressure_manager.clone(),
        );

        let auth_agg = {
            let safe_client_metrics_base = SafeClientMetricsBase::new(&prometheus_registry);
            let auth_agg_metrics = Arc::new(AuthAggMetrics::new(&prometheus_registry));
            Arc::new(ArcSwap::new(Arc::new(
                AuthorityAggregator::new_from_epoch_start_state(
                    epoch_start_configuration.epoch_start_state(),
                    &committee_store,
                    safe_client_metrics_base,
                    auth_agg_metrics,
                ),
            )))
        };

        let chain_id = ChainIdentifier::from(*genesis.checkpoint().digest());
        let chain = match config.chain_override_for_testing {
            Some(chain) => chain,
            None => ChainIdentifier::from(*genesis.checkpoint().digest()).chain(),
        };

        let epoch_options = default_db_options().optimize_db_for_write_throughput(4);
        let epoch_store = AuthorityPerEpochStore::new(
            config.protocol_public_key(),
            committee.clone(),
            &config.db_path().join("store"),
            Some(epoch_options.options),
            EpochMetrics::new(&registry_service.default_registry()),
            epoch_start_configuration,
            cache_traits.backing_package_store.clone(),
            cache_traits.object_store.clone(),
            cache_metrics,
            signature_verifier_metrics,
            &config.expensive_safety_check_config,
            (chain_id, chain),
            checkpoint_store
                .get_highest_executed_checkpoint_seq_number()
                .expect("checkpoint store read cannot fail")
                .unwrap_or(0),
            Arc::new(SubmittedTransactionCacheMetrics::new(
                &registry_service.default_registry(),
            )),
        )?;

        info!("created epoch store");

        replay_log!(
            "Beginning replay run. Epoch: {:?}, Protocol config: {:?}",
            epoch_store.epoch(),
            epoch_store.protocol_config()
        );

        // the database is empty at genesis time
        if is_genesis {
            info!("checking SUI conservation at genesis");
            // When we are opening the db table, the only time when it's safe to
            // check SUI conservation is at genesis. Otherwise we may be in the middle of
            // an epoch and the SUI conservation check will fail. This also initialize
            // the expected_network_sui_amount table.
            cache_traits
                .reconfig_api
                .expensive_check_sui_conservation(&epoch_store)
                .expect("SUI conservation check cannot fail at genesis");
        }

        let effective_buffer_stake = epoch_store.get_effective_buffer_stake_bps();
        let default_buffer_stake = epoch_store
            .protocol_config()
            .buffer_stake_for_protocol_upgrade_bps();
        if effective_buffer_stake != default_buffer_stake {
            warn!(
                ?effective_buffer_stake,
                ?default_buffer_stake,
                "buffer_stake_for_protocol_upgrade_bps is currently overridden"
            );
        }

        checkpoint_store.insert_genesis_checkpoint(
            genesis.checkpoint(),
            genesis.checkpoint_contents().clone(),
            &epoch_store,
        );

        info!("creating state sync store");
        let state_sync_store = RocksDbStore::new(
            cache_traits.clone(),
            committee_store.clone(),
            checkpoint_store.clone(),
        );

        let index_store = if is_full_node && config.enable_index_processing {
            info!("creating index store");
            Some(Arc::new(IndexStore::new(
                config.db_path().join("indexes"),
                &prometheus_registry,
                epoch_store
                    .protocol_config()
                    .max_move_identifier_len_as_option(),
                config.remove_deprecated_tables,
            )))
        } else {
            None
        };

        let rpc_index = if is_full_node && config.rpc().is_some_and(|rpc| rpc.enable_indexing()) {
            Some(Arc::new(
                RpcIndexStore::new(
                    &config.db_path(),
                    &store,
                    &checkpoint_store,
                    &epoch_store,
                    &cache_traits.backing_package_store,
                    pruner_watermarks.checkpoint_id.clone(),
                    config.rpc().cloned().unwrap_or_default(),
                )
                .await,
            ))
        } else {
            None
        };

        let chain_identifier = epoch_store.get_chain_identifier();

        info!("creating archive reader");
        // Create network
        // TODO only configure validators as seed/preferred peers for validators and not for
        // fullnodes once we've had a chance to re-work fullnode configuration generation.
        let (trusted_peer_change_tx, trusted_peer_change_rx) = watch::channel(Default::default());
        let (randomness_tx, randomness_rx) = mpsc::channel(
            config
                .p2p_config
                .randomness
                .clone()
                .unwrap_or_default()
                .mailbox_capacity(),
        );
        let P2pComponents {
            p2p_network,
            known_peers,
            discovery_handle,
            state_sync_handle,
            randomness_handle,
        } = Self::create_p2p_network(
            &config,
            state_sync_store.clone(),
            chain_identifier,
            trusted_peer_change_rx,
            randomness_tx,
            &prometheus_registry,
        )?;

        // We must explicitly send this instead of relying on the initial value to trigger
        // watch value change, so that state-sync is able to process it.
        send_trusted_peer_change(
            &config,
            &trusted_peer_change_tx,
            epoch_store.epoch_start_state(),
        )
        .expect("Initial trusted peers must be set");

        info!("start snapshot upload");
        // Start uploading state snapshot to remote store
        let state_snapshot_handle = Self::start_state_snapshot(
            &config,
            &prometheus_registry,
            checkpoint_store.clone(),
            chain_identifier,
        )?;

        // Start uploading db checkpoints to remote store
        info!("start db checkpoint");
        let (db_checkpoint_config, db_checkpoint_handle) = Self::start_db_checkpoint(
            &config,
            &prometheus_registry,
            state_snapshot_handle.is_some(),
        )?;

        if !epoch_store
            .protocol_config()
            .simplified_unwrap_then_delete()
        {
            // We cannot prune tombstones if simplified_unwrap_then_delete is not enabled.
            config
                .authority_store_pruning_config
                .set_killswitch_tombstone_pruning(true);
        }

        let authority_name = config.protocol_public_key();
        let validator_tx_finalizer =
            config
                .enable_validator_tx_finalizer
                .then_some(Arc::new(ValidatorTxFinalizer::new(
                    auth_agg.clone(),
                    authority_name,
                    &prometheus_registry,
                )));

        info!("create authority state");
        let state = AuthorityState::new(
            authority_name,
            secret,
            config.supported_protocol_versions.unwrap(),
            store.clone(),
            cache_traits.clone(),
            epoch_store.clone(),
            committee_store.clone(),
            index_store.clone(),
            rpc_index,
            checkpoint_store.clone(),
            &prometheus_registry,
            genesis.objects(),
            &db_checkpoint_config,
            config.clone(),
            validator_tx_finalizer,
            chain_identifier,
            pruner_db,
            config.policy_config.clone(),
            config.firewall_config.clone(),
            pruner_watermarks,
        )
        .await;
        // ensure genesis txn was executed
        if epoch_store.epoch() == 0 {
            let txn = &genesis.transaction();
            let span = error_span!("genesis_txn", tx_digest = ?txn.digest());
            let transaction =
                sui_types::executable_transaction::VerifiedExecutableTransaction::new_unchecked(
                    sui_types::executable_transaction::ExecutableTransaction::new_from_data_and_sig(
                        genesis.transaction().data().clone(),
                        sui_types::executable_transaction::CertificateProof::Checkpoint(0, 0),
                    ),
                );
            state
                .try_execute_immediately(
                    &transaction,
                    ExecutionEnv::new().with_scheduling_source(SchedulingSource::NonFastPath),
                    &epoch_store,
                )
                .instrument(span)
                .await
                .unwrap();
        }

        // Start the loop that receives new randomness and generates transactions for it.
        RandomnessRoundReceiver::spawn(state.clone(), randomness_rx);

        if config
            .expensive_safety_check_config
            .enable_secondary_index_checks()
            && let Some(indexes) = state.indexes.clone()
        {
            sui_core::verify_indexes::verify_indexes(
                state.get_global_state_hash_store().as_ref(),
                indexes,
            )
            .expect("secondary indexes are inconsistent");
        }

        let (end_of_epoch_channel, end_of_epoch_receiver) =
            broadcast::channel(config.end_of_epoch_broadcast_channel_capacity);

        let transaction_orchestrator = if is_full_node && run_with_range.is_none() {
            Some(Arc::new(TransactionOrchestrator::new_with_auth_aggregator(
                auth_agg.load_full(),
                state.clone(),
                end_of_epoch_receiver,
                &config.db_path(),
                &prometheus_registry,
                &config,
            )))
        } else {
            None
        };

        let (http_servers, subscription_service_checkpoint_sender) = build_http_servers(
            state.clone(),
            state_sync_store,
            &transaction_orchestrator.clone(),
            &config,
            &prometheus_registry,
            server_version,
        )
        .await?;

        let global_state_hasher = Arc::new(GlobalStateHasher::new(
            cache_traits.global_state_hash_store.clone(),
            GlobalStateHashMetrics::new(&prometheus_registry),
        ));

        let authority_names_to_peer_ids = epoch_store
            .epoch_start_state()
            .get_authority_names_to_peer_ids();

        let network_connection_metrics = mysten_network::quinn_metrics::QuinnConnectionMetrics::new(
            "sui",
            &registry_service.default_registry(),
        );

        let authority_names_to_peer_ids = ArcSwap::from_pointee(authority_names_to_peer_ids);

        let connection_monitor_handle =
            mysten_network::anemo_connection_monitor::AnemoConnectionMonitor::spawn(
                p2p_network.downgrade(),
                Arc::new(network_connection_metrics),
                known_peers,
            );

        let connection_monitor_status = ConnectionMonitorStatus {
            connection_statuses: connection_monitor_handle.connection_statuses(),
            authority_names_to_peer_ids,
        };

        let connection_monitor_status = Arc::new(connection_monitor_status);
        let sui_node_metrics = Arc::new(SuiNodeMetrics::new(&registry_service.default_registry()));

        sui_node_metrics
            .binary_max_protocol_version
            .set(ProtocolVersion::MAX.as_u64() as i64);
        sui_node_metrics
            .configured_max_protocol_version
            .set(config.supported_protocol_versions.unwrap().max.as_u64() as i64);

        let validator_components = if state.is_validator(&epoch_store) {
            let (components, _) = futures::join!(
                Self::construct_validator_components(
                    config.clone(),
                    state.clone(),
                    committee,
                    epoch_store.clone(),
                    checkpoint_store.clone(),
                    state_sync_handle.clone(),
                    randomness_handle.clone(),
                    Arc::downgrade(&global_state_hasher),
                    backpressure_manager.clone(),
                    connection_monitor_status.clone(),
                    &registry_service,
                    sui_node_metrics.clone(),
                    checkpoint_metrics.clone(),
                ),
                Self::reexecute_pending_consensus_certs(&epoch_store, &state,)
            );
            let mut components = components?;

            components.consensus_adapter.submit_recovered(&epoch_store);

            // Start the gRPC server
            components.validator_server_handle = components.validator_server_handle.start().await;

            Some(components)
        } else {
            None
        };

        // setup shutdown channel
        let (shutdown_channel, _) = broadcast::channel::<Option<RunWithRange>>(1);

        let node = Self {
            config,
            validator_components: Mutex::new(validator_components),
            http_servers,
            state,
            transaction_orchestrator,
            registry_service,
            metrics: sui_node_metrics,
            checkpoint_metrics,

            _discovery: discovery_handle,
            _connection_monitor_handle: connection_monitor_handle,
            state_sync_handle,
            randomness_handle,
            checkpoint_store,
            global_state_hasher: Mutex::new(Some(global_state_hasher)),
            end_of_epoch_channel,
            connection_monitor_status,
            trusted_peer_change_tx,
            backpressure_manager,

            _db_checkpoint_handle: db_checkpoint_handle,

            #[cfg(msim)]
            sim_state: Default::default(),

            _state_snapshot_uploader_handle: state_snapshot_handle,
            shutdown_channel_tx: shutdown_channel,

            auth_agg,
            subscription_service_checkpoint_sender,
        };

        info!("SuiNode started!");
        let node = Arc::new(node);
        let node_copy = node.clone();
        spawn_monitored_task!(async move {
            let result = Self::monitor_reconfiguration(node_copy, epoch_store).await;
            if let Err(error) = result {
                warn!("Reconfiguration finished with error {:?}", error);
            }
        });

        Ok(node)
    }

    pub fn subscribe_to_epoch_change(&self) -> broadcast::Receiver<SuiSystemState> {
        self.end_of_epoch_channel.subscribe()
    }

    pub fn subscribe_to_shutdown_channel(&self) -> broadcast::Receiver<Option<RunWithRange>> {
        self.shutdown_channel_tx.subscribe()
    }

    pub fn current_epoch_for_testing(&self) -> EpochId {
        self.state.current_epoch_for_testing()
    }

    pub fn db_checkpoint_path(&self) -> PathBuf {
        self.config.db_checkpoint_path()
    }

    // Init reconfig process by starting to reject user certs
    pub async fn close_epoch(&self, epoch_store: &Arc<AuthorityPerEpochStore>) -> SuiResult {
        info!("close_epoch (current epoch = {})", epoch_store.epoch());
        self.validator_components
            .lock()
            .await
            .as_ref()
            .ok_or_else(|| SuiError::from("Node is not a validator"))?
            .consensus_adapter
            .close_epoch(epoch_store);
        Ok(())
    }

    pub fn clear_override_protocol_upgrade_buffer_stake(&self, epoch: EpochId) -> SuiResult {
        self.state
            .clear_override_protocol_upgrade_buffer_stake(epoch)
    }

    pub fn set_override_protocol_upgrade_buffer_stake(
        &self,
        epoch: EpochId,
        buffer_stake_bps: u64,
    ) -> SuiResult {
        self.state
            .set_override_protocol_upgrade_buffer_stake(epoch, buffer_stake_bps)
    }

    // Testing-only API to start epoch close process.
    // For production code, please use the non-testing version.
    pub async fn close_epoch_for_testing(&self) -> SuiResult {
        let epoch_store = self.state.epoch_store_for_testing();
        self.close_epoch(&epoch_store).await
    }

    fn start_state_snapshot(
        config: &NodeConfig,
        prometheus_registry: &Registry,
        checkpoint_store: Arc<CheckpointStore>,
        chain_identifier: ChainIdentifier,
    ) -> Result<Option<tokio::sync::broadcast::Sender<()>>> {
        if let Some(remote_store_config) = &config.state_snapshot_write_config.object_store_config {
            let snapshot_uploader = StateSnapshotUploader::new(
                &config.db_checkpoint_path(),
                &config.snapshot_path(),
                remote_store_config.clone(),
                60,
                prometheus_registry,
                checkpoint_store,
                chain_identifier,
            )?;
            Ok(Some(snapshot_uploader.start()))
        } else {
            Ok(None)
        }
    }

    fn start_db_checkpoint(
        config: &NodeConfig,
        prometheus_registry: &Registry,
        state_snapshot_enabled: bool,
    ) -> Result<(
        DBCheckpointConfig,
        Option<tokio::sync::broadcast::Sender<()>>,
    )> {
        let checkpoint_path = Some(
            config
                .db_checkpoint_config
                .checkpoint_path
                .clone()
                .unwrap_or_else(|| config.db_checkpoint_path()),
        );
        let db_checkpoint_config = if config.db_checkpoint_config.checkpoint_path.is_none() {
            DBCheckpointConfig {
                checkpoint_path,
                perform_db_checkpoints_at_epoch_end: if state_snapshot_enabled {
                    true
                } else {
                    config
                        .db_checkpoint_config
                        .perform_db_checkpoints_at_epoch_end
                },
                ..config.db_checkpoint_config.clone()
            }
        } else {
            config.db_checkpoint_config.clone()
        };

        match (
            db_checkpoint_config.object_store_config.as_ref(),
            state_snapshot_enabled,
        ) {
            // If db checkpoint config object store not specified but
            // state snapshot object store is specified, create handler
            // anyway for marking db checkpoints as completed so that they
            // can be uploaded as state snapshots.
            (None, false) => Ok((db_checkpoint_config, None)),
            (_, _) => {
                let handler = DBCheckpointHandler::new(
                    &db_checkpoint_config.checkpoint_path.clone().unwrap(),
                    db_checkpoint_config.object_store_config.as_ref(),
                    60,
                    db_checkpoint_config
                        .prune_and_compact_before_upload
                        .unwrap_or(true),
                    config.authority_store_pruning_config.clone(),
                    prometheus_registry,
                    state_snapshot_enabled,
                )?;
                Ok((
                    db_checkpoint_config,
                    Some(DBCheckpointHandler::start(handler)),
                ))
            }
        }
    }

    fn create_p2p_network(
        config: &NodeConfig,
        state_sync_store: RocksDbStore,
        chain_identifier: ChainIdentifier,
        trusted_peer_change_rx: watch::Receiver<TrustedPeerChangeEvent>,
        randomness_tx: mpsc::Sender<(EpochId, RandomnessRound, Vec<u8>)>,
        prometheus_registry: &Registry,
    ) -> Result<P2pComponents> {
        let (state_sync, state_sync_server) = state_sync::Builder::new()
            .config(config.p2p_config.state_sync.clone().unwrap_or_default())
            .store(state_sync_store)
            .archive_config(config.archive_reader_config())
            .with_metrics(prometheus_registry)
            .build();

        let (discovery, discovery_server) = discovery::Builder::new(trusted_peer_change_rx)
            .config(config.p2p_config.clone())
            .build();

        let discovery_config = config.p2p_config.discovery.clone().unwrap_or_default();
        let known_peers: HashMap<PeerId, String> = discovery_config
            .allowlisted_peers
            .clone()
            .into_iter()
            .map(|ap| (ap.peer_id, "allowlisted_peer".to_string()))
            .chain(config.p2p_config.seed_peers.iter().filter_map(|peer| {
                peer.peer_id
                    .map(|peer_id| (peer_id, "seed_peer".to_string()))
            }))
            .collect();

        let (randomness, randomness_router) =
            randomness::Builder::new(config.protocol_public_key(), randomness_tx)
                .config(config.p2p_config.randomness.clone().unwrap_or_default())
                .with_metrics(prometheus_registry)
                .build();

        let p2p_network = {
            let routes = anemo::Router::new()
                .add_rpc_service(discovery_server)
                .add_rpc_service(state_sync_server);
            let routes = routes.merge(randomness_router);

            let inbound_network_metrics =
                mysten_network::metrics::NetworkMetrics::new("sui", "inbound", prometheus_registry);
            let outbound_network_metrics = mysten_network::metrics::NetworkMetrics::new(
                "sui",
                "outbound",
                prometheus_registry,
            );

            let service = ServiceBuilder::new()
                .layer(
                    TraceLayer::new_for_server_errors()
                        .make_span_with(DefaultMakeSpan::new().level(tracing::Level::INFO))
                        .on_failure(DefaultOnFailure::new().level(tracing::Level::WARN)),
                )
                .layer(CallbackLayer::new(
                    mysten_network::metrics::MetricsMakeCallbackHandler::new(
                        Arc::new(inbound_network_metrics),
                        config.p2p_config.excessive_message_size(),
                    ),
                ))
                .service(routes);

            let outbound_layer = ServiceBuilder::new()
                .layer(
                    TraceLayer::new_for_client_and_server_errors()
                        .make_span_with(DefaultMakeSpan::new().level(tracing::Level::INFO))
                        .on_failure(DefaultOnFailure::new().level(tracing::Level::WARN)),
                )
                .layer(CallbackLayer::new(
                    mysten_network::metrics::MetricsMakeCallbackHandler::new(
                        Arc::new(outbound_network_metrics),
                        config.p2p_config.excessive_message_size(),
                    ),
                ))
                .into_inner();

            let mut anemo_config = config.p2p_config.anemo_config.clone().unwrap_or_default();
            // Set the max_frame_size to be 1 GB to work around the issue of there being too many
            // staking events in the epoch change txn.
            anemo_config.max_frame_size = Some(1 << 30);

            // Set a higher default value for socket send/receive buffers if not already
            // configured.
            let mut quic_config = anemo_config.quic.unwrap_or_default();
            if quic_config.socket_send_buffer_size.is_none() {
                quic_config.socket_send_buffer_size = Some(20 << 20);
            }
            if quic_config.socket_receive_buffer_size.is_none() {
                quic_config.socket_receive_buffer_size = Some(20 << 20);
            }
            quic_config.allow_failed_socket_buffer_size_setting = true;

            // Set high-performance defaults for quinn transport.
            // With 200MiB buffer size and ~500ms RTT, max throughput ~400MiB/s.
            if quic_config.max_concurrent_bidi_streams.is_none() {
                quic_config.max_concurrent_bidi_streams = Some(500);
            }
            if quic_config.max_concurrent_uni_streams.is_none() {
                quic_config.max_concurrent_uni_streams = Some(500);
            }
            if quic_config.stream_receive_window.is_none() {
                quic_config.stream_receive_window = Some(100 << 20);
            }
            if quic_config.receive_window.is_none() {
                quic_config.receive_window = Some(200 << 20);
            }
            if quic_config.send_window.is_none() {
                quic_config.send_window = Some(200 << 20);
            }
            if quic_config.crypto_buffer_size.is_none() {
                quic_config.crypto_buffer_size = Some(1 << 20);
            }
            if quic_config.max_idle_timeout_ms.is_none() {
                quic_config.max_idle_timeout_ms = Some(30_000);
            }
            if quic_config.keep_alive_interval_ms.is_none() {
                quic_config.keep_alive_interval_ms = Some(5_000);
            }
            anemo_config.quic = Some(quic_config);

            let server_name = format!("sui-{}", chain_identifier);
            let network = Network::bind(config.p2p_config.listen_address)
                .server_name(&server_name)
                .private_key(config.network_key_pair().copy().private().0.to_bytes())
                .config(anemo_config)
                .outbound_request_layer(outbound_layer)
                .start(service)?;
            info!(
                server_name = server_name,
                "P2p network started on {}",
                network.local_addr()
            );

            network
        };

        let discovery_handle =
            discovery.start(p2p_network.clone(), config.network_key_pair().copy());
        let state_sync_handle = state_sync.start(p2p_network.clone());
        let randomness_handle = randomness.start(p2p_network.clone());

        Ok(P2pComponents {
            p2p_network,
            known_peers,
            discovery_handle,
            state_sync_handle,
            randomness_handle,
        })
    }

    async fn construct_validator_components(
        config: NodeConfig,
        state: Arc<AuthorityState>,
        committee: Arc<Committee>,
        epoch_store: Arc<AuthorityPerEpochStore>,
        checkpoint_store: Arc<CheckpointStore>,
        state_sync_handle: state_sync::Handle,
        randomness_handle: randomness::Handle,
        global_state_hasher: Weak<GlobalStateHasher>,
        backpressure_manager: Arc<BackpressureManager>,
        connection_monitor_status: Arc<ConnectionMonitorStatus>,
        registry_service: &RegistryService,
        sui_node_metrics: Arc<SuiNodeMetrics>,
        checkpoint_metrics: Arc<CheckpointMetrics>,
    ) -> Result<ValidatorComponents> {
        let mut config_clone = config.clone();
        let consensus_config = config_clone
            .consensus_config
            .as_mut()
            .ok_or_else(|| anyhow!("Validator is missing consensus config"))?;

        let client = Arc::new(UpdatableConsensusClient::new());
        let consensus_adapter = Arc::new(Self::construct_consensus_adapter(
            &committee,
            consensus_config,
            state.name,
            connection_monitor_status.clone(),
            &registry_service.default_registry(),
            epoch_store.protocol_config().clone(),
            client.clone(),
            checkpoint_store.clone(),
        ));
        let consensus_manager = Arc::new(ConsensusManager::new(
            &config,
            consensus_config,
            registry_service,
            client,
        ));

        // This only gets started up once, not on every epoch. (Make call to remove every epoch.)
        let consensus_store_pruner = ConsensusStorePruner::new(
            consensus_manager.get_storage_base_path(),
            consensus_config.db_retention_epochs(),
            consensus_config.db_pruner_period(),
            &registry_service.default_registry(),
        );

        let sui_tx_validator_metrics =
            SuiTxValidatorMetrics::new(&registry_service.default_registry());

        let validator_server_handle = Self::start_grpc_validator_service(
            &config,
            state.clone(),
            consensus_adapter.clone(),
            &registry_service.default_registry(),
        )
        .await?;

        // Starts an overload monitor that monitors the execution of the authority.
        // Don't start the overload monitor when max_load_shedding_percentage is 0.
        let validator_overload_monitor_handle = if config
            .authority_overload_config
            .max_load_shedding_percentage
            > 0
        {
            let authority_state = Arc::downgrade(&state);
            let overload_config = config.authority_overload_config.clone();
            fail_point!("starting_overload_monitor");
            Some(spawn_monitored_task!(overload_monitor(
                authority_state,
                overload_config,
            )))
        } else {
            None
        };

        Self::start_epoch_specific_validator_components(
            &config,
            state.clone(),
            consensus_adapter,
            checkpoint_store,
            epoch_store,
            state_sync_handle,
            randomness_handle,
            consensus_manager,
            consensus_store_pruner,
            global_state_hasher,
            backpressure_manager,
            validator_server_handle,
            validator_overload_monitor_handle,
            checkpoint_metrics,
            sui_node_metrics,
            sui_tx_validator_metrics,
        )
        .await
    }

    async fn start_epoch_specific_validator_components(
        config: &NodeConfig,
        state: Arc<AuthorityState>,
        consensus_adapter: Arc<ConsensusAdapter>,
        checkpoint_store: Arc<CheckpointStore>,
        epoch_store: Arc<AuthorityPerEpochStore>,
        state_sync_handle: state_sync::Handle,
        randomness_handle: randomness::Handle,
        consensus_manager: Arc<ConsensusManager>,
        consensus_store_pruner: ConsensusStorePruner,
        state_hasher: Weak<GlobalStateHasher>,
        backpressure_manager: Arc<BackpressureManager>,
        validator_server_handle: SpawnOnce,
        validator_overload_monitor_handle: Option<JoinHandle<()>>,
        checkpoint_metrics: Arc<CheckpointMetrics>,
        sui_node_metrics: Arc<SuiNodeMetrics>,
        sui_tx_validator_metrics: Arc<SuiTxValidatorMetrics>,
    ) -> Result<ValidatorComponents> {
        let checkpoint_service = Self::build_checkpoint_service(
            config,
            consensus_adapter.clone(),
            checkpoint_store.clone(),
            epoch_store.clone(),
            state.clone(),
            state_sync_handle,
            state_hasher,
            checkpoint_metrics.clone(),
        );

        // create a new map that gets injected into both the consensus handler and the consensus adapter
        // the consensus handler will write values forwarded from consensus, and the consensus adapter
        // will read the values to make decisions about which validator submits a transaction to consensus
        let low_scoring_authorities = Arc::new(ArcSwap::new(Arc::new(HashMap::new())));

        consensus_adapter.swap_low_scoring_authorities(low_scoring_authorities.clone());

        if epoch_store.randomness_state_enabled() {
            let randomness_manager = RandomnessManager::try_new(
                Arc::downgrade(&epoch_store),
                Box::new(consensus_adapter.clone()),
                randomness_handle,
                config.protocol_key_pair(),
            )
            .await;
            if let Some(randomness_manager) = randomness_manager {
                epoch_store
                    .set_randomness_manager(randomness_manager)
                    .await?;
            }
        }

        ExecutionTimeObserver::spawn(
            epoch_store.clone(),
            Box::new(consensus_adapter.clone()),
            config
                .execution_time_observer_config
                .clone()
                .unwrap_or_default(),
        );

        let throughput_calculator = Arc::new(ConsensusThroughputCalculator::new(
            None,
            state.metrics.clone(),
        ));

        let throughput_profiler = Arc::new(ConsensusThroughputProfiler::new(
            throughput_calculator.clone(),
            None,
            None,
            state.metrics.clone(),
            ThroughputProfileRanges::from_chain(epoch_store.get_chain_identifier()),
        ));

        consensus_adapter.swap_throughput_profiler(throughput_profiler);

        let consensus_handler_initializer = ConsensusHandlerInitializer::new(
            state.clone(),
            checkpoint_service.clone(),
            epoch_store.clone(),
            consensus_adapter.clone(),
            low_scoring_authorities,
            throughput_calculator,
            backpressure_manager,
        );

        info!("Starting consensus manager asynchronously");

        // Spawn consensus startup asynchronously to avoid blocking other components
        tokio::spawn({
            let config = config.clone();
            let epoch_store = epoch_store.clone();
            let sui_tx_validator = SuiTxValidator::new(
                state.clone(),
                consensus_adapter.clone(),
                checkpoint_service.clone(),
                sui_tx_validator_metrics.clone(),
            );
            let consensus_manager = consensus_manager.clone();
            async move {
                consensus_manager
                    .start(
                        &config,
                        epoch_store,
                        consensus_handler_initializer,
                        sui_tx_validator,
                    )
                    .await;
            }
        });
        let replay_waiter = consensus_manager.replay_waiter();

        info!("Spawning checkpoint service");
        let replay_waiter = if std::env::var("DISABLE_REPLAY_WAITER").is_ok() {
            None
        } else {
            Some(replay_waiter)
        };
        checkpoint_service
            .spawn(epoch_store.clone(), replay_waiter)
            .await;

        if epoch_store.authenticator_state_enabled() {
            Self::start_jwk_updater(
                config,
                sui_node_metrics,
                state.name,
                epoch_store.clone(),
                consensus_adapter.clone(),
            );
        }

        Ok(ValidatorComponents {
            validator_server_handle,
            validator_overload_monitor_handle,
            consensus_manager,
            consensus_store_pruner,
            consensus_adapter,
            checkpoint_metrics,
            sui_tx_validator_metrics,
        })
    }

    fn build_checkpoint_service(
        config: &NodeConfig,
        consensus_adapter: Arc<ConsensusAdapter>,
        checkpoint_store: Arc<CheckpointStore>,
        epoch_store: Arc<AuthorityPerEpochStore>,
        state: Arc<AuthorityState>,
        state_sync_handle: state_sync::Handle,
        state_hasher: Weak<GlobalStateHasher>,
        checkpoint_metrics: Arc<CheckpointMetrics>,
    ) -> Arc<CheckpointService> {
        let epoch_start_timestamp_ms = epoch_store.epoch_start_state().epoch_start_timestamp_ms();
        let epoch_duration_ms = epoch_store.epoch_start_state().epoch_duration_ms();

        debug!(
            "Starting checkpoint service with epoch start timestamp {}
            and epoch duration {}",
            epoch_start_timestamp_ms, epoch_duration_ms
        );

        let checkpoint_output = Box::new(SubmitCheckpointToConsensus {
            sender: consensus_adapter,
            signer: state.secret.clone(),
            authority: config.protocol_public_key(),
            next_reconfiguration_timestamp_ms: epoch_start_timestamp_ms
                .checked_add(epoch_duration_ms)
                .expect("Overflow calculating next_reconfiguration_timestamp_ms"),
            metrics: checkpoint_metrics.clone(),
        });

        let certified_checkpoint_output = SendCheckpointToStateSync::new(state_sync_handle);
        let max_tx_per_checkpoint = max_tx_per_checkpoint(epoch_store.protocol_config());
        let max_checkpoint_size_bytes =
            epoch_store.protocol_config().max_checkpoint_size_bytes() as usize;

        CheckpointService::build(
            state.clone(),
            checkpoint_store,
            epoch_store,
            state.get_transaction_cache_reader().clone(),
            state_hasher,
            checkpoint_output,
            Box::new(certified_checkpoint_output),
            checkpoint_metrics,
            max_tx_per_checkpoint,
            max_checkpoint_size_bytes,
        )
    }

    fn construct_consensus_adapter(
        committee: &Committee,
        consensus_config: &ConsensusConfig,
        authority: AuthorityName,
        connection_monitor_status: Arc<ConnectionMonitorStatus>,
        prometheus_registry: &Registry,
        protocol_config: ProtocolConfig,
        consensus_client: Arc<dyn ConsensusClient>,
        checkpoint_store: Arc<CheckpointStore>,
    ) -> ConsensusAdapter {
        let ca_metrics = ConsensusAdapterMetrics::new(prometheus_registry);
        // The consensus adapter allows the authority to send user certificates through consensus.

        ConsensusAdapter::new(
            consensus_client,
            checkpoint_store,
            authority,
            connection_monitor_status,
            consensus_config.max_pending_transactions(),
            consensus_config.max_pending_transactions() * 2 / committee.num_members(),
            consensus_config.max_submit_position,
            consensus_config.submit_delay_step_override(),
            ca_metrics,
            protocol_config,
        )
    }

    async fn start_grpc_validator_service(
        config: &NodeConfig,
        state: Arc<AuthorityState>,
        consensus_adapter: Arc<ConsensusAdapter>,
        prometheus_registry: &Registry,
    ) -> Result<SpawnOnce> {
        let validator_service = ValidatorService::new(
            state.clone(),
            consensus_adapter,
            Arc::new(ValidatorServiceMetrics::new(prometheus_registry)),
            config.policy_config.clone().map(|p| p.client_id_source),
        );

        let mut server_conf = mysten_network::config::Config::new();
        server_conf.connect_timeout = Some(DEFAULT_GRPC_CONNECT_TIMEOUT);
        server_conf.http2_keepalive_interval = Some(DEFAULT_GRPC_CONNECT_TIMEOUT);
        server_conf.http2_keepalive_timeout = Some(DEFAULT_GRPC_CONNECT_TIMEOUT);
        server_conf.global_concurrency_limit = config.grpc_concurrency_limit;
        server_conf.load_shed = config.grpc_load_shed;
        let mut server_builder =
            ServerBuilder::from_config(&server_conf, GrpcMetrics::new(prometheus_registry));

        server_builder = server_builder.add_service(ValidatorServer::new(validator_service));

        let tls_config = sui_tls::create_rustls_server_config(
            config.network_key_pair().copy().private(),
            SUI_TLS_SERVER_NAME.to_string(),
        );

        let network_address = config.network_address().clone();

        let (ready_tx, ready_rx) = oneshot::channel();

        Ok(SpawnOnce::new(ready_rx, async move {
            let server = server_builder
                .bind(&network_address, Some(tls_config))
                .await
                .unwrap_or_else(|err| panic!("Failed to bind to {network_address}: {err}"));
            let local_addr = server.local_addr();
            info!("Listening to traffic on {local_addr}");
            ready_tx.send(()).unwrap();
            if let Err(err) = server.serve().await {
                info!("Server stopped: {err}");
            }
            info!("Server stopped");
        }))
    }

    /// Re-executes pending consensus certificates, which may not have been committed to disk
    /// before the node restarted. This is necessary for the following reasons:
    ///
    /// 1. For any transaction for which we returned signed effects to a client, we must ensure
    ///    that we have re-executed the transaction before we begin accepting grpc requests.
    ///    Otherwise we would appear to have forgotten about the transaction.
    /// 2. While this is running, we are concurrently waiting for all previously built checkpoints
    ///    to be rebuilt. Since there may be dependencies in either direction (from checkpointed
    ///    consensus transactions to pending consensus transactions, or vice versa), we must
    ///    re-execute pending consensus transactions to ensure that both processes can complete.
    /// 3. Also note that for any pending consensus transactions for which we wrote a signed effects
    ///    digest to disk, we must re-execute using that digest as the expected effects digest,
    ///    to ensure that we cannot arrive at different effects than what we previously signed.
    async fn reexecute_pending_consensus_certs(
        epoch_store: &Arc<AuthorityPerEpochStore>,
        state: &Arc<AuthorityState>,
    ) {
        let mut pending_consensus_certificates = Vec::new();
        let mut additional_certs = Vec::new();

        for tx in epoch_store.get_all_pending_consensus_transactions() {
            match tx.kind {
                // Shared object txns cannot be re-executed at this point, because we must wait for
                // consensus replay to assign shared object versions.
                ConsensusTransactionKind::CertifiedTransaction(tx) if !tx.is_consensus_tx() => {
                    let tx = *tx;
                    // new_unchecked is safe because we never submit a transaction to consensus
                    // without verifying it
                    let tx = VerifiedExecutableTransaction::new_from_certificate(
                        VerifiedCertificate::new_unchecked(tx),
                    );
                    // we only need to re-execute if we previously signed the effects (which indicates we
                    // returned the effects to a client).
                    if let Some(fx_digest) = epoch_store
                        .get_signed_effects_digest(tx.digest())
                        .expect("db error")
                    {
                        pending_consensus_certificates.push((
                            Schedulable::Transaction(tx),
                            ExecutionEnv::new().with_expected_effects_digest(fx_digest),
                        ));
                    } else {
                        additional_certs.push((
                            Schedulable::Transaction(tx),
                            ExecutionEnv::new()
                                .with_scheduling_source(SchedulingSource::NonFastPath),
                        ));
                    }
                }
                _ => (),
            }
        }

        let digests = pending_consensus_certificates
            .iter()
            // unwrap_digest okay because only user certs are in pending_consensus_certificates
            .map(|(tx, _)| *tx.key().unwrap_digest())
            .collect::<Vec<_>>();

        info!(
            "reexecuting {} pending consensus certificates: {:?}",
            digests.len(),
            digests
        );

        state
            .execution_scheduler()
            .enqueue(pending_consensus_certificates, epoch_store);
        state
            .execution_scheduler()
            .enqueue(additional_certs, epoch_store);

        // If this times out, the validator will still almost certainly start up fine. But, it is
        // possible that it may temporarily "forget" about transactions that it had previously
        // executed. This could confuse clients in some circumstances. However, the transactions
        // are still in pending_consensus_certificates, so we cannot lose any finality guarantees.
        let timeout = if cfg!(msim) { 120 } else { 60 };
        if tokio::time::timeout(
            std::time::Duration::from_secs(timeout),
            state
                .get_transaction_cache_reader()
                .notify_read_executed_effects_digests(
                    "SuiNode::notify_read_executed_effects_digests",
                    &digests,
                ),
        )
        .await
        .is_err()
        {
            // Log all the digests that were not executed to help debugging.
            let executed_effects_digests = state
                .get_transaction_cache_reader()
                .multi_get_executed_effects_digests(&digests);
            let pending_digests = digests
                .iter()
                .zip(executed_effects_digests.iter())
                .filter_map(|(digest, executed_effects_digest)| {
                    if executed_effects_digest.is_none() {
                        Some(digest)
                    } else {
                        None
                    }
                })
                .collect::<Vec<_>>();
            debug_fatal!(
                "Timed out waiting for effects digests to be executed: {:?}",
                pending_digests
            );
        }
    }

    pub fn state(&self) -> Arc<AuthorityState> {
        self.state.clone()
    }

    // Only used for testing because of how epoch store is loaded.
    pub fn reference_gas_price_for_testing(&self) -> Result<u64, anyhow::Error> {
        self.state.reference_gas_price_for_testing()
    }

    pub fn clone_committee_store(&self) -> Arc<CommitteeStore> {
        self.state.committee_store().clone()
    }

    /*
    pub fn clone_authority_store(&self) -> Arc<AuthorityStore> {
        self.state.db()
    }
    */

    /// Clone an AuthorityAggregator currently used in this node, if the node is a fullnode.
    /// After reconfig, Transaction Driver builds a new AuthorityAggregator. The caller
    /// of this function will mostly likely want to call this again
    /// to get a fresh one.
    pub fn clone_authority_aggregator(
        &self,
    ) -> Option<Arc<AuthorityAggregator<NetworkAuthorityClient>>> {
        self.transaction_orchestrator
            .as_ref()
            .map(|to| to.clone_authority_aggregator())
    }

    pub fn transaction_orchestrator(
        &self,
    ) -> Option<Arc<TransactionOrchestrator<NetworkAuthorityClient>>> {
        self.transaction_orchestrator.clone()
    }

    /// This function awaits the completion of checkpoint execution of the current epoch,
    /// after which it initiates reconfiguration of the entire system.
    pub async fn monitor_reconfiguration(
        self: Arc<Self>,
        mut epoch_store: Arc<AuthorityPerEpochStore>,
    ) -> Result<()> {
        let checkpoint_executor_metrics =
            CheckpointExecutorMetrics::new(&self.registry_service.default_registry());

        loop {
            let mut hasher_guard = self.global_state_hasher.lock().await;
            let hasher = hasher_guard.take().unwrap();
            info!(
                "Creating checkpoint executor for epoch {}",
                epoch_store.epoch()
            );
            let checkpoint_executor = CheckpointExecutor::new(
                epoch_store.clone(),
                self.checkpoint_store.clone(),
                self.state.clone(),
                hasher.clone(),
                self.backpressure_manager.clone(),
                self.config.checkpoint_executor_config.clone(),
                checkpoint_executor_metrics.clone(),
                self.subscription_service_checkpoint_sender.clone(),
            );

            let run_with_range = self.config.run_with_range;

            let cur_epoch_store = self.state.load_epoch_store_one_call_per_task();

            // Update the current protocol version metric.
            self.metrics
                .current_protocol_version
                .set(cur_epoch_store.protocol_config().version.as_u64() as i64);

            // Advertise capabilities to committee, if we are a validator.
            if let Some(components) = &*self.validator_components.lock().await {
                // TODO: without this sleep, the consensus message is not delivered reliably.
                tokio::time::sleep(Duration::from_millis(1)).await;

                let config = cur_epoch_store.protocol_config();
                let mut supported_protocol_versions = self
                    .config
                    .supported_protocol_versions
                    .expect("Supported versions should be populated")
                    // no need to send digests of versions less than the current version
                    .truncate_below(config.version);

                while supported_protocol_versions.max > config.version {
                    let proposed_protocol_config = ProtocolConfig::get_for_version(
                        supported_protocol_versions.max,
                        cur_epoch_store.get_chain(),
                    );

                    if proposed_protocol_config.enable_accumulators()
                        && !epoch_store.accumulator_root_exists()
                    {
                        error!(
                            "cannot upgrade to protocol version {:?} because accumulator root does not exist",
                            supported_protocol_versions.max
                        );
                        supported_protocol_versions.max = supported_protocol_versions.max.prev();
                    } else {
                        break;
                    }
                }

                let binary_config = config.binary_config(None);
                let transaction = if config.authority_capabilities_v2() {
                    ConsensusTransaction::new_capability_notification_v2(
                        AuthorityCapabilitiesV2::new(
                            self.state.name,
                            cur_epoch_store.get_chain_identifier().chain(),
                            supported_protocol_versions,
                            self.state
                                .get_available_system_packages(&binary_config)
                                .await,
                        ),
                    )
                } else {
                    ConsensusTransaction::new_capability_notification(AuthorityCapabilitiesV1::new(
                        self.state.name,
                        self.config
                            .supported_protocol_versions
                            .expect("Supported versions should be populated"),
                        self.state
                            .get_available_system_packages(&binary_config)
                            .await,
                    ))
                };
                info!(?transaction, "submitting capabilities to consensus");
                components.consensus_adapter.submit(
                    transaction,
                    None,
                    &cur_epoch_store,
                    None,
                    None,
                )?;
            }

            let stop_condition = checkpoint_executor.run_epoch(run_with_range).await;

            if stop_condition == StopReason::RunWithRangeCondition {
                SuiNode::shutdown(&self).await;
                self.shutdown_channel_tx
                    .send(run_with_range)
                    .expect("RunWithRangeCondition met but failed to send shutdown message");
                return Ok(());
            }

            // Safe to call because we are in the middle of reconfiguration.
            let latest_system_state = self
                .state
                .get_object_cache_reader()
                .get_sui_system_state_object_unsafe()
                .expect("Read Sui System State object cannot fail");

            #[cfg(msim)]
            if !self
                .sim_state
                .sim_safe_mode_expected
                .load(Ordering::Relaxed)
            {
                debug_assert!(!latest_system_state.safe_mode());
            }

            #[cfg(not(msim))]
            debug_assert!(!latest_system_state.safe_mode());

            if let Err(err) = self.end_of_epoch_channel.send(latest_system_state.clone())
                && self.state.is_fullnode(&cur_epoch_store)
            {
                warn!(
                    "Failed to send end of epoch notification to subscriber: {:?}",
                    err
                );
            }

            cur_epoch_store.record_is_safe_mode_metric(latest_system_state.safe_mode());
            let new_epoch_start_state = latest_system_state.into_epoch_start_state();

            self.auth_agg.store(Arc::new(
                self.auth_agg
                    .load()
                    .recreate_with_new_epoch_start_state(&new_epoch_start_state),
            ));

            let next_epoch_committee = new_epoch_start_state.get_sui_committee();
            let next_epoch = next_epoch_committee.epoch();
            assert_eq!(cur_epoch_store.epoch() + 1, next_epoch);

            info!(
                next_epoch,
                "Finished executing all checkpoints in epoch. About to reconfigure the system."
            );

            fail_point_async!("reconfig_delay");

            // We save the connection monitor status map regardless of validator / fullnode status
            // so that we don't need to restart the connection monitor every epoch.
            // Update the mappings that will be used by the consensus adapter if it exists or is
            // about to be created.
            let authority_names_to_peer_ids =
                new_epoch_start_state.get_authority_names_to_peer_ids();
            self.connection_monitor_status
                .update_mapping_for_epoch(authority_names_to_peer_ids);

            cur_epoch_store.record_epoch_reconfig_start_time_metric();

            let _ = send_trusted_peer_change(
                &self.config,
                &self.trusted_peer_change_tx,
                &new_epoch_start_state,
            );

            let mut validator_components_lock_guard = self.validator_components.lock().await;

            // The following code handles 4 different cases, depending on whether the node
            // was a validator in the previous epoch, and whether the node is a validator
            // in the new epoch.
            let new_epoch_store = self
                .reconfigure_state(
                    &self.state,
                    &cur_epoch_store,
                    next_epoch_committee.clone(),
                    new_epoch_start_state,
                    hasher.clone(),
                )
                .await;

            let new_validator_components = if let Some(ValidatorComponents {
                validator_server_handle,
                validator_overload_monitor_handle,
                consensus_manager,
                consensus_store_pruner,
                consensus_adapter,
                checkpoint_metrics,
                sui_tx_validator_metrics,
            }) = validator_components_lock_guard.take()
            {
                info!("Reconfiguring the validator.");

                consensus_manager.shutdown().await;
                info!("Consensus has shut down.");

                info!("Epoch store finished reconfiguration.");

                // No other components should be holding a strong reference to state hasher
                // at this point. Confirm here before we swap in the new hasher.
                let global_state_hasher_metrics = Arc::into_inner(hasher)
                    .expect("Object state hasher should have no other references at this point")
                    .metrics();
                let new_hasher = Arc::new(GlobalStateHasher::new(
                    self.state.get_global_state_hash_store().clone(),
                    global_state_hasher_metrics,
                ));
                let weak_hasher = Arc::downgrade(&new_hasher);
                *hasher_guard = Some(new_hasher);

                consensus_store_pruner.prune(next_epoch).await;

                if self.state.is_validator(&new_epoch_store) {
                    // Only restart consensus if this node is still a validator in the new epoch.
                    Some(
                        Self::start_epoch_specific_validator_components(
                            &self.config,
                            self.state.clone(),
                            consensus_adapter,
                            self.checkpoint_store.clone(),
                            new_epoch_store.clone(),
                            self.state_sync_handle.clone(),
                            self.randomness_handle.clone(),
                            consensus_manager,
                            consensus_store_pruner,
                            weak_hasher,
                            self.backpressure_manager.clone(),
                            validator_server_handle,
                            validator_overload_monitor_handle,
                            checkpoint_metrics,
                            self.metrics.clone(),
                            sui_tx_validator_metrics,
                        )
                        .await?,
                    )
                } else {
                    info!("This node is no longer a validator after reconfiguration");
                    None
                }
            } else {
                // No other components should be holding a strong reference to state hasher
                // at this point. Confirm here before we swap in the new hasher.
                let global_state_hasher_metrics = Arc::into_inner(hasher)
                    .expect("Object state hasher should have no other references at this point")
                    .metrics();
                let new_hasher = Arc::new(GlobalStateHasher::new(
                    self.state.get_global_state_hash_store().clone(),
                    global_state_hasher_metrics,
                ));
                let weak_hasher = Arc::downgrade(&new_hasher);
                *hasher_guard = Some(new_hasher);

                if self.state.is_validator(&new_epoch_store) {
                    info!("Promoting the node from fullnode to validator, starting grpc server");

                    let mut components = Self::construct_validator_components(
                        self.config.clone(),
                        self.state.clone(),
                        Arc::new(next_epoch_committee.clone()),
                        new_epoch_store.clone(),
                        self.checkpoint_store.clone(),
                        self.state_sync_handle.clone(),
                        self.randomness_handle.clone(),
                        weak_hasher,
                        self.backpressure_manager.clone(),
                        self.connection_monitor_status.clone(),
                        &self.registry_service,
                        self.metrics.clone(),
                        self.checkpoint_metrics.clone(),
                    )
                    .await?;

                    components.validator_server_handle =
                        components.validator_server_handle.start().await;

                    Some(components)
                } else {
                    None
                }
            };
            *validator_components_lock_guard = new_validator_components;

            // Force releasing current epoch store DB handle, because the
            // Arc<AuthorityPerEpochStore> may linger.
            cur_epoch_store.release_db_handles();

            if cfg!(msim)
                && !matches!(
                    self.config
                        .authority_store_pruning_config
                        .num_epochs_to_retain_for_checkpoints(),
                    None | Some(u64::MAX) | Some(0)
                )
            {
                self.state
                    .prune_checkpoints_for_eligible_epochs_for_testing(
                        self.config.clone(),
                        sui_core::authority::authority_store_pruner::AuthorityStorePruningMetrics::new_for_test(),
                    )
                    .await?;
            }

            epoch_store = new_epoch_store;
            info!("Reconfiguration finished");
        }
    }

    async fn shutdown(&self) {
        if let Some(validator_components) = &*self.validator_components.lock().await {
            validator_components.consensus_manager.shutdown().await;
        }
    }

    async fn reconfigure_state(
        &self,
        state: &Arc<AuthorityState>,
        cur_epoch_store: &AuthorityPerEpochStore,
        next_epoch_committee: Committee,
        next_epoch_start_system_state: EpochStartSystemState,
        global_state_hasher: Arc<GlobalStateHasher>,
    ) -> Arc<AuthorityPerEpochStore> {
        let next_epoch = next_epoch_committee.epoch();

        let last_checkpoint = self
            .checkpoint_store
            .get_epoch_last_checkpoint(cur_epoch_store.epoch())
            .expect("Error loading last checkpoint for current epoch")
            .expect("Could not load last checkpoint for current epoch");

        let last_checkpoint_seq = *last_checkpoint.sequence_number();

        assert_eq!(
            Some(last_checkpoint_seq),
            self.checkpoint_store
                .get_highest_executed_checkpoint_seq_number()
                .expect("Error loading highest executed checkpoint sequence number")
        );

        let epoch_start_configuration = EpochStartConfiguration::new(
            next_epoch_start_system_state,
            *last_checkpoint.digest(),
            state.get_object_store().as_ref(),
            EpochFlag::default_flags_for_new_epoch(&state.config),
        )
        .expect("EpochStartConfiguration construction cannot fail");

        let new_epoch_store = self
            .state
            .reconfigure(
                cur_epoch_store,
                self.config.supported_protocol_versions.unwrap(),
                next_epoch_committee,
                epoch_start_configuration,
                global_state_hasher,
                &self.config.expensive_safety_check_config,
                last_checkpoint_seq,
            )
            .await
            .expect("Reconfigure authority state cannot fail");
        info!(next_epoch, "Node State has been reconfigured");
        assert_eq!(next_epoch, new_epoch_store.epoch());
        self.state.get_reconfig_api().update_epoch_flags_metrics(
            cur_epoch_store.epoch_start_config().flags(),
            new_epoch_store.epoch_start_config().flags(),
        );

        new_epoch_store
    }

    pub fn get_config(&self) -> &NodeConfig {
        &self.config
    }

    pub fn randomness_handle(&self) -> randomness::Handle {
        self.randomness_handle.clone()
    }

    /// Get a short prefix of a digest for metric labels
    fn get_digest_prefix(digest: impl std::fmt::Display) -> String {
        let digest_str = digest.to_string();
        if digest_str.len() >= 8 {
            digest_str[0..8].to_string()
        } else {
            digest_str
        }
    }

    /// Check for previously detected forks and handle them appropriately.
    /// For validators with fork recovery config, clear the fork if it matches the recovery config.
    /// For all other cases, block node startup if a fork is detected.
    async fn check_and_recover_forks(
        checkpoint_store: &CheckpointStore,
        checkpoint_metrics: &CheckpointMetrics,
        is_validator: bool,
        fork_recovery: Option<&ForkRecoveryConfig>,
    ) -> Result<()> {
        // Fork detection and recovery is only relevant for validators
        // Fullnodes should sync from validators and don't need fork checking
        if !is_validator {
            return Ok(());
        }

        // Try to recover from forks if recovery config is provided
        if let Some(recovery) = fork_recovery {
            Self::try_recover_checkpoint_fork(checkpoint_store, recovery)?;
            Self::try_recover_transaction_fork(checkpoint_store, recovery)?;
        }

        if let Some((checkpoint_seq, checkpoint_digest)) = checkpoint_store
            .get_checkpoint_fork_detected()
            .map_err(|e| {
                error!("Failed to check for checkpoint fork: {:?}", e);
                e
            })?
        {
            Self::handle_checkpoint_fork(
                checkpoint_seq,
                checkpoint_digest,
                checkpoint_metrics,
                fork_recovery,
            )
            .await?;
        }
        if let Some((tx_digest, expected_effects, actual_effects)) = checkpoint_store
            .get_transaction_fork_detected()
            .map_err(|e| {
                error!("Failed to check for transaction fork: {:?}", e);
                e
            })?
        {
            Self::handle_transaction_fork(
                tx_digest,
                expected_effects,
                actual_effects,
                checkpoint_metrics,
                fork_recovery,
            )
            .await?;
        }

        Ok(())
    }

    fn try_recover_checkpoint_fork(
        checkpoint_store: &CheckpointStore,
        recovery: &ForkRecoveryConfig,
    ) -> Result<()> {
        // If configured overrides include a checkpoint whose locally computed digest mismatches,
        // clear locally computed checkpoints from that sequence (inclusive).
        for (seq, expected_digest_str) in &recovery.checkpoint_overrides {
            let Ok(expected_digest) = CheckpointDigest::from_str(expected_digest_str) else {
                anyhow::bail!(
                    "Invalid checkpoint digest override for seq {}: {}",
                    seq,
                    expected_digest_str
                );
            };

            if let Some(local_summary) = checkpoint_store.get_locally_computed_checkpoint(*seq)? {
                let local_digest = sui_types::message_envelope::Message::digest(&local_summary);
                if local_digest != expected_digest {
                    info!(
                        seq,
                        local = %Self::get_digest_prefix(local_digest),
                        expected = %Self::get_digest_prefix(expected_digest),
                        "Fork recovery: clearing locally_computed_checkpoints from {} due to digest mismatch",
                        seq
                    );
                    checkpoint_store
                        .clear_locally_computed_checkpoints_from(*seq)
                        .context(
                            "Failed to clear locally computed checkpoints from override seq",
                        )?;
                }
            }
        }

        if let Some((checkpoint_seq, checkpoint_digest)) =
            checkpoint_store.get_checkpoint_fork_detected()?
            && recovery.checkpoint_overrides.contains_key(&checkpoint_seq)
        {
            info!(
                "Fork recovery enabled: clearing checkpoint fork at seq {} with digest {:?}",
                checkpoint_seq, checkpoint_digest
            );
            checkpoint_store
                .clear_checkpoint_fork_detected()
                .expect("Failed to clear checkpoint fork detected marker");
        }
        Ok(())
    }

    fn try_recover_transaction_fork(
        checkpoint_store: &CheckpointStore,
        recovery: &ForkRecoveryConfig,
    ) -> Result<()> {
        if recovery.transaction_overrides.is_empty() {
            return Ok(());
        }

        if let Some((tx_digest, _, _)) = checkpoint_store.get_transaction_fork_detected()?
            && recovery
                .transaction_overrides
                .contains_key(&tx_digest.to_string())
        {
            info!(
                "Fork recovery enabled: clearing transaction fork for tx {:?}",
                tx_digest
            );
            checkpoint_store
                .clear_transaction_fork_detected()
                .expect("Failed to clear transaction fork detected marker");
        }
        Ok(())
    }

    fn get_current_timestamp() -> u64 {
        std::time::SystemTime::now()
            .duration_since(std::time::SystemTime::UNIX_EPOCH)
            .unwrap()
            .as_secs()
    }

    async fn handle_checkpoint_fork(
        checkpoint_seq: u64,
        checkpoint_digest: CheckpointDigest,
        checkpoint_metrics: &CheckpointMetrics,
        fork_recovery: Option<&ForkRecoveryConfig>,
    ) -> Result<()> {
        checkpoint_metrics
            .checkpoint_fork_crash_mode
            .with_label_values(&[
                &checkpoint_seq.to_string(),
                &Self::get_digest_prefix(checkpoint_digest),
                &Self::get_current_timestamp().to_string(),
            ])
            .set(1);

        let behavior = fork_recovery
            .map(|fr| fr.fork_crash_behavior)
            .unwrap_or_default();

        match behavior {
            ForkCrashBehavior::AwaitForkRecovery => {
                error!(
                    checkpoint_seq = checkpoint_seq,
                    checkpoint_digest = ?checkpoint_digest,
                    "Checkpoint fork detected! Node startup halted. Sleeping indefinitely."
                );
                futures::future::pending::<()>().await;
                unreachable!("pending() should never return");
            }
            ForkCrashBehavior::ReturnError => {
                error!(
                    checkpoint_seq = checkpoint_seq,
                    checkpoint_digest = ?checkpoint_digest,
                    "Checkpoint fork detected! Returning error."
                );
                Err(anyhow::anyhow!(
                    "Checkpoint fork detected! checkpoint_seq: {}, checkpoint_digest: {:?}",
                    checkpoint_seq,
                    checkpoint_digest
                ))
            }
        }
    }

    async fn handle_transaction_fork(
        tx_digest: TransactionDigest,
        expected_effects_digest: TransactionEffectsDigest,
        actual_effects_digest: TransactionEffectsDigest,
        checkpoint_metrics: &CheckpointMetrics,
        fork_recovery: Option<&ForkRecoveryConfig>,
    ) -> Result<()> {
        checkpoint_metrics
            .transaction_fork_crash_mode
            .with_label_values(&[
                &Self::get_digest_prefix(tx_digest),
                &Self::get_digest_prefix(expected_effects_digest),
                &Self::get_digest_prefix(actual_effects_digest),
                &Self::get_current_timestamp().to_string(),
            ])
            .set(1);

        let behavior = fork_recovery
            .map(|fr| fr.fork_crash_behavior)
            .unwrap_or_default();

        match behavior {
            ForkCrashBehavior::AwaitForkRecovery => {
                error!(
                    tx_digest = ?tx_digest,
                    expected_effects_digest = ?expected_effects_digest,
                    actual_effects_digest = ?actual_effects_digest,
                    "Transaction fork detected! Node startup halted. Sleeping indefinitely."
                );
                futures::future::pending::<()>().await;
                unreachable!("pending() should never return");
            }
            ForkCrashBehavior::ReturnError => {
                error!(
                    tx_digest = ?tx_digest,
                    expected_effects_digest = ?expected_effects_digest,
                    actual_effects_digest = ?actual_effects_digest,
                    "Transaction fork detected! Returning error."
                );
                Err(anyhow::anyhow!(
                    "Transaction fork detected! tx_digest: {:?}, expected_effects: {:?}, actual_effects: {:?}",
                    tx_digest,
                    expected_effects_digest,
                    actual_effects_digest
                ))
            }
        }
    }
}

#[cfg(not(msim))]
impl SuiNode {
    async fn fetch_jwks(
        _authority: AuthorityName,
        provider: &OIDCProvider,
    ) -> SuiResult<Vec<(JwkId, JWK)>> {
        use fastcrypto_zkp::bn254::zk_login::fetch_jwks;
        use sui_types::error::SuiErrorKind;
        let client = reqwest::Client::new();
        fetch_jwks(provider, &client, true)
            .await
            .map_err(|_| SuiErrorKind::JWKRetrievalError.into())
    }
}

#[cfg(msim)]
impl SuiNode {
    pub fn get_sim_node_id(&self) -> sui_simulator::task::NodeId {
        self.sim_state.sim_node.id()
    }

    pub fn set_safe_mode_expected(&self, new_value: bool) {
        info!("Setting safe mode expected to {}", new_value);
        self.sim_state
            .sim_safe_mode_expected
            .store(new_value, Ordering::Relaxed);
    }

    #[allow(unused_variables)]
    async fn fetch_jwks(
        authority: AuthorityName,
        provider: &OIDCProvider,
    ) -> SuiResult<Vec<(JwkId, JWK)>> {
        get_jwk_injector()(authority, provider)
    }
}

enum SpawnOnce {
    // Mutex is only needed to make SpawnOnce Send
    Unstarted(oneshot::Receiver<()>, Mutex<BoxFuture<'static, ()>>),
    #[allow(unused)]
    Started(JoinHandle<()>),
}

impl SpawnOnce {
    pub fn new(
        ready_rx: oneshot::Receiver<()>,
        future: impl Future<Output = ()> + Send + 'static,
    ) -> Self {
        Self::Unstarted(ready_rx, Mutex::new(Box::pin(future)))
    }

    pub async fn start(self) -> Self {
        match self {
            Self::Unstarted(ready_rx, future) => {
                let future = future.into_inner();
                let handle = tokio::spawn(future);
                ready_rx.await.unwrap();
                Self::Started(handle)
            }
            Self::Started(_) => self,
        }
    }
}

/// Notify state-sync that a new list of trusted peers are now available.
fn send_trusted_peer_change(
    config: &NodeConfig,
    sender: &watch::Sender<TrustedPeerChangeEvent>,
    epoch_state_state: &EpochStartSystemState,
) -> Result<(), watch::error::SendError<TrustedPeerChangeEvent>> {
    sender
        .send(TrustedPeerChangeEvent {
            new_peers: epoch_state_state.get_validator_as_p2p_peers(config.protocol_public_key()),
        })
        .tap_err(|err| {
            warn!(
                "Failed to send validator peer information to state sync: {:?}",
                err
            );
        })
}

fn build_kv_store(
    state: &Arc<AuthorityState>,
    config: &NodeConfig,
    registry: &Registry,
) -> Result<Arc<TransactionKeyValueStore>> {
    let metrics = KeyValueStoreMetrics::new(registry);
    let db_store = TransactionKeyValueStore::new("rocksdb", metrics.clone(), state.clone());

    let base_url = &config.transaction_kv_store_read_config.base_url;

    if base_url.is_empty() {
        info!("no http kv store url provided, using local db only");
        return Ok(Arc::new(db_store));
    }

    let base_url: url::Url = base_url.parse().tap_err(|e| {
        error!(
            "failed to parse config.transaction_kv_store_config.base_url ({:?}) as url: {}",
            base_url, e
        )
    })?;

    let network_str = match state.get_chain_identifier().chain() {
        Chain::Mainnet => "/mainnet",
        _ => {
            info!("using local db only for kv store");
            return Ok(Arc::new(db_store));
        }
    };

    let base_url = base_url.join(network_str)?.to_string();
    let http_store = HttpKVStore::new_kv(
        &base_url,
        config.transaction_kv_store_read_config.cache_size,
        metrics.clone(),
    )?;
    info!("using local key-value store with fallback to http key-value store");
    Ok(Arc::new(FallbackTransactionKVStore::new_kv(
        db_store,
        http_store,
        metrics,
        "json_rpc_fallback",
    )))
}

async fn build_http_servers(
    state: Arc<AuthorityState>,
    store: RocksDbStore,
    transaction_orchestrator: &Option<Arc<TransactionOrchestrator<NetworkAuthorityClient>>>,
    config: &NodeConfig,
    prometheus_registry: &Registry,
    server_version: ServerVersion,
) -> Result<(HttpServers, Option<tokio::sync::mpsc::Sender<Checkpoint>>)> {
    // Validators do not expose these APIs
    if config.consensus_config().is_some() {
        return Ok((HttpServers::default(), None));
    }

    let mut router = axum::Router::new();

    let json_rpc_router = {
        let traffic_controller = state.traffic_controller.clone();
        let mut server = JsonRpcServerBuilder::new(
            env!("CARGO_PKG_VERSION"),
            prometheus_registry,
            traffic_controller,
            config.policy_config.clone(),
        );

        let kv_store = build_kv_store(&state, config, prometheus_registry)?;

        let metrics = Arc::new(JsonRpcMetrics::new(prometheus_registry));
        server.register_module(ReadApi::new(
            state.clone(),
            kv_store.clone(),
            metrics.clone(),
        ))?;
        server.register_module(CoinReadApi::new(
            state.clone(),
            kv_store.clone(),
            metrics.clone(),
        ))?;

        // if run_with_range is enabled we want to prevent any transactions
        // run_with_range = None is normal operating conditions
        if config.run_with_range.is_none() {
            server.register_module(TransactionBuilderApi::new(state.clone()))?;
        }
        server.register_module(GovernanceReadApi::new(state.clone(), metrics.clone()))?;
        server.register_module(BridgeReadApi::new(state.clone(), metrics.clone()))?;

        if let Some(transaction_orchestrator) = transaction_orchestrator {
            server.register_module(TransactionExecutionApi::new(
                state.clone(),
                transaction_orchestrator.clone(),
                metrics.clone(),
            ))?;
        }

        let name_service_config =
            if let (Some(package_address), Some(registry_id), Some(reverse_registry_id)) = (
                config.name_service_package_address,
                config.name_service_registry_id,
                config.name_service_reverse_registry_id,
            ) {
                sui_name_service::NameServiceConfig::new(
                    package_address,
                    registry_id,
                    reverse_registry_id,
                )
            } else {
                match state.get_chain_identifier().chain() {
                    Chain::Mainnet => sui_name_service::NameServiceConfig::mainnet(),
                    Chain::Testnet => sui_name_service::NameServiceConfig::testnet(),
                    Chain::Unknown => sui_name_service::NameServiceConfig::default(),
                }
            };

        server.register_module(IndexerApi::new(
            state.clone(),
            ReadApi::new(state.clone(), kv_store.clone(), metrics.clone()),
            kv_store,
            name_service_config,
            metrics,
            config.indexer_max_subscriptions,
        ))?;
        server.register_module(MoveUtils::new(state.clone()))?;

        let server_type = config.jsonrpc_server_type();

        server.to_router(server_type).await?
    };

    router = router.merge(json_rpc_router);

    let (subscription_service_checkpoint_sender, subscription_service_handle) =
        SubscriptionService::build(prometheus_registry);
    let rpc_router = {
        let mut rpc_service =
            sui_rpc_api::RpcService::new(Arc::new(RestReadStore::new(state.clone(), store)));
        rpc_service.with_server_version(server_version);

        if let Some(config) = config.rpc.clone() {
            rpc_service.with_config(config);
        }

        rpc_service.with_metrics(RpcMetrics::new(prometheus_registry));
        rpc_service.with_subscription_service(subscription_service_handle);

        if let Some(transaction_orchestrator) = transaction_orchestrator {
            rpc_service.with_executor(transaction_orchestrator.clone())
        }

        rpc_service.into_router().await
    };

    let layers = ServiceBuilder::new()
        .map_request(|mut request: axum::http::Request<_>| {
            if let Some(connect_info) = request.extensions().get::<sui_http::ConnectInfo>() {
                let axum_connect_info = axum::extract::ConnectInfo(connect_info.remote_addr);
                request.extensions_mut().insert(axum_connect_info);
            }
            request
        })
        .layer(axum::middleware::from_fn(server_timing_middleware))
        // Setup a permissive CORS policy
        .layer(
            tower_http::cors::CorsLayer::new()
                .allow_methods([http::Method::GET, http::Method::POST])
                .allow_origin(tower_http::cors::Any)
                .allow_headers(tower_http::cors::Any),
        );

    router = router.merge(rpc_router).layer(layers);

    let https = if let Some((tls_config, https_address)) = config
        .rpc()
        .and_then(|config| config.tls_config().map(|tls| (tls, config.https_address())))
    {
        let https = sui_http::Builder::new()
            .tls_single_cert(tls_config.cert(), tls_config.key())
            .and_then(|builder| builder.serve(https_address, router.clone()))
            .map_err(|e| anyhow::anyhow!(e))?;

        info!(
            https_address =? https.local_addr(),
            "HTTPS rpc server listening on {}",
            https.local_addr()
        );

        Some(https)
    } else {
        None
    };

    let http = sui_http::Builder::new()
        .serve(&config.json_rpc_address, router)
        .map_err(|e| anyhow::anyhow!(e))?;

    info!(
        http_address =? http.local_addr(),
        "HTTP rpc server listening on {}",
        http.local_addr()
    );

    Ok((
        HttpServers {
            http: Some(http),
            https,
        },
        Some(subscription_service_checkpoint_sender),
    ))
}

#[cfg(not(test))]
fn max_tx_per_checkpoint(protocol_config: &ProtocolConfig) -> usize {
    protocol_config.max_transactions_per_checkpoint() as usize
}

#[cfg(test)]
fn max_tx_per_checkpoint(_: &ProtocolConfig) -> usize {
    2
}

#[derive(Default)]
struct HttpServers {
    #[allow(unused)]
    http: Option<sui_http::ServerHandle>,
    #[allow(unused)]
    https: Option<sui_http::ServerHandle>,
}

#[cfg(test)]
mod tests {
    use super::*;
    use prometheus::Registry;
    use std::collections::BTreeMap;
    use sui_config::node::{ForkCrashBehavior, ForkRecoveryConfig};
    use sui_core::checkpoints::{CheckpointMetrics, CheckpointStore};
    use sui_types::digests::{CheckpointDigest, TransactionDigest, TransactionEffectsDigest};

    #[tokio::test]
    async fn test_fork_error_and_recovery_both_paths() {
        let checkpoint_store = CheckpointStore::new_for_tests();
        let checkpoint_metrics = CheckpointMetrics::new(&Registry::new());

        // ---------- Checkpoint fork path ----------
        let seq_num = 42;
        let digest = CheckpointDigest::random();
        checkpoint_store
            .record_checkpoint_fork_detected(seq_num, digest)
            .unwrap();

        let fork_recovery = ForkRecoveryConfig {
            transaction_overrides: Default::default(),
            checkpoint_overrides: Default::default(),
            fork_crash_behavior: ForkCrashBehavior::ReturnError,
        };

        let r = SuiNode::check_and_recover_forks(
            &checkpoint_store,
            &checkpoint_metrics,
            true,
            Some(&fork_recovery),
        )
        .await;
        assert!(r.is_err());
        assert!(
            r.unwrap_err()
                .to_string()
                .contains("Checkpoint fork detected")
        );

        let mut checkpoint_overrides = BTreeMap::new();
        checkpoint_overrides.insert(seq_num, digest.to_string());
        let fork_recovery_with_override = ForkRecoveryConfig {
            transaction_overrides: Default::default(),
            checkpoint_overrides,
            fork_crash_behavior: ForkCrashBehavior::ReturnError,
        };
        let r = SuiNode::check_and_recover_forks(
            &checkpoint_store,
            &checkpoint_metrics,
            true,
            Some(&fork_recovery_with_override),
        )
        .await;
        assert!(r.is_ok());
        assert!(
            checkpoint_store
                .get_checkpoint_fork_detected()
                .unwrap()
                .is_none()
        );

        // ---------- Transaction fork path ----------
        let tx_digest = TransactionDigest::random();
        let expected_effects = TransactionEffectsDigest::random();
        let actual_effects = TransactionEffectsDigest::random();
        checkpoint_store
            .record_transaction_fork_detected(tx_digest, expected_effects, actual_effects)
            .unwrap();

        let fork_recovery = ForkRecoveryConfig {
            transaction_overrides: Default::default(),
            checkpoint_overrides: Default::default(),
            fork_crash_behavior: ForkCrashBehavior::ReturnError,
        };
        let r = SuiNode::check_and_recover_forks(
            &checkpoint_store,
            &checkpoint_metrics,
            true,
            Some(&fork_recovery),
        )
        .await;
        assert!(r.is_err());
        assert!(
            r.unwrap_err()
                .to_string()
                .contains("Transaction fork detected")
        );

        let mut transaction_overrides = BTreeMap::new();
        transaction_overrides.insert(tx_digest.to_string(), actual_effects.to_string());
        let fork_recovery_with_override = ForkRecoveryConfig {
            transaction_overrides,
            checkpoint_overrides: Default::default(),
            fork_crash_behavior: ForkCrashBehavior::ReturnError,
        };
        let r = SuiNode::check_and_recover_forks(
            &checkpoint_store,
            &checkpoint_metrics,
            true,
            Some(&fork_recovery_with_override),
        )
        .await;
        assert!(r.is_ok());
        assert!(
            checkpoint_store
                .get_transaction_fork_detected()
                .unwrap()
                .is_none()
        );
    }
}