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::in_test_configuration;
use prometheus::Registry;
use std::collections::{BTreeSet, HashMap, HashSet};
use std::fmt;
use std::future::Future;
use std::path::Path;
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::admission_queue::{
    AdmissionQueueContext, AdmissionQueueManager, AdmissionQueueMetrics,
};
use sui_core::authority::ExecutionEnv;
use sui_core::authority::authority_store_tables::AuthorityPerpetualTablesOptions;
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::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_core::randomness_round_receiver::{RandomnessRoundReceiver, RandomnessRoundReceiverHandle};
use sui_network::endpoint_manager::{AddressSource, EndpointId};
use sui_network::validator::server::SUI_TLS_SERVER_NAME;
use sui_types::full_checkpoint_content::Checkpoint;
use sui_types::node_role::NodeRole;

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::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::messages_consensus::AuthorityCapabilitiesV2;
use sui_types::sui_system_state::SuiSystemState;
use tap::tap::TapFallible;
use tokio::sync::oneshot;
use tokio::sync::{Mutex, broadcast, mpsc};
use tokio::task::JoinHandle;
use tower::ServiceBuilder;
use tracing::{Instrument, error_span, info};
use tracing::{debug, error, warn};

// Logs at debug level in test configuration, info level otherwise.
// JWK logs cause significant volume in tests, but are insignificant in prod,
// so we keep them at info
macro_rules! jwk_log {
    ($($arg:tt)+) => {
        if in_test_configuration() {
            debug!($($arg)+);
        } else {
            info!($($arg)+);
        }
    };
}

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::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, CheckpointOutput, CheckpointService, CheckpointStore, LogCheckpointOutput,
    SendCheckpointToStateSync, SubmitCheckpointToConsensus,
};
use sui_core::consensus_adapter::{ConsensusAdapter, ConsensusAdapterMetrics};
use sui_core::consensus_manager::ConsensusManager;
use sui_core::consensus_throughput_calculator::ConsensusThroughputCalculator;
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_store_embed::EmbeddedRpcStore;
use sui_core::signature_verifier::SignatureVerifierMetrics;
use sui_core::storage::RocksDbStore;
use sui_core::storage::RpcStoreReadStore;
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_arg, fail_point_async, replay_log};
use sui_network::api::ValidatorServer;
use sui_network::discovery;
use sui_network::endpoint_manager::EndpointManager;
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::{ConsensusTransaction, check_total_jwk_size};
use sui_types::storage::RpcStateReader;
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 address_prober;
pub mod admin;
pub mod db_shell;
mod handle;
pub mod metrics;

pub struct ValidatorComponents {
    validator_server_handle: Option<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>,
    admission_queue: Option<AdmissionQueueContext>,
}

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,
    endpoint_manager: EndpointManager,
}

#[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::PrunerWatermarks;
use sui_core::{
    consensus_handler::ConsensusHandlerInitializer, safe_client::SafeClientMetricsBase,
};

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>>>,

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

    /// EndpointManager for updating peer network addresses.
    endpoint_manager: EndpointManager,

    /// Handle to the discovery-shared address prober (`None` when disabled).
    address_prober: Option<address_prober::Handle>,

    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>>,

    /// Handle shared with RandomnessManager and the consensus layer.
    randomness_receiver_handle: Arc<RandomnessRoundReceiverHandle>,

    /// 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::broadcast::Sender<Arc<Checkpoint>>>,

    /// The embedded `sui-rpc-store`, present when the node is a fullnode
    /// with indexing enabled. Held for the node's lifetime so its tip
    /// indexer keeps running (dropping it aborts the indexer). Exposed
    /// through [`SuiNode::embedded_rpc_store`] for introspection.
    embedded_rpc_store: Option<EmbeddedRpcStore>,
}

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 {
                        jwk_log!("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() {
                                    jwk_log!("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>> {
        // Fail fast on config errors before starting any node components.
        if let Some(prober_config) = &config.address_prober {
            prober_config.validate()?;
        }

        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 prometheus_registry = registry_service.default_registry();
        let node_role = config.intended_node_role();

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

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

        // Initialize db sync-to-disk setting from config (falls back to env var if not set)
        typed_store::init_write_sync(config.enable_db_sync_to_disk);

        // 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());

        #[allow(unused_mut)]
        let mut build_version = server_version.version.to_string();
        fail_point_arg!("override_binary_version", |version: std::sync::Arc<
            std::sync::Mutex<String>,
        >| {
            #[cfg(msim)]
            {
                build_version = version.lock().unwrap().clone();
            }
        });
        // Embedded in any fork marker this run records, so recovery can refuse to clear a fork
        // under the same binary version that produced it.
        checkpoint_store.set_binary_version(&build_version);

        if node_role.runs_consensus() {
            Self::check_and_recover_forks(
                &checkpoint_store,
                &checkpoint_metrics,
                config.fork_recovery.as_ref(),
                &build_version,
            )
            .await?;
        }

        // By default, only enable write stall on nodes that run consensus.
        let enable_write_stall = config
            .enable_db_write_stall
            .unwrap_or(node_role.runs_consensus());
        // The tidehunter objects compactor retains only the latest version per
        // ObjectID and is mutually exclusive with the object pruner. Enable it
        // for validators (which always disable the pruner), and also for any
        // node configured with `num_epochs_to_retain = 0` — that aggressive
        // setting is what the compactor replaces. The pruner is force-disabled
        // in `AuthorityStorePruner::new` whenever this is true.
        let enable_objects_compactor = node_role.is_validator()
            || config.authority_store_pruning_config.num_epochs_to_retain == 0;
        let perpetual_tables_options = AuthorityPerpetualTablesOptions {
            enable_write_stall,
            enable_objects_compactor,
        };
        let perpetual_tables = Arc::new(AuthorityPerpetualTables::open(
            &config.db_store_path(),
            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);
            Arc::new(ArcSwap::new(Arc::new(
                AuthorityAggregator::new_from_epoch_start_state(
                    epoch_start_configuration.epoch_start_state(),
                    &committee_store,
                    safe_client_metrics_base,
                ),
            )))
        };

        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 highest_executed_checkpoint = checkpoint_store
            .get_highest_executed_checkpoint_seq_number()
            .expect("checkpoint store read cannot fail")
            .unwrap_or(0);

        let previous_epoch_last_checkpoint = if cur_epoch == 0 {
            0
        } else {
            checkpoint_store
                .get_epoch_last_checkpoint_seq_number(cur_epoch - 1)
                .expect("checkpoint store read cannot fail")
                .unwrap_or(highest_executed_checkpoint)
        };

        let epoch_options = default_db_options().optimize_db_for_write_throughput(4, false);
        let epoch_store = AuthorityPerEpochStore::new(
            config.protocol_public_key(),
            committee.clone(),
            &config.db_store_path(),
            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),
            highest_executed_checkpoint,
            previous_epoch_last_checkpoint,
            Arc::new(SubmittedTransactionCacheMetrics::new(
                &registry_service.default_registry(),
            )),
            config.fullnode_sync_mode,
        )?;

        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 node_role.is_fullnode() && config.enable_index_processing {
            info!("creating jsonrpc 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 chain_identifier = epoch_store.get_chain_identifier();

        // The embedded `sui-rpc-store` is the node's index backend: when
        // indexing is enabled it builds the derived-index and ledger-history
        // column families (indexed independently of the authority store) and
        // serves the index read paths from the embedded store. Raw chain data
        // is still served from the perpetual store.
        let mut embedded_rpc_store =
            if node_role.is_fullnode() && config.rpc().is_some_and(|rpc| rpc.enable_indexing()) {
                info!("creating embedded rpc-store");
                // The embedded `sui-rpc-store` replaced the legacy `rpc-index`
                // backend; remove its now-dead on-disk directory if a prior
                // version left one behind.
                remove_legacy_rpc_index_store(&config.db_path());
                // The tip indexer pulls checkpoints from the node's local
                // checkpoint / perpetual stores via a dedicated read handle.
                let ingestion_source = RocksDbStore::new(
                    cache_traits.clone(),
                    committee_store.clone(),
                    checkpoint_store.clone(),
                );
                let embedded_rpc_store = EmbeddedRpcStore::bootstrap(
                    &config,
                    &store,
                    &checkpoint_store,
                    ingestion_source,
                    chain_identifier,
                    &prometheus_registry,
                )
                .await?;
                Some(embedded_rpc_store)
            } else {
                None
            };

        info!("creating archive reader");
        // Create network
        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,
            endpoint_manager,
        } = Self::create_p2p_network(
            &config,
            state_sync_store.clone(),
            chain_identifier,
            randomness_tx,
            &prometheus_registry,
        )?;

        // Inject configured peer address overrides.
        for peer in &config.p2p_config.peer_address_overrides {
            endpoint_manager
                .update_endpoint(
                    EndpointId::P2p(peer.peer_id),
                    AddressSource::Config,
                    peer.addresses.clone(),
                )
                .expect("Updating peer address overrides should not fail");
        }

        // Send initial peer addresses to the p2p network.
        update_peer_addresses(
            &config,
            &endpoint_manager,
            epoch_store.epoch_start_state(),
            None,
        );

        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();

        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(),
            embedded_rpc_store.as_ref().map(|embedded| embedded.store()),
            checkpoint_store.clone(),
            &prometheus_registry,
            genesis.objects(),
            &db_checkpoint_config,
            config.clone(),
            chain_identifier,
            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),
                    ),
                );
            let _enter = span.enter();
            state
                .try_execute_immediately(&transaction, ExecutionEnv::new(), &epoch_store)
                .unwrap();
        }

        // Start the loop that receives new randomness and generates transactions for it.
        // The returned is long-lived (node lifetime).
        let randomness_receiver_handle =
            RandomnessRoundReceiver::spawn(state.clone(), randomness_rx);

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

        let transaction_orchestrator = if node_role.is_fullnode() && 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,
            node_role,
            embedded_rpc_store.as_ref(),
        )
        .await?;

        // Start the embedded rpc-store's tip indexer. It follows the tip
        // via the checkpoint executor's broadcast stream and backfills
        // any gap from the perpetual store. Spawned on a background task
        // (see `spawn_indexer`) so node startup does not block on the
        // first checkpoint, which the executor only produces after this
        // function returns.
        if let Some(embedded) = embedded_rpc_store.as_mut() {
            embedded.spawn_indexer(
                subscription_service_checkpoint_sender.clone(),
                prometheus_registry.clone(),
            );
        }

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

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

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

        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 node_role = epoch_store.node_role();
        let validator_components = if node_role.runs_consensus() {
            let mut components = 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(),
                &registry_service,
                sui_node_metrics.clone(),
                checkpoint_metrics.clone(),
                node_role,
                randomness_receiver_handle.clone(),
            )
            .await?;

            if node_role.is_validator() {
                components
                    .consensus_adapter
                    .recover_end_of_publish(&epoch_store);

                // Start the gRPC server
                components.validator_server_handle = Some(
                    components
                        .validator_server_handle
                        .take()
                        .unwrap()
                        .start()
                        .await,
                );

                // Set the consensus address updater so that we can update the consensus peer addresses when requested.
                endpoint_manager
                    .set_consensus_address_updater(components.consensus_manager.clone());
            } else {
                info!("Starting node as Observer — connecting to configured peers");
            }

            Some(components)
        } else {
            None
        };

        let address_prober = if Self::address_prober_enabled(&config) {
            let handle = address_prober::Builder::new()
                .config(config.address_prober.clone().unwrap_or_default())
                .with_metrics(&prometheus_registry)
                .build()
                .start(
                    p2p_network.clone(),
                    discovery_handle.sender(),
                    consensus_config::NetworkKeyPair::new(config.network_key_pair().copy()),
                );
            // Seed the current epoch if we are starting as a validator.
            if node_role.is_validator()
                && let Some(components) = &validator_components
            {
                handle.update_epoch(
                    epoch_store.epoch(),
                    epoch_store.epoch_start_state().get_consensus_committee(),
                    components.consensus_manager.clone(),
                );
            }
            Some(handle)
        } 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,
            endpoint_manager,
            backpressure_manager,
            address_prober,

            _db_checkpoint_handle: db_checkpoint_handle,

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

            _state_snapshot_uploader_handle: state_snapshot_handle,
            shutdown_channel_tx: shutdown_channel,
            randomness_receiver_handle,

            auth_agg,
            subscription_service_checkpoint_sender,
            embedded_rpc_store,
        };

        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,
                config.state_snapshot_write_config.archive_interval_epochs,
            )?;
            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,
        randomness_tx: mpsc::Sender<(EpochId, RandomnessRound, Vec<u8>)>,
        prometheus_registry: &Registry,
    ) -> Result<P2pComponents> {
        let mut p2p_config = config.p2p_config.clone();
        {
            let disc = p2p_config.discovery.get_or_insert_with(Default::default);
            if disc.peer_addr_store_path.is_none() {
                disc.peer_addr_store_path =
                    Some(config.db_path().join("discovery_peer_cache.yaml"));
            }
        }
        let mut discovery_builder = discovery::Builder::new().config(p2p_config.clone());
        if let Some(consensus_config) = &config.consensus_config {
            let effective_addr = consensus_config
                .external_address
                .as_ref()
                .or(consensus_config.listen_address.as_ref());
            if let Some(addr) = effective_addr {
                discovery_builder = discovery_builder.consensus_external_address(addr.clone());
            }
        }
        let (discovery, discovery_server, endpoint_manager) = discovery_builder.build();
        let discovery_sender = discovery.sender();

        let (state_sync, state_sync_router) = state_sync::Builder::new()
            .config(config.p2p_config.state_sync.clone().unwrap_or_default())
            .store(state_sync_store)
            .archive_config(config.archive_reader_config())
            .discovery_sender(discovery_sender)
            .with_metrics(prometheus_registry)
            .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)
                .merge(state_sync_router);
            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();
            // Inbound requests on this network are small (signatures, queries, summaries).
            // Cap request frames at 1 MiB.
            anemo_config.max_request_frame_size = Some(1 << 20);
            // Responses can be larger (checkpoint contents).
            // Cap response frames at 128 MiB.
            anemo_config.max_response_frame_size = Some(128 << 20);

            // 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(10_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,
            endpoint_manager,
        })
    }

    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>,
        registry_service: &RegistryService,
        sui_node_metrics: Arc<SuiNodeMetrics>,
        checkpoint_metrics: Arc<CheckpointMetrics>,
        node_role: NodeRole,
        randomness_receiver_handle: Arc<RandomnessRoundReceiverHandle>,
    ) -> Result<ValidatorComponents> {
        let mut config_clone = config.clone();
        let consensus_config = config_clone
            .consensus_config
            .as_mut()
            .ok_or_else(|| anyhow!("Node is missing consensus config"))?;

        let client = Arc::new(UpdatableConsensusClient::new());
        let inflight_slot_freed_notify = Arc::new(tokio::sync::Notify::new());
        let consensus_adapter = Arc::new(Self::construct_consensus_adapter(
            &committee,
            consensus_config,
            state.name,
            &registry_service.default_registry(),
            client.clone(),
            checkpoint_store.clone(),
            inflight_slot_freed_notify.clone(),
        ));

        let consensus_manager = Arc::new(ConsensusManager::new(
            &config,
            consensus_config,
            registry_service,
            client,
            node_role,
        ));

        // 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, admission_queue) = if node_role.is_validator() {
            let (handle, queue) = Self::start_grpc_validator_service(
                &config,
                state.clone(),
                consensus_adapter.clone(),
                epoch_store.clone(),
                &registry_service.default_registry(),
                inflight_slot_freed_notify,
            )
            .await?;
            (Some(handle), queue)
        } else {
            (None, None)
        };

        // 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 node_role.is_validator()
            && 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,
            randomness_receiver_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,
            admission_queue,
            node_role,
        )
        .await
    }

    fn address_prober_enabled(config: &NodeConfig) -> bool {
        let prober_enabled = config
            .address_prober
            .as_ref()
            .map(|c| c.enabled())
            .unwrap_or(true);
        let v3_enabled = config
            .p2p_config
            .discovery
            .as_ref()
            .is_some_and(|d| d.use_get_known_peers_v3());
        prober_enabled && v3_enabled
    }

    fn update_address_prober_epoch(
        &self,
        epoch_store: &AuthorityPerEpochStore,
        consensus_manager: &Arc<ConsensusManager>,
    ) {
        if !epoch_store.is_validator() {
            return;
        }
        if let Some(handle) = &self.address_prober {
            handle.update_epoch(
                epoch_store.epoch(),
                epoch_store.epoch_start_state().get_consensus_committee(),
                consensus_manager.clone(),
            );
        }
    }

    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,
        randomness_receiver_handle: Arc<RandomnessRoundReceiverHandle>,
        consensus_manager: Arc<ConsensusManager>,
        consensus_store_pruner: ConsensusStorePruner,
        state_hasher: Weak<GlobalStateHasher>,
        backpressure_manager: Arc<BackpressureManager>,
        validator_server_handle: Option<SpawnOnce>,
        validator_overload_monitor_handle: Option<JoinHandle<()>>,
        checkpoint_metrics: Arc<CheckpointMetrics>,
        sui_node_metrics: Arc<SuiNodeMetrics>,
        sui_tx_validator_metrics: Arc<SuiTxValidatorMetrics>,
        admission_queue: Option<AdmissionQueueContext>,
        node_role: NodeRole,
    ) -> 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(),
            node_role,
        );

        // Clear the VSS public key from the previous epoch so any randomness round
        // signatures buffer in the channel until the new DKG completes.
        randomness_receiver_handle.clear_public_key();

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

        if node_role.is_validator() {
            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 consensus_handler_initializer = ConsensusHandlerInitializer::new(
            state.clone(),
            checkpoint_service.clone(),
            epoch_store.clone(),
            throughput_calculator,
            backpressure_manager,
            config.congestion_log.clone(),
        );

        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(),
                epoch_store.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,
                        Some(randomness_receiver_handle),
                    )
                    .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 node_role.is_validator() && epoch_store.authenticator_state_enabled() {
            Self::start_jwk_updater(
                config,
                sui_node_metrics,
                state.name,
                epoch_store.clone(),
                consensus_adapter.clone(),
            );
        }

        if let Some(ctx) = &admission_queue {
            ctx.rotate_for_epoch(epoch_store);
        }

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

    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>,
        node_role: NodeRole,
    ) -> Arc<CheckpointService> {
        let checkpoint_output: Box<dyn CheckpointOutput> = if node_role.is_validator() {
            Box::new(SubmitCheckpointToConsensus::new(
                consensus_adapter,
                state.secret.clone(),
                config.protocol_public_key(),
                checkpoint_metrics.clone(),
            ))
        } else {
            Box::new(LogCheckpointOutput::new(checkpoint_metrics.clone()))
        };

        let certified_checkpoint_output = SendCheckpointToStateSync::new(state_sync_handle);

        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,
        )
    }

    fn construct_consensus_adapter(
        committee: &Committee,
        consensus_config: &ConsensusConfig,
        authority: AuthorityName,
        prometheus_registry: &Registry,
        consensus_client: Arc<dyn ConsensusClient>,
        checkpoint_store: Arc<CheckpointStore>,
        inflight_slot_freed_notify: Arc<tokio::sync::Notify>,
    ) -> 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,
            consensus_config.max_pending_transactions(),
            consensus_config.max_pending_transactions() * 2 / committee.num_members(),
            ca_metrics,
            inflight_slot_freed_notify,
        )
    }

    async fn start_grpc_validator_service(
        config: &NodeConfig,
        state: Arc<AuthorityState>,
        consensus_adapter: Arc<ConsensusAdapter>,
        epoch_store: Arc<AuthorityPerEpochStore>,
        prometheus_registry: &Registry,
        inflight_slot_freed_notify: Arc<tokio::sync::Notify>,
    ) -> Result<(SpawnOnce, Option<AdmissionQueueContext>)> {
        let overload_config = &config.authority_overload_config;
        let admission_queue = overload_config.admission_queue_enabled.then(|| {
            let manager = Arc::new(AdmissionQueueManager::new(
                consensus_adapter.clone(),
                Arc::new(AdmissionQueueMetrics::new(prometheus_registry)),
                overload_config.admission_queue_capacity_fraction,
                overload_config.admission_queue_failover_timeout,
                inflight_slot_freed_notify,
            ));
            AdmissionQueueContext::spawn(manager, epoch_store)
        });
        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),
            admission_queue.clone(),
        );

        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();

        let spawn_once = 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");
        });
        Ok((spawn_once, admission_queue))
    }

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

    /// The embedded `sui-rpc-store` index backend, when the node is a
    /// fullnode with indexing enabled. Exposes the startup bootstrap
    /// decision and per-cohort watermarks for introspection (used by
    /// tests to observe restore/resume behavior across restarts without
    /// going through the RPC surface).
    pub fn embedded_rpc_store(&self) -> Option<&EmbeddedRpcStore> {
        self.embedded_rpc_store.as_ref()
    }

    #[cfg(any(test, msim))]
    pub fn connection_monitor_handle_for_testing(
        &self,
    ) -> &mysten_network::anemo_connection_monitor::ConnectionMonitorHandle {
        &self._connection_monitor_handle
    }

    #[cfg(any(test, msim))]
    pub fn address_prober_metrics_for_testing(
        &self,
    ) -> std::sync::Arc<address_prober::AddressProberMetrics> {
        self.address_prober
            .as_ref()
            .expect("address prober should be running in tests")
            .metrics_for_testing()
    }

    #[cfg(feature = "testing")]
    pub fn prometheus_metrics_for_testing(&self) -> Vec<prometheus::proto::MetricFamily> {
        self.registry_service.default_registry().gather()
    }

    pub fn node_role(&self) -> NodeRole {
        self.state.load_epoch_store_one_call_per_task().node_role()
    }

    // 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_checkpoint_store(&self) -> Arc<CheckpointStore> {
        self.checkpoint_store.clone()
    }

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

    pub fn clone_consensus_store(
        &self,
    ) -> Option<Arc<consensus_core::storage::rocksdb_store::RocksDBStore>> {
        self.validator_components
            .try_lock()
            .ok()?
            .as_ref()?
            .consensus_manager
            .consensus_store()
    }

    /// 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.
            // FullNodes that state sync via consensus will also have validator components, by they are not supposed to submit any capabilities.
            if let Some(components) = &*self.validator_components.lock().await
                && cur_epoch_store.is_validator()
            {
                // 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 = 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,
                    ),
                );
                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");

            cur_epoch_store.record_epoch_reconfig_start_time_metric();

            update_peer_addresses(
                &self.config,
                &self.endpoint_manager,
                &new_epoch_start_state,
                Some(cur_epoch_store.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_role = new_epoch_store.node_role();

            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,
                admission_queue,
            }) = validator_components_lock_guard.take()
            {
                info!("Reconfiguring node (was running consensus).");

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

                if let Some(handle) = &self.address_prober {
                    handle.leave_committee();
                }

                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 new_role.runs_consensus() {
                    info!("Restarting consensus as {new_role}");
                    let components = 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(),
                        self.randomness_receiver_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,
                        admission_queue,
                        new_role,
                    )
                    .await?;
                    self.update_address_prober_epoch(
                        &new_epoch_store,
                        &components.consensus_manager,
                    );
                    Some(components)
                } else {
                    info!(
                        "This node has new role {new_role} and no longer runs consensus 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 new_role.runs_consensus() {
                    info!("Promoting node to {new_role}, starting consensus components");

                    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.registry_service,
                        self.metrics.clone(),
                        self.checkpoint_metrics.clone(),
                        new_role,
                        self.randomness_receiver_handle.clone(),
                    )
                    .await?;

                    if new_role.is_validator() {
                        components.validator_server_handle = Some(
                            components
                                .validator_server_handle
                                .take()
                                .unwrap()
                                .start()
                                .await,
                        );

                        self.endpoint_manager
                            .set_consensus_address_updater(components.consensus_manager.clone());
                    }

                    self.update_address_prober_epoch(
                        &new_epoch_store,
                        &components.consensus_manager,
                    );
                    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()
    }

    pub fn state_sync_handle(&self) -> state_sync::Handle {
        self.state_sync_handle.clone()
    }

    pub fn endpoint_manager(&self) -> &EndpointManager {
        &self.endpoint_manager
    }

    pub async fn address_prober_report(&self) -> Option<address_prober::ProbeReport> {
        match &self.address_prober {
            Some(handle) => handle.probe_report().await,
            None => None,
        }
    }

    /// 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,
        fork_recovery: Option<&ForkRecoveryConfig>,
        build_version: &str,
    ) -> Result<()> {
        // Manual recovery from operator-supplied overrides; runs regardless of fork_crash_behavior
        // and only acts on the checkpoints / transactions explicitly listed in the config.
        if let Some(recovery) = fork_recovery {
            Self::try_recover_checkpoint_fork(checkpoint_store, recovery)?;
            Self::try_recover_transaction_fork(checkpoint_store, recovery)?;
        }

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

        match behavior {
            ForkCrashBehavior::RecoverOncePerVersion => {
                Self::try_recover_forks(checkpoint_store, checkpoint_metrics, build_version)?;
            }
            ForkCrashBehavior::AwaitForkRecovery | ForkCrashBehavior::ReturnError => {}
        }

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

        Ok(())
    }

    /// Manual recovery: for each `seq -> digest` override, if the locally computed checkpoint at
    /// `seq` differs, clear locally computed checkpoints from `seq` (and the checkpoint fork marker)
    /// so the node rebuilds toward the operator-specified digest.
    fn try_recover_checkpoint_fork(
        checkpoint_store: &CheckpointStore,
        recovery: &ForkRecoveryConfig,
    ) -> Result<()> {
        if recovery.checkpoint_overrides.is_empty() {
            return Ok(());
        }

        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(fork_info) = checkpoint_store.get_checkpoint_fork_detected()?
            && recovery
                .checkpoint_overrides
                .contains_key(&fork_info.checkpoint_seq)
        {
            info!(
                "Fork recovery enabled: clearing checkpoint fork at seq {} with digest {:?}",
                fork_info.checkpoint_seq, fork_info.checkpoint_digest
            );
            checkpoint_store
                .clear_checkpoint_fork_detected()
                .expect("Failed to clear checkpoint fork detected marker");
        }
        Ok(())
    }

    /// Manual recovery: if the forked transaction is listed in transaction_overrides, clear its fork
    /// marker so the node proceeds on restart.
    fn try_recover_transaction_fork(
        checkpoint_store: &CheckpointStore,
        recovery: &ForkRecoveryConfig,
    ) -> Result<()> {
        if recovery.transaction_overrides.is_empty() {
            return Ok(());
        }

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

    /// Auto-recovery: clear fork markers (the affected seq/tx is read from the markers) so the
    /// node re-derives canonically. A marker is cleared only if both gates pass:
    ///
    /// - Version gate: the marker was recorded by a different binary version than the one now
    ///   running. The binary that forked would deterministically fork again, so clearing under
    ///   it would only add a second equivocation; the node hangs until a corrected binary is
    ///   deployed.
    /// - Certification gate: the marker records the certified checkpoint the node diverged
    ///   from. Markers carry it only when detection compared against a certificate already
    ///   durably persisted locally, so its presence proves the network certified the canonical
    ///   outcome. Recovery is deliberate equivocation — the node may have already signed the
    ///   forked result and will sign a different one after re-deriving — which is safe only
    ///   under that proof: a quorum certificate is irrevocable (a conflicting certificate would
    ///   require f+1 double-signers), so re-signing can no longer influence what finalizes.
    ///   Self-divergence markers (the node disagreeing with its own prior result rather than a
    ///   certificate) carry no certified reference and never pass; the node halts awaiting
    ///   operator intervention.
    fn try_recover_forks(
        checkpoint_store: &CheckpointStore,
        checkpoint_metrics: &CheckpointMetrics,
        build_version: &str,
    ) -> Result<()> {
        if let Some(fork_info) = checkpoint_store.get_checkpoint_fork_detected()? {
            if fork_info.binary_version == build_version {
                error!(
                    checkpoint_seq = fork_info.checkpoint_seq,
                    build_version,
                    "Fork recovery blocked: this binary version produced the checkpoint fork and \
                     would fork again. Halting; deploy a corrected binary to recover."
                );
                checkpoint_metrics
                    .fork_auto_recovery_awaiting_new_binary
                    .set(1);
            } else if fork_info.certified_checkpoint_digest.is_none() {
                // The builder re-derived a previously computed checkpoint differently: the fork
                // is against the node's own prior result, not a certified checkpoint, so there
                // is no canonical outcome to converge toward.
                error!(
                    checkpoint_seq = fork_info.checkpoint_seq,
                    checkpoint_digest = ?fork_info.checkpoint_digest,
                    "Fork recovery blocked: the builder re-derived its own previous checkpoint \
                     differently, so there is no certified checkpoint proving the canonical \
                     outcome to converge toward. Halting awaiting operator intervention."
                );
                checkpoint_metrics
                    .fork_auto_recovery_blocked_uncertified
                    .set(1);
            } else {
                info!(
                    checkpoint_seq = fork_info.checkpoint_seq,
                    checkpoint_digest = ?fork_info.checkpoint_digest,
                    forked_binary_version = ?fork_info.binary_version,
                    build_version,
                    "Fork recovery: clearing checkpoint fork and locally computed checkpoints \
                     from the forked sequence so the builder rebuilds toward the certified \
                     checkpoint"
                );
                checkpoint_store
                    .clear_locally_computed_checkpoints_from(fork_info.checkpoint_seq)
                    .context("Failed to clear locally computed checkpoints during fork recovery")?;
                checkpoint_store.clear_checkpoint_fork_detected()?;
                checkpoint_metrics.checkpoint_fork_auto_recovered.set(1);
            }
        }

        if let Some(fork_info) = checkpoint_store.get_transaction_fork_detected()? {
            if fork_info.binary_version == build_version {
                error!(
                    tx_digest = ?fork_info.tx_digest,
                    build_version,
                    "Fork recovery blocked: this binary version produced the transaction fork and \
                     would fork again. Halting; deploy a corrected binary to recover."
                );
                checkpoint_metrics
                    .fork_auto_recovery_awaiting_new_binary
                    .set(1);
            } else if fork_info.certified_checkpoint_seq.is_none() {
                error!(
                    tx_digest = ?fork_info.tx_digest,
                    "Fork recovery blocked: the expected effects of the forked transaction did \
                     not come from a certified checkpoint (they came from this validator's own \
                     previously signed effects), so the network has not provably certified the \
                     canonical outcome. Halting awaiting operator intervention."
                );
                checkpoint_metrics
                    .fork_auto_recovery_blocked_uncertified
                    .set(1);
            } else {
                info!(
                    tx_digest = ?fork_info.tx_digest,
                    expected_effects = ?fork_info.expected_effects_digest,
                    actual_effects = ?fork_info.actual_effects_digest,
                    certified_checkpoint_seq = ?fork_info.certified_checkpoint_seq,
                    forked_binary_version = ?fork_info.binary_version,
                    build_version,
                    "Fork recovery: clearing transaction fork; re-execution will converge toward \
                     the canonical certified effects"
                );
                checkpoint_store.clear_transaction_fork_detected()?;
                checkpoint_metrics.transaction_fork_auto_recovered.set(1);
            }
        }

        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 | ForkCrashBehavior::RecoverOncePerVersion => {
                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 | ForkCrashBehavior::RecoverOncePerVersion => {
                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,
        }
    }
}

/// Updates trusted peer addresses in the p2p network (for nodes configured as validators).
/// When `prev_epoch_start_state` is provided, validators that are no longer in the committee
/// have their Chain addresses cleared.
fn update_peer_addresses(
    config: &NodeConfig,
    endpoint_manager: &EndpointManager,
    epoch_start_state: &EpochStartSystemState,
    prev_epoch_start_state: Option<&EpochStartSystemState>,
) {
    if config.consensus_config().is_none() {
        return;
    }
    let new_peers: HashSet<PeerId> = epoch_start_state
        .get_validator_as_p2p_peers(config.protocol_public_key())
        .into_iter()
        .map(|(peer_id, address)| {
            endpoint_manager
                .update_endpoint(
                    EndpointId::P2p(peer_id),
                    AddressSource::Chain,
                    vec![address],
                )
                .expect("Updating peer addresses should not fail");
            peer_id
        })
        .collect();

    // Clear Chain addresses for validators that left the committee.
    if let Some(prev) = prev_epoch_start_state {
        for (peer_id, _) in prev.get_validator_as_p2p_peers(config.protocol_public_key()) {
            if !new_peers.contains(&peer_id) {
                endpoint_manager
                    .update_endpoint(EndpointId::P2p(peer_id), AddressSource::Chain, vec![])
                    .expect("Clearing peer addresses should not fail");
            }
        }
    }
}

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_json_rpc_router(
    state: &Arc<AuthorityState>,
    transaction_orchestrator: &Option<Arc<TransactionOrchestrator<NetworkAuthorityClient>>>,
    config: &NodeConfig,
    prometheus_registry: &Registry,
) -> Result<axum::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();

    Ok(server.to_router(server_type).await?)
}

/// Remove the on-disk directory of the legacy `rpc-index` backend.
///
/// The embedded `sui-rpc-store` replaced the `RpcIndexStore` backend, which
/// wrote to `<db_path>/rpc-index`; that data is now dead. Remove it on startup
/// so a node upgraded from an older version does not leave it lingering and
/// wasting disk. Best-effort: a node that never ran the legacy backend has
/// nothing to remove, and a failure to remove stale data must not block
/// startup.
fn remove_legacy_rpc_index_store(db_path: &Path) {
    let legacy_dir = db_path.join("rpc-index");
    match std::fs::remove_dir_all(&legacy_dir) {
        Ok(()) => info!(
            "removed legacy rpc-index directory {}",
            legacy_dir.display()
        ),
        // The common case: the node never ran the legacy backend, or it was
        // already cleaned up on a prior startup.
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
        Err(e) => warn!(
            "failed to remove legacy rpc-index directory {}: {e:?}",
            legacy_dir.display()
        ),
    }
}

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

    info!("starting rpc service with config: {:?}", config.rpc);

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

    // The JSON-RPC service can be disabled independently of the gRPC/REST
    // service and of JSON-RPC indexing, so that a node can keep indexing
    // without exposing the JSON-RPC endpoints.
    if config.json_rpc_enabled() {
        router = router.merge(
            build_json_rpc_router(
                &state,
                transaction_orchestrator,
                config,
                prometheus_registry,
            )
            .await?,
        );
    } else {
        info!("json-rpc service is disabled");
    }

    // When the embedded rpc-store is active, gate checkpoint delivery on the
    // index so a client that waits for a checkpoint can immediately read its
    // indexed state (matching the legacy synchronously-committed index).
    let indexed_checkpoint = embedded_rpc_store.map(|embedded| embedded.indexed_checkpoint_fn());
    let subscription_watermark_interval = config
        .rpc
        .as_ref()
        .and_then(|rpc| rpc.subscription_watermark_interval);
    let subscription_max_subscribers = config
        .rpc
        .as_ref()
        .and_then(|rpc| rpc.subscription_max_subscribers);
    let subscription_shards = config.rpc.as_ref().and_then(|rpc| rpc.subscription_shards);
    let (subscription_service_checkpoint_sender, subscription_service_handle) =
        SubscriptionService::build(
            prometheus_registry,
            indexed_checkpoint,
            subscription_watermark_interval,
            subscription_max_subscribers,
            subscription_shards,
        );
    let rpc_router = {
        // Serve the index read paths from the embedded rpc-store when it
        // is enabled. Raw chain data comes from the perpetual / checkpoint
        // stores either way.
        let reader: Arc<dyn RpcStateReader> = match embedded_rpc_store {
            Some(embedded) => Arc::new(RpcStoreReadStore::new(
                state.clone(),
                store,
                embedded.reader(),
            )),
            None => Arc::new(RestReadStore::new(state.clone(), store)),
        };
        let mut rpc_service = sui_rpc_api::RpcService::new(reader);
        rpc_service.with_server_version(server_version);

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

        rpc_service.with_metrics(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)
                .expose_headers(tower_http::cors::Any),
        );

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

    // On top of sui-http's hardened defaults (bounded concurrent streams;
    // transport keepalives stay disabled by default), bound connection
    // lifetime: GOAWAY at the configured age and force-close after the grace
    // period. The hard close is the only server-side mechanism that reclaims
    // streams wedged behind HTTP/2 flow-control windows that a stalled peer
    // never reopens, and connection age also bounds how long a vanished peer
    // can pin connection state, which keepalives would otherwise detect.
    let server_config = {
        let rpc_config = config.rpc().cloned().unwrap_or_default();
        let mut server_config = sui_http::Config::default()
            .max_connection_age_grace(rpc_config.max_connection_age_grace());
        if let Some(age) = rpc_config.max_connection_age() {
            server_config = server_config.max_connection_age(age);
        }
        server_config
    };

    let https = if let Some((tls_config, https_address)) = config
        .rpc()
        .and_then(|config| config.tls_config().map(|tls| (tls, config.https_address())))
    {
        let tls_server_config = https_rustls_config(tls_config.cert(), tls_config.key())?;
        let https = sui_http::Builder::new()
            .config(server_config.clone())
            .tls_config(tls_server_config)
            .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()
        .config(server_config)
        .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),
    ))
}

/// Builds the HTTPS RPC server's rustls config from PEM files, pinning the
/// ring crypto provider.
///
/// `sui_http::Builder::tls_single_cert` resolves the provider from rustls
/// crate features and panics at runtime when more than one provider feature is
/// enabled in the final binary (e.g. `aws-lc-rs` is pulled in through
/// `aws-config` in the `sui` CLI), so the provider is pinned explicitly here
/// instead.
fn https_rustls_config(cert: &str, key: &str) -> Result<sui_http::rustls::ServerConfig> {
    use sui_http::rustls;
    use sui_http::rustls::pki_types::pem::PemObject;

    let certs = rustls::pki_types::CertificateDer::pem_file_iter(cert)
        .with_context(|| format!("failed to read TLS certificate chain from {cert}"))?
        .collect::<Result<Vec<_>, _>>()
        .with_context(|| format!("failed to parse TLS certificate chain from {cert}"))?;
    let private_key = rustls::pki_types::PrivateKeyDer::from_pem_file(key)
        .with_context(|| format!("failed to read TLS private key from {key}"))?;
    let config = rustls::ServerConfig::builder_with_provider(Arc::new(
        rustls::crypto::ring::default_provider(),
    ))
    .with_protocol_versions(rustls::DEFAULT_VERSIONS)?
    .with_no_client_auth()
    .with_single_cert(certs, private_key)?;
    Ok(config)
}

#[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};

    // A present legacy `rpc-index` directory is removed, while its siblings
    // (notably the still-used jsonrpc `indexes` store) are left untouched, and a
    // missing directory is a no-op.
    #[test]
    fn removes_only_the_legacy_rpc_index_directory() {
        let db = tempfile::tempdir().unwrap();
        let legacy = db.path().join("rpc-index");
        let sibling = db.path().join("indexes");
        std::fs::create_dir(&legacy).unwrap();
        std::fs::create_dir(&sibling).unwrap();
        std::fs::write(legacy.join("CURRENT"), b"stale").unwrap();

        remove_legacy_rpc_index_store(db.path());
        assert!(
            !legacy.exists(),
            "legacy rpc-index directory should be gone"
        );
        assert!(sibling.exists(), "sibling stores must be left untouched");

        // Idempotent: a second run (nothing to remove) does not error or touch
        // the siblings.
        remove_legacy_rpc_index_store(db.path());
        assert!(!legacy.exists());
        assert!(sibling.exists());
    }

    // Halt / ReturnError never clear markers; ReturnError surfaces the fork as a startup error.
    #[tokio::test]
    async fn test_return_error_does_not_recover() {
        let checkpoint_store = CheckpointStore::new_for_tests();
        let checkpoint_metrics = CheckpointMetrics::new(&Registry::new());
        let cfg = ForkRecoveryConfig {
            transaction_overrides: Default::default(),
            checkpoint_overrides: Default::default(),
            fork_crash_behavior: ForkCrashBehavior::ReturnError,
        };

        // Checkpoint fork.
        checkpoint_store
            .record_checkpoint_fork_detected(
                42,
                CheckpointDigest::random(),
                Some(CheckpointDigest::random()),
            )
            .unwrap();
        let r = SuiNode::check_and_recover_forks(
            &checkpoint_store,
            &checkpoint_metrics,
            Some(&cfg),
            "v1",
        )
        .await;
        assert!(
            r.unwrap_err()
                .to_string()
                .contains("Checkpoint fork detected")
        );
        assert!(
            checkpoint_store
                .get_checkpoint_fork_detected()
                .unwrap()
                .is_some()
        );
        checkpoint_store.clear_checkpoint_fork_detected().unwrap();

        // Transaction fork.
        checkpoint_store
            .record_transaction_fork_detected(
                TransactionDigest::random(),
                TransactionEffectsDigest::random(),
                TransactionEffectsDigest::random(),
                Some(1),
            )
            .unwrap();
        let r = SuiNode::check_and_recover_forks(
            &checkpoint_store,
            &checkpoint_metrics,
            Some(&cfg),
            "v1",
        )
        .await;
        assert!(
            r.unwrap_err()
                .to_string()
                .contains("Transaction fork detected")
        );
    }

    // A fork marker carrying the currently running binary version is never cleared — the binary
    // that forked would deterministically fork again — so the node hangs until a corrected
    // binary (different version) runs recovery.
    #[tokio::test]
    async fn test_same_binary_version_does_not_recover() {
        let checkpoint_store = CheckpointStore::new_for_tests();
        let checkpoint_metrics = CheckpointMetrics::new(&Registry::new());
        let seq = 7;
        // The fork is recorded by binary "v1", against a certified checkpoint.
        checkpoint_store.set_binary_version("v1");
        checkpoint_store
            .record_checkpoint_fork_detected(
                seq,
                CheckpointDigest::random(),
                Some(CheckpointDigest::random()),
            )
            .unwrap();

        // Restarting the same binary: recovery refused despite certification.
        SuiNode::try_recover_forks(&checkpoint_store, &checkpoint_metrics, "v1").unwrap();
        assert!(
            checkpoint_store
                .get_checkpoint_fork_detected()
                .unwrap()
                .is_some()
        );
        assert_eq!(
            checkpoint_metrics
                .fork_auto_recovery_awaiting_new_binary
                .get(),
            1
        );
        assert_eq!(checkpoint_metrics.checkpoint_fork_auto_recovered.get(), 0);

        // Corrected binary (new version): recovers.
        SuiNode::try_recover_forks(&checkpoint_store, &checkpoint_metrics, "v2").unwrap();
        assert!(
            checkpoint_store
                .get_checkpoint_fork_detected()
                .unwrap()
                .is_none()
        );
        assert_eq!(checkpoint_metrics.checkpoint_fork_auto_recovered.get(), 1);
    }

    // The default behavior (RecoverOncePerVersion) recovers with no fork-recovery config present.
    #[tokio::test]
    async fn test_default_recovers() {
        let checkpoint_store = CheckpointStore::new_for_tests();
        let checkpoint_metrics = CheckpointMetrics::new(&Registry::new());

        let tx_digest = TransactionDigest::random();
        // The fork was recorded by binary "v1"; its expected effects came from certified
        // checkpoint 3, so recovery under "v2" is permitted.
        checkpoint_store.set_binary_version("v1");
        checkpoint_store
            .record_transaction_fork_detected(
                tx_digest,
                TransactionEffectsDigest::random(),
                TransactionEffectsDigest::random(),
                Some(3),
            )
            .unwrap();

        let r =
            SuiNode::check_and_recover_forks(&checkpoint_store, &checkpoint_metrics, None, "v2")
                .await;
        assert!(r.is_ok());
        assert!(
            checkpoint_store
                .get_transaction_fork_detected()
                .unwrap()
                .is_none()
        );
        assert_eq!(checkpoint_metrics.transaction_fork_auto_recovered.get(), 1);
    }

    // checkpoint_overrides clears only the checkpoint fork marker (when the forked seq is listed); it
    // is decoupled from the transaction fork marker, which is cleared by transaction_overrides.
    #[tokio::test]
    async fn test_checkpoint_overrides_clear_checkpoint_marker_only() {
        let checkpoint_store = CheckpointStore::new_for_tests();
        let seq = 9;

        checkpoint_store
            .record_checkpoint_fork_detected(
                seq,
                CheckpointDigest::random(),
                Some(CheckpointDigest::random()),
            )
            .unwrap();
        checkpoint_store
            .record_transaction_fork_detected(
                TransactionDigest::random(),
                TransactionEffectsDigest::random(),
                TransactionEffectsDigest::random(),
                None,
            )
            .unwrap();

        // No overrides: both markers are left intact.
        SuiNode::try_recover_checkpoint_fork(&checkpoint_store, &ForkRecoveryConfig::default())
            .unwrap();
        assert!(
            checkpoint_store
                .get_checkpoint_fork_detected()
                .unwrap()
                .is_some()
        );
        assert!(
            checkpoint_store
                .get_transaction_fork_detected()
                .unwrap()
                .is_some()
        );

        // Override for the forked seq: clears the checkpoint marker but leaves the transaction marker.
        let mut checkpoint_overrides = BTreeMap::new();
        checkpoint_overrides.insert(seq, CheckpointDigest::random().to_string());
        let cfg = ForkRecoveryConfig {
            transaction_overrides: Default::default(),
            checkpoint_overrides,
            fork_crash_behavior: ForkCrashBehavior::AwaitForkRecovery,
        };
        SuiNode::try_recover_checkpoint_fork(&checkpoint_store, &cfg).unwrap();
        assert!(
            checkpoint_store
                .get_checkpoint_fork_detected()
                .unwrap()
                .is_none()
        );
        assert!(
            checkpoint_store
                .get_transaction_fork_detected()
                .unwrap()
                .is_some(),
            "checkpoint_overrides must not touch the transaction fork marker"
        );
    }

    // transaction_overrides clears the transaction fork marker when the forked tx is listed.
    #[tokio::test]
    async fn test_transaction_overrides_clear_transaction_marker() {
        let checkpoint_store = CheckpointStore::new_for_tests();
        let tx_digest = TransactionDigest::random();
        checkpoint_store
            .record_transaction_fork_detected(
                tx_digest,
                TransactionEffectsDigest::random(),
                TransactionEffectsDigest::random(),
                None,
            )
            .unwrap();

        // Unrelated override: marker stays.
        let mut transaction_overrides = BTreeMap::new();
        transaction_overrides.insert(TransactionDigest::random().to_string(), String::new());
        let cfg = ForkRecoveryConfig {
            transaction_overrides,
            checkpoint_overrides: Default::default(),
            fork_crash_behavior: ForkCrashBehavior::AwaitForkRecovery,
        };
        SuiNode::try_recover_transaction_fork(&checkpoint_store, &cfg).unwrap();
        assert!(
            checkpoint_store
                .get_transaction_fork_detected()
                .unwrap()
                .is_some()
        );

        // Override for the forked tx: marker cleared.
        let mut transaction_overrides = BTreeMap::new();
        transaction_overrides.insert(tx_digest.to_string(), String::new());
        let cfg = ForkRecoveryConfig {
            transaction_overrides,
            checkpoint_overrides: Default::default(),
            fork_crash_behavior: ForkCrashBehavior::AwaitForkRecovery,
        };
        SuiNode::try_recover_transaction_fork(&checkpoint_store, &cfg).unwrap();
        assert!(
            checkpoint_store
                .get_transaction_fork_detected()
                .unwrap()
                .is_none()
        );
    }

    // Under RecoverOncePerVersion, a checkpoint override clears the fork via the manual path
    // even when the auto path would refuse (here: the fork was recorded by the currently running
    // binary version and the sequence is not certified).
    #[tokio::test]
    async fn test_override_clears_fork_auto_recovery_refuses() {
        let checkpoint_store = CheckpointStore::new_for_tests();
        let checkpoint_metrics = CheckpointMetrics::new(&Registry::new());
        let seq = 5;
        checkpoint_store.set_binary_version("v1");
        checkpoint_store
            .record_checkpoint_fork_detected(
                seq,
                CheckpointDigest::random(),
                Some(CheckpointDigest::random()),
            )
            .unwrap();

        let mut checkpoint_overrides = BTreeMap::new();
        checkpoint_overrides.insert(seq, CheckpointDigest::random().to_string());
        let cfg = ForkRecoveryConfig {
            transaction_overrides: Default::default(),
            checkpoint_overrides,
            fork_crash_behavior: ForkCrashBehavior::RecoverOncePerVersion,
        };

        SuiNode::check_and_recover_forks(&checkpoint_store, &checkpoint_metrics, Some(&cfg), "v1")
            .await
            .unwrap();

        assert!(
            checkpoint_store
                .get_checkpoint_fork_detected()
                .unwrap()
                .is_none()
        );
    }

    // A self-divergence checkpoint fork (the builder re-derived its own previous checkpoint
    // differently; no certified digest in the marker) is never auto-recovered, even under a new
    // binary version, because neither result is proven canonical. It requires operator
    // overrides.
    #[tokio::test]
    async fn test_self_divergence_checkpoint_fork_blocks_recovery() {
        let checkpoint_store = CheckpointStore::new_for_tests();
        let checkpoint_metrics = CheckpointMetrics::new(&Registry::new());
        let seq = 17;
        checkpoint_store.set_binary_version("v1");
        checkpoint_store
            .record_checkpoint_fork_detected(seq, CheckpointDigest::random(), None)
            .unwrap();

        SuiNode::try_recover_forks(&checkpoint_store, &checkpoint_metrics, "v2").unwrap();

        assert!(
            checkpoint_store
                .get_checkpoint_fork_detected()
                .unwrap()
                .is_some()
        );
        assert_eq!(
            checkpoint_metrics
                .fork_auto_recovery_blocked_uncertified
                .get(),
            1
        );
        assert_eq!(checkpoint_metrics.checkpoint_fork_auto_recovered.get(), 0);
    }

    // A transaction fork whose expected effects did not come from a certified checkpoint (i.e.
    // they came from this validator's own previously signed effects) is never auto-recovered,
    // even on a new binary version.
    #[tokio::test]
    async fn test_uncertified_transaction_fork_blocks_recovery() {
        let checkpoint_store = CheckpointStore::new_for_tests();
        let checkpoint_metrics = CheckpointMetrics::new(&Registry::new());
        checkpoint_store.set_binary_version("v1");
        checkpoint_store
            .record_transaction_fork_detected(
                TransactionDigest::random(),
                TransactionEffectsDigest::random(),
                TransactionEffectsDigest::random(),
                None,
            )
            .unwrap();

        SuiNode::try_recover_forks(&checkpoint_store, &checkpoint_metrics, "v2").unwrap();
        SuiNode::try_recover_forks(&checkpoint_store, &checkpoint_metrics, "v3").unwrap();

        assert!(
            checkpoint_store
                .get_transaction_fork_detected()
                .unwrap()
                .is_some()
        );
        assert_eq!(
            checkpoint_metrics
                .fork_auto_recovery_blocked_uncertified
                .get(),
            1
        );
    }
}